Merge pull request 'A module's process may hold a bus account of its own, and a module stream may be a work queue (ADR 0436, issues 1170, 1177-1181)' (#390) from feat/mail-sender-controller into main

internal/inventory/busrecords.go

@@ -187,6 +187,12 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
 		Logs: logsOf(m),
 		// And the streams it owns (novox/hq ADR 0363).
 		Streams: streamsOf(m),
+		// And its own work-queue streams, each with the one process its bus.queues names as the
+		// consumer (novox/hq ADR 0436 §1, §2, §6).
+		Queues: queueOwnersOf(m),
+		// And every process of its own that declares bus, each granted a bus account of its own (novox/hq
+		// ADR 0436 §1, §2).
+		Processes: ProcessesOf(m),
 		// And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them.
 		Checks: catalogue.HealthChecks(m),
 		// And whether it runs as an account of its own (novox/hq ADR 0259 §8).
@@ -422,7 +428,67 @@ func streamsOf(m catalogue.Manifest) []broker.ModuleStream {
 			events = append(events, broker.WirePattern(e))
 		}
 		out = append(out, broker.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events,
-			MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil})
+			MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil,
+			WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates})
+	}
+	return out
+}
+
+// queueOwnersOf is every work-queue stream of a module's own that one of its processes names in bus.queues,
+// with that process's id and the stream's declared max-deliveries (novox/hq ADR 0436 §1, §2, §6). In
+// resource order, so composition is repeatable; a queue naming a stream the module does not own, or one two
+// processes name, is caught at registration (catalogue.WorkQueueStreamProblems) and never reaches here.
+func queueOwnersOf(m catalogue.Manifest) []broker.QueueOwner {
+	buses := m.ProcessBuses()
+	if len(buses) == 0 {
+		return nil
+	}
+	byName := map[string]catalogue.StreamDeclaration{}
+	for _, s := range m.Streams {
+		byName[s.Name] = s
+	}
+	var out []broker.QueueOwner
+	for _, r := range m.Resources {
+		id := fmt.Sprint(r["id"])
+		b, declared := buses[id]
+		if !declared {
+			continue
+		}
+		for _, q := range b.Queues {
+			s, known := byName[q]
+			if !known || s.Retention != catalogue.StreamRetentionWorkQueue {
+				continue
+			}
+			out = append(out, broker.QueueOwner{Process: id, Stream: s.BusName(), MaxDeliver: s.MaxDeliveries})
+		}
+	}
+	return out
+}
+
+// ProcessesOf is every process of a module that declares bus, as the broker composes its own account from
+// (novox/hq ADR 0436 §1, §2): its own emits, its own work-queue consumers (queueOwnersOf, filtered to each
+// process's own), its own state, whether it reads its module's own asks, and the tools and seat verbs it
+// calls.
+func ProcessesOf(m catalogue.Manifest) []broker.ProcessPrincipal {
+	buses := m.ProcessBuses()
+	if len(buses) == 0 {
+		return nil
+	}
+	queues := queueOwnersOf(m)
+	var out []broker.ProcessPrincipal
+	for _, r := range m.Resources {
+		id := fmt.Sprint(r["id"])
+		b, declared := buses[id]
+		if !declared {
+			continue
+		}
+		pr := broker.ProcessPrincipal{Process: id, Emits: b.Emits, State: b.State, Asks: b.Asks, Invokes: b.Invokes}
+		for _, q := range queues {
+			if q.Process == id {
+				pr.Queues = append(pr.Queues, q)
+			}
+		}
+		out = append(out, pr)
 	}
 	return out
 }

internal/inventory/bususers.go

@@ -68,6 +68,9 @@ const (
 	// BusProvisioner is a machine's provisioner client (novox/hq ADR 0349 clause 11, design 57 P3): an account of
 	// its own, named for the machine.
 	BusProvisioner = "provisioner"
+	// BusProcess is one process of a module that declares bus (novox/hq ADR 0436 §1, §2): an account of its
+	// own, named for the machine, the module and the process.
+	BusProcess = "process"
 )
 
 // MintBusPassword makes a bus password and records its hash under a username, replacing whatever was

internal/inventory/streams_test.go

@@ -43,3 +43,47 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) {
 		t.Fatalf("the provider's stream: %+v", got)
 	}
 }
+
+// 1177/AC1: a catalogue stream declaring retention work-queue is carried into the broker's ModuleStream as
+// WorkQueue, with its duplicate window.
+func TestAWorkQueueStreamIsCarriedIntoTheBrokersModuleStream(t *testing.T) {
+	sender := catalogue.Manifest{Module: "mesh-mail", Streams: []catalogue.StreamDeclaration{{Name: "mail-sends",
+		Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64,
+		Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5, Duplicates: 24}}}
+	got := streamsOf(sender)
+	if len(got) != 1 || got[0].Name != "MAIL_SENDS" || !got[0].WorkQueue || got[0].Duplicates != 24 {
+		t.Fatalf("the sender's stream: %+v", got)
+	}
+}
+
+// 1177/AC3: a catalogue stream with no retention is carried into the broker as not a work queue.
+func TestANonWorkQueueStreamIsCarriedAsNotAWorkQueue(t *testing.T) {
+	plain := catalogue.Manifest{Module: "feed", Streams: []catalogue.StreamDeclaration{{Name: "items",
+		Events: []string{"item.**"}}}}
+	got := streamsOf(plain)
+	if len(got) != 1 || got[0].WorkQueue || got[0].Duplicates != 0 {
+		t.Fatalf("the plain stream: %+v", got)
+	}
+}
+
+// 1178/AC1: the process a work-queue stream's bus.queues names is read from the manifest, by its resource
+// id, with the stream's own max-deliveries — what the broker composes the one durable consumer from.
+func TestTheQueueOwnersOfAWorkQueueStreamAreReadFromItsProcess(t *testing.T) {
+	sender := catalogue.Manifest{Module: "mesh-mail", Emits: []string{"mail.send.sent", "mail.send.failed"},
+		Streams: []catalogue.StreamDeclaration{{Name: "mail-sends", Events: []string{"mail.send.approved"},
+			Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5}},
+		Resources: []map[string]any{
+			{"id": "account", "type": "user", "name": "mesh-mail-send"},
+			{"id": "send", "type": "process", "run": []any{"./mesh-mail", "send"}, "user": "mesh-mail-send",
+				"bus": map[string]any{"emits": []any{"mail.send.sent", "mail.send.failed"},
+					"queues": []any{"mail-sends"}}},
+		}}
+	got := queueOwnersOf(sender)
+	if len(got) != 1 || got[0].Process != "send" || got[0].Stream != "MAIL_SENDS" || got[0].MaxDeliver != 5 {
+		t.Fatalf("the sender's queue owners: %+v", got)
+	}
+	// And a module with no process declaring bus has none.
+	if got := queueOwnersOf(catalogue.Manifest{Module: "feed"}); got != nil {
+		t.Fatalf("a module with no process bus has queue owners: %+v", got)
+	}
+}