From d36e0ea27809463dbf8868b3597d764d39619e59 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 18:34:44 +0200 Subject: [PATCH] =?UTF-8?q?Announce=20the=20controller's=20states,=20keep?= =?UTF-8?q?=20every=20module's=20word,=20and=20grant=20a=20read=20only=20w?= =?UTF-8?q?here=20its=20owner=20announced=20the=20state=20readable;=20the?= =?UTF-8?q?=20copied=20list=20of=20the=20controller's=20mirrors=20goes=20(?= =?UTF-8?q?issue=201790,=20D1-D4,=20ADR=200495,=20ADR=200513=20=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-controller/build.go | 14 +- cmd/mesh-controller/main.go | 2 + cmd/mesh-controller/push.go | 8 + cmd/mesh-controller/readable.go | 8 +- cmd/mesh-controller/states_announced.go | 225 ++++++++++++++++++ .../states_announced_integration_test.go | 118 +++++++++ cmd/mesh-controller/states_announced_test.go | 56 +++++ cmd/mesh-controller/status.go | 12 + internal/broker/announced.go | 81 +++++++ internal/broker/announced_test.go | 85 +++++++ .../broker/announced_test_helpers_test.go | 13 + internal/broker/checks_running_mirror_test.go | 4 +- internal/broker/controller_buckets.go | 22 +- internal/broker/controller_buckets_test.go | 6 +- internal/broker/delivery_parts_grant_test.go | 4 +- internal/broker/nats.go | 13 + internal/broker/plans_building_mirror_test.go | 4 +- internal/broker/process_grants_test.go | 2 + internal/broker/state.go | 6 + internal/broker/testdata/composed.conf | 2 +- internal/broker/users.go | 7 + internal/broker/writers.go | 24 ++ internal/broker/writers_test.go | 3 + internal/inventory/busrecords.go | 28 +++ internal/recordmirror/readers_test.go | 11 +- 25 files changed, 732 insertions(+), 26 deletions(-) create mode 100644 cmd/mesh-controller/states_announced.go create mode 100644 cmd/mesh-controller/states_announced_integration_test.go create mode 100644 cmd/mesh-controller/states_announced_test.go create mode 100644 internal/broker/announced.go create mode 100644 internal/broker/announced_test.go create mode 100644 internal/broker/announced_test_helpers_test.go diff --git a/cmd/mesh-controller/build.go b/cmd/mesh-controller/build.go index a9c640cc..c09a49e0 100644 --- a/cmd/mesh-controller/build.go +++ b/cmd/mesh-controller/build.go @@ -683,11 +683,15 @@ func publishable(result link.BuildResult, follows string) error { // answers is what the three questions came back with, read once. type answers struct { - wrong []inventory.Doing - nodes []inventory.Node - quiet []inventory.Node - behind map[string][]string - sources map[string]inventory.Source + // notAnnounced are the owners some module reads the state of that have announced nothing yet (novox/hq ADR 0495, + // issue 1790): read from `reads` as before until they do; notAnnouncedWhy why that could not be read. + notAnnounced []string + notAnnouncedWhy string + wrong []inventory.Doing + nodes []inventory.Node + quiet []inventory.Node + behind map[string][]string + sources map[string]inventory.Source // waiting is every machine not running what the mesh would send it. waiting []inventory.Machine // recordedWaits is, per machine, every move of a recorded module that waits there for a person (novox/hq ADR diff --git a/cmd/mesh-controller/main.go b/cmd/mesh-controller/main.go index 9eab8a5e..0b506f2a 100644 --- a/cmd/mesh-controller/main.go +++ b/cmd/mesh-controller/main.go @@ -54,6 +54,8 @@ func run() error { return fmt.Errorf("no command given") } + // The bus's users are composed from what each module announced it keeps (novox/hq ADR 0495, issue 1790). + inventory.Announcements = readAnnouncements // Every connection this process dials says what it is (novox/hq issue 327). broker.ConnectionName = connectionName(args[0], os.Getenv(verbVar)) diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index e3997d88..510f650a 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -289,6 +289,14 @@ func serve(ctx context.Context) (err error) { // And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by // the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build. watchNewModules(ctx, open) + // And says what it keeps, and keeps what every module says it keeps (novox/hq ADR 0495, issue 1790): a changed word + // sends the machine holding the bus its new user list. + if err := announceOwnStates(ctx, server.JetStream()); err != nil { + fmt.Printf("states: %v\n", err) + } + if err := keepAnnouncements(ctx, server.JetStream(), open); err != nil { + fmt.Printf("states: the states the modules announce are not kept: %v\n", err) + } // Every call carries the lease's epoch, and its record is written only under the lease (novox/hq // to-be 45 §6). link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) }) diff --git a/cmd/mesh-controller/readable.go b/cmd/mesh-controller/readable.go index 82657485..30a32f0d 100644 --- a/cmd/mesh-controller/readable.go +++ b/cmd/mesh-controller/readable.go @@ -53,6 +53,11 @@ type meshStatus struct { // Per machine, and data. One node failing must never take the document away from a reader // asking about the others. Unresolved []machineUnresolved `json:"unresolved"` + // NotAnnounced are the modules others read the state of that have not said what they keep (novox/hq ADR 0495, + // issue 1790): their readers are granted from `reads` as before until they do. NotAnnouncedWhy is why it could not + // be read. + NotAnnounced []string `json:"not-announced,omitempty"` + NotAnnouncedWhy string `json:"not-announced-why,omitempty"` // Network is why the private network could not be computed, when it could not; absent when it // could. Almost always a consequence of Unresolved: a node that does not resolve is not on the // network, and a mesh whose hub is that node has no hub. @@ -247,7 +252,8 @@ func statusAsJSON(asked answers) ([]byte, error) { out := meshStatus{Machines: len(nodes), Wrong: []machineDoing{}, Quiet: []machineQuiet{}, Behind: []moduleBehind{}, Waiting: []machineWaiting{}, Reported: []machineReported{}, Unresolved: []machineUnresolved{}, - Network: asked.network, Adopted: adoptedNodes(nodes), Plans: planStatuses(asked.plans, time.Now(), asked.paused, asked.tierBounds)} + Network: asked.network, Adopted: adoptedNodes(nodes), Plans: planStatuses(asked.plans, time.Now(), asked.paused, asked.tierBounds), + NotAnnounced: asked.notAnnounced, NotAnnouncedWhy: asked.notAnnouncedWhy} // In a stated order, so two readings of an unchanged mesh are the same document. untakenNodes := make([]string, 0, len(asked.untaken)) for name := range asked.untaken { diff --git a/cmd/mesh-controller/states_announced.go b/cmd/mesh-controller/states_announced.go new file mode 100644 index 00000000..8e4012a1 --- /dev/null +++ b/cmd/mesh-controller/states_announced.go @@ -0,0 +1,225 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + + "git.novox.be/novox/mesh-sdk/go/announce" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" +) + +// What each module keeps, and who may read it (novox/hq ADR 0495 clauses 2 to 5, ADR 0513 §3; issue 1790, D1-D3). +// +// Every module says on the bus which states it keeps and which of them others may read (announce.Subject), from its +// code; the controller says its own (D1). The serving controller keeps the last word of each module in its bucket +// (D2), and the bus's users are composed from it: a reader's `reads: .` is granted only when the owner +// announced that state readable; an owner that has announced nothing yet is read from `reads` as before (the +// transition), and `status` names it. A changed word composes the bus's users again: the machine holding the bus is +// sent its new user list (D3). No timer: the words wake it. + +// announceOwnStates says what the controller keeps, on its own event subject (D1). +func announceOwnStates(ctx context.Context, js *broker.JetStream) error { + host, _ := os.Hostname() + body, err := announce.Encode(broker.ControllerAnnouncement(host)) + if err != nil { + return err + } + api, err := jetstream.New(js.Conn()) + if err != nil { + return err + } + if _, err := api.Publish(ctx, announce.Subject(broker.ControllerSeat), body); err != nil { + return fmt.Errorf("saying what the controller keeps: %w", err) + } + return nil +} + +// keepAnnouncements follows every module's word on what it keeps (D2) and keeps the last of each, from the last each +// module said: one module's word in the bucket under its name. A word whose module is not the subject's is refused, +// said; a changed word composes the bus's users again (D3). +func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores) error { + api, err := jetstream.New(js.Conn()) + if err != nil { + return err + } + kv, err := api.KeyValue(ctx, broker.StatesAnnouncedBucket) + if err != nil { + return fmt.Errorf("the bucket of the states announced: %w", err) + } + stream, err := api.Stream(ctx, broker.EventsStream) + if err != nil { + return fmt.Errorf("the events stream: %w", err) + } + consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{announce.SubjectAll}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return fmt.Errorf("following the states announced: %w", err) + } + consuming, err := consumer.Consume(func(m jetstream.Msg) { + changed, err := keepAnnouncement(ctx, kv, m.Subject(), m.Data()) + if err != nil { + fmt.Printf("states: %v\n", err) + return + } + if changed { + go composeAnnouncedReads(context.WithoutCancel(ctx), open) + } + }) + if err != nil { + return fmt.Errorf("following the states announced: %w", err) + } + go func() { + <-ctx.Done() + consuming.Stop() + }() + return nil +} + +// announcedKeeper is the bucket of the states announced, as keepAnnouncement writes it. +type announcedKeeper interface { + Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) + Put(ctx context.Context, key string, value []byte) (uint64, error) +} + +// keepAnnouncement keeps one module's word, and says whether it changed what the bucket held. +func keepAnnouncement(ctx context.Context, kv announcedKeeper, subject string, body []byte) (bool, error) { + s, err := announce.Decode(body) + if err != nil { + return false, fmt.Errorf("a word on %s is not read: %w", subject, err) + } + if subject != announce.Subject(s.Module) { + return false, fmt.Errorf("%s says what %s keeps, and a module says only what it keeps itself: refused", subject, + s.Module) + } + key, err := announce.Key(s.Module) + if err != nil { + return false, err + } + stored, err := announce.Encode(s) + if err != nil { + return false, err + } + if was, err := kv.Get(ctx, key); err == nil { + if string(was.Value()) == string(stored) { + return false, nil + } + } else if !errors.Is(err, jetstream.ErrKeyNotFound) { + return false, fmt.Errorf("the last word of %s cannot be read: %w", s.Module, err) + } + if _, err := kv.Put(ctx, key, stored); err != nil { + return false, fmt.Errorf("the word of %s cannot be kept: %w", s.Module, err) + } + fmt.Printf("states: %s keeps %s\n", s.Module, statesWords(s)) + return true, nil +} + +// statesWords is an announcement as a person reads it: each state, readable ones marked. +func statesWords(s announce.States) string { + if len(s.States) == 0 { + return "nothing" + } + var out []string + for _, k := range s.States { + if k.Readable { + out = append(out, k.Name+" (readable)") + } else { + out = append(out, k.Name) + } + } + return strings.Join(out, ", ") +} + +// readAnnouncements is the last word of every module, by module (what inventory.Announcements reads): none when the +// bucket is not there yet, which reads every owner as not yet announced. +func readAnnouncements(ctx context.Context) (map[string]announce.States, error) { + js, err := aBus() + if err != nil { + return nil, err + } + defer js.Close() + api, err := jetstream.New(js.Conn()) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, broker.StatesAnnouncedBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + // empty-on-error: not an error; no module announced yet, so every read is granted from `reads` as before + return map[string]announce.States{}, nil + } + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + // empty-on-error: not an error; no module announced yet + return map[string]announce.States{}, nil + } + if err != nil { + return nil, err + } + out := map[string]announce.States{} + for _, key := range keys { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return nil, err + } + s, err := announce.Decode(e.Value()) + if err != nil { + return nil, fmt.Errorf("the kept word of %s is not read: %w", key, err) + } + out[s.Module] = s + } + return out, nil +} + +// composeAnnouncedReads sends the machine holding the bus its user list when a changed word changed it (D3): as a new +// controller's user list is carried (carryUserList), not when that machine is held. +func composeAnnouncedReads(ctx context.Context, open *stores) { + holder, behind, err := brokerBehind(ctx, open, nil) + if err != nil { + fmt.Printf("states: whether the bus's user list is behind the states announced cannot be read: %v\n", err) + return + } + if holder == "" || !behind { + return + } + held, err := heldMachines(ctx, open, []string{holder}) + if err != nil { + fmt.Printf("states: whether %s may be sent its new user list cannot be read: %v\n", holder, err) + return + } + if why, isHeld := held[holder]; isHeld { + fmt.Printf("states: the user list the states announced compose is not sent to %s, which holds the bus: %s — "+ + "`push %s` sends it\n", holder, strings.Join(why, "; "), holder) + return + } + if _, err := sendRollout(ctx, open, []string{holder}); err != nil { + fmt.Printf("states: the user list the states announced compose could not be sent to %s: %v\n", holder, err) + return + } + fmt.Printf("states: sent %s, which holds the bus, the user list the states announced compose\n", holder) +} + +// notYetAnnounced are the owners some module reads the state of and that have announced nothing yet: what `status` +// names while the transition lasts. +func notYetAnnounced(ctx context.Context, inv *inventory.Inventory) ([]string, error) { + if inventory.Announcements == nil { + return nil, nil + } + records, err := inv.BusRecords(ctx) + if err != nil { + return nil, err + } + if records.Announced == nil { + return nil, errors.New("the states announced cannot be read") + } + return broker.NotAnnounced(records), nil +} diff --git a/cmd/mesh-controller/states_announced_integration_test.go b/cmd/mesh-controller/states_announced_integration_test.go new file mode 100644 index 00000000..0e64ca7f --- /dev/null +++ b/cmd/mesh-controller/states_announced_integration_test.go @@ -0,0 +1,118 @@ +//go:build integration + +package main + +import ( + "context" + "slices" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/testbus" +) + +// D1-D3 whole against a real bus and the store (ADR 0495 checks §2, §3, §5; ADR 0513 §3; issue 1790): the controller +// says what it keeps and keeps what every module says; a reader of a state its owner announced readable is granted it, +// one its owner announced and did not make readable is refused and said, and one of an owner that announced nothing +// yet is granted as before and named as not yet announced. +func TestTheStatesAnnouncedGrantTheReads(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + servingBus.Store(js) + t.Cleanup(func() { servingBus.Store(nil) }) + was := inventory.Announcements + inventory.Announcements = readAnnouncements + t.Cleanup(func() { inventory.Announcements = was }) + + // The owner keeps two states, one readable; the reader reads both and a state of a module that says nothing. + register := func(m catalogue.Manifest) { + t.Helper() + if err := inv.RegisterModule(ctx, m, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", + Path: "modules/" + m.Module, BuiltFrom: "c0"}); err != nil { + t.Fatal(err) + } + } + register(catalogue.Manifest{Module: "owner", Version: "1", State: []catalogue.StateDeclaration{{Name: "open"}, {Name: "closed"}}}) + register(catalogue.Manifest{Module: "quiet", Version: "1", State: []catalogue.StateDeclaration{{Name: "any"}}}) + register(catalogue.Manifest{Module: "reader", Version: "1", Reads: []string{"owner.open", "owner.closed", "quiet.any"}}) + if _, err := inv.AddNode(ctx, "one"); err != nil { + t.Fatal(err) + } + for _, m := range []string{"owner", "quiet", "reader"} { + if _, err := inv.Assign(ctx, "one", m); err != nil { + t.Fatal(err) + } + } + + watching, stop := context.WithCancel(context.Background()) + t.Cleanup(stop) + if err := announceOwnStates(ctx, js); err != nil { + t.Fatal(err) + } + if err := keepAnnouncements(watching, js, open); err != nil { + t.Fatal(err) + } + api, _ := jetstream.New(js.Conn()) + word, _ := announce.Encode(announce.States{Module: "owner", Node: "one", + States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}}) + if _, err := api.Publish(ctx, announce.Subject("owner"), word); err != nil { + t.Fatal(err) + } + becomes(t, 10*time.Second, "both words kept", func() bool { + a, err := readAnnouncements(ctx) + return err == nil && len(a) == 2 && a["owner"].Readable("open") && a[broker.ControllerSeat].Readable(broker.PlansBuilding) + }) + + records, err := inv.BusRecords(ctx) + if err != nil { + t.Fatal(err) + } + var reader broker.Declared + for _, d := range records.Assigned["one"] { + if d.Module == "reader" { + reader = d + } + } + if !slices.Equal(reader.Reads, []string{"owner.open", "quiet.any"}) || !slices.Equal(reader.ReadsRefused, []string{"owner.closed"}) { + t.Fatalf("0495 §3: the reader is granted %v and refused %v", reader.Reads, reader.ReadsRefused) + } + if got := broker.NotAnnounced(records); !slices.Equal(got, []string{"quiet"}) { + t.Fatalf("not yet announced: %v", got) + } + users, err := broker.Users(records) + if err != nil { + t.Fatal(err) + } + for _, u := range users { + if u.Kind != broker.KindModule || u.Module != "reader" { + continue + } + perms, err := broker.PermissionsFor(u) + if err != nil { + t.Fatal(err) + } + if slices.ContainsFunc(perms.Publish, func(s string) bool { return s == "$JS.API.DIRECT.GET.KV_owner_closed.>" }) { + t.Fatal("0495 §3: the reader may read the state its owner did not announce readable") + } + if !slices.Contains(perms.Publish, "$JS.API.DIRECT.GET.KV_owner_open.>") { + t.Fatal("0495 §3: the reader may not read the state its owner announced readable") + } + } +} diff --git a/cmd/mesh-controller/states_announced_test.go b/cmd/mesh-controller/states_announced_test.go new file mode 100644 index 00000000..eb3f920f --- /dev/null +++ b/cmd/mesh-controller/states_announced_test.go @@ -0,0 +1,56 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/announce" + "github.com/nats-io/nats.go/jetstream" +) + +// announcedInMemory is the bucket of the states announced, in memory. +type announcedInMemory map[string][]byte + +type announcedEntry struct { + jetstream.KeyValueEntry + value []byte +} + +func (e announcedEntry) Value() []byte { return e.value } + +func (k announcedInMemory) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + v, ok := k[key] + if !ok { + return nil, jetstream.ErrKeyNotFound + } + return announcedEntry{value: v}, nil +} + +func (k announcedInMemory) Put(_ context.Context, key string, value []byte) (uint64, error) { + k[key] = value + return uint64(time.Now().UnixNano()), nil +} + +// D2 (ADR 0495 clause 2; issue 1790): a module's word is kept under its name, changed only when it changes, and a word +// about another module than its subject's is refused. +func TestAModulesWordIsKeptUnderItsNameAndOnlyItsOwn(t *testing.T) { + k := announcedInMemory{} + word, _ := announce.Encode(announce.States{Module: "keeper", States: []announce.State{{Name: "things", Readable: true}}}) + changed, err := keepAnnouncement(t.Context(), k, announce.Subject("keeper"), word) + if err != nil || !changed || string(k["keeper"]) != string(word) { + t.Fatalf("the first word: %v %v %q", changed, err, k["keeper"]) + } + if changed, err := keepAnnouncement(t.Context(), k, announce.Subject("keeper"), word); err != nil || changed { + t.Fatalf("the same word again changed the bucket: %v %v", changed, err) + } + other, _ := announce.Encode(announce.States{Module: "victim", States: []announce.State{{Name: "secrets", Readable: true}}}) + if _, err := keepAnnouncement(t.Context(), k, announce.Subject("keeper"), other); err == nil || + !strings.Contains(err.Error(), "only what it keeps itself") { + t.Fatalf("a word about another module was kept: %v", err) + } + if _, found := k["victim"]; found { + t.Fatal("a word about another module was kept") + } +} diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index 5d262e36..93191908 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -85,6 +85,13 @@ func printStatus(asked answers) error { // **What is wrong leads** (novox/hq to-be 45 §2): every open condition, urgent first, oldest // first, silenced ones with when their silence ends. printConditions(asked.conditions, asked.conditionsUnread, time.Now()) + if len(asked.notAnnounced) > 0 { + fmt.Printf("%d module(s) others read the state of have not said what they keep, and are read from `reads` as "+ + "before until they do (ADR 0495): %s\n\n", len(asked.notAnnounced), strings.Join(asked.notAnnounced, ", ")) + } + if asked.notAnnouncedWhy != "" { + fmt.Printf("whose states are read without an announcement cannot be said: %s\n\n", asked.notAnnouncedWhy) + } if len(asked.refused) > 0 { // First of what follows. A machine that cannot be worked out is not running an old @@ -435,6 +442,11 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { if err != nil { return answers{}, err } + // And whose states are still read without an announcement (novox/hq ADR 0495, issue 1790): said, never a reason + // for the rest of the answer to fail. + if out.notAnnounced, err = notYetAnnounced(ctx, inv); err != nil { + out.notAnnounced, out.notAnnouncedWhy = nil, err.Error() + } // And what each machine is holding rather than running, by the module that would run it. Read // from what the machine itself last reported, not from what take-time computed: the machine is // the only thing that knows what it found (novox/hq 04-ISSUES/125). diff --git a/internal/broker/announced.go b/internal/broker/announced.go new file mode 100644 index 00000000..9d1fdcae --- /dev/null +++ b/internal/broker/announced.go @@ -0,0 +1,81 @@ +package broker + +import ( + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// Who may read a module's state is what the module announced (novox/hq ADR 0495 clauses 2 and 3, ADR 0513 §3; issue +// 1790). A module says on the bus which states it keeps and which others may read (announce.Subject); the controller +// keeps the last word of each (announce.KeptBucket) and grants a reader's `reads: .` only when the owner +// announced that state readable. An owner that has announced nothing yet is read as before, from the reader's `reads` +// (the transition, until every module announces), and is named as not yet announced. + +// StatesAnnouncedBucket is the controller's bucket of the last announcement of each module (announce.KeptBucket). +const StatesAnnouncedBucket = announce.KeptBucket + +// controllerReadable are the controller's states others may read: the record mirrors, the chain's buckets, the checks +// running and the plans building. +var controllerReadable = []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, + ChecksRunning, PlansBuilding} + +// ControllerAnnouncement is what the controller announces it keeps (ADR 0495 clause 2): every bucket of its own by its local +// name, the ones others may read marked readable. It announces itself, as any module does from its code. +func ControllerAnnouncement(node string) announce.States { + s := announce.States{Module: ControllerSeat, Node: node} + readable := map[string]bool{} + for _, n := range controllerReadable { + readable[n] = true + } + for _, bucket := range ControllerBuckets() { + local, mine := strings.CutPrefix(bucket, ControllerSeat+"_") + if !mine { + continue + } + s.States = append(s.States, announce.State{Name: local, Readable: readable[local]}) + } + sort.Slice(s.States, func(i, j int) bool { return s.States[i].Name < s.States[j].Name }) + return s +} + +// ReadsAnnounced splits a reader's `reads` (`.`) by what each owner announced: granted when the owner +// announced the state readable, or announced nothing yet (the transition); refused when the owner announced and did +// not announce that state readable. +func ReadsAnnounced(reads []string, announced map[string]announce.States) (granted, refused []string) { + for _, r := range reads { + owner, name, ok := strings.Cut(r, ".") + s, said := announced[owner] + switch { + case !ok || !said || s.Readable(name): + granted = append(granted, r) + default: + refused = append(refused, r) + } + } + return granted, refused +} + +// NotAnnounced are the owners some module reads the state of and that have announced nothing yet, sorted: read from the +// readers' `reads` as before, until they announce. +func NotAnnounced(r Records) []string { + seen := map[string]bool{} + var out []string + for _, ds := range r.Assigned { + for _, d := range ds { + for _, read := range append(append([]string(nil), d.Reads...), d.ReadsRefused...) { + owner, _, ok := strings.Cut(read, ".") + if !ok || seen[owner] { + continue + } + if _, said := r.Announced[owner]; !said { + seen[owner] = true + out = append(out, owner) + } + } + } + } + sort.Strings(out) + return out +} diff --git a/internal/broker/announced_test.go b/internal/broker/announced_test.go new file mode 100644 index 00000000..cca553df --- /dev/null +++ b/internal/broker/announced_test.go @@ -0,0 +1,85 @@ +package broker + +import ( + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// D3 (ADR 0495 clause 3, ADR 0513 §3; issue 1790): a reader's `reads` is granted where the owner announced the state +// readable, refused where it announced and did not, and granted as before where the owner announced nothing yet. +func TestAReadIsGrantedFromItsOwnersAnnouncement(t *testing.T) { + announced := map[string]announce.States{ + "owner": {Module: "owner", States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}}, + } + granted, refused := ReadsAnnounced([]string{"owner.open", "owner.closed", "owner.unsaid", "quiet.any"}, announced) + if !slices.Equal(granted, []string{"owner.open", "quiet.any"}) || !slices.Equal(refused, []string{"owner.closed", "owner.unsaid"}) { + t.Fatalf("granted %v, refused %v", granted, refused) + } + r := Records{Announced: announced, Assigned: map[string][]Declared{"one": {{Module: "reader", Reads: granted, + ReadsRefused: refused}}}} + if got := NotAnnounced(r); !slices.Equal(got, []string{"quiet"}) { + t.Fatalf("not yet announced: %v", got) + } +} + +// D1 (ADR 0495 clause 2): the controller announces every bucket of its own, and readable exactly the record mirrors, +// the chain's buckets, the checks running and the plans building. +func TestTheControllerAnnouncesWhatItKeeps(t *testing.T) { + s := ControllerAnnouncement("novox") + if _, err := announce.Encode(s); err != nil { + t.Fatal(err) + } + var readable []string + for _, k := range s.States { + if k.Readable { + readable = append(readable, k.Name) + } + } + want := []string{ChainBuilds, ChecksRunning, MirrorDeclaredConfig, PlansBuilding, MirrorPredefined, ChainRollouts, + MirrorAssignments} + slices.Sort(want) + if !slices.Equal(readable, want) { + t.Fatalf("readable %v, want %v", readable, want) + } + for _, name := range []string{"lease", "calls", "states-announced", "merge-window"} { + if !s.Keeps(name) || s.Readable(name) { + t.Errorf("%s: kept %v, readable %v", name, s.Keeps(name), s.Readable(name)) + } + } + perms, err := PermissionsFor(Principal{Kind: KindController}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(perms.Publish, announce.Subject(ControllerSeat)) { + t.Fatal("the controller may not say what it keeps") + } +} + +// D2: every account that keeps state may say so on its own subject, a process account included, and no other's. +func TestAnAccountKeepingStateMaySayWhatItKeeps(t *testing.T) { + keeper, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "keeper", PasswordHash: "x", + State: []string{"things"}}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(keeper.Publish, announce.Subject("keeper")) || slices.Contains(keeper.Publish, announce.Subject("other")) { + t.Fatalf("the keeper's announcement grants: %v", keeper.Publish) + } + process, err := PermissionsFor(Principal{Kind: KindProcess, Node: "one", Module: "keeper", Process: "send", + PasswordHash: "x", State: []string{"sends"}}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(process.Publish, announce.Subject("keeper")) { + t.Fatal("a process keeping state may not say what it keeps") + } + none, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "stateless", PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if slices.Contains(none.Publish, announce.Subject("stateless")) { + t.Fatal("a module keeping nothing may say it keeps something") + } +} diff --git a/internal/broker/announced_test_helpers_test.go b/internal/broker/announced_test_helpers_test.go new file mode 100644 index 00000000..e8b1b5a2 --- /dev/null +++ b/internal/broker/announced_test_helpers_test.go @@ -0,0 +1,13 @@ +package broker + +// controllerReadableStates are the controller's states its announcement makes readable (ADR 0495 clause 2): what a +// module may name in `reads` after `mesh-controller.`. +func controllerReadableStates() []string { + var out []string + for _, s := range ControllerAnnouncement("").States { + if s.Readable { + out = append(out, s.Name) + } + } + return out +} diff --git a/internal/broker/checks_running_mirror_test.go b/internal/broker/checks_running_mirror_test.go index a19f1dcf..aec691ea 100644 --- a/internal/broker/checks_running_mirror_test.go +++ b/internal/broker/checks_running_mirror_test.go @@ -8,8 +8,8 @@ import ( // 1155/AC1: a module whose manifest reads mesh-controller.checks-running may watch and get every key of the bucket and // write none. func TestAModuleReadingChecksRunningReadsAndWritesNone(t *testing.T) { // 1155/AC1 - if !slices.Contains(ControllerMirrors(), "checks-running") || BucketName(ControllerSeat, "checks-running") != CheckRunningBucket { - t.Fatalf("a module cannot name %s in `reads`: %v", CheckRunningBucket, ControllerMirrors()) + if !slices.Contains(controllerReadableStates(), "checks-running") || BucketName(ControllerSeat, "checks-running") != CheckRunningBucket { + t.Fatalf("a module cannot name %s in `reads`: %v", CheckRunningBucket, controllerReadableStates()) } perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-watchman", PasswordHash: "x", Reads: []string{"mesh-controller.checks-running"}}) diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index fe588c01..f3aad638 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -150,12 +150,6 @@ const ( // ChainBuckets are the three buckets of the chain: builds, rollouts and assignments. func ChainBuckets() []string { return []string{BuildsBucket, RolloutsBucket, AssignmentsBucket} } -// ControllerMirrors are the local names of the controller's buckets a module may read: the record mirrors and the -// chain's buckets. -func ControllerMirrors() []string { - return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, ChecksRunning, PlansBuilding} -} - // MirrorsWrittenByAnother are the controller's mirrors whose writer is not the controller, each with its reason: the rule // that a mirror is written by the controller alone (TestTheMirrorsListAndTheWritersTableAgree) has these exceptions and no // other. @@ -224,14 +218,14 @@ func IsControllerBucket(bucket string) bool { bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket || bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket || - bucket == PlansBuildingBucket || bucket == delivery.HaveBucket + bucket == PlansBuildingBucket || bucket == delivery.HaveBucket || bucket == StatesAnnouncedBucket } // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket, - DeliveryMergesBucket, PlansBuildingBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket} + DeliveryMergesBucket, PlansBuildingBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket, StatesAnnouncedBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -390,6 +384,18 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", PlansBuildingBucket, err) } + // **The states each module announced** (novox/hq ADR 0495 clauses 2 and 3, issue 1790): one key a module, its last word. + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: StatesAnnouncedBucket, + Description: "the states each module announced it keeps and which of them others may read (novox/hq ADR 0495): " + + "`` holds its last announcement; what a reader's `reads` is granted from; written by the controller alone", + History: 1, + MaxValueSize: 64 << 10, + MaxBytes: 16 << 20, + Storage: jetstream.FileStorage, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", StatesAnnouncedBucket, err) + } // **The merges mesh-delivery announced: keys that age** (ADR 0420 §7, issue 1025). One message a key; a week. if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ Bucket: DeliveryMergesBucket, diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 228f05c0..f0637377 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -100,7 +100,7 @@ func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) { // stands for their declaration, so the report of state nothing declares never names one. func TestAListedMirrorIsNotReportedAsUndeclared(t *testing.T) { names := []string{"gone_state"} - for _, m := range ControllerMirrors() { + for _, m := range controllerReadableStates() { names = append(names, BucketName(ControllerSeat, m)) } undeclared, err := RaiseBuckets(fakeBuckets{names: names}, nil) @@ -132,7 +132,7 @@ func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) { } } listed := map[string]bool{} - for _, m := range ControllerMirrors() { + for _, m := range controllerReadableStates() { bucket := BucketName(ControllerSeat, m) listed[bucket] = true if why, other := MirrorsWrittenByAnother[m]; other { // 1155/AC3: the exception, with its reason @@ -292,7 +292,7 @@ func TestEveryChainBucketIsOnTheControllersListAndHasAWriterRow(t *testing.T) { t.Errorf("%s is not on the controller's list of buckets", bucket) } local := strings.TrimPrefix(bucket, ControllerSeat+"_") - if !slices.Contains(ControllerMirrors(), local) { + if !slices.Contains(controllerReadableStates(), local) { t.Errorf("a module cannot name %s in `reads`", bucket) } rows := 0 diff --git a/internal/broker/delivery_parts_grant_test.go b/internal/broker/delivery_parts_grant_test.go index fbfbe001..49edefeb 100644 --- a/internal/broker/delivery_parts_grant_test.go +++ b/internal/broker/delivery_parts_grant_test.go @@ -2,6 +2,7 @@ package broker import ( "encoding/json" + "git.novox.be/novox/mesh-sdk/go/announce" "slices" "sort" "strings" @@ -407,7 +408,8 @@ func withPrefix(patterns []string, prefixes ...string) []string { func eventsSaid(p Permissions) []string { var out []string for _, s := range withPrefix(p.Publish, "mesh.mod.", "mesh.stream.") { - if strings.Contains(s, ".event.") { + // A module keeping state says what it keeps (ADR 0495 clause 2): no event of its declaration. + if strings.Contains(s, ".event.") && !strings.HasSuffix(s, ".event."+announce.Event) { out = append(out, s) } } diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 0f8466fa..e61dfbd0 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -15,6 +15,7 @@ package broker import ( "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/announce" "regexp" "slices" "sort" @@ -389,6 +390,16 @@ type DegradableGrant struct { // works. Every other grant stays required — missing one of those is a controller that comes up and is // refused on the first thing it tries, with no fallback that covers it. var DegradableGrants = []DegradableGrant{ + { + Subject: announce.Subject(ControllerSeat), + Reason: "refused: the controller's own announcement waits for its first push, and until it is kept every reader of " + + "its states is granted from its `reads` as before (novox/hq ADR 0495, issue 1790)", + }, + { + Subject: "$KV." + StatesAnnouncedBucket + ".>", + Reason: "refused: no announcement is kept until the controller's first push, and until then every reader is " + + "granted from its `reads` as before (novox/hq ADR 0495, issue 1790)", + }, { Subject: "mesh.seat." + ControllerSeat + ".tool.push", Reason: "refused: the ask of a held move (Release) is answered, and the refusal told, while the operator's own " + @@ -751,6 +762,8 @@ func PermissionsFor(p Principal) (Permissions, error) { // **And its own buckets** (novox/hq to-be 45 §1): the calls it served and the acts done by // hand, which it alone writes. A put is a publish to the bucket's subject, which `$JS.API.>` // does not cover; each bucket named, not `$KV.>`, which would let it write any module's state. + // And says what it keeps, as every module does (novox/hq ADR 0495 clause 2, issue 1790). + pub = append(pub, announce.Subject(ControllerSeat)) for _, bucket := range ControllerBuckets() { if bucket == delivery.HaveBucket { continue // each node's node-engine writes its own keys of it, the controller none (ADR 0429 §1) diff --git a/internal/broker/plans_building_mirror_test.go b/internal/broker/plans_building_mirror_test.go index 43efd6db..9b57f4d3 100644 --- a/internal/broker/plans_building_mirror_test.go +++ b/internal/broker/plans_building_mirror_test.go @@ -8,8 +8,8 @@ import ( // Issue 1687 (ADR 0483 clause 2): a module whose manifest reads mesh-controller.plans-building may name it in `reads`, and // reads the bucket and writes none; the controller writes it, so it is no mirror written by another. func TestAModuleReadingPlansBuildingReadsAndWritesNone(t *testing.T) { - if !slices.Contains(ControllerMirrors(), "plans-building") || BucketName(ControllerSeat, PlansBuilding) != PlansBuildingBucket { - t.Fatalf("a module cannot name %s in `reads`: %v", PlansBuildingBucket, ControllerMirrors()) + if !slices.Contains(controllerReadableStates(), "plans-building") || BucketName(ControllerSeat, PlansBuilding) != PlansBuildingBucket { + t.Fatalf("a module cannot name %s in `reads`: %v", PlansBuildingBucket, controllerReadableStates()) } if _, other := MirrorsWrittenByAnother[PlansBuilding]; other { t.Error("the controller writes the plans building itself") diff --git a/internal/broker/process_grants_test.go b/internal/broker/process_grants_test.go index 7866df30..93259427 100644 --- a/internal/broker/process_grants_test.go +++ b/internal/broker/process_grants_test.go @@ -48,6 +48,8 @@ func TestAProcessIsGrantedExactlyItsBusDeclaration(t *testing.T) { "$KV.mesh-mail_sends.>", "mesh.mod.mesh-mail.event.mail.send.failed", "mesh.mod.mesh-mail.event.mail.send.sent", + // It keeps state, so it says what it keeps (novox/hq ADR 0495 clause 2, issue 1790). + "mesh.mod.mesh-mail.event.states", "mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send", "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", "mesh.seat.operator-channel.tool.tell", diff --git a/internal/broker/state.go b/internal/broker/state.go index 25c7c7ab..b89b3217 100644 --- a/internal/broker/state.go +++ b/internal/broker/state.go @@ -2,6 +2,7 @@ package broker import ( "fmt" + "git.novox.be/novox/mesh-sdk/go/announce" "sort" "strings" "time" @@ -147,6 +148,11 @@ func stateGrants(a stateAccess) []string { read(bucket, "") out = append(out, "$KV."+bucket+".>") } + // **And it says what it keeps** (novox/hq ADR 0495 clause 2, issue 1790): a module keeping state announces it on its + // own event subject, from its code; refused, its readers are read as before. + if len(a.Keeps) > 0 && safeSubject.MatchString(a.Module) { + out = append(out, announce.Subject(a.Module)) + } for _, r := range a.Reads { if bucket, ok := bucketOfRead(r); ok { read(bucket, "") diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 4e2ba482..c3f0da3e 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_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$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-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.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"] } + 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_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$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.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/users.go b/internal/broker/users.go index a63e281e..123e436e 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -2,6 +2,7 @@ package broker import ( "fmt" + "git.novox.be/novox/mesh-sdk/go/announce" "slices" "sort" ) @@ -56,6 +57,9 @@ type Declared struct { HearsAStream bool // Reads are other modules' state it reads, each `.` (novox/hq ADR 0201). Reads []string + // ReadsRefused are the reads its manifest names that the owner's announcement does not make readable (novox/hq ADR + // 0495 clause 3, issue 1790): granted nothing, kept to be said. + ReadsRefused []string // KeyedReads are keys of other modules' state it reads, each one key alone: what a seat's holder is // granted for the pieces the modules beside it offer (novox/hq ADR 0260). KeyedReads []KeyedRead @@ -126,6 +130,9 @@ type Records struct { // account agents run as, judged unable to become root by its node-engine, and serves no login shell // execute. A machine absent is not free: no record of a pass is no pass. RootFree map[string]bool + // Announced is each module's last announcement of the states it keeps (novox/hq ADR 0495, issue 1790), as the + // controller keeps it; a module absent has announced nothing yet. Nil where none was read. + Announced map[string]announce.States } // Users is every user the composed file should contain, in the order it will be written. diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 9dd7ed3f..b31d0192 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -87,6 +87,22 @@ func ownModule(p Principal, pattern string) bool { return false } +// ownAnnouncer is a module, a process of it or the node tools carrying it saying what that module keeps, or the +// controller saying what it keeps itself (novox/hq ADR 0495 clause 2): each under its own name alone. +func ownAnnouncer(p Principal, pattern string) bool { + tokens := strings.Split(pattern, ".") + if len(tokens) < 3 { + return false + } + switch p.Kind { + case KindController: + return tokens[2] == ControllerSeat + case KindProcess: + return p.Module == tokens[2] + } + return ownModule(p, pattern) +} + // ownProvisioner is a machine's provisioner client saying a report about its own machine: the third token of // `prov.applied..…` is its machine (novox/hq ADR 0349 §4, design 57 P3). func ownProvisioner(p Principal, pattern string) bool { @@ -181,6 +197,14 @@ var WritersTable = []WriterRow{ KeptIn: "key-value " + PlansBuildingBucket + ", aged out when not refreshed", Others: "the build seat's holders read it and take no new check while a key is there (ADR 0483)", Subjects: kvOf(PlansBuildingBucket), Writes: isController}, + // What a module keeps and who may read it (novox/hq ADR 0495 clauses 2 and 3, issue 1790): said by the module, kept by + // the controller. + {State: "a module's states announced", Writer: "the module itself, from its code, under its own name", + KeptIn: "the bus (EVENTS)", Others: "the controller keeps the last of each and grants readers from it", + Subjects: []string{"mesh.mod.*.event.states"}, Writes: ownAnnouncer}, + {State: "the states announced", Writer: "controller, from each module's announcement", + KeptIn: "key-value " + StatesAnnouncedBucket, Others: "the composition of the bus's users reads it", + Subjects: kvOf(StatesAnnouncedBucket), 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 24128be7..d01005ce 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -31,6 +31,9 @@ var designRows = []string{ "a pull request's check running", // The plans in their build phase, a bucket the build seat's holders read (ADR 0483 clause 2, issue 1687). "the plans-building mirror", + // What each module keeps and who may read it, said by the module and kept by the controller (ADR 0495, issue 1790). + "a module's states announced", + "the states announced", "a merge announced", "a pull request's head announced", "a pull request's merge check", diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index a3fdd616..de25fb28 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -3,6 +3,7 @@ package inventory import ( "context" "fmt" + "git.novox.be/novox/mesh-sdk/go/announce" "slices" "strings" @@ -16,6 +17,10 @@ import ( // apart for the reason that package keeps its own types: a permission must be a function of what a // module declared, and a query that decided anything would be a second place authority came from. +// Announcements reads the last announcement of each module (novox/hq ADR 0495, issue 1790), set by the controller that +// reaches the bus; nil, nothing is read and every read is granted from `reads`. +var Announcements func(ctx context.Context) (map[string]announce.States, error) + // BusRecords is every fact the composer needs about who may reach the bus. // // **A module's authority comes from the manifest, not from the assignment.** The assignment says @@ -110,6 +115,29 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) { } } + // **Who may read a state is what its owner announced** (novox/hq ADR 0495 clause 3, issue 1790): a read the owner's + // announcement does not make readable is granted nothing, and said; an owner that announced nothing yet is read as + // before. Announcements that cannot be read are said, and every read is granted as before: the bus's users are + // composed either way. + if Announcements != nil { + announced, err := Announcements(ctx) + if err != nil { + fmt.Printf("the states the modules announced cannot be read, so every read is granted from `reads` as before: %v\n", err) + } else { + out.Announced = announced + for node, ds := range out.Assigned { + for k := range ds { + granted, refused := broker.ReadsAnnounced(ds[k].Reads, announced) + ds[k].Reads, ds[k].ReadsRefused = granted, refused + for _, r := range refused { + fmt.Printf("%s on %s reads %s, which its owner does not announce readable: not granted\n", + ds[k].Module, node, r) + } + } + } + } + } + enrolling, err := i.NodesWithALiveToken(ctx) if err != nil { return broker.Records{}, err diff --git a/internal/recordmirror/readers_test.go b/internal/recordmirror/readers_test.go index 91736ad0..83a9faf2 100644 --- a/internal/recordmirror/readers_test.go +++ b/internal/recordmirror/readers_test.go @@ -28,6 +28,9 @@ var mayNameAMirror = []string{ // The grant of a machine's own keys of the assignments mirror to its provisioner client (design 57 P3). "internal/broker/provisioner.go", "internal/broker/writers.go", + // The controller's announcement of what it keeps, the mirrors readable (novox/hq ADR 0495, issue 1790): it names them + // to announce them, never reads them. + "internal/broker/announced.go", } // The packages whose names stand for a mirror, by import path, and the names in each. @@ -38,7 +41,7 @@ const ( var mirrorNames = map[string][]string{ brokerPath: {"AssignmentsBucket", "DeclaredConfigBucket", "MirrorAssignments", "MirrorDeclaredConfig", - "ControllerMirrors"}, + "controllerReadable"}, recordmirrorPath: {"Assignments", "DeclaredConfig", "Mirrors"}, } @@ -54,8 +57,10 @@ var mirrorNames = map[string][]string{ // allowed file. Those are deliberate evasions rather than ordinary code; a reviewer judges them. func TestOnlyTheAuditAndAMirrorsWriterReadAMirror(t *testing.T) { var buckets []string - for _, m := range broker.ControllerMirrors() { - buckets = append(buckets, broker.BucketName(broker.ControllerSeat, m)) + for _, s := range broker.ControllerAnnouncement("").States { + if s.Readable { + buckets = append(buckets, broker.BucketName(broker.ControllerSeat, s.Name)) + } } root := filepath.Join("..", "..") found := mirrorReaders(t, root, buckets)