diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 99035909..1e8efc6f 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -1629,11 +1629,6 @@ func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string) if err != nil { return err } - // And each work queue's cancelled set (novox/hq ADR 0219), so a holder taking an ask can ask - // whether it was cancelled the moment it took it. - if err := broker.RaiseCancelledSets(js, inventory.MeshSeats()); err != nil { - return err - } // And the controller's own buckets (novox/hq to-be 45 §1): the calls it serves and the acts done // by hand, kept where a restart of this process does not take them. if err := js.EnsureControllerBuckets(); err != nil { diff --git a/internal/broker/cancelled.go b/internal/broker/cancelled.go index afd00088..e69b2561 100644 --- a/internal/broker/cancelled.go +++ b/internal/broker/cancelled.go @@ -1,92 +1,11 @@ package broker -import ( - "context" - "fmt" - "strings" - "time" +import "strings" - "github.com/nats-io/nats.go/jetstream" -) +// A seat's cancelled set (novox/hq ADR 0219) is a bucket the mesh no longer raises or grants (R4g, ADR 0525 clause 6): the +// controller asks no work queue, so nothing is cancelled. A bus that still holds one is not reported as state nothing declares. -// A seat's cancelled set (novox/hq ADR 0219). -// -// **Cancelling an ask is deleting its message from the seat's work queue** — and that alone has a -// race no ordering on one side closes: a holder may fetch the ask in the moment between the -// controller reading the queue and deleting the message, and build what a person cancelled. So the -// controller first writes the ask's id into the seat's cancelled set, and a holder looks its ask up -// there after taking it and before building: listed, it terminates the ask and announces it failed -// rather than starting it. -// -// **A key-value bucket, because that is the shape the bus already has for "a small set the -// controller writes and many read cheaply"** (ADR 0201's state buckets): one direct read by key per -// ask taken — no consumer, no subscription a holder must keep, nothing that grows a holder's grants -// beyond one read subject. Kept beside the seat's own stream and worker and named like them, so the -// three objects of one work queue read as one family; asserted by the controller with the queue, -// and aged out with it — an id cancelled a week ago names an ask the queue no longer holds. - -// CancelledSetName is the bucket holding a seat's cancelled asks. -func CancelledSetName(seat string) string { return "SEAT_" + upperSnake(seat) + "_cancelled" } - -// hasCancelledSet is whether a seat's queue can be cancelled from: the work queues the controller -// asks, whose asks it alone shows and changes. A module's own seat's queue is that module's affair. -func hasCancelledSet(seat string) bool { - for _, s := range seatsTheControllerAsks { - if s == seat { - return true - } - } - return false -} - -// IsCancelledSet says a bucket is a seat's cancelled set rather than a module's state — the mesh's -// own, and never one to report as state nothing declares. +// IsCancelledSet says a bucket is a seat's cancelled set rather than a module's state. func IsCancelledSet(bucket string) bool { return strings.HasPrefix(bucket, "SEAT_") && strings.HasSuffix(bucket, "_cancelled") } - -// cancelledSetAge is how long a cancelled id is kept: as long as the seat's queue keeps an ask. -const cancelledSetAge = 7 * 24 * time.Hour - -// A CancelledSetAsserter is what raising the cancelled sets needs of a connection. -type CancelledSetAsserter interface { - EnsureCancelledSet(seat string) error -} - -// RaiseCancelledSets asserts the cancelled set of every work queue the controller asks. -func RaiseCancelledSets(a CancelledSetAsserter, seats []DeclaredSeat) error { - for _, s := range seats { - if len(s.Accepts) == 0 || !hasCancelledSet(s.Name) { - continue - } - if err := a.EnsureCancelledSet(s.Name); err != nil { - return fmt.Errorf("asserting the cancelled set of %s: %w", s.Name, err) - } - } - return nil -} - -// EnsureCancelledSet creates a seat's cancelled set if absent and brings its options to match. -// Direct reads on, which is how a holder looks an id up with one request. -func (j *JetStream) EnsureCancelledSet(seat string) error { - js, err := jetstream.New(j.conn) - if err != nil { - return err - } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: CancelledSetName(seat), - Description: fmt.Sprintf("the asks of the %s seat cancelled by hand (novox/hq ADR 0219): written "+ - "by the controller before it deletes an ask from the queue, read by a holder on taking one, "+ - "so an ask fetched in that moment is ended rather than built", seat), - History: 1, - TTL: cancelledSetAge, - MaxValueSize: 4 * 1024, - MaxBytes: 4 * 1024 * 1024, - Storage: jetstream.FileStorage, - }); err != nil { - return fmt.Errorf("asserting bucket %s: %w", CancelledSetName(seat), err) - } - return nil -} diff --git a/internal/broker/cancelled_test.go b/internal/broker/cancelled_test.go index 5bf4a0a6..629d1ced 100644 --- a/internal/broker/cancelled_test.go +++ b/internal/broker/cancelled_test.go @@ -2,60 +2,6 @@ package broker import "testing" -// A build agent reads its seat's cancelled set by key and nothing more, answers its verbs on its own -// machine's subjects, and says whether it is paused under its own machine's name; the controller may -// ask any machine's holder its verbs (novox/hq ADR 0219). -func TestTheBuildQueueIsControlledWithTheGrantsItNeedsAndNoMore(t *testing.T) { - seat := Seat{Name: "node-build-agent", Scope: "node", Accepts: []string{"build"}, - Emits: []string{"started", "built", "log.*", "paused.*"}, - Serves: []string{"current", "kill", "pause", "resume"}} - holder, err := PermissionsFor(Principal{Kind: KindModule, Node: "ace", Module: "build-agent", - Holds: []Seat{seat}, PasswordHash: "x"}) - if err != nil { - t.Fatal(err) - } - has(t, holder.Publish, "$JS.API.DIRECT.GET.KV_SEAT_NODE_BUILD_AGENT_cancelled.$KV.SEAT_NODE_BUILD_AGENT_cancelled.>") - hasNot(t, holder.Publish, "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>") - has(t, holder.Publish, "mesh.seat.node-build-agent.event.paused.ace") - hasNot(t, holder.Publish, "mesh.seat.node-build-agent.event.paused.*") - has(t, holder.Publish, "mesh.seat.node-build-agent.event.log.*") - has(t, holder.Subscribe, "mesh.seat.node-build-agent.tool.kill.ace") - hasNot(t, holder.Subscribe, "mesh.seat.node-build-agent.tool.kill.g14") - - // A module's own seat's queue is its own affair: no cancelled set, no grant for one. - other, _ := PermissionsFor(Principal{Kind: KindModule, Node: "ace", Module: "telegram", - Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}, PasswordHash: "x"}) - hasNot(t, other.Publish, "$JS.API.DIRECT.GET.KV_SEAT_TELEGRAM_SENDER_cancelled.$KV.SEAT_TELEGRAM_SENDER_cancelled.>") - - controller, _ := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) - has(t, controller.Publish, "mesh.seat.node-build-agent.tool.>") - - if CancelledSetName("node-build-agent") != "SEAT_NODE_BUILD_AGENT_cancelled" || - !IsCancelledSet("SEAT_NODE_BUILD_AGENT_cancelled") || IsCancelledSet("build-agent_cancelled") { - t.Error("the cancelled set is not named as its seat's family") - } -} - -// Only the work queues the controller asks get a cancelled set. -func TestOnlyTheControllersQueuesHaveACancelledSet(t *testing.T) { - var asserted []string - a := asserterFunc(func(seat string) error { asserted = append(asserted, seat); return nil }) - if err := RaiseCancelledSets(a, []DeclaredSeat{ - {Name: "node-build-agent", Accepts: []string{"build"}}, - {Name: "telegram-sender", Accepts: []string{"send"}}, - {Name: "mesh-controller"}, - }); err != nil { - t.Fatal(err) - } - if len(asserted) != 1 || asserted[0] != "node-build-agent" { - t.Fatalf("asserted %v", asserted) - } -} - -type asserterFunc func(string) error - -func (f asserterFunc) EnsureCancelledSet(seat string) error { return f(seat) } - // A seat's cancelled set is never reported as state nothing declares. func TestACancelledSetIsNotUndeclaredState(t *testing.T) { undeclared, err := RaiseBuckets(fakeBuckets{names: []string{"SEAT_NODE_BUILD_AGENT_cancelled", "gone_state"}}, nil) diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 2488ea44..799a46d7 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -20,11 +20,6 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) { want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>", "$KV." + AssignmentsBucket + ".>", "$KV." + DeclaredConfigBucket + ".>", "$KV." + PredefinedBucket + ".>"} - for _, seat := range seatsTheControllerAsks { - if hasCancelledSet(seat) { - want = append(want, "$KV."+CancelledSetName(seat)+".>") - } - } for _, subject := range want { if !slices.Contains(p.Publish, subject) { t.Errorf("the controller may not publish %s, so it cannot write that bucket", subject) diff --git a/internal/broker/genesis_template_test.go b/internal/broker/genesis_template_test.go index 0f25ea63..bcf1cf1f 100644 --- a/internal/broker/genesis_template_test.go +++ b/internal/broker/genesis_template_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "regexp" + "slices" "sort" "strings" "testing" @@ -44,6 +45,12 @@ func TestTheInstallersFirstUserListIsWhatTheControllerWouldCompose(t *testing.T) } // And nothing wider than what it derives, or genesis quietly grants a privilege the composition // takes away again on the first push. + // Except the retired build seats' grants, which the installer's list still names until genesis stops installing the + // builder (issue 1152): the controller derives none of them (R4g, ADR 0525 §10), the first push takes them away. + carriedPub = slices.DeleteFunc(slices.Clone(carriedPub), func(s string) bool { + return strings.HasPrefix(s, "mesh.seat.node-build-agent.") || strings.HasPrefix(s, "mesh.seat.mesh-build-machine.") || + strings.HasPrefix(s, "$KV.SEAT_NODE_BUILD_AGENT_") || strings.HasPrefix(s, "$KV.SEAT_MESH_BUILD_MACHINE_") + }) if diff := missing(carriedPub, want.Publish); len(diff) > 0 { t.Errorf("the installer's user list lets the controller publish %v, which it does not derive", diff) } diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 6cf5af43..8f96f164 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -222,29 +222,6 @@ type Principal struct { PasswordHash string } -// seatsTheControllerAsks are the roles the mesh's own flows submit work to. Named rather than -// derived from the seat set: the controller is not a module and declares no `uses`, so its side of a -// seat has to be stated, and a list is what makes "which roles does the mesh itself talk to" answerable. -// Both build roles while the handover runs (novox/hq ADR 0190): the controller asks whichever has a -// holder, and the retired one has one until build-agent replaces the builder. The second entry -// goes with the retired seat row. -var seatsTheControllerAsks = []string{"node-build-agent", "mesh-build-machine"} - -// TheControllersAsk says whether a message of stream, published on subject, is an ask the controller -// itself makes: one on the accept subject of a seat in seatsTheControllerAsks, kept in that seat's own -// work queue. Such an ask, given up on by the seat's worker, can be delivered again with the authority -// the controller already holds — the publish on that seat's accepts and the stream API to remove the -// 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 { - for _, seat := range seatsTheControllerAsks { - if stream == seatStreamName(seat) && strings.HasPrefix(subject, "mesh.seat."+seat+".accept.") { - return true - } - } - return false -} - // SeatVerb is one verb of one seat, on every machine holding it. type SeatVerb struct{ Seat, Verb string } @@ -646,19 +623,6 @@ func PermissionsFor(p Principal) (Permissions, error) { // like the catalogue does — which is why no holder needs to publish into anybody's inbox. // A node-scoped seat's work subject carries no node (novox/hq ADR 0190): the ask goes to // the role, and whichever machine holding it is idle takes it. - for _, seat := range seatsTheControllerAsks { - pub = append(pub, "mesh.seat."+seat+".accept.>") - // **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. - pub = append(pub, "mesh.seat."+seat+".tool.>") - // **And its cancelled set** (novox/hq ADR 0219, issue 269): a cancel writes the ask's id - // there before it deletes the ask, and a write is a publish to the bucket's subject, which - // `$JS.API.>` does not cover — so every cancel timed out, refused by this list. - if hasCancelledSet(seat) { - pub = append(pub, "$KV."+CancelledSetName(seat)+".>") - } - } // **And what the mesh says it did** (novox/hq ADR 0134). The control plane states its own // facts under the seat it holds, because a role's events belong to the role and keep their // address while the holder is replaced. Named one by one rather than as a whole namespace: @@ -995,15 +959,6 @@ func PermissionsFor(p Principal) (Permissions, error) { "$JS.API.CONSUMER.INFO."+stream+"."+worker, "$JS.API.CONSUMER.MSG.NEXT."+stream+"."+worker, "$JS.ACK."+stream+"."+worker+".>") - // **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: - // the set is the controller's, and only the work queues the controller asks — and so may - // cancel from — have one. - if hasCancelledSet(s.Name) { - set := CancelledSetName(s.Name) - pub = append(pub, "$JS.API.DIRECT.GET.KV_"+set+".$KV."+set+".>") - } for _, a := range s.Accepts { sub = append(sub, seatSubject(s, "accept", a)) } diff --git a/internal/broker/nats_test.go b/internal/broker/nats_test.go index b0243280..62ec6620 100644 --- a/internal/broker/nats_test.go +++ b/internal/broker/nats_test.go @@ -481,10 +481,16 @@ func TestANodeSeatsWorkSubjectCarriesNoNode(t *testing.T) { hasNot(t, perms.Subscribe, "mesh.seat.node-build-agent.accept.build.anchor") // And its tools still carry the machine. has(t, perms.Subscribe, "mesh.seat.node-build-agent.tool.status.anchor") - // The controller asks the role, not a machine. + // The controller asks no build seat any more (R4g, ADR 0525 §10): no publish to any subject of the retired roles. controller, err := PermissionsFor(Principal{Kind: KindController}) if err != nil { t.Fatal(err) } - has(t, controller.Publish, "mesh.seat.node-build-agent.accept.>") + for _, seat := range []string{"node-build-agent", "mesh-build-machine"} { + for _, p := range controller.Publish { + if strings.HasPrefix(p, "mesh.seat."+seat+".") || strings.Contains(p, "SEAT_"+upperSnake(seat)) { + t.Errorf("the controller may still publish %s, a subject of the retired seat %s", p, seat) + } + } + } } diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index a5ec0774..5d4ce16e 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] 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_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "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-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "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.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "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-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "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-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/link/deadletters.go b/internal/link/deadletters.go index 94de7d22..e450b8bc 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -356,11 +356,8 @@ func deadLetterWhole(js nats.JetStreamContext, id uint64) (DeadLetter, error) { // it: an event under that consumer's own again subject on EVENTS — and only when the consumer exists and // filters that subject, so a message is never let go as delivered while nobody receives it. // -// An ask the controller itself made (broker.TheControllersAsk) goes back on its own subject: a seat's -// work queue has one worker per subject, so only the worker that gave it up takes it, and DeliverAgain -// removes the original from the queue first, so there are never two (novox/hq issue 334); only while the -// worker that gave it up is on the bus. Any other stream's -// message is refused, with why — but a module's own stream's, which goes back as an event while its consumer is +// The controller makes no ask of a seat's work queue any more (R4g, ADR 0525 §10), so none goes back on its own subject. +// Any other stream's message is refused, with why — but a module's own stream's, which goes back as an event while its consumer is // still granted it (novox/hq ADR 0363 §7): an ask to a seat the controller does not ask would need a publish it is // not granted, in its asker's name (ADR 0264's consequences, ADR 0259 §3), and a stream that is neither // would reach every consumer of its subject. @@ -378,17 +375,6 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // age is left. return "", fmt.Errorf("dead letter %d is from %s, a work queue: it is only dropped, never delivered "+ "again (novox/hq ADR 0436 clause 6). Drop it", d.ID, d.Stream) - case broker.TheControllersAsk(d.Stream, d.Subject): - // The seat's worker takes it, and only while it is on the bus: without one the queue would keep - // the ask for a holder that may never come, and it would be let go as delivered meanwhile. - if _, err := js.ConsumerInfo(d.Stream, d.Consumer); errors.Is(err, nats.ErrConsumerNotFound) { - return "", fmt.Errorf("dead letter %d was given up by %s, which is no longer on the bus, so the ask "+ - "would only wait in %s for a holder. Nothing was done, and it is still kept: deliver it again once "+ - "the seat has a holder, or drop it", d.ID, d.Who, d.Stream) - } else if err != nil { - return "", fmt.Errorf("whether %s can receive dead letter %d cannot be read: %w", d.Who, d.ID, err) - } - return d.Subject, nil case d.Stream != broker.EventsStream && broker.OnAModuleStream(d.Subject): // **A module stream's event goes back to its consumer only while it is still granted** (novox/hq ADR // 0363 §7): its consumer on that stream must still be there and still filter the subject, or delivering diff --git a/internal/link/deadletters_ask_test.go b/internal/link/deadletters_ask_test.go deleted file mode 100644 index 9be18d47..00000000 --- a/internal/link/deadletters_ask_test.go +++ /dev/null @@ -1,252 +0,0 @@ -//go:build integration - -package link - -import ( - "errors" - "strings" - "testing" - - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - - "github.com/novox/mesh-controller/internal/broker" - "github.com/novox/mesh-controller/internal/testbus" -) - -// What a seat's worker gave up on, when the ask is one the controller itself makes (novox/hq issue 334). -// -// The controller already publishes on the accept subjects of the seats it asks (broker's -// seatsTheControllerAsks) and reaches the stream API, so delivering its own ask again needs no grant it -// does not hold: the original is removed from the work queue by its sequence, and the kept copy is -// published on the ask's own subject, where the seat's one worker takes it. An ask to any other seat is -// still refused, and stays kept: delivering it would need a publish the controller is not granted -// (ADR 0264's consequences), in the asker's name (ADR 0259 §3). - -// TheBuildMachine is the retired build seat the controller's own asks went to; BuildWorkOf is where one lands. -const TheBuildMachine = "node-build-agent" - -func BuildWorkOf(seat string) string { return "mesh.seat." + seat + ".accept.build" } - -// aBusWithTheBuildRole raises a bus of its own with the retired build seat's queue and worker: the seat the controller's -// own asks went to, whose grants the installer's carried user list still names (R4g keeps them until it drops them). -func aBusWithTheBuildRole(t *testing.T) *broker.JetStream { - t.Helper() - js, err := broker.Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(js.Close) - seats := []broker.DeclaredSeat{{Name: TheBuildMachine, Accepts: []string{"build"}, Emits: []string{"built"}}} - if err := broker.AssertMeshStreams(js); err != nil { - t.Fatal(err) - } - if err := broker.RaiseSeats(js, seats, map[string]broker.Holder{ - TheBuildMachine: {Node: "anchor", Module: "builder"}, - }); err != nil { - t.Fatal(err) - } - t.Cleanup(func() { - _ = js.Context().DeleteStream("SEAT_NODE_BUILD_AGENT") - for _, s := range broker.MeshStreams() { - _ = js.Context().PurgeStream(s.Name) - } - }) - return js -} - -// theBuildWorker is the build seat's worker as a holder pulls from it. -func theBuildWorker(t *testing.T, js *broker.JetStream) jetstream.Consumer { - t.Helper() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - stream := "SEAT_NODE_BUILD_AGENT" - worker, err := api.Consumer(t.Context(), stream, stream+"_worker") - if err != nil { - t.Fatal(err) - } - return worker -} - -func TestAnAskTheControllerMadeIsDeliveredAgainToTheSeatsWorker(t *testing.T) { - js := aBusWithTheBuildRole(t) - keeping(t, js) - worker := theBuildWorker(t, js) - - subject := BuildWorkOf(TheBuildMachine) - ask := &nats.Msg{Subject: subject, Data: []byte(`{"module":"x"}`), Header: nats.Header{}} - ask.Header.Set(nats.MsgIdHdr, "build-1") - if _, err := js.Context().PublishMsg(ask); err != nil { - t.Fatal(err) - } - d := givenUp(t, js, worker) - if d.Stream != "SEAT_NODE_BUILD_AGENT" || d.Subject != subject || d.Lost != "" { - t.Fatalf("kept as %+v", d) - } - if _, err := js.Context().GetMsg(d.Stream, d.Sequence); err != nil { - t.Fatalf("the original is not in the work queue before it is delivered again: %v", err) - } - - _, to, err := DeliverAgain(js.Context(), d.ID) - if err != nil { - t.Fatal(err) - } - if to != subject { - t.Fatalf("delivered again on %s, not the ask's own subject %s", to, subject) - } - if _, err := js.Context().GetMsg(d.Stream, d.Sequence); !errors.Is(err, nats.ErrMsgNotFound) { - t.Fatalf("the original is still in the work queue beside its copy: %v", err) - } - again := next(t, worker) - if again == nil { - t.Fatal("the seat's worker was not handed the ask again") - } - if string(again.Data()) != `{"module":"x"}` || again.Headers().Get(AgainHeader) == "" { - t.Fatalf("handed again as %s %v", again.Data(), again.Headers()) - } - _ = again.Ack() - if m := next(t, worker); m != nil { - t.Fatalf("the worker was handed it twice: %s", m.Subject()) - } - if _, err := DeadLetterNamed(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) { - t.Fatalf("still kept after it was delivered again: %v", err) - } - if _, _, err := DeliverAgain(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) { - t.Fatalf("a second delivery answered %v", err) - } -} - -// An ask to a seat the controller does not ask is refused, says why, and nothing is done. -func TestAnAskTheControllerDidNotMakeIsStillOnlyDropped(t *testing.T) { - js := aBus(t) - for _, d := range []DeadLetter{ - {ID: 2, Stream: "SEAT_TELEGRAM_SENDER", Consumer: "SEAT_TELEGRAM_SENDER_worker", - Subject: "mesh.seat.telegram-sender.accept.send"}, - // The subject of a seat the controller asks, on a stream that is not that seat's queue. - {ID: 3, Stream: "SEAT_TELEGRAM_SENDER", Consumer: "SEAT_TELEGRAM_SENDER_worker", - Subject: "mesh.seat.node-build-agent.accept.build"}, - } { - if to, err := AgainTo(js.Context(), d); err == nil { - t.Errorf("%s on %s was given %s to be delivered again on", d.Subject, d.Stream, to) - } else if !strings.Contains(err.Error(), "Drop it") { - t.Errorf("refused without saying what to do: %v", err) - } - } -} - -// aBuildAskGivenUp publishes one build ask and lets the worker give it up. -func aBuildAskGivenUp(t *testing.T, js *broker.JetStream, worker jetstream.Consumer, body string) DeadLetter { - t.Helper() - if _, err := js.Context().Publish(BuildWorkOf(TheBuildMachine), []byte(body)); err != nil { - t.Fatal(err) - } - return givenUp(t, js, worker) -} - -// A copy that cannot be published: the original is already out of the queue, the dead letter is kept and -// the answer says both; asked again once it can be published, it is delivered once. -func TestAnAskWhoseCopyIsRefusedStaysKeptAndIsDeliveredOnTheNextTry(t *testing.T) { - js := aBusWithTheBuildRole(t) - keeping(t, js) - worker := theBuildWorker(t, js) - d := aBuildAskGivenUp(t, js, worker, `{"module":"x"}`) - - // The queue stops taking the ask's subject, so the copy's publish is refused by the server. - info, err := js.Context().StreamInfo(d.Stream) - if err != nil { - t.Fatal(err) - } - cfg := info.Config - taking := cfg.Subjects - cfg.Subjects = []string{"mesh.seat." + TheBuildMachine + ".accept.nothing"} - if _, err := js.Context().UpdateStream(&cfg); err != nil { - t.Fatal(err) - } - _, _, err = DeliverAgain(js.Context(), d.ID) - if err == nil || !strings.Contains(err.Error(), "still kept") || !strings.Contains(err.Error(), "was removed") { - t.Fatalf("a refused copy answered %v", err) - } - if _, err := js.Context().GetMsg(d.Stream, d.Sequence); !errors.Is(err, nats.ErrMsgNotFound) { - t.Fatalf("the original is still in the queue: %v", err) - } - if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil { - t.Fatalf("a refused copy let the dead letter go: %v", err) - } - - cfg.Subjects = taking - if _, err := js.Context().UpdateStream(&cfg); err != nil { - t.Fatal(err) - } - retried, _, err := DeliverAgain(js.Context(), d.ID) - if err != nil { - t.Fatal(err) - } - if !strings.Contains(retried.Original, "no longer held") { - t.Fatalf("the retry said of the original: %q", retried.Original) - } - if again := next(t, worker); again == nil || string(again.Data()) != `{"module":"x"}` { - t.Fatal("the retry did not hand the ask to the worker") - } else { - _ = again.Ack() - } - if m := next(t, worker); m != nil { - t.Fatalf("handed twice: %s", m.Data()) - } -} - -// A queue made again numbers from one: the old dead letter's sequence then names a live ask, which is left -// alone. -func TestAnAskWhoseSequenceNamesAnotherMessageLeavesThatOneAlone(t *testing.T) { - js := aBusWithTheBuildRole(t) - keeping(t, js) - d := aBuildAskGivenUp(t, js, theBuildWorker(t, js), `{"module":"old"}`) - - // The build handover's way: the seat's queue deleted and made again, with its worker. - if err := js.Context().DeleteStream(d.Stream); err != nil { - t.Fatal(err) - } - if err := broker.RaiseSeats(js, []broker.DeclaredSeat{{Name: TheBuildMachine, Accepts: []string{"build"}, - Emits: []string{"built"}}}, map[string]broker.Holder{TheBuildMachine: {Node: "anchor", Module: "builder"}}); err != nil { - t.Fatal(err) - } - live, err := js.Context().Publish(BuildWorkOf(TheBuildMachine), []byte(`{"module":"live"}`)) - if err != nil { - t.Fatal(err) - } - if live.Sequence != d.Sequence { - t.Fatalf("the live ask is message %d, the dead letter names %d: the test does not set up the collision", - live.Sequence, d.Sequence) - } - - delivered, _, err := DeliverAgain(js.Context(), d.ID) - if err != nil { - t.Fatal(err) - } - if !strings.Contains(delivered.Original, "another message") { - t.Fatalf("the answer said of the original: %q", delivered.Original) - } - if held, err := js.Context().GetMsg(d.Stream, d.Sequence); err != nil || string(held.Data) != `{"module":"live"}` { - t.Fatalf("the live ask at that sequence was touched: %v", err) - } -} - -// No worker on the seat's queue: refused, kept, and nothing published. -func TestAnAskIsNotDeliveredAgainWhileTheSeatHasNoWorker(t *testing.T) { - js := aBusWithTheBuildRole(t) - keeping(t, js) - d := aBuildAskGivenUp(t, js, theBuildWorker(t, js), `{"module":"x"}`) - if err := js.Context().DeleteConsumer(d.Stream, d.Consumer); err != nil { - t.Fatal(err) - } - if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "wait in") { - t.Fatalf("delivered with no worker: %v", err) - } - if _, err := js.Context().GetMsg(d.Stream, d.Sequence); err != nil { - t.Fatalf("a refusal removed the original: %v", err) - } - if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil { - t.Fatalf("a refusal let the dead letter go: %v", err) - } -}