From b5475cfa5c3daffd52b5b25eb7063e745b976026 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 21:57:31 +0100 Subject: [PATCH] Subscribe as soon as the runtime serves, never after a guessed wait, so no event sent at a start is missed (issue 774) mesh-healers slept 2 s before its first subscribe to the keeper's conditions; its whole test delivered condition-raised once the lease was read, often before that, and the event was dropped: "never: the would-act written" in about two runs of three on main, failing the group check. Eight modules had the same shape. Quick fix: sleep after a failed subscribe in mesh-healers alone. Thorough fix (chosen, the kind is in eight modules): every loop that asks the runtime for events or a watch waits for stdio.Served() and subscribes at once, sleeping only after a refusal (mesh-healers, mesh-config, mesh-mail, mesh-issues, mesh-provisioner, mesh-watcher, mesh-delivery-gate); mesh-knowledge, whose pinned mesh-sdk has no Served, subscribes first and sleeps after a refusal. The healers' whole test delivers the event only once the bundle has subscribed it, as the runtime's consumer holds it until then. --- modules/mesh-config/cmd/mesh-config/main.go | 6 +++-- .../cmd/mesh-delivery-gate/main.go | 2 +- modules/mesh-healers/cmd/mesh-healers/main.go | 6 +++-- .../mesh-healers/whole_integration_test.go | 24 +++++++++++++++++++ modules/mesh-issues/cmd/mesh-issues/main.go | 3 ++- .../mesh-knowledge/cmd/mesh-knowledge/main.go | 2 +- modules/mesh-mail/cmd/mesh-mail/main.go | 3 ++- .../cmd/mesh-provisioner/main.go | 3 ++- modules/mesh-watcher/cmd/mesh-watcher/main.go | 2 +- 9 files changed, 41 insertions(+), 10 deletions(-) diff --git a/modules/mesh-config/cmd/mesh-config/main.go b/modules/mesh-config/cmd/mesh-config/main.go index c0bff6a7..1256b0fa 100644 --- a/modules/mesh-config/cmd/mesh-config/main.go +++ b/modules/mesh-config/cmd/mesh-config/main.go @@ -175,8 +175,8 @@ func resolving(c *Config) { // reads the router's record of every waiting ask once a minute besides, so an answer heard nowhere is still taken. func hearing(c *Config) { go func() { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error { var w asks.Warrant if json.Unmarshal(e.Body, &w) == nil { @@ -191,12 +191,13 @@ func hearing(c *Config) { return } logf("[mesh-config] does not hear the operator's answers yet (%v); asking again in %s", err, wait) + time.Sleep(wait) } }() for _, pattern := range []string{"gitea.pull.merged", "gitea.pull.closed", "gitea.forge.pull-request.closed"} { go func(pattern string) { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { var p PullEvent if json.Unmarshal(e.Body, &p) != nil { @@ -212,6 +213,7 @@ func hearing(c *Config) { return } logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait) + time.Sleep(wait) } }(pattern) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index 52d68da4..8143b44e 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -91,7 +91,7 @@ func main() { // events from then on (novox/hq ADR 0406 §1, issue 975): no tick, no forge poll, no reconcile loop. Each step is retried, // each failure said, never given up on quietly. func run(h *Holder, l *listening) { - time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it + <-stdio.Served() // Serve first: the state is reached through it, and an event is heard from as soon as it serves loadKept(h, l) go h.Work() for _, pattern := range followed { diff --git a/modules/mesh-healers/cmd/mesh-healers/main.go b/modules/mesh-healers/cmd/mesh-healers/main.go index fb1e50b5..70fad411 100644 --- a/modules/mesh-healers/cmd/mesh-healers/main.go +++ b/modules/mesh-healers/cmd/mesh-healers/main.go @@ -74,8 +74,8 @@ func main() { }) // S15 over both logs: the seat's act-recorded on a durable consumer, beside the controller's bucket's watch. go func() { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(RecordedEvent, func(e stdio.Envelope) error { h.ActRecorded(e.Body) return nil @@ -84,18 +84,20 @@ func main() { return } logf("[%s] does not hear %s yet (%v); asking again in %s", Module, RecordedEvent, err, wait) + time.Sleep(wait) } }() for _, event := range []string{RaisedEvent, ChangedEvent} { go func(event string) { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(event, func(e stdio.Envelope) error { return h.Heard(e.Body) }) if err == nil { logf("[%s] hears %s", Module, event) return } logf("[%s] does not hear %s yet (%v); asking again in %s", Module, event, err, wait) + time.Sleep(wait) } }(event) } diff --git a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go index db056254..081e878c 100644 --- a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go @@ -51,6 +51,21 @@ type player struct { next int pending map[string]chan wire asked []string + // subscribed are the patterns the bundle subscribed, as the runtime takes them: an event is delivered only once its + // pattern is taken, as the runtime's consumer would hold it until then. + subscribed []string +} + +// hears says the bundle has subscribed pattern. +func (p *player) hears(pattern string) bool { + p.mu.Lock() + defer p.mu.Unlock() + for _, s := range p.subscribed { + if s == pattern { + return true + } + } + return false } func (p *player) send(m map[string]any) { @@ -198,7 +213,14 @@ func (p *player) handle(m wire) { p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}}) } case "mesh/subscribe": + var sub struct { + Pattern string `json:"pattern"` + } + _ = json.Unmarshal(m.Params, &sub) p.answer(m.ID, map[string]any{}, nil) + p.mu.Lock() + p.subscribed = append(p.subscribed, sub.Pattern) + p.mu.Unlock() case "mesh/state.keys": kv, err := p.kv(a.State) if err != nil { @@ -368,6 +390,8 @@ func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { return a != nil && strings.Contains(fmt.Sprint(a["why"]), "names a controller") }) + // The runtime's consumer holds an event until the bundle subscribes: delivered once it hears them. + eventually(t, "the healers hearing the keeper's conditions", func() bool { return p.hears(RaisedEvent) && p.hears(ChangedEvent) }) // The controller holds the lease: the induced failure is written as would, never acted on. p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records")) eventually(t, "the would-act written", func() bool { return len(keys(Module+"_"+WouldState)) == 1 }) diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index 08654a2d..9e0760fd 100644 --- a/modules/mesh-issues/cmd/mesh-issues/main.go +++ b/modules/mesh-issues/cmd/mesh-issues/main.go @@ -163,14 +163,15 @@ func hearOpened(t *Tracker) { } hearings = append(hearings, hearing{DeliveredEvent, t.OnDelivered}) for _, s := range hearings { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(s.pattern, s.handle) if err == nil { logf("[mesh-issues] the triage hears %s", s.pattern) break } logf("[mesh-issues] the triage does not hear %s yet (%v); asking again in %s", s.pattern, err, wait) + time.Sleep(wait) } } } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index 9c6ab06a..bf04e2de 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -114,7 +114,6 @@ func callerOf(args map[string]any) string { return stdio.CallerOf(args) } // hear subscribes to the router's word on this module's asks, asked again until the runtime takes it. func hear(k *KB) { for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error { var w asks.Warrant if json.Unmarshal(e.Body, &w) == nil { @@ -127,6 +126,7 @@ func hear(k *KB) { return } logf("[mesh-knowledge] does not hear the operator's answers yet (%v); asking again in %s", err, wait) + time.Sleep(wait) } } diff --git a/modules/mesh-mail/cmd/mesh-mail/main.go b/modules/mesh-mail/cmd/mesh-mail/main.go index cb66ae89..499bb8fc 100644 --- a/modules/mesh-mail/cmd/mesh-mail/main.go +++ b/modules/mesh-mail/cmd/mesh-mail/main.go @@ -250,8 +250,8 @@ func waitsTellTicker(b *Bundle) { // approved send goes out as soon as the operator answers it, never only when something later calls // mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790). func hear(b *Bundle) { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error { var w asks.Warrant if json.Unmarshal(e.Body, &w) == nil { @@ -264,6 +264,7 @@ func hear(b *Bundle) { return } logf("[mesh-mail] does not hear the operator's answers yet (%v); asking again in %s", err, wait) + time.Sleep(wait) } } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index 95d795be..fda63ee2 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -136,8 +136,8 @@ func main() { return nil }) go func() { + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) err := stdio.Subscribe(asks.Seat+".decided."+Module, func(stdio.Envelope) error { for _, said := range h.Decide() { logf("[%s] %s", Module, said) @@ -148,6 +148,7 @@ func main() { return } logf("[%s] does not hear the operator's answers yet (%v); asking again in %s", Module, err, wait) + time.Sleep(wait) } }() if err := stdio.Serve("", append(Tools(s), holderTools(h)...)); err != nil { diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index a4fb0b55..8ab0e45d 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -114,7 +114,7 @@ func run(w *Watcher, l *listening, node string) { w.Heartbeat(b) return nil } - time.Sleep(500 * time.Millisecond) + <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: a heartbeat sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(ControllerSeat+".*", handle) if err == nil {