From 6aeb401095290e056d7ae8d1f10ff771bdc244f6 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:29:30 +0100 Subject: [PATCH 1/2] buildrun: read an ask's step at every step end of its build module, so a deaf watch loses no end mesh-catalog #553's verdict stayed running after its step passed on shanks at 19:09:31Z on 2026-10-28: the watch of that one step key, opened at 18:46:05Z, heard neither the start nor the end and never closed. HeardEnds stands between Asker and each ask's watch: nudged at each step end of the same build module (HearEnds: one watch per steps state) and at a reconnect, every waited ask reads its keys now; a watch that closes is opened again, at most 3 times. The checker and the builder both wait through it. Tests: a section of the asker's test (deaf, nudged by another module and by its own, a watch that closes); counts unchanged. --- go/delivery/buildrun/asks_test.go | 48 ++++++ go/delivery/buildrun/heardends.go | 234 ++++++++++++++++++++++++++++++ 2 files changed, 282 insertions(+) create mode 100644 go/delivery/buildrun/heardends.go diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 6905d2e..7453b63 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -109,6 +109,31 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if err != nil || !IsLost(step) { t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err) } + // The lost check of 2026-10-28: a step whose own watch went deaf is read at the next nudge (a step of its build module + // ended) and answered; a nudge of another build module reads nothing; a watch that closes is opened again. + d := &FakeAsks{} + heard := &HeardEnds{Inner: deafWatch{}, Read: d} + k = Asker{Publisher: d, Waiter: heard, Bound: time.Minute} + d.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "shanks", State: delivery.StepPassed}) + got := make(chan delivery.Step, 1) + go func() { s, _ := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); got <- s }() + for heard.Waiting() == 0 { + time.Sleep(time.Millisecond) + } + heard.Nudge(NodeBuildModule) + select { + case s := <-got: + t.Fatalf("a nudge of another build module answered %+v", s) + case <-time.After(20 * time.Millisecond): + } + heard.Nudge(GoBuildModule) + if s := <-got; s.State != delivery.StepPassed || s.Node != "shanks" { + t.Fatalf("a step its watch never handed over: %+v", s) + } + closing := &HeardEnds{Inner: &closingWatch{inner: d}, Read: d} + if s, err := (Asker{Publisher: d, Waiter: closing}).Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || s.State != delivery.StepPassed { + t.Fatalf("a watch that closed once: %+v, %v", s, err) + } for _, row := range []struct { h *Hearing want string @@ -165,3 +190,26 @@ func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, hear func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n } func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) } + +// deafWatch is a watch of an ask's step that stays open and hears nothing. +type deafWatch struct{} + +func (deafWatch) WatchAsk(context.Context, string, string) (<-chan delivery.Step, error) { + return make(chan delivery.Step), nil +} + +// closingWatch closes its first watch at once, and hands the later ones to inner. +type closingWatch struct { + inner AskWaiter + closed bool +} + +func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + if !c.closed { + c.closed = true + ch := make(chan delivery.Step) + close(ch) + return ch, nil + } + return c.inner.WatchAsk(ctx, module, askID) +} diff --git a/go/delivery/buildrun/heardends.go b/go/delivery/buildrun/heardends.go new file mode 100644 index 0000000..74f5dad --- /dev/null +++ b/go/delivery/buildrun/heardends.go @@ -0,0 +1,234 @@ +package buildrun + +import ( + "context" + "fmt" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A step's end is never only its own watch's to hear** (the lost check of 2026-10-28: mesh-catalog #553's verdict +// stayed running after its step mvsgthyjed5786 had passed on shanks at 19:09:31Z; the watch of that one key, opened at +// 18:46:05Z, heard neither the step's start nor its end and never closed, while a watch opened at 18:58:48Z heard its own +// step end). HeardEnds stands between the asker (Asker) and the watch of each ask's step key: when it is +// nudged, every ask it waits on reads its step keys now (a read, not the watch) and hands what it reads to its wait, so a +// step that ended is judged even when its own watch went deaf. A watch that closes under a wait is opened again, at most +// reopenTries times. It is nudged on an event (Nudge): any step of the same build module ending (one watch of each steps +// state, HearEnds), a reconnect to the bus, and whatever else the asker hears (the checker: checker_ask_again). +// It holds no timer. Every asker of the build seats waits through it (the checker and the builder, HearEnds). + +// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.`. +type StepReader interface { + AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) +} + +// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed. +const reopenTries = 3 + +// HeardEnds is an AskWaiter over Inner that also reads an ask's steps when nudged. Safe for concurrent use. +type HeardEnds struct { + Inner AskWaiter + Read StepReader + Logf func(format string, args ...any) + + mu sync.Mutex + waiting map[string]*waitedAsk // by ask id +} + +type waitedAsk struct { + module string + nudge chan struct{} +} + +func (h *HeardEnds) logf(format string, args ...any) { + if h.Logf != nil { + h.Logf(format, args...) + } +} + +// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now. +func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + in, err := h.Inner.WatchAsk(ctx, module, askID) + if err != nil { + return nil, err + } + w := &waitedAsk{module: module, nudge: make(chan struct{}, 1)} + h.mu.Lock() + if h.waiting == nil { + h.waiting = map[string]*waitedAsk{} + } + h.waiting[askID] = w + h.mu.Unlock() + out := make(chan delivery.Step) + go func() { + defer close(out) + defer func() { + h.mu.Lock() + if h.waiting[askID] == w { + delete(h.waiting, askID) + } + h.mu.Unlock() + }() + send := func(s delivery.Step) bool { + select { + case out <- s: + return true + case <-ctx.Done(): + return false + } + } + reopened := 0 + for { + select { + case <-ctx.Done(): + return + case s, ok := <-in: + if !ok { + if ctx.Err() != nil || reopened >= reopenTries { + h.logf("the watch of the ask %s's step on %s closed %d times; its wait ends", askID, module, reopened+1) + return + } + reopened++ + if in, err = h.Inner.WatchAsk(ctx, module, askID); err != nil { + h.logf("the watch of the ask %s's step on %s closed and cannot be opened again, so its wait ends: %v", askID, module, err) + return + } + h.logf("the watch of the ask %s's step on %s closed; opened again (%d of %d)", askID, module, reopened, reopenTries) + continue + } + if !send(s) { + return + } + case <-w.nudge: + if h.Read == nil { + continue + } + steps, err := h.Read.AskSteps(ctx, module, askID) + if err != nil { + h.logf("the steps of the ask %s on %s cannot be read: %v", askID, module, err) + continue + } + for _, s := range steps { + if s.State != delivery.StepRunning { + h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State) + } + if !send(s) { + return + } + } + } + } + }() + return out, nil +} + +// Nudge has every ask waited on of module read its steps now; module "" nudges every ask. +func (h *HeardEnds) Nudge(module string) { + h.mu.Lock() + defer h.mu.Unlock() + for _, w := range h.waiting { + if module == "" || w.module == module { + select { + case w.nudge <- struct{}{}: + default: // a nudge is pending already + } + } + } +} + +// Waiting is how many asks are waited on now. +func (h *HeardEnds) Waiting() int { + h.mu.Lock() + defer h.mu.Unlock() + return len(h.waiting) +} + +// AskSteps is StepReader's: the keys `*.` of the build module's steps state, read now. +func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) { + pattern, err := delivery.AskStepWatch(askID) + if err != nil { + return nil, err + } + kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + lister, err := kv.ListKeysFiltered(ctx, pattern) + if err != nil { + return nil, err + } + var out []delivery.Step + for key := range lister.Keys() { + e, err := kv.Get(ctx, key) + if err != nil { + continue // deleted between the list and the read + } + if step, err := delivery.DecodeStep(e.Value()); err == nil { + out = append(out, step) + } + } + return out, nil +} + +// HearEnds is the HeardEnds every asker of the build seats waits through: over inner (the asks' own watches, or a +// LostAskWaiter's inner), reading with asks, and nudged at each step end of each of modules (one watch of each steps state, +// updates only; one that closes is opened again while it heard something since it was opened, and said and left +// otherwise). The asker nudges it at a reconnect too (Nudge("")). +func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules []string, logf func(string, ...any)) *HeardEnds { + if logf == nil { + logf = func(string, ...any) {} + } + h := &HeardEnds{Inner: inner, Read: asks, Logf: logf} + for _, module := range modules { + kv, err := asks.JS.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + logf("the steps of %s cannot be read (%v): an ask whose own watch goes deaf is read only at a reconnect", module, err) + continue + } + go h.follow(ctx, kv, module) + } + return h +} + +func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module string) { + for ctx.Err() == nil { + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + h.logf("the steps of %s cannot be watched, so an ask whose own watch goes deaf is read only at a reconnect: %v", module, err) + return + } + heard := false + for e := range w.Updates() { + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + heard = true + if step, err := delivery.DecodeStep(e.Value()); err == nil && step.State != delivery.StepRunning { + h.Nudge(module) + } + } + _ = w.Stop() + if ctx.Err() != nil { + return + } + h.Nudge(module) // a step may have ended while the watch was closing + if !heard { + h.logf("the watch of the steps of %s closed having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module) + return + } + h.logf("the watch of the steps of %s closed; opened again", module) + } +} + +// AskSteps is StepReader's: the last step written for the ask, if any. +func (f *FakeAsks) AskSteps(_ context.Context, _ string, askID string) ([]delivery.Step, error) { + f.mu.Lock() + defer f.mu.Unlock() + if s := f.steps[askID]; len(s) > 0 { + return []delivery.Step{s[len(s)-1]}, nil + } + return nil, nil +} -- 2.54.0 From f91340fe82ef16601f40bc192d8ef5d87fe1f4f2 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 20:35:02 +0100 Subject: [PATCH 2/2] buildrun: open a steps watch again that closed having heard nothing, up to 3 closes in a row A watch that closed at once was left for good, so an ask whose own watch went deaf was read only at a reconnect (review of fix/lost-check). --- go/delivery/buildrun/heardends.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/go/delivery/buildrun/heardends.go b/go/delivery/buildrun/heardends.go index 74f5dad..25b8cbb 100644 --- a/go/delivery/buildrun/heardends.go +++ b/go/delivery/buildrun/heardends.go @@ -175,8 +175,7 @@ func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]de // HearEnds is the HeardEnds every asker of the build seats waits through: over inner (the asks' own watches, or a // LostAskWaiter's inner), reading with asks, and nudged at each step end of each of modules (one watch of each steps state, -// updates only; one that closes is opened again while it heard something since it was opened, and said and left -// otherwise). The asker nudges it at a reconnect too (Nudge("")). +// updates only; one that closes is opened again, and said and left after reopenTries closes in a row that heard nothing). The asker nudges it at a reconnect too (Nudge("")). func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules []string, logf func(string, ...any)) *HeardEnds { if logf == nil { logf = func(string, ...any) {} @@ -194,6 +193,7 @@ func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules } func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module string) { + silent := 0 // closes in a row having heard nothing for ctx.Err() == nil { w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) if err != nil { @@ -215,8 +215,10 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st return } h.Nudge(module) // a step may have ended while the watch was closing - if !heard { - h.logf("the watch of the steps of %s closed having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module) + if heard { + silent = 0 + } else if silent++; silent >= reopenTries { + h.logf("the watch of the steps of %s closed %d times in a row having heard nothing, so it is left: an ask whose own watch goes deaf is read only at a reconnect", module, silent) return } h.logf("the watch of the steps of %s closed; opened again", module) -- 2.54.0