Give a work-queue stream its one consumer, drop-only dead letters (issue 1178)

internal/broker/derived.go

@@ -190,18 +190,33 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 	var out []ModuleConsumer
 	seen := map[string]bool{}
 	add := func(p Principal) {
-		c, needed := ConsumerFor(p)
-		if !needed || seen[p.Node+"/"+p.Module] {
+		key := p.Node + "/" + p.Module
+		if seen[key] {
 			return
 		}
-		seen[p.Node+"/"+p.Module] = true
-		out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c})
-		// And one on each module stream it hears (novox/hq ADR 0363 §5).
-		for _, h := range p.Hears {
-			if len(h.Filters) == 0 {
-				continue
+		c, needed := ConsumerFor(p)
+		// **A module with a work-queue consumer of its own needs no other reason to be visited here**
+		// (novox/hq ADR 0436 §1, §2, §6): mesh-mail's sender consumes and watches nothing of its own, so
+		// ConsumerFor alone would have found nothing needed and skipped it before its one queue consumer
+		// was ever composed.
+		if !needed && len(p.Queues) == 0 {
+			return
+		}
+		seen[key] = true
+		if needed {
+			out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c})
+			// And one on each module stream it hears (novox/hq ADR 0363 §5).
+			for _, h := range p.Hears {
+				if len(h.Filters) == 0 {
+					continue
+				}
+				out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)})
 			}
-			out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)})
+		}
+		// And the one durable consumer of each of its own work-queue streams, for the one process its
+		// bus.queues names (novox/hq ADR 0436 §1, §2, §6).
+		for _, q := range p.Queues {
+			out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: queueConsumer(p, q)})
 		}
 	}
 	for _, p := range users {

internal/broker/modstream_work_queue_test.go

@@ -36,3 +36,21 @@ func TestANonWorkQueueStreamStillDropsItsOldest(t *testing.T) {
 		t.Fatalf("composed as %+v; want limits retention, never discarding new, no duplicate window", s)
 	}
 }
+
+// 1178/AC1: the process a work-queue stream's bus.queues names gets exactly one durable consumer,
+// <node>_<module>_<process id>, with the stream's declared max-deliver — composed even though mesh-mail's
+// sender otherwise consumes and watches nothing of its own, which alone would give ConsumerFor nothing
+// needed. Explicit acknowledgement is EnsureConsumer's own rule for every consumer it makes (jetstream.go),
+// so it is not re-asserted here.
+func TestTheOneProcessAWorkQueuesBusNamesGetsItsOwnConsumer(t *testing.T) {
+	sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x",
+		Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}}
+	got := ConsumersOf([]Principal{sender})
+	if len(got) != 1 {
+		t.Fatalf("mesh-mail's consumers: %+v", got)
+	}
+	c := got[0].Consumer
+	if c.Name != "novox_mesh-mail_send" || c.Stream != "MAIL_SENDS" || c.MaxDeliver != 5 {
+		t.Fatalf("the queue's consumer: %+v", c)
+	}
+}

internal/broker/modstreams.go

@@ -94,10 +94,44 @@ func (s ModuleStream) Stream() Stream {
 		out.Retention = RetentionWorkQueue
 		out.DiscardNew = true
 		out.DuplicatesSeconds = s.Duplicates * 60 * 60
+		// Said on the bus too, reusing the one flag a restricted stream's dead letters already age and hide
+		// by (novox/hq ADR 0436 §6, ADR 0376 §3-4): a work-queue stream's grant is always empty (ADR 0436
+		// §6 refuses one that carries a grant), so this never collides with a trusted-config restriction.
+		out.Restricted = true
 	}
 	return out
 }
 
+// QueueOwner is the one process of a module that is the durable consumer of one of its own work-queue
+// streams (novox/hq ADR 0436 §1, §2, §6): named in that process's bus.queues, and checked at registration
+// to be the only one (catalogue.WorkQueueStreamProblems).
+type QueueOwner struct {
+	// Process is the process resource's id — its bus account's name, with the node and module (novox/hq ADR
+	// 0436 §2).
+	Process string
+	// Stream is the work-queue stream's name on the bus.
+	Stream string
+	// MaxDeliver bounds how many times a message is handed over before it is given up as a dead letter
+	// (novox/hq ADR 0436 §6).
+	MaxDeliver int
+}
+
+// queueConsumer is the durable consumer the one process a module's bus.queues names is granted on its own
+// work-queue stream (novox/hq ADR 0436 §1, §2, §6): no filter, since it is the stream's only consumer —
+// unlike heardConsumer, which is one of possibly several hearing a module's ordinary stream.
+func queueConsumer(p Principal, q QueueOwner) Consumer {
+	return Consumer{
+		Name:           p.Node + "_" + p.Module + "_" + q.Process,
+		Stream:         q.Stream,
+		AckWaitSeconds: 30,
+		MaxDeliver:     q.MaxDeliver,
+		Why: fmt.Sprintf("%s's process %s is the one consumer %s's bus.queues names for its work-queue "+
+			"stream %s (novox/hq ADR 0436 §6); after max-deliver %d it gives a message up, and the "+
+			"controller keeps it in %s, offered only drop", p.Module, q.Process, p.Module, q.Stream,
+			q.MaxDeliver, DeadLettersStream),
+	}
+}
+
 // ModuleStreams is every declared module stream as the controller asserts it, in name order.
 func ModuleStreams(declared []ModuleStream) []Stream {
 	sorted := append([]ModuleStream(nil), declared...)

internal/broker/modstreams_workqueue_integration_test.go

@@ -0,0 +1,101 @@
+//go:build integration
+
+package broker
+
+import (
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// 1178/AC4, on a real bus: the controller ensuring mesh-mail's work-queue stream and its one consumer gives
+// the bus's own guarantees — a duplicate publish within the window is taken once, a message naked past
+// max-deliver is not handed over again, and a full stream refuses a new publish (ADR 0436 §6).
+//
+//	go test -tags integration ./internal/broker/ -run TestAWorkQueueStreamOnARealBus
+func TestAWorkQueueStreamOnARealBus(t *testing.T) {
+	js, err := Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+
+	// Small enough that a handful of messages fill it, so "the stream is full" is reachable within a test.
+	stream := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
+		MaxAgeDays: 1, MaxMiB: 1, WorkQueue: true, Duplicates: 1}
+	if err := js.EnsureStream(stream.Stream()); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = js.Context().DeleteStream(stream.Name) })
+
+	owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5}
+	sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"}
+	consumer := queueConsumer(sender, owner)
+	consumer.AckWaitSeconds = 1
+	if err := js.EnsureConsumer(consumer); err != nil {
+		t.Fatal(err)
+	}
+
+	subject := ModuleStreamSubject("mesh-mail", "mail.send.approved")
+
+	// A duplicate publish within the window is kept once (ADR 0436 §6's duplicate window).
+	publish := func(id string, size int) (*nats.PubAck, error) {
+		return js.Context().Publish(subject, make([]byte, size), nats.MsgId(id))
+	}
+	if _, err := publish("dup-1", 16); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := publish("dup-1", 16); err != nil {
+		t.Fatal(err)
+	}
+	if info, err := js.Context().StreamInfo(stream.Name); err != nil || info.State.Msgs != 1 {
+		t.Fatalf("a message published twice with the same id: %v msgs, err %v; want 1", info, err)
+	}
+
+	// Naked five times, the consumer is not handed the message a sixth time.
+	api, err := jetstream.New(js.Conn())
+	if err != nil {
+		t.Fatal(err)
+	}
+	reading, err := api.Consumer(t.Context(), stream.Name, consumer.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < owner.MaxDeliver; i++ {
+		batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
+		if err != nil {
+			t.Fatal(err)
+		}
+		got := false
+		for msg := range batch.Messages() {
+			got = true
+			_ = msg.Nak()
+		}
+		if !got {
+			t.Fatalf("delivery %d of %d never arrived", i+1, owner.MaxDeliver)
+		}
+	}
+	batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
+	if err != nil {
+		t.Fatal(err)
+	}
+	for range batch.Messages() {
+		t.Fatalf("the message was delivered a %dth time, past its max-deliver of %d", owner.MaxDeliver+1, owner.MaxDeliver)
+	}
+
+	// Full, the stream refuses a new publish rather than dropping what it holds (DiscardNew).
+	var fullErr error
+	for i := 0; i < 1<<16; i++ {
+		if _, err := publish("fill-"+time.Now().String(), 1<<16); err != nil {
+			fullErr = err
+			break
+		}
+	}
+	if fullErr == nil {
+		t.Fatal("a work-queue stream accepted every publish; want it to refuse one once full")
+	}
+}

internal/broker/nats.go

@@ -170,6 +170,9 @@ type Principal struct {
 	// subjects it was granted (novox/hq ADR 0363).
 	Streams []ModuleStream
 	Hears   []Heard
+	// Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436
+	// §1, §2, §6).
+	Queues []QueueOwner
 
 	// SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its
 	// whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing

internal/broker/users.go

@@ -46,6 +46,9 @@ type Declared struct {
 	Logs []Log
 	// Streams are the streams it owns, each its events' one place (novox/hq ADR 0363).
 	Streams []ModuleStream
+	// Queues are its own work-queue streams and the one process each is the consumer of (novox/hq ADR 0436
+	// §1, §2, §6): one durable consumer per entry, named `<node>_<module>_<process id>`.
+	Queues []QueueOwner
 	// Hears are the module streams it hears on this machine, each with the subjects it was granted
 	// (novox/hq ADR 0363 §5); HearsAStream says its `consumes` names a module that owns a stream, granted
 	// or not, so its membership is issued again when a grant changes.
@@ -155,7 +158,7 @@ func Users(r Records) ([]Principal, error) {
 				Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes,
 				State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus,
 				PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads,
-				Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d),
+				Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues,
 			})
 			// **The vault's holder asks the desks and the issuing providers itself** (issue 723), under the
 			// account its own runtime runs on (ADR 0259 §8's runs-as): the one module granted it.