From 65d5ed33671d35cb482a7a62296717a7c09afa27 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 13:51:35 +0100 Subject: [PATCH 1/2] broker: a user of a seat pulled by kind reads its seat's dead letters, as mesh-sdk buscompose composes it (issue 734) The checker hears from DEAD_LETTERS which of its asks a build seat gave up on, and ends the try or asks the check again; the controller composes today's user list, so its copy of the composer carries the grant too. NATS-native research piece: none moved; the composition itself moves to mesh-grants (ADR 0547), and this keeps the copy in step with mesh-sdk. --- internal/broker/seattraffic.go | 23 +++++++++++++++++++++-- internal/broker/seattraffic_test.go | 12 ++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/internal/broker/seattraffic.go b/internal/broker/seattraffic.go index eea8e19c..5b68ca20 100644 --- a/internal/broker/seattraffic.go +++ b/internal/broker/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) { diff --git a/internal/broker/seattraffic_test.go b/internal/broker/seattraffic_test.go index cfe8218b..8f98305f 100644 --- a/internal/broker/seattraffic_test.go +++ b/internal/broker/seattraffic_test.go @@ -403,6 +403,18 @@ func TestTheBuilderMayAskABuildSeatAndASessionMayNot(t *testing.T) { t.Errorf("the builder may not publish %s", s) } } + // Issue 734: the asker hears which of its asks the seat's workers gave up on, from DEAD_LETTERS, and no other stream's. + 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.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.STREAM.PURGE.DEAD_LETTERS": false, + } { + if got := allowed(builder.Publish, subject); got != want { + t.Errorf("the builder may publish %s: %v, want %v", subject, got, want) + } + } session := perms(t, Principal{Kind: KindPerson, Module: "jochen", Invokes: []string{"mesh-controller.status"}}) if allowed(session.Publish, "mesh.seat.node-build-go.accept.delivery") { t.Error("a session's account may put an ask") From ccc7ba334c82554d809b5b38d8eb0eb5f36be7be Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 14:17:00 +0100 Subject: [PATCH 2/2] broker: DeadLetterReads grants only what a pulled ordered consumer needs, as mesh-sdk buscompose (issue 734, review round 1) NATS-native research piece: none moved; the copy stays in step with mesh-sdk until mesh-grants composes the list. --- internal/broker/seattraffic.go | 5 ++--- internal/broker/seattraffic_test.go | 1 + 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/broker/seattraffic.go b/internal/broker/seattraffic.go index 5b68ca20..ad2bffe4 100644 --- a/internal/broker/seattraffic.go +++ b/internal/broker/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/internal/broker/seattraffic_test.go b/internal/broker/seattraffic_test.go index 8f98305f..01b8a1d9 100644 --- a/internal/broker/seattraffic_test.go +++ b/internal/broker/seattraffic_test.go @@ -410,6 +410,7 @@ func TestTheBuilderMayAskABuildSeatAndASessionMayNot(t *testing.T) { "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true, "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false, "$JS.API.STREAM.PURGE.DEAD_LETTERS": false, + "$JS.FC.DEAD_LETTERS.x1": false, } { if got := allowed(builder.Publish, subject); got != want { t.Errorf("the builder may publish %s: %v, want %v", subject, got, want)