From e7f0dc612e6e62dca9f558d0cde80ed69f86ca3a Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:51:35 +0100 Subject: [PATCH] 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) {