From e7f0dc612e6e62dca9f558d0cde80ed69f86ca3a Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:51:35 +0100 Subject: [PATCH 1/4] buscompose: a user of a seat pulled by kind reads its seat's dead letters, so an asker hears an ask given up (issue 734) A check ask given up during a bus reload left its try running for ever: the asker had no grant to learn that DEAD_LETTERS holds it. DeadLetterReads grants an ordered consumer on DEAD_LETTERS filtered to the seat stream's copies, and nothing else of the stream. --- go/buscompose/buscompose_test.go | 30 ++++++++++++++++++++++++++++++ go/buscompose/seattraffic.go | 23 +++++++++++++++++++++-- 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index f5244ba..1a685ed 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -69,6 +69,36 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { } }) } + // An asker on a seat pulled by kind hears what that seat's workers gave up on, and no other stream's dead letters + // (novox/hq issue 734: a check ask given up during a bus reload left its try running for ever). + t.Run("an asker reads its seat's dead letters", func(t *testing.T) { + build := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true} + perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-checker", Uses: []Seat{build}}) + if err != nil { + t.Fatal(err) + } + for subject, want := range map[string]bool{ + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>": true, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.w.next": true, + "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1": true, + "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false, + "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1": false, + "$JS.API.STREAM.PURGE.DEAD_LETTERS": false, + } { + if got := MayPublish(perms, subject); got != want { + t.Errorf("may publish %s: %v, want %v", subject, got, want) + } + } + plain, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop", + Uses: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}}) + if err != nil { + t.Fatal(err) + } + if MayPublish(plain, "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1") { + t.Error("a user of a seat not pulled by kind reads DEAD_LETTERS") + } + }) t.Run("the controller's golden server configuration", func(t *testing.T) { seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}} got, err := Compose( diff --git a/go/buscompose/seattraffic.go b/go/buscompose/seattraffic.go index 1b7095c..d7b1430 100644 --- a/go/buscompose/seattraffic.go +++ b/go/buscompose/seattraffic.go @@ -24,8 +24,9 @@ import ( // subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded // holder asks with its own kind; the modules that watch the seat answer. // -// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only -// (`$KV...>`), so an asker reads the state of its own asks and no other asker's. +// And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only +// (`$KV...>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by +// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734). // BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written // here as the broker mirrors the catalogue rather than importing it. @@ -129,6 +130,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { } t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>") } + if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) { + t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...) + } } for _, w := range watches { if w.Kinded { @@ -151,6 +155,21 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { return t } +// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers +// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that +// seat's stream (DeadLetterSubject(stream, *)), pulled or pushed, its deletion and its flow control. A copy carries the ask's own message id +// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead +// letters and no other stream's, and changes none. +func DeadLetterReads(stream string) []string { + return []string{ + "$JS.API.STREAM.INFO." + DeadLettersStream, + "$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"), + "$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*", + "$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".>", + "$JS.FC." + DeadLettersStream + ".>", + } +} + // seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a // worker is pulled and acknowledged through and a record is read through. func seatTrafficGrants(t SeatTraffic) (pub, sub []string) { From d96b9fd6606977a0d509bb3fb3df94e53b18d007 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:00:46 +0100 Subject: [PATCH 2/4] buildrun: an ask the build seat gave up on ends its asker's wait as failed with why, for every asker of the build seats (issue 734) The checker's try and the builder's lanes waited for ever on an ask given up during a bus reload: no build node writes a step for it. LostAskWaiter hands the asker a failed step on the node bus when DEAD_LETTERS keeps the ask's copy (HearDeadLetters, granted by DeadLetterReads), and under NotTakenBound when no node took the ask within the bound. --- go/delivery/buildrun/asks_test.go | 29 +++ go/delivery/buildrun/lostask.go | 296 ++++++++++++++++++++++++++++++ 2 files changed, 325 insertions(+) create mode 100644 go/delivery/buildrun/lostask.go diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 3365202..44281b1 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -48,6 +48,35 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if len(g.Asks()) != 0 { t.Fatal("a wait put an ask") } + // Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed + // with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written + // first is answered as the node wrote it. + dead := &DeadLetters{} + lw := LostAskWaiter{Inner: g, Dead: dead} + k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute} + g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()}) + given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"}, + "Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}}) + if !ok { + t.Fatal("a dead letter's headers were not read") + } + go dead.Heard(given) + step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970")) + if err != nil || step.State != delivery.StepFailed || !IsLost(step) || + !strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") { + t.Fatalf("a given-up ask: %+v, %v", step, err) + } + if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) { + t.Fatalf("a given-up ask's error: %v", err) + } + step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971")) + if err != nil || !IsNotTaken(step) || IsLost(step) { + t.Fatalf("an ask no node took: %+v, %v", step, err) + } + g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed}) + if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" { + t.Fatalf("a step written: %+v, %v", step, err) + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go new file mode 100644 index 0000000..52f7ee8 --- /dev/null +++ b/go/delivery/buildrun/lostask.go @@ -0,0 +1,296 @@ +package buildrun + +import ( + "context" + "fmt" + "strconv" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask +// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node +// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under +// mesh.events.dead.., with the ask's message id (its ask id), how often it was handed out and when it +// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays +// running and holds a try place, a check's verdict stays running and its pull request waits. +// +// An asker of the build seats (the checker, the builder) hears those copies (HearDeadLetters: an ordered consumer on +// DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose +// DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its +// reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the +// builder asks its part of a build again). +// +// Belt and braces, where the asker says so in the context (NotTakenBound; the checker's tries): an ask no build node took +// within a bound ends the same way, saying NotTakenWords. An ask waits in the stream as long as the backlog is otherwise +// (ADR 0525 §9): a backlog is no fault. + +// LostNode is the node a step ended for a lost ask names: no build node wrote it. +const LostNode = "bus" + +// The words of a step ended for its ask. +const ( + LostWords = "its ask was given up by the build seat" + NotTakenWords = "no build node took its ask" +) + +// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link +// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up. +const ( + DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + DeadHeaderDeliveries = "Mesh-Dead-Deliveries" + DeadHeaderGaveUp = "Mesh-Dead-Gave-Up" + DeadHeaderStream = "Mesh-Dead-Stream" +) + +// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat +// stream's copies there (buscompose DeadLetterSubject). +const DeadLettersStream = "DEAD_LETTERS" + +// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under. +func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" } + +// SeatStream is the stream of a build seat's language (buscompose seatStreamName of node-build-). +func SeatStream(language string) string { return "SEAT_NODE_BUILD_" + strings.ToUpper(language) } + +// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it. +type GivenUp struct { + AskID string + Stream string + Deliveries int + At time.Time +} + +// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id. +func GivenUpOf(header map[string][]string) (GivenUp, bool) { + get := func(k string) string { + if v := header[k]; len(v) > 0 { + return strings.TrimSpace(v[0]) + } + return "" + } + g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)} + if g.AskID == "" { + return GivenUp{}, false + } + g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries)) + g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp)) + return g, true +} + +// Words are what a step ended for the given-up ask says. +func (g GivenUp) Words() string { + at := "at an unknown time" + if !g.At.IsZero() { + at = "at " + g.At.UTC().Format(time.RFC3339) + } + return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream) +} + +// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard. +const keptGivenUp = 1024 + +// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use. +type DeadLetters struct { + mu sync.Mutex + heard map[string]GivenUp + order []string + watchers map[string][]chan GivenUp +} + +// Heard takes one given-up ask in. +func (d *DeadLetters) Heard(g GivenUp) { + d.mu.Lock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if _, ok := d.heard[g.AskID]; !ok { + d.order = append(d.order, g.AskID) + if len(d.order) > keptGivenUp { + delete(d.heard, d.order[0]) + d.order = d.order[1:] + } + } + d.heard[g.AskID] = g + ws := d.watchers[g.AskID] + delete(d.watchers, g.AskID) + d.mu.Unlock() + for _, w := range ws { + w <- g // buffered, one each + } +} + +// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel. +func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) { + ch := make(chan GivenUp, 1) + d.mu.Lock() + defer d.mu.Unlock() + if d.heard == nil { + d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{} + } + if g, ok := d.heard[askID]; ok { + ch <- g + return ch, func() {} + } + d.watchers[askID] = append(d.watchers[askID], ch) + return ch, func() { + d.mu.Lock() + defer d.mu.Unlock() + ws := d.watchers[askID] + for i, w := range ws { + if w == ch { + d.watchers[askID] = append(ws[:i], ws[i+1:]...) + break + } + } + if len(d.watchers[askID]) == 0 { + delete(d.watchers, askID) + } + } +} + +type notTakenKey struct{} + +// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's). +func NotTakenBound(ctx context.Context, bound time.Duration) context.Context { + return context.WithValue(ctx, notTakenKey{}, bound) +} + +func notTakenBoundOf(ctx context.Context) time.Duration { + d, _ := ctx.Value(notTakenKey{}).(time.Duration) + return d +} + +// LostAskWaiter is an AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under +// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on +// the node "bus". Every step the build module writes is handed on as it is. +type LostAskWaiter struct { + Inner AskWaiter + Dead *DeadLetters + Now func() time.Time +} + +func (w LostAskWaiter) now() time.Time { + if w.Now != nil { + return w.Now() + } + return time.Now() +} + +// WatchAsk is AskWaiter's. +func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + inner, err := w.Inner.WatchAsk(ctx, module, askID) + if err != nil { + return nil, err + } + var dead <-chan GivenUp + cancel := func() {} + if w.Dead != nil { + dead, cancel = w.Dead.watch(askID) + } + // The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask. + var notTaken <-chan struct{} + bound := notTakenBoundOf(ctx) + stopBound := func() {} + if bound > 0 { + var nt context.Context + nt, stopBound = context.WithTimeout(context.Background(), bound) + notTaken = nt.Done() + } + out := make(chan delivery.Step) + go func() { + defer close(out) + defer cancel() + defer func() { stopBound() }() + send := func(s delivery.Step) bool { + select { + case out <- s: + return true + case <-ctx.Done(): + return false + } + } + ended := func(reason string) delivery.Step { + at := w.now().UTC() + return delivery.Step{ID: askID, Node: LostNode, State: delivery.StepFailed, Reason: reason, Ended: at} + } + for { + select { + case <-ctx.Done(): + return + case s, ok := <-inner: + if !ok { + return + } + if notTaken != nil { + stopBound() // a node took it: its step is bounded by the asker from its start + notTaken = nil + } + if !send(s) { + return + } + case g := <-dead: + dead = nil + send(ended(g.Words())) + return + case <-notTaken: + send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound))) + return + } + } + }() + return out, nil +} + +// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords). +func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) } + +// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound. +func IsNotTaken(s delivery.Step) bool { + return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords) +} + +// HearDeadLetters follows, for each language's build seat stream, the copies DEAD_LETTERS keeps of the asks its workers +// gave up on, into dead: an ordered consumer from now, which resumes where it was after a reconnect, and starts here again +// when it is reset before it heard any, so a bus reload loses none. Where the bus refuses it (the grant not yet composed) +// it says so and goes on; the stop it answers ends every consumer it started. +func HearDeadLetters(ctx context.Context, js jetstream.JetStream, languages []string, dead *DeadLetters, logf func(string, ...any)) func() { + var stops []func() + since := time.Now() + for _, language := range languages { + stream := SeatStream(language) + filter := DeadLetterFilter(stream) + notHeard := func(err error) { + logf("the asks %s gives up on are not heard (%s of %s), so a wait for one ends only at its bound: %v", + stream, filter, DeadLettersStream, err) + } + cons, err := js.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) + if err != nil { + notHeard(err) + continue + } + cc, err := cons.Consume(func(m jetstream.Msg) { + if g, ok := GivenUpOf(m.Headers()); ok { + logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream, + g.Deliveries, g.At.Format(time.RFC3339)) + dead.Heard(g) + } + }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", DeadLettersStream, err) })) + if err != nil { + notHeard(err) + continue + } + stops = append(stops, cc.Stop) + } + return func() { + for _, stop := range stops { + stop() + } + } +} From 9b01c94a9e1c02fdabfa559f79c2a5dbdc1fab70 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:16:58 +0100 Subject: [PATCH 3/4] buildrun: the asker hears DEAD_LETTERS again on its next reconnect or ask, in the background, and says whether it hears; buscompose grants only what a pulled ordered consumer needs (issue 734, review round 1) An asker started before its grant was composed was refused once and stayed deaf until it started again, silently. DeadLetterHearing tries in the background and again on an event (Again: a reconnect, an ask waited for while deaf), and tells each change of Hearing so the module's health tool can say it. DeadLetterReads drops the flow control a pull consumer does not use and deletes one consumer by name. --- go/buscompose/buscompose_test.go | 1 + go/buscompose/seattraffic.go | 5 +- go/delivery/buildrun/asks_test.go | 52 +++++++ go/delivery/buildrun/lostask.go | 231 +++++++++++++++++++++++++----- 4 files changed, 249 insertions(+), 40 deletions(-) diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 1a685ed..e7ff21c 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -85,6 +85,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false, "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1": false, "$JS.API.STREAM.PURGE.DEAD_LETTERS": false, + "$JS.FC.DEAD_LETTERS.x1": false, } { if got := MayPublish(perms, subject); got != want { t.Errorf("may publish %s: %v, want %v", subject, got, want) diff --git a/go/buscompose/seattraffic.go b/go/buscompose/seattraffic.go index d7b1430..e89b208 100644 --- a/go/buscompose/seattraffic.go +++ b/go/buscompose/seattraffic.go @@ -157,7 +157,7 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { // DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers // gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that -// seat's stream (DeadLetterSubject(stream, *)), pulled or pushed, its deletion and its flow control. A copy carries the ask's own message id +// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id // (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead // letters and no other stream's, and changes none. func DeadLetterReads(stream string) []string { @@ -165,8 +165,7 @@ func DeadLetterReads(stream string) []string { "$JS.API.STREAM.INFO." + DeadLettersStream, "$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"), "$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*", - "$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".>", - "$JS.FC." + DeadLettersStream + ".>", + "$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*", } } diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 44281b1..f5ea8c3 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -2,7 +2,9 @@ package buildrun import ( "context" + "errors" "strings" + "sync" "testing" "time" @@ -77,6 +79,33 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" { t.Fatalf("a step written: %+v, %v", step, err) } + // Review round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and + // tries again on the next ask waited for; heard, it says so and hands the given-up asks on. + fol := &refusingFollower{refuse: 1} + var mu sync.Mutex + var told []Hearing + hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil, + func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() }) + hearing.Wait() + if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") { + t.Fatalf("a refused hearing says %q", why) + } + k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute} + g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed}) + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil { + t.Fatal(err) + } + hearing.Wait() + mu.Lock() + if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 { + t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls()) + } + mu.Unlock() + fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5}) + step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974")) + if err != nil || !IsLost(step) { + t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err) + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the @@ -98,3 +127,26 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) { t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err) } } + +// refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed. +type refusingFollower struct { + mu sync.Mutex + refuse int + n int + hand func(GivenUp) +} + +func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) { + f.mu.Lock() + defer f.mu.Unlock() + f.n++ + if f.n <= f.refuse { + return nil, errors.New("permissions violation") + } + f.hand = heard + return func() {}, nil +} + +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) } diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go index 52f7ee8..1e70245 100644 --- a/go/delivery/buildrun/lostask.go +++ b/go/delivery/buildrun/lostask.go @@ -20,7 +20,7 @@ import ( // was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays // running and holds a try place, a check's verdict stays running and its pull request waits. // -// An asker of the build seats (the checker, the builder) hears those copies (HearDeadLetters: an ordered consumer on +// An asker of the build seats (the checker, the builder) hears those copies (DeadLetterHearing: an ordered consumer on // DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose // DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its // reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the @@ -172,7 +172,13 @@ func notTakenBoundOf(ctx context.Context) time.Duration { type LostAskWaiter struct { Inner AskWaiter Dead *DeadLetters - Now func() time.Time + // Hearing, when set, is asked to hear DEAD_LETTERS again each time an ask is waited for while it cannot: an ask put is + // the event that makes hearing matter. + Hearing interface { + Deaf() string + Again() + } + Now func() time.Time } func (w LostAskWaiter) now() time.Time { @@ -184,6 +190,9 @@ func (w LostAskWaiter) now() time.Time { // WatchAsk is AskWaiter's. func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { + if w.Hearing != nil && w.Hearing.Deaf() != "" { + w.Hearing.Again() + } inner, err := w.Inner.WatchAsk(ctx, module, askID) if err != nil { return nil, err @@ -255,42 +264,190 @@ func IsNotTaken(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords) } -// HearDeadLetters follows, for each language's build seat stream, the copies DEAD_LETTERS keeps of the asks its workers -// gave up on, into dead: an ordered consumer from now, which resumes where it was after a reconnect, and starts here again -// when it is reset before it heard any, so a bus reload loses none. Where the bus refuses it (the grant not yet composed) -// it says so and goes on; the stop it answers ends every consumer it started. -func HearDeadLetters(ctx context.Context, js jetstream.JetStream, languages []string, dead *DeadLetters, logf func(string, ...any)) func() { - var stops []func() - since := time.Now() - for _, language := range languages { - stream := SeatStream(language) - filter := DeadLetterFilter(stream) - notHeard := func(err error) { - logf("the asks %s gives up on are not heard (%s of %s), so a wait for one ends only at its bound: %v", - stream, filter, DeadLettersStream, err) - } - cons, err := js.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{ - FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) - if err != nil { - notHeard(err) - continue - } - cc, err := cons.Consume(func(m jetstream.Msg) { - if g, ok := GivenUpOf(m.Headers()); ok { - logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream, - g.Deliveries, g.At.Format(time.RFC3339)) - dead.Heard(g) - } - }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", DeadLettersStream, err) })) - if err != nil { - notHeard(err) - continue - } - stops = append(stops, cc.Stop) +// **The asker hears DEAD_LETTERS, tries again when it cannot, and says whether it hears** (issue 734). The read is +// granted with the asker's seat traffic (buscompose DeadLetterReads); an asker started before its grant was composed is +// refused, and an asker that only tried once would stay deaf until it starts again, silently. DeadLetterHearing tries in +// the background, never holding the start up, and again on an event (Again): the asker's reconnect to the bus, and every +// ask waited for while it is deaf (LostAskWaiter.Hearing); no timer. Each change of whether it hears is said (Said), so +// the module's health tool says it: a deaf asker is not healthy. + +// Hearing is what a DeadLetterHearing says: whether it hears every build seat stream it follows, and why not. +type Hearing struct { + Heard bool `json:"heard"` + Streams []string `json:"streams"` + Why string `json:"why,omitempty"` + At time.Time `json:"at"` +} + +// HearingKey is the key a module keeps its Hearing under, in its own state (the checker's and the builder's hearing). +const HearingKey = "dead-letters" + +// DeadLetterFollower starts following one seat stream's copies in DEAD_LETTERS from since, handing each given-up ask to +// heard; it answers the stop, or why it cannot. +type DeadLetterFollower interface { + Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) +} + +// JetStreamDeadLetters is the bus side of a DeadLetterFollower: an ordered consumer on DEAD_LETTERS filtered to the +// stream's copies, which resumes where it was after a reconnect and, reset before it heard any, starts at since again. +type JetStreamDeadLetters struct{ JS jetstream.JetStream } + +// Follow is DeadLetterFollower's. +func (j JetStreamDeadLetters) Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) { + cons, err := j.JS.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{DeadLetterFilter(stream)}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since}) + if err != nil { + return nil, err } - return func() { - for _, stop := range stops { - stop() + cc, err := cons.Consume(func(m jetstream.Msg) { + if g, ok := GivenUpOf(m.Headers()); ok { + heard(g) } + }) + if err != nil { + return nil, err + } + return cc.Stop, nil +} + +// DeadLetterHearing follows the build seats' given-up asks into Dead. Make it with NewDeadLetterHearing. +type DeadLetterHearing struct { + follower DeadLetterFollower + streams []string + dead *DeadLetters + logf func(string, ...any) + said func(Hearing) + ctx context.Context + since time.Time + + mu sync.Mutex + stops map[string]func() + why map[string]string + trying bool + again bool + told *Hearing + wg sync.WaitGroup +} + +// NewDeadLetterHearing starts hearing the given-up asks of each language's build seat stream, in the background. said +// is told each change of whether it hears (nil tells nothing); ctx ends it. +func NewDeadLetterHearing(ctx context.Context, follower DeadLetterFollower, languages []string, dead *DeadLetters, + logf func(string, ...any), said func(Hearing)) *DeadLetterHearing { + h := &DeadLetterHearing{follower: follower, dead: dead, logf: logf, said: said, ctx: ctx, since: time.Now(), + stops: map[string]func(){}, why: map[string]string{}} + for _, l := range languages { + s := SeatStream(l) + h.streams = append(h.streams, s) + h.why[s] = "not tried yet" + } + if h.logf == nil { + h.logf = func(string, ...any) {} + } + h.Again() + return h +} + +// Deaf says why a stream is not heard, or "" when every one is. +func (h *DeadLetterHearing) Deaf() string { + h.mu.Lock() + defer h.mu.Unlock() + return h.deafLocked() +} + +func (h *DeadLetterHearing) deafLocked() string { + var out []string + for _, s := range h.streams { + if why, deaf := h.why[s]; deaf { + out = append(out, s+": "+why) + } + } + if len(out) == 0 { + return "" + } + return DeadLettersStream + " is not heard, so an ask a build seat gives up on is waited for until its bound: " + + strings.Join(out, "; ") +} + +// Again tries, in the background, every stream not heard yet; a call while it tries makes it try once more after. +func (h *DeadLetterHearing) Again() { + h.mu.Lock() + if h.trying { + h.again = true + h.mu.Unlock() + return + } + h.trying = true + h.mu.Unlock() + h.wg.Add(1) + go func() { + defer h.wg.Done() + for { + h.try() + h.mu.Lock() + if !h.again || h.ctx.Err() != nil { + h.trying = false + h.mu.Unlock() + return + } + h.again = false + h.mu.Unlock() + } + }() +} + +// Wait returns when no try runs. +func (h *DeadLetterHearing) Wait() { h.wg.Wait() } + +// Stop ends every stream's following. +func (h *DeadLetterHearing) Stop() { + h.mu.Lock() + defer h.mu.Unlock() + for s, stop := range h.stops { + stop() + delete(h.stops, s) + } +} + +func (h *DeadLetterHearing) try() { + for _, s := range h.streams { + h.mu.Lock() + _, deaf := h.why[s] + h.mu.Unlock() + if !deaf || h.ctx.Err() != nil { + continue + } + stream := s + stop, err := h.follower.Follow(h.ctx, stream, h.since, func(g GivenUp) { + h.logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream, + g.Deliveries, g.At.Format(time.RFC3339)) + h.dead.Heard(g) + }) + h.mu.Lock() + if err != nil { + h.why[stream] = err.Error() + h.logf("the asks %s gives up on are not heard (%s of %s); tried again at the next reconnect or ask: %v", + stream, DeadLetterFilter(stream), DeadLettersStream, err) + } else { + delete(h.why, stream) + h.stops[stream] = stop + h.logf("the asks %s gives up on are heard from %s", stream, DeadLettersStream) + } + h.mu.Unlock() + } + h.tell() +} + +// tell says the hearing when it changed. +func (h *DeadLetterHearing) tell() { + h.mu.Lock() + why := h.deafLocked() + now := Hearing{Heard: why == "", Streams: append([]string(nil), h.streams...), Why: why, At: time.Now().UTC()} + same := h.told != nil && h.told.Heard == now.Heard && h.told.Why == now.Why + if !same { + h.told = &now + } + h.mu.Unlock() + if !same && h.said != nil { + h.said(now) } } From 3203358c4cb5fa5bed223f7ce85577bbb70a346e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:18:41 +0100 Subject: [PATCH 4/4] buildrun: an asker keeps whether it hears DEAD_LETTERS, and its health tool answers from it (issue 734, review round 1) A deaf asker must be loud: SayHearing keeps each Hearing in the asker's state hearing, and HearingHealth answers the module's health tool from it, unhealthy with why while DEAD_LETTERS is not heard. --- go/delivery/buildrun/asks_test.go | 14 +++++++++++ go/delivery/buildrun/lostask.go | 42 +++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index f5ea8c3..0f8540a 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -2,6 +2,7 @@ package buildrun import ( "context" + "encoding/json" "errors" "strings" "sync" @@ -9,6 +10,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) func goAsk(id string) delivery.StepAsk { @@ -106,6 +108,18 @@ 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) } + for _, row := range []struct { + h *Hearing + want string + }{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} { + var raw []byte + if row.h != nil { + raw, _ = json.Marshal(row.h) + } + if a := HearingHealth(raw); a.State != row.want || a.Why == "" { + t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want) + } + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go index 1e70245..90a9ece 100644 --- a/go/delivery/buildrun/lostask.go +++ b/go/delivery/buildrun/lostask.go @@ -2,6 +2,7 @@ package buildrun import ( "context" + "encoding/json" "fmt" "strconv" "strings" @@ -11,6 +12,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) // **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask @@ -451,3 +453,43 @@ func (h *DeadLetterHearing) tell() { h.said(now) } } + +// HearingState is the name of the state an asker keeps its Hearing in (`_hearing`, key HearingKey), which its +// health tool reads. +const HearingState = "hearing" + +// SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and +// says so once. +func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) { + if kv == nil { + if logf != nil { + logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState) + } + return nil + } + return func(h Hearing) { + data, err := json.Marshal(h) + if err == nil { + _, err = kv.Put(ctx, HearingKey, data) + } + if err != nil && logf != nil { + logf("whether %s is heard could not be kept: %v", DeadLettersStream, err) + } + } +} + +// HearingHealth is the health answer of an asker's kept Hearing (raw, empty when none is kept): healthy while it hears +// DEAD_LETTERS, unhealthy saying why while it does not or has not said. +func HearingHealth(raw []byte) health.Answer { + if len(raw) == 0 { + return health.Unhealthy("it has not said whether it hears " + DeadLettersStream + " (issue 734): it is starting, or its state " + HearingState + " is not there") + } + var h Hearing + if err := json.Unmarshal(raw, &h); err != nil { + return health.Unhealthy("what it keeps of hearing " + DeadLettersStream + " is not a hearing: " + err.Error()) + } + if !h.Heard { + return health.Unhealthy(h.Why + " (since " + h.At.Format(time.RFC3339) + "; tried again at its next reconnect or ask)") + } + return health.Healthy("it hears " + DeadLettersStream + " for " + strings.Join(h.Streams, ", ")) +}