diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index b9e2a576..a0a5b560 100644 --- a/cmd/mesh-controller/signals.go +++ b/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 diff --git a/internal/broker/checks.go b/internal/broker/checks.go new file mode 100644 index 00000000..e9efd275 --- /dev/null +++ b/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..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 +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index bb3f82c1..7507be33 100644 --- a/internal/broker/nats.go +++ b/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: diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 70b76520..f8377211 100644 --- a/internal/broker/streams.go +++ b/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). diff --git a/internal/broker/streams_test.go b/internal/broker/streams_test.go index 7a52d422..68831f41 100644 --- a/internal/broker/streams_test.go +++ b/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() { diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index d76e8659..dadf2781 100644 --- a/internal/broker/testdata/composed.conf +++ b/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" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 1b03593f..437612e1 100644 --- a/internal/broker/writers.go +++ b/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. diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index 0777967d..6f4c585f 100644 --- a/internal/broker/writers_test.go +++ b/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", diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index a3e3da68..012178b6 100644 --- a/internal/catalogue/streams.go +++ b/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 { diff --git a/internal/link/advisories.go b/internal/link/advisories.go index ced95ffa..5d0ec818 100644 --- a/internal/link/advisories.go +++ b/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"