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

@@ -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(

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.<bucket>.<module>.>`), 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.<bucket>.<module>.>`), 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) {