Take the controller's publish grants on node-build-agent and mesh-build-machine away, with the cancelled sets and the redelivery of its own asks: it asks no build seat (ADR 0525 §10, C3a)
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -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" }
|
||||
} }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user