issue 974: two work-queue streams for the checks of pull requests (CHECKS_READY, CHECKS_OTHER), their workers, grants and writer

cmd/mesh-controller/signals.go

@@ -671,6 +671,8 @@ func consumerWho(stream, consumer string) string {
 	switch {
 	case consumer == broker.ControllerName || consumer == broker.MergesConsumerName:
 		return "the controller"
+	case broker.IsCheckStream(stream):
+		return "the holder of " + broker.TheBuildSeat
 	case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(consumer, "_worker"):
 		seat := strings.ToLower(strings.ReplaceAll(strings.TrimSuffix(strings.TrimPrefix(consumer, "SEAT_"), "_worker"), "_", "-"))
 		return "the holder of " + seat

internal/broker/checks.go

@@ -0,0 +1,112 @@
+package broker
+
+import "strings"
+
+// The pull requests' checks have work queues of their own (novox/hq ADR 0406 §3, issue 974).
+//
+// A delivery build goes on the build seat's own queue, `mesh.seat.node-build-agent.accept.build`. A check of a pull
+// request marked ready goes on a second work-queue stream and every other check on a third. A builder takes from the
+// first, then the second, then the third; the others are taken in the order asked. **Each check stream has one subject
+// per pull request and keeps only the newest request** (a limit of one message a subject, the older discarded), so a
+// newer request replaces one still waiting and a pull request has at most one waiting check: the stream's rule, not a
+// pass of the controller that moves asks about. Nothing pauses the worker or deletes and republishes a request.
+
+const (
+	// ChecksReadyStream holds the checks of pull requests marked ready; ChecksOtherStream every other check.
+	ChecksReadyStream = "CHECKS_READY"
+	ChecksOtherStream = "CHECKS_OTHER"
+	// ChecksReadyWorker and ChecksOtherWorker are the durable consumers the build seat's holders share on them, as
+	// they share the seat's worker on its own queue (novox/hq ADR 0190).
+	ChecksReadyWorker = "CHECKS_READY_worker"
+	ChecksOtherWorker = "CHECKS_OTHER_worker"
+	// TheBuildSeat is the role whose checks these streams queue. The retired build role's checks stay on its own queue.
+	TheBuildSeat = "node-build-agent"
+)
+
+// PullsBucket is mesh-delivery's key per pull request: its head, the trunk commit its current check is asked against and
+// its verdict (ADR 0406 §4). A running check watches its key.
+var PullsBucket = BucketName("mesh-delivery", "pulls")
+
+// CheckSubjectPrefix is where a seat's checks of one kind are asked: `mesh.seat.<seat>.check.ready.` or `.other.`. The
+// `check` token keeps the subjects out of the seat's own queue, which takes `accept.>`.
+func CheckSubjectPrefix(seat string, ready bool) string {
+	kind := "other"
+	if ready {
+		kind = "ready"
+	}
+	return "mesh.seat." + seat + ".check." + kind + "."
+}
+
+// CheckStreams are the two check queues, in the order a builder takes from them.
+func CheckStreams() []Stream {
+	return []Stream{
+		{
+			Name:              ChecksReadyStream,
+			Subjects:          []string{CheckSubjectPrefix(TheBuildSeat, true) + ">"},
+			Retention:         RetentionWorkQueue,
+			MaxAge:            7 * 24 * 60 * 60,
+			MaxMsgsPerSubject: 1,
+			Why: "the checks of pull requests marked ready, taken after a delivery build and before any other check " +
+				"(novox/hq ADR 0406 §3); one subject a pull request, the newest request kept, so a newer one replaces a " +
+				"request still waiting",
+		},
+		{
+			Name:              ChecksOtherStream,
+			Subjects:          []string{CheckSubjectPrefix(TheBuildSeat, false) + ">"},
+			Retention:         RetentionWorkQueue,
+			MaxAge:            7 * 24 * 60 * 60,
+			MaxMsgsPerSubject: 1,
+			Why: "every other check, taken last and in the order asked (novox/hq ADR 0406 §3); one subject a pull " +
+				"request, the newest request kept; one left behind for a head already judged is dropped when taken",
+		},
+	}
+}
+
+// CheckWorkers are the consumers the build seat's holders pull the check queues through: as the seat's own worker is,
+// one for every holder, one ask at a time, acknowledged when the check is done.
+func CheckWorkers() []Consumer {
+	worker := func(name, stream string, ready bool) Consumer {
+		return Consumer{
+			Name:           name,
+			Stream:         stream,
+			Filters:        []string{CheckSubjectPrefix(TheBuildSeat, ready) + ">"},
+			AckWaitSeconds: 60,
+			MaxDeliver:     5,
+			Why: "every holder of " + TheBuildSeat + " pulls one check at a time from this worker and acknowledges after " +
+				"the check is done, so a crash mid-check redelivers rather than loses (novox/hq ADR 0406 §3)",
+		}
+	}
+	return []Consumer{
+		worker(ChecksReadyWorker, ChecksReadyStream, true),
+		worker(ChecksOtherWorker, ChecksOtherStream, false),
+	}
+}
+
+// holderCheckGrants are what a holder of the build seat is granted to take checks: ask about the two workers, pull from
+// them and acknowledge, and nothing of the subjects themselves.
+func holderCheckGrants() []string {
+	var pub []string
+	for _, w := range CheckWorkers() {
+		pub = append(pub,
+			"$JS.API.CONSUMER.INFO."+w.Stream+"."+w.Name,
+			"$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Name,
+			"$JS.ACK."+w.Stream+"."+w.Name+".>")
+	}
+	return pub
+}
+
+// IsCheckStream says a stream is one of the check queues.
+func IsCheckStream(stream string) bool {
+	return stream == ChecksReadyStream || stream == ChecksOtherStream
+}
+
+// isCheckAsk says a message of a check queue is an ask the controller itself makes.
+func isCheckAsk(stream, subject string) bool {
+	switch stream {
+	case ChecksReadyStream:
+		return strings.HasPrefix(subject, CheckSubjectPrefix(TheBuildSeat, true))
+	case ChecksOtherStream:
+		return strings.HasPrefix(subject, CheckSubjectPrefix(TheBuildSeat, false))
+	}
+	return false
+}

internal/broker/nats.go

@@ -217,6 +217,9 @@ var seatsTheControllerAsks = []string{"node-build-agent", "mesh-build-machine"}
 // original — and no other can (novox/hq issue 334, ADR 0264's consequences: no grant over the seats'
 // queues).
 func TheControllersAsk(stream, subject string) bool {
+	if isCheckAsk(stream, subject) {
+		return true
+	}
 	for _, seat := range seatsTheControllerAsks {
 		if stream == seatStreamName(seat) && strings.HasPrefix(subject, "mesh.seat."+seat+".accept.") {
 			return true
@@ -378,6 +381,12 @@ var DegradableGrants = []DegradableGrant{
 		Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " +
 			"window setting says (ADR 0407 decision 1, issue 973)",
 	},
+	{
+		Subject: "mesh.seat." + TheBuildSeat + ".check.>",
+		Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " +
+			"first push, and a check asked before that is asked again by the next announcement of its head (ADR 0406 §3, " +
+			"issue 974)",
+	},
 	{
 		Subject: "mesh.seat." + ControllerSeat + ".event.sent",
 		Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " +
@@ -575,6 +584,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// the role, and whichever machine holding it is idle takes it.
 		for _, seat := range seatsTheControllerAsks {
 			pub = append(pub, "mesh.seat."+seat+".accept.>")
+			// **And its checks, on the two queues of their own** (novox/hq ADR 0406 §3, issue 974): one subject a pull
+			// request on each, under `check`; the retired role's checks stay on its accepts.
+			if seat == TheBuildSeat {
+				pub = append(pub, "mesh.seat."+seat+".check.>")
+			}
 			// **And its holders' verbs, on every machine** (novox/hq ADR 0219): what a holder is
 			// building, kill it, pause it, resume it. The queue is the controller's to show and to
 			// change, and what one machine is doing with an ask it took only that machine can say.
@@ -883,6 +897,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
 				"$JS.API.CONSUMER.INFO."+stream+"."+worker,
 				"$JS.API.CONSUMER.MSG.NEXT."+stream+"."+worker,
 				"$JS.ACK."+stream+"."+worker+".>")
+			// And the checks' two workers, for the build role they are queued for (novox/hq ADR 0406 §3).
+			if s.Name == TheBuildSeat {
+				pub = append(pub, holderCheckGrants()...)
+			}
 			// **And whether an ask it took was cancelled** (novox/hq ADR 0219): one key of the
 			// seat's cancelled set, read directly by its id, so a holder that fetched an ask in the
 			// moment the controller cancelled it ends it instead of building it. Read, never written:

internal/broker/streams.go

@@ -147,7 +147,7 @@ func OriginalOfAgain(subject string) (string, bool) {
 // every tool invocation in the mesh, and a tool call must never be persisted (design 25 §3 keeps
 // tools on core NATS, where a lost call is a timeout the caller already handles).
 func MeshStreams() []Stream {
-	return []Stream{
+	return append([]Stream{
 		{
 			Name: "CONTROL",
 			// A build's outcome is no longer here: it is the build-machine seat's own event, so one
@@ -221,7 +221,7 @@ func MeshStreams() []Stream {
 				"failed with the reason (novox/hq ADR 0349 §4): what a walk's end and a gate wait for (ADR 0362, " +
 				"ADR 0349 §15), read by subject; a week, as the events are",
 		},
-	}
+	}, CheckStreams()...)
 }
 
 // ProvAppliedStream keeps what each machine's provisioner client last said of each module's files, newest per
@@ -465,7 +465,7 @@ const EventsStream = "EVENTS"
 // held through a store restart — the exact message the stream exists to protect — some minutes
 // before the controller had finished deciding about it.
 func MeshConsumers() []Consumer {
-	return []Consumer{
+	return append([]Consumer{
 		{
 			Name:           ControllerName,
 			Stream:         "CONTROL",
@@ -525,7 +525,7 @@ func MeshConsumers() []Consumer {
 			Why: "the controller copies each message a consumer gave up on into DEAD_LETTERS; no max-deliver, " +
 				"because a notice it gave up on would lose the message it is about",
 		},
-	}
+	}, CheckWorkers()...)
 }
 
 // NoticesConsumer is the controller's consumer on DEAD_LETTER_NOTICES (novox/hq issue 330).

internal/broker/streams_test.go

@@ -134,6 +134,9 @@ func TestEachStreamCarriesTheRetentionItsShapeNeeds(t *testing.T) {
 		// What each machine's provisioner client last said of each module's files (design 57 P6): the newest
 		// word per module, what a walk's end reads.
 		"PROV_APPLIED": RetentionLastPerSubject,
+		// The checks of pull requests, in two work queues of their own (novox/hq ADR 0406 §3, issue 974).
+		"CHECKS_READY": RetentionWorkQueue,
+		"CHECKS_OTHER": RetentionWorkQueue,
 	}
 	got := map[string]Retention{}
 	for _, s := range MeshStreams() {

internal/broker/testdata/composed.conf

@@ -24,7 +24,7 @@ accounts {
     jetstream: enabled
     users = [
       { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
-        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
+        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
         subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }

internal/broker/writers.go

@@ -148,6 +148,11 @@ var WritersTable = []WriterRow{
 		Subjects: []string{"mesh.seat.node-build-agent.event.built", "mesh.seat.mesh-build-machine.event.built"},
 		Writes:   holdsSeatOf,
 		Shared:   "every machine holding the build seat answers the asks it took; each outcome names its ask"},
+	{State: "a pull request's check asked", Writer: "controller",
+		KeptIn:   "the work queues " + ChecksReadyStream + " and " + ChecksOtherStream + ", one subject a pull request, the newest kept",
+		Others:   "the build seat's holders take it",
+		Subjects: []string{CheckSubjectPrefix(TheBuildSeat, true) + ">", CheckSubjectPrefix(TheBuildSeat, false) + ">"},
+		Writes:   isController},
 	{State: "a merge announced", Writer: "one announcer per forge (the hook, or the poll when the hook is absent — never both)",
 		KeptIn: "the bus", Others: "—", Subjects: []string{"mesh.mod.*.event.pull.merged"}, Writes: ownModule},
 	// A pull request's head, before it merges (novox/hq to-be 45 §9): the same one announcer.

internal/broker/writers_test.go

@@ -27,6 +27,7 @@ var designRows = []string{
 	"the healers' acts and their brake",
 	"stream definitions and bus permissions",
 	"builds and their outcomes",
+	"a pull request's check asked",
 	"a merge announced",
 	"a pull request's head announced",
 	"a pull request's merge check",

internal/catalogue/streams.go

@@ -36,7 +36,7 @@ var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"}
 
 // MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
 // package cannot import; a test keeps them agreeing.
-var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED"}
+var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
 
 // StreamDeclaration is one stream a module owns.
 type StreamDeclaration struct {

internal/link/advisories.go

@@ -143,7 +143,7 @@ func MeshNamed(stream, name string) bool {
 		return false // the mesh defines no durable consumer on a bucket's stream, or the memberships'
 	case stream == "NODES":
 		return true
-	case strings.HasPrefix(stream, "SEAT_"):
+	case strings.HasPrefix(stream, "SEAT_") || broker.IsCheckStream(stream):
 		return strings.HasSuffix(name, "_worker")
 	case name == broker.ControllerName || name == broker.MergesConsumerName:
 		return true
@@ -160,6 +160,8 @@ func ConsumerInWords(stream, name string) string {
 		return "the controller's consumer on " + stream
 	case stream == "NODES":
 		return "how " + name + " hears what it should be (its declaration consumer)"
+	case broker.IsCheckStream(stream):
+		return "the worker every holder of " + broker.TheBuildSeat + " takes its checks from (" + stream + ")"
 	case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(name, "_worker"):
 		seat := strings.ToLower(strings.ReplaceAll(strings.TrimPrefix(stream, "SEAT_"), "_", "-"))
 		return "the worker every holder of " + seat + " takes its asks from"