Merge pull request 'Seat users pulled by kind may read their own seat's dead letters (issue 734)' (#472) from fix/734-lost-ask into main

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
+// 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.
+// (`$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,20 @@ 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, 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 {
+	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 + ".*",
+	}
+}
+
 // 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) {

internal/broker/seattraffic_test.go

@@ -403,6 +403,19 @@ 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,
+		"$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)
+		}
+	}
 	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")