Subscribe as soon as the runtime serves, never after a guessed wait, so no event sent at a start is missed (issue 774)
mesh/delivery merged: source.main moved to novox/mesh-catalog@5d9cead336ab
mesh/delivery-group group fix/held-module-and-assigner-contract merged: every member is merged
mesh/check checking b5475cfa5c3d on 395a37dddab9 | novox/mesh-catalog#571 in group fix/held-module-and-assigner-contract

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.
This commit is contained in:
2026-10-30 21:57:31 +01:00
parent b8ce4f0c0a
commit b5475cfa5c
9 changed files with 41 additions and 10 deletions
+4 -2
View File
@@ -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)
}
@@ -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 {
@@ -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)
}
@@ -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 })
+2 -1
View File
@@ -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)
}
}
}
@@ -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)
}
}
+2 -1
View File
@@ -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)
}
}
@@ -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 {
@@ -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 {