diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go new file mode 100644 index 00000000..fb1f526f --- /dev/null +++ b/cmd/mesh-controller/composed_keys.go @@ -0,0 +1,225 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/builderkeys" + "github.com/novox/mesh-controller/internal/inventory" +) + +// A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7, +// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's +// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the +// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's +// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now +// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone, +// written into its want at once by the assigner), and writes `grants..`. The assigner's sixth and seventh +// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every +// module once (B5's start pass). + +// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for. +type doneKeySteps struct { + // held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go). + held func(ctx context.Context, module string) (bool, error) + // trunk is the indexer's trunk value of the module. + trunk func(ctx context.Context, module string) (sdkdelivery.Module, error) + // registeredAt says the module is registered at the commit. + registeredAt func(ctx context.Context, module, commit string) (bool, error) + // takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when + // the key is not done yet. + takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error + // nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none. + nodesOf func(ctx context.Context, module string) ([]string, error) + busHolder func(ctx context.Context) (string, error) + // compose writes the nodes' compositions; grants writes the grants issued. + compose func(ctx context.Context, nodes []string) error + grants func(ctx context.Context) error +} + +// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was +// nothing to do. +func (s doneKeySteps) take(ctx context.Context, module string) (string, error) { + held, err := s.held(ctx, module) + if err != nil || !held { + return "", err + } + trunk, err := s.trunk(ctx, module) + if errors.Is(err, builderkeys.ErrNoValue) { + return "", nil + } + if err != nil { + return "", err + } + if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) { + return "", nil + } + at, err := s.registeredAt(ctx, module, trunk.Commit) + if err != nil { + return "", err + } + said := "" + if !at { + if err := s.takeIn(ctx, module, trunk); err != nil { + return "", err + } + said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit)) + } + nodes, err := s.nodesOf(ctx, module) + if err != nil { + return said, err + } + if len(nodes) == 0 { + return said, nil + } + // The module's nodes first: composing them makes what the new manifest needs, its credentials among it. + if err := s.compose(ctx, nodes); err != nil { + return said, err + } + // Then the bus's user list, which carries the module's permissions at its new fingerprint. + holder, err := s.busHolder(ctx) + if err != nil { + return said, err + } + if holder != "" && !inList(nodes, holder) { + if err := s.compose(ctx, []string{holder}); err != nil { + return said, err + } + } + // Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7). + if err := s.grants(ctx); err != nil { + return said, err + } + if said == "" { + said = fmt.Sprintf("%s at %s", module, short(trunk.FP2)) + } + return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil +} + +func inList(list []string, s string) bool { + for _, v := range list { + if v == s { + return true + } + } + return false +} + +// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module. +var doneKeysMu sync.Map + +// takeDoneKey is a woken module's take-in on the live stores and buckets, said. +func takeDoneKey(ctx context.Context, open *stores, module string) { + mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{}) + mu.(*sync.Mutex).Lock() + defer mu.(*sync.Mutex).Unlock() + keys, closeKeys, err := openBuilderKeys() + if err != nil { + fmt.Printf("done keys: %s: the builder's keys cannot be read: %v\n", module, err) + return + } + defer closeKeys() + said, err := liveDoneKeySteps(open, keys).take(ctx, module) + switch { + case errors.Is(err, errKeyWaits): + // Said by nobody: the key's next write wakes this again. + case err != nil: + fmt.Printf("done keys: %s: %v\n", module, err) + case said != "": + fmt.Printf("done keys: %s\n", said) + } +} + +// liveDoneKeySteps are the steps on the controller's own stores and the bus. +func liveDoneKeySteps(open *stores, keys builderkeys.Reader) doneKeySteps { + inv := open.inventory + return doneKeySteps{ + held: func(ctx context.Context, module string) (bool, error) { + _, err := inv.SourceOf(ctx, module) + if errors.Is(err, inventory.ErrNoSuchModule) { + return false, nil + } + return err == nil, err + }, + trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) { + return keys.ModuleAt(ctx, module, "") + }, + registeredAt: func(ctx context.Context, module, commit string) (bool, error) { + src, err := inv.SourceOf(ctx, module) + if err != nil { + return false, err + } + return sameCommitID(src.BuiltFrom, commit), nil + }, + takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error { + _, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}}) + if errors.Is(err, inventory.ErrSuperseded) { + return nil + } + return err + }, + nodesOf: func(ctx context.Context, module string) ([]string, error) { + entries, err := inv.Catalogued(ctx) + if err != nil { + return nil, err + } + for _, e := range entries { + if e.Manifest.Module == module { + return e.On, nil + } + } + return nil, nil + }, + busHolder: func(ctx context.Context) (string, error) { + holders, err := seatHolders(ctx, inv) + if err != nil { + return "", err + } + return holders[catalogue.BrokerSeat].Node, nil + }, + compose: func(ctx context.Context, nodes []string) error { + _, err := composeModules(ctx, open, nodes) + return err + }, + grants: func(ctx context.Context) error { + bus, err := aBus() + if err != nil { + return err + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return err + } + grantsIssued(ctx, inv, js, "done keys: ") + return nil + }, + } +} + +// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx +// ends: never on the watch's own goroutine. +func watchDoneKeys(ctx context.Context, open *stores) { + keys, closeKeys, err := openBuilderKeys() + if err != nil { + fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) + return + } + if err := builderkeys.Watch(ctx, keys.JS, func(module string) { + inBackground(func() { takeDoneKey(ctx, open, module) }) + }); err != nil { + closeKeys() + fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) + return + } + go func() { + <-ctx.Done() + closeKeys() + }() +} diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go new file mode 100644 index 00000000..83e7f575 --- /dev/null +++ b/cmd/mesh-controller/composed_keys_test.go @@ -0,0 +1,116 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/novox/mesh-controller/internal/builderkeys" +) + +// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to +// nodes, the bus held on holder; every act is written to acts in order. +func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps { + return doneKeySteps{ + held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil }, + trunk: func(context.Context, string) (sdkdelivery.Module, error) { + if fp == "" { + return sdkdelivery.Module{}, builderkeys.ErrNoValue + } + return sdkdelivery.Module{Commit: commit, FP2: fp}, nil + }, + registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil }, + takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error { + *acts = append(*acts, "take in "+module+" at "+trunk.Commit) + return nil + }, + nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil }, + busHolder: func(context.Context) (string, error) { return holder, nil }, + compose: func(_ context.Context, ns []string) error { + *acts = append(*acts, "compose "+strings.Join(ns, ",")) + return nil + }, + grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil }, + } +} + +// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are +// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist +// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is +// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left +// alone. +func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) { + fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) + var acts []string + said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail") + if err != nil { + t.Fatal(err) + } + if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" { + t.Errorf("the acts were %q", got) + } + if !strings.Contains(said, "registered from the builder's done key") { + t.Errorf("said %q", said) + } + + acts = nil + if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil { + t.Fatal(err) + } + if got := strings.Join(acts, "; "); got != "compose novox; grants" { + t.Errorf("a module registered at its trunk already: %q", got) + } + + for name, s := range map[string]doneKeySteps{ + "assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"), + "no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"), + } { + acts = nil + if _, err := s.take(context.Background(), "mesh-mail"); err != nil { + t.Fatal(err) + } + if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" { + t.Errorf("%s: %v", name, acts) + } + if name == "no trunk value" && len(acts) != 0 { + t.Errorf("%s: %v", name, acts) + } + } + acts = nil + if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 { + t.Errorf("a module the catalogue does not hold: %v %v", acts, err) + } +} + +// 1945/AC4: the bus's user list is a file of the bus module's composition (ADR 0539 clause 7): a change of the composed +// users is a new revision of that composition alone, every other composition of the node keeping its digest. +func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testing.T) { + body, fps := capturedBody(t, "g14") + fps["nats"] = sdkdelivery.SourcePrefix2 + strings.Repeat("e", 64) + withUsers := func(users string) map[string]sdkdelivery.Composed { + var d map[string]any + if err := json.Unmarshal(body, &d); err != nil { + t.Fatal(err) + } + // As the composer writes it for the module holding mesh-broker (mesh-sdk catalogue: BusUsersID, owned by the module). + d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file", + "module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"}) + raw, _ := json.Marshal(d) + comps, _, err := compositionsOf("g14", raw, fps, nil) + if err != nil { + t.Fatal(err) + } + return comps + } + before, after := withUsers("authorization { users = [ {user: a} ] }\n"), withUsers("authorization { users = [ {user: a}, {user: b} ] }\n") + natsKey, _ := sdkdelivery.ComposedKey("g14", "nats", fps["nats"]) + for key, c := range before { + d1, _ := sdkdelivery.ComposedDigestOf(c) + d2, _ := sdkdelivery.ComposedDigestOf(after[key]) + if changed := d1 != d2; changed != (key == natsKey) { + t.Errorf("%s changed %v; only the bus module's composition changes with its user list", key, changed) + } + } +} diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go new file mode 100644 index 00000000..4e4947e3 --- /dev/null +++ b/cmd/mesh-controller/composed_start.go @@ -0,0 +1,228 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/link" +) + +// What the controller does at its start under the lease (novox/hq ADR 0539 clauses 10 and 13): +// +// - it deletes the buckets of the plans' state this release deleted, each said once in its log and as its event +// `bucket-deleted`: a later start finds none and says nothing; +// - it composes every node, so every assigned module has its composition at the fingerprint it is registered at +// and each node its own (ADR 0539 clause 11: the new controller writes every composition at its start); +// - and it composes, on the control-node, the controller it replaced at that controller's own fingerprint, from that +// build's manifest, so the operator's way back is one key at the terminal (clause 10). + +// planStateBucketDeleter is what deleting the plans' buckets needs of the bus: jetstream.JetStream, or a test's. +type planStateBucketDeleter interface { + DeleteKeyValue(ctx context.Context, bucket string) error +} + +// deletePlanStateBuckets deletes each bucket of the plans' state the bus still holds and says each it deleted; it +// answers those it deleted. A bucket not there is no error and is not said. +func deletePlanStateBuckets(ctx context.Context, js planStateBucketDeleter, say func(bucket string)) ([]string, error) { + var deleted []string + var errs []error + for _, bucket := range broker.PlanStateBuckets { + err := js.DeleteKeyValue(ctx, bucket) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound), errors.Is(err, jetstream.ErrStreamNotFound): + continue + case err != nil: + errs = append(errs, fmt.Errorf("deleting %s: %w", bucket, err)) + continue + } + deleted = append(deleted, bucket) + say(bucket) + } + return deleted, errors.Join(errs...) +} + +// BucketDeleted is the body of the controller's event `bucket-deleted`. +type BucketDeleted struct { + Bucket string `json:"bucket"` + Why string `json:"why"` + At time.Time `json:"at"` +} + +// sayBucketDeleted says a deleted bucket in the log and as the controller's event. +func sayBucketDeleted(ctx context.Context, bucket string) { + why := "a bucket of the plans' state, which release 6 deleted (novox/hq ADR 0539 clause 13)" + fmt.Printf("start: deleted %s: %s\n", bucket, why) + bus := checkEvents + if bus == nil { + return + } + body, err := json.Marshal(BucketDeleted{Bucket: bucket, Why: why, At: time.Now().UTC()}) + if err != nil { + return + } + saying, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer stop() + if err := bus.PublishSeatEvent(saying, link.MeshControllerSeat, "bucket-deleted", body); err != nil { + fmt.Printf("start: that %s was deleted could not be said as the event bucket-deleted: %v\n", bucket, err) + } +} + +// composeAtStart is the start's pass: the plans' buckets deleted, every node composed, the replaced controller +// composed. Said, never fatal: what is not composed now is composed by the next act that touches it. +func composeAtStart(ctx context.Context, open *stores) { + if bus, err := aBus(); err != nil { + fmt.Printf("start: the plans' buckets are not looked at: %v\n", err) + } else { + if js, err := jetstream.New(bus.Conn()); err == nil { + if _, err := deletePlanStateBuckets(ctx, js, func(b string) { sayBucketDeleted(ctx, b) }); err != nil { + fmt.Printf("start: %v\n", err) + } + } + bus.Close() + } + nodes, err := open.inventory.Nodes(ctx) + if err != nil { + fmt.Printf("start: nothing is composed: %v\n", err) + return + } + names := make([]string, 0, len(nodes)) + for _, n := range nodes { + names = append(names, n.Name) + } + sort.Strings(names) + if _, err := composeModules(ctx, open, names); err != nil { + fmt.Printf("start: %v\n", err) + } + if err := composeReplacedController(ctx, open); err != nil { + fmt.Printf("start: the controller this one replaced is not composed: %v\n", err) + } +} + +// replacedController is the controller a control-node ran before this one: the build the mesh last sent it and that +// build's fingerprint and manifest. +type replacedController struct { + node, fp string + manifest catalogue.Manifest +} + +// replacedControllers are, for each node the controller is assigned to, the controller build it was last sent when its +// fingerprint is of the new form and not the one the controller is registered at now. +func replacedControllers(ctx context.Context, open *stores) ([]replacedController, error) { + inv := open.inventory + fps, err := currentFingerprints(ctx, inv) + if err != nil { + return nil, err + } + entries, err := inv.Catalogued(ctx) + if err != nil { + return nil, err + } + var on []string + for _, e := range entries { + if e.Manifest.Module == controllerModule { + on = e.On + } + } + var out []replacedController + for _, node := range on { + sent, known, err := inv.SentBuilds(ctx, node) + if err != nil || !known || sent[controllerModule] == "" { + continue + } + build, found, err := inv.BuildByID(ctx, sent[controllerModule]) + if err != nil || !found || build.SourceFingerprint == fps[controllerModule] || + !strings.HasPrefix(build.SourceFingerprint, sdkdelivery.SourcePrefix2) || len(build.Manifest) == 0 { + continue + } + var m catalogue.Manifest + if err := json.Unmarshal(build.Manifest, &m); err != nil { + return nil, fmt.Errorf("the manifest of %s's build %s: %w", controllerModule, build.ID, err) + } + out = append(out, replacedController{node: node, fp: build.SourceFingerprint, manifest: m}) + } + return out, nil +} + +// composeReplacedController writes, on each control-node, the controller's composition at the fingerprint of the one +// it replaced: the node composed with that build's manifest in place of the controller's registered one, in the +// reading mode (nothing made), and only the controller's composition of it kept. +func composeReplacedController(ctx context.Context, open *stores) error { + if _, err := theLease.epoch(ctx); err != nil { + return err + } + replaced, err := replacedControllers(ctx, open) + if err != nil || len(replaced) == 0 { + return err + } + bus, err := aBus() + if err != nil { + return err + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return err + } + sign, err := controllerSigner(ctx, open) + if err != nil { + return err + } + var errs []error + for _, r := range replaced { + in, err := inputFor(ctx, open) + if err != nil { + return err + } + c, err := replacedComposition(in.Input, r) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", r.node, err)) + continue + } + at, err := openComposedPlace(ctx, js, bus.Conn(), r.node) + if err != nil { + errs = append(errs, err) + continue + } + at.sign = sign + key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp) + if wrote, err := writeComposed(ctx, at, key, c); err != nil { + errs = append(errs, err) + } else if wrote { + fmt.Printf("start: composed %s, the controller this one replaced, on %s: the way back is want.%s.%s naming it\n", + short(r.fp), r.node, r.node, controllerModule) + } + } + return errors.Join(errs...) +} + +// replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest +// in place of the controller's registered one. +func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) { + shelf := make(map[string]catalogue.Manifest, len(in.Catalogue)) + for k, v := range in.Catalogue { + shelf[k] = v + } + shelf[controllerModule] = r.manifest + in.Catalogue = shelf + all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading) + if err != nil { + return sdkdelivery.Composed{}, err + } + key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp) + c, ok := all[key] + if !ok { + return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule) + } + return c, nil +} diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go new file mode 100644 index 00000000..2e3393d6 --- /dev/null +++ b/cmd/mesh-controller/composed_start_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// fakeBuckets is the bus's buckets by name, for deleting. +type fakeBuckets map[string]bool + +func (f fakeBuckets) DeleteKeyValue(_ context.Context, bucket string) error { + if !f[bucket] { + return jetstream.ErrBucketNotFound + } + delete(f, bucket) + return nil +} + +// 1945/AC14 (unit half): the first start deletes the merge window's, the silent sends' and the building plans' buckets +// and says each once; a later start finds none and says nothing; a bucket of anything else is left alone. +func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) { + bus := fakeBuckets{"mesh-controller_merge-window": true, "mesh-controller_rollout-silent": true, + "mesh-controller_plans-building": true, "mesh-controller_grants": true} + var said []string + deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }) + if err != nil || len(deleted) != 3 || strings.Join(said, " ") != strings.Join(broker.PlanStateBuckets, " ") { + t.Fatalf("first start: deleted %v, said %v, %v", deleted, said, err) + } + if !bus["mesh-controller_grants"] || len(bus) != 1 { + t.Errorf("the bus holds %v after the first start", bus) + } + said = nil + if deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }); err != nil || len(deleted)+len(said) != 0 { + t.Errorf("a later start deleted %v and said %v (%v)", deleted, said, err) + } +} + +// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its +// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares. +func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) { + now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{ + {"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}} + before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{ + {"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}} + in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now}, + Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}} + fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64) + c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before}) + if err != nil { + t.Fatal(err) + } + if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 { + t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources)) + } + var f struct{ Content string } + _ = json.Unmarshal(c.Resources[0], &f) + if f.Content != "five\n" { + t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content) + } + if in.Catalogue[controllerModule].Version != "6" { + t.Errorf("composing the replaced controller changed the Input's catalogue") + } +} diff --git a/cmd/mesh-controller/new_modules.go b/cmd/mesh-controller/new_modules.go index 77a26108..e00b58d7 100644 --- a/cmd/mesh-controller/new_modules.go +++ b/cmd/mesh-controller/new_modules.go @@ -203,6 +203,11 @@ func watchNewModules(ctx context.Context, open *stores) { inBackground(func() { takeNewModules(ctx, open) }) + // And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq + // ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node + // composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted. + inBackground(func() { composeAtStart(ctx, open) }) + watchDoneKeys(ctx, open) keys, closeKeys, err := openBuilderKeys() if err != nil { fmt.Printf("new modules: the builder's keys are not watched: %v\n", err) diff --git a/go.mod b/go.mod index 792c08ac..e59dab93 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-controller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144241-e64486b0e6ab + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024150354-5da87a0dda89 github.com/jackc/pgx/v5 v5.10.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/go.sum b/go.sum index 5aa4866c..dc16e277 100644 --- a/go.sum +++ b/go.sum @@ -10,6 +10,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144045-64a502d16aac h1:jInLKz6g git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144045-64a502d16aac/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144241-e64486b0e6ab h1:D5VBezf/5LJeOphCvCr4gT3v7b52ZORoD1Q88Dl/Ajk= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144241-e64486b0e6ab/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024150354-5da87a0dda89 h1:9FENYqhynDUf7ByotHEdDjmvyFEPCYmteHjN8gRE3ao= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024150354-5da87a0dda89/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/broker/announced.go b/internal/broker/announced.go index dd0ec5f9..1c8304a9 100644 --- a/internal/broker/announced.go +++ b/internal/broker/announced.go @@ -17,10 +17,10 @@ import ( const StatesAnnouncedBucket = announce.KeptBucket // controllerReadable are the controller's states others may read: the record mirrors (the packages each node declares -// among them, which the pacman module reads, ADR 0507 §5), the chain's buckets and the grants issued, which the assigner -// reads (ADR 0528 clause 2). +// among them, which the pacman module reads, ADR 0507 §5), the chain's buckets, and the grants issued and the +// compositions' digests, which the assigner reads (ADR 0528 clause 2, ADR 0539 clause 6). var controllerReadable = []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, - MirrorDeclaredPackages, ControllerGrants} + MirrorDeclaredPackages, ControllerGrants, ControllerComposed} // 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. diff --git a/internal/broker/announced_test.go b/internal/broker/announced_test.go index f538c557..ccb4fdb0 100644 --- a/internal/broker/announced_test.go +++ b/internal/broker/announced_test.go @@ -39,16 +39,22 @@ func TestTheControllerAnnouncesWhatItKeeps(t *testing.T) { } } want := []string{ChainBuilds, MirrorDeclaredConfig, MirrorPredefined, ChainRollouts, - MirrorAssignments, MirrorDeclaredPackages, ControllerGrants} + MirrorAssignments, MirrorDeclaredPackages, ControllerGrants, ControllerComposed} 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"} { + for _, name := range []string{"lease", "calls", "states-announced"} { if !s.Keeps(name) || s.Readable(name) { t.Errorf("%s: kept %v, readable %v", name, s.Keeps(name), s.Readable(name)) } } + // The buckets of plan state release 6 deleted are no longer kept (ADR 0539 clause 13). + for _, name := range []string{"merge-window", "rollout-silent", "plans-building"} { + if s.Keeps(name) { + t.Errorf("%s is still kept", name) + } + } perms, err := PermissionsFor(Principal{Kind: KindController}) if err != nil { t.Fatal(err) diff --git a/internal/broker/composed.go b/internal/broker/composed.go index 1c3fbeb2..16dc1a37 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -14,6 +14,10 @@ import ( // assigner reads. The controller writes both, by compare-and-set while it holds the lease; each keeps five revisions // of a key. The names and layouts are mesh-sdk's (delivery.ComposedBucketFor, ControllerComposedBucket, ComposedKey). +// ControllerComposed is the local name of the compositions' digests, as a reader names it in `reads` after +// `mesh-controller.`: delivery.ControllerComposedBucket. +const ControllerComposed = "composed" + // ComposedValueBytes is the most a composition may hold: a file over 64 KiB travels by digest, so a module's // composition is far smaller; just under the server's largest message, which also carries the subject and headers. const ComposedValueBytes = 1<<20 - 8<<10 @@ -113,3 +117,7 @@ func NodeAgentAsks(node string) ([]string, error) { } return out, nil } + +// NodeAgentHears are the events a node's node-engine subscribes while it applies a want: the backup holder's word that a +// backup is taken (`backed-up`), from whichever module holds node-backup on its machine (restic today). +var NodeAgentHears = []string{"mesh.mod.*.event.backed-up"} diff --git a/internal/broker/composed_test.go b/internal/broker/composed_test.go index 8b3569bd..3ddb231b 100644 --- a/internal/broker/composed_test.go +++ b/internal/broker/composed_test.go @@ -65,6 +65,9 @@ func TestANodeAgentAsksItsOwnBackupHolder(t *testing.T) { t.Errorf("a node-engine of ace may not publish %s", want) } } + if !strings.Contains(strings.Join(p.Subscribe, " "), "mesh.mod.*.event.backed-up") { + t.Errorf("a node-engine of ace does not hear a backup taken: %v", p.Subscribe) + } if strings.Contains(all, "node-backup.tool.now.*") || strings.Contains(all, "KV_composed-g14") { t.Errorf("a node-engine of ace reaches another machine: %v", p.Publish) } diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index f7dd0a2b..13a2966a 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -39,17 +39,6 @@ var ( // each ask by its id, its options and the actions they stand for, how it ended and whether the // controller acted on its warrant — so a restart neither asks twice nor acts twice. AskedBucket = BucketName(ControllerSeat, "asked") - // MergeWindowBucket holds the clock of every open batch's merge window (novox/hq ADR 0407 decision 1): per - // batch a key `.quiet`, made again by each merge into it and living the window, and a key - // `.most`, made once when it opens and living the window's most. The first to expire closes the - // batch, and the controller's watcher hears the delete marker. Written by the lease holder alone. - MergeWindowBucket = BucketName(ControllerSeat, "merge-window") - // RolloutSilentBucket holds the wait of every machine a rollout has sent a tier to (novox/hq ADR 0406 §5, issue - // 974): per send a key `..` that lives ApplySilentAfter, made when the controller sends that - // machine and deleted by the machine's report on that send. Only a missing report lets it expire, and the - // controller's watcher hearing the expiry ends the wait and names the machine silent. Written by the lease - // holder alone. - RolloutSilentBucket = BucketName(ControllerSeat, "rollout-silent") // DeliveryMergesBucket holds the merges mesh-delivery announced it made, each with the ask whose Approve it merged // on (novox/hq ADR 0420 §7, issue 1025): per merge a key `///` holding the // announcement, kept DeliveryMergesKeptFor. What lets the build of a moved module register it from its new @@ -152,17 +141,11 @@ const ( MirrorMarkerTTL = 24 * time.Hour ) -// MergeWindowMarkerTTL is how long the marker an expired window key leaves is kept: long enough for a watcher -// that is up to hear it, short enough to leave nothing behind. -const MergeWindowMarkerTTL = time.Minute - -// RolloutSilentMarkerTTL is how long the marker an expired silent key leaves is kept, for the same reason. -const RolloutSilentMarkerTTL = time.Minute - -// RolloutSilentKeptFor is the age of the silent keys' bucket: every report deletes a key and the delete leaves a marker, which -// with one value per key would stay for ever and fill the bucket. A key lives a quarter of an hour at most, so an hour -// takes nothing live. -const RolloutSilentKeptFor = time.Hour +// PlanStateBuckets are the buckets of the plans' state that release 6 deletes (novox/hq ADR 0539 clause 13): the merge +// window's clock, the silent sends' waits and the building plans' hold. The controller deletes each at its first start +// and says it once (composed_start.go); none is asserted, granted or written any more. +var PlanStateBuckets = []string{BucketName(ControllerSeat, "merge-window"), BucketName(ControllerSeat, "rollout-silent"), + BucketName(ControllerSeat, "plans-building")} // AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own. const AskedKeptFor = 30 * 24 * time.Hour @@ -193,17 +176,17 @@ func IsControllerBucket(bucket string) bool { bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || bucket == DeclaredPackagesBucket || - bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket || bucket == delivery.HaveBucket || bucket == StatesAnnouncedBucket || - bucket == GrantsBucket + bucket == GrantsBucket || bucket == delivery.ControllerComposedBucket } // 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, DeclaredPackagesBucket, MergeWindowBucket, RolloutSilentBucket, - DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket, StatesAnnouncedBucket, GrantsBucket} + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, DeclaredPackagesBucket, + DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket, StatesAnnouncedBucket, GrantsBucket, + delivery.ControllerComposedBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -300,36 +283,9 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err) } - // **The merge window: keys that expire.** A key's own time to live is the window; the marker an expired key - // leaves is what a watcher hears, kept a minute (ADR 0407 decision 1). - if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: MergeWindowBucket, - Description: "the clock of every open batch's merge window (novox/hq ADR 0407): `.quiet` lives the " + - "window and is made again by each merge, `.most` lives the most; the first to expire closes the " + - "batch; written by the controller alone", - History: 1, - MaxValueSize: 4 << 10, - MaxBytes: 8 << 20, - Storage: jetstream.FileStorage, - LimitMarkerTTL: MergeWindowMarkerTTL, - }); err != nil { - return fmt.Errorf("asserting bucket %s: %w", MergeWindowBucket, err) - } - // **The rollout's silent keys: keys that expire** (ADR 0406 §5, issue 974). A key lives as long as the machine - // may stay silent; the machine's report deletes it, and a marker of expiry is what the watcher hears. - if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ - Bucket: RolloutSilentBucket, - Description: "the wait of every machine a rollout sent a tier to (novox/hq ADR 0406): `..` " + - "lives the time a machine may stay silent and is deleted by its report; one that expires names the " + - "machine silent; written by the controller alone", - History: 1, - TTL: RolloutSilentKeptFor, - MaxValueSize: 4 << 10, - MaxBytes: 8 << 20, - Storage: jetstream.FileStorage, - LimitMarkerTTL: RolloutSilentMarkerTTL, - }); err != nil { - return fmt.Errorf("asserting bucket %s: %w", RolloutSilentBucket, err) + // **The compositions' digests** (ADR 0539 clauses 2 and 6): what the assigner reads before it writes a want. + if _, err := js.CreateOrUpdateKeyValue(ctx, ComposedDigestsConfig()); err != nil { + return fmt.Errorf("asserting bucket %s: %w", delivery.ControllerComposedBucket, 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{ diff --git a/internal/broker/controller_buckets_integration_test.go b/internal/broker/controller_buckets_integration_test.go index 9f6e8d5d..bf8a1ca4 100644 --- a/internal/broker/controller_buckets_integration_test.go +++ b/internal/broker/controller_buckets_integration_test.go @@ -4,7 +4,6 @@ package broker import ( "context" - "slices" "testing" "time" @@ -42,152 +41,6 @@ func TestWhatTheControllerAskedIsBounded(t *testing.T) { } } -func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) { - controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) - if err != nil { - t.Fatal(err) - } - if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") { - t.Error("the controller may not write the merge window's bucket") - } - if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) { - t.Error("the merge window's bucket is not asserted as the controller's own") - } - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - kv, err := api.KeyValue(ctx, MergeWindowBucket) - if err != nil { - t.Fatal(err) - } - w, err := kv.WatchAll(ctx) - if err != nil { - t.Fatal(err) - } - defer w.Stop() - if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { - t.Fatal(err) - } - for { - select { - case e := <-w.Updates(): - if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut { - return - } - case <-ctx.Done(): - t.Fatal("no watcher heard the key expire") - } - } -} - -func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T) { - controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) - if err != nil { - t.Fatal(err) - } - if !slices.Contains(controller.Publish, "$KV."+RolloutSilentBucket+".>") { - t.Error("the controller may not write the rollout's silent keys' bucket") - } - if !IsControllerBucket(RolloutSilentBucket) || !slices.Contains(ControllerBuckets(), RolloutSilentBucket) { - t.Error("the rollout's silent keys' bucket is not asserted as the controller's own") - } - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - kv, err := api.KeyValue(ctx, RolloutSilentBucket) - if err != nil { - t.Fatal(err) - } - w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) - if err != nil { - t.Fatal(err) - } - defer w.Stop() - if _, err := kv.Create(ctx, "plan-1.0.ace", []byte("x"), jetstream.KeyTTL(time.Hour)); err != nil { - t.Fatal(err) - } - if err := kv.Delete(ctx, "plan-1.0.ace"); err != nil { - t.Fatal(err) - } - if _, err := kv.Create(ctx, "plan-1.0.g14", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { - t.Fatal(err) - } - for { - select { - case e := <-w.Updates(): - if e == nil || e.Operation() == jetstream.KeyValuePut { - continue - } - if e.Key() == "plan-1.0.ace" && e.Operation() != jetstream.KeyValueDelete { - t.Fatalf("a report's delete was heard as %v", e.Operation()) - } - if e.Key() == "plan-1.0.g14" { - if e.Operation() != jetstream.KeyValuePurge { - t.Fatalf("an expiry was heard as %v, not a purge", e.Operation()) - } - return - } - case <-ctx.Done(): - t.Fatal("no watcher heard the key expire") - } - } -} - -func TestTheSilentKeysBucketAgesAndAnOlderOneIsBroughtToIt(t *testing.T) { - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - // As the first build made it: no age. - if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: RolloutSilentBucket, History: 1, - MaxValueSize: 4 << 10, MaxBytes: 8 << 20, Storage: jetstream.FileStorage, LimitMarkerTTL: RolloutSilentMarkerTTL}); err != nil { - t.Fatal(err) - } - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatalf("a bucket made without the age is not brought to it: %v", err) - } - kv, err := api.KeyValue(ctx, RolloutSilentBucket) - if err != nil { - t.Fatal(err) - } - status, err := kv.Status(ctx) - if err != nil { - t.Fatal(err) - } - if status.TTL() != RolloutSilentKeptFor { - t.Errorf("the bucket ages at %v; want %v", status.TTL(), RolloutSilentKeptFor) - } -} - func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) { js, err := Dial(testbus.URL(t)) if err != nil { diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 799a46d7..a492d856 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -115,7 +115,8 @@ func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) { module := Principal{Kind: KindModule, Module: ControllerSeat, Node: "one"} rowed := map[string]bool{} for _, row := range WritersTable { - if !strings.HasSuffix(row.State, " mirror") && !strings.HasSuffix(row.State, " of the chain") && row.State != "the grants issued" { + if !strings.HasSuffix(row.State, " mirror") && !strings.HasSuffix(row.State, " of the chain") && row.State != "the grants issued" && + row.State != "the compositions and their digests" { continue } for _, subject := range row.Subjects { diff --git a/internal/broker/log_test.go b/internal/broker/log_test.go index b1988cdd..ca618ca6 100644 --- a/internal/broker/log_test.go +++ b/internal/broker/log_test.go @@ -171,8 +171,8 @@ func TestRaisingLogsReportsWhatNothingDeclares(t *testing.T) { // ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket // delete, or purges a stream, but for the controller lease's own stream, which purges its own history // below a sequence (internal/lease), and the controller's record mirrors, which purge the key of a record -// the store no longer holds (internal/recordmirror; novox/hq ADR 0289, design 54 §2). A new call is a decision, -// not a refactor. +// the store no longer holds (internal/recordmirror; novox/hq ADR 0289, design 54 §2), and the buckets of the plans' +// state the controller deletes once at its first start (ADR 0539 clause 13). A new call is a decision, not a refactor. func TestNoPathDeletesALogOrABucket(t *testing.T) { removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true, "DeleteObjectStore": true} @@ -199,7 +199,10 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) { name := sel.Sel.Name decided := strings.Contains(filepath.ToSlash(path), "internal/lease/") || strings.Contains(filepath.ToSlash(path), "internal/recordmirror/") - if removers[name] || (name == "Purge" && !decided) { + // And the buckets of the plans' state release 6 deleted, at the controller's first start, each said (novox/hq + // ADR 0539 clause 13): the one decided deletion of a bucket. + release6 := strings.HasSuffix(filepath.ToSlash(path), "cmd/mesh-controller/composed_start.go") && name == "DeleteKeyValue" + if (removers[name] && !release6) || (name == "Purge" && !decided) { found = append(found, path+": "+name) } return true diff --git a/internal/broker/nats.go b/internal/broker/nats.go index af7d26a0..3ff528fd 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -396,11 +396,6 @@ var DegradableGrants = []DegradableGrant{ Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at " + "its first push, which is when it acknowledges a merge (ADR 0407 decision 3, issue 973)", }, - { - Subject: "$KV." + MergeWindowBucket + ".>", - Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " + - "window setting says (ADR 0407 decision 1, issue 973)", - }, { Subject: "$KV." + DeclaredPackagesBucket + ".>", Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + @@ -443,11 +438,6 @@ var DegradableGrants = []DegradableGrant{ Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + "first push, and writes the chain only after that, holding the lease (ADR 0408 §2 and §8, issue 1089)", }, - { - Subject: "$KV." + RolloutSilentBucket + ".>", - Reason: "refused: the controller looks at the plans when a wait for a machine's report is up, by a timer of its own " + - "for that wait (ADR 0406 §5, issue 974)", - }, } // perMachineEvents are a node-scoped seat's events about the holder itself, whose last token is the @@ -750,8 +740,6 @@ func PermissionsFor(p Principal) (Permissions, error) { } pub = append(pub, composed) } - // And the compositions' digests, which the assigner reads (ADR 0539 clauses 2 and 6). - pub = append(pub, "$KV."+delivery.ControllerComposedBucket+".>") // **And what the bus says about itself, read-only** (novox/hq to-be 45 §3, S9): a durable // consumer that gave up on a message, or one that was deleted. The server already publishes // both in the mesh's own account; the controller says each as a condition in the mesh's words. @@ -871,6 +859,9 @@ func PermissionsFor(p Principal) (Permissions, error) { return Permissions{}, err } pub = append(pub, backups...) + // And the backup holder's word that it backed the module up (ADR 0539 clause 16), whichever module holds + // node-backup: one event, by its name. + sub = append(sub, NodeAgentHears...) // And, while the vault serves it, the vault's move-in in its own name and nobody else's (issue 676): the // values sealed to this machine, which only it opens, handed in sealed to the vault's key for the call. if p.MovesIntoTheVault { diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 90941656..e0511377 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.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$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.artifacts-gone", "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"] } + 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_composed.>", "$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_predefined.>", "$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.artifacts-gone", "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" } } } @@ -39,7 +39,7 @@ accounts { } } { user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: { publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.CREATE.KV_composed-one.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_wants.*.$KV.mesh-delivery-assigner_wants.want.one.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.one.>", "$JS.API.CONSUMER.CREATE.OBJ_declaration-files-one.*.$O.declaration-files-one.C.*", "$JS.API.CONSUMER.DELETE.KV_composed-one.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_wants.>", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files-one.*", "$JS.API.CONSUMER.INFO.NODES.one", "$JS.API.DIRECT.GET.KV_composed-one.>", "$JS.API.DIRECT.GET.KV_mesh-delivery-assigner_wants.$KV.mesh-delivery-assigner_wants.want.one.>", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.one.>", "$JS.API.DIRECT.GET.OBJ_declaration-files-one.$O.declaration-files-one.M.*", "$JS.API.STREAM.INFO.KV_composed-one", "$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_wants", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_composed-one.>", "$JS.FC.KV_mesh-delivery-assigner_wants.>", "$JS.FC.KV_node-engine_have.>", "$JS.FC.OBJ_declaration-files-one.>", "$KV.node-engine_have.one.>", "$SRV.PING.node-tools.one", "mesh.control.one.>", "mesh.seat.node-backup.call.backed-up.one.node~one", "mesh.seat.node-backup.call.now.one.node~one", "mesh.seat.node-backup.tool.backed-up.one", "mesh.seat.node-backup.tool.now.one"] } - subscribe: { allow: ["_DELIVER.one", "_DELIVER.one.>", "_INBOX.node.one.>", "mesh.node.one.ask.hand-over", "mesh.node.one.ask.judge-directories", "mesh.node.one.ask.move-secret", "mesh.node.one.ask.report", "mesh.node.one.ask.setuid-search", "mesh.node.one.declare"] } + subscribe: { allow: ["_DELIVER.one", "_DELIVER.one.>", "_INBOX.node.one.>", "mesh.mod.*.event.backed-up", "mesh.node.one.ask.hand-over", "mesh.node.one.ask.judge-directories", "mesh.node.one.ask.move-secret", "mesh.node.one.ask.report", "mesh.node.one.ask.setuid-search", "mesh.node.one.declare"] } allow_responses: { max: 1, ttl: "1m" } } } { user: "one.mesh-delivery-watchman", password: "$2a$11$wwwwwwwwwwwwwwwwwwwwww", permissions: { diff --git a/internal/broker/writers.go b/internal/broker/writers.go index b904ed1e..ec5342b7 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -142,16 +142,16 @@ var WritersTable = []WriterRow{ Subjects: []string{"mesh.control.*.report", "mesh.control.*.health", "mesh.control.*.cli"}, Writes: ownMachine}, {State: "the controller lease", Writer: "the controller instance holding it", KeptIn: "key-value " + LeaseBucket, Others: "a candidate waits", Subjects: kvOf(LeaseBucket), Writes: isController}, - // And the clock of an open batch's merge window, a plan before it is one (novox/hq ADR 0407 decision 1), and the - // wait of a rollout for each machine it sent (ADR 0406 §5, issue 974): keys that expire in buckets of their own, - // written by the lease holder. - {State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision", - KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket + - ", a rollout's wait for a machine's report in key-value " + RolloutSilentBucket + - ", and the merges mesh-delivery announced, for a moved module's registration (ADR 0420 §7), in key-value " + - DeliveryMergesBucket, Others: "read through plans", - Subjects: append(append(kvOf(MergeWindowBucket), kvOf(RolloutSilentBucket)...), kvOf(DeliveryMergesBucket)...), - Writes: isController}, + // The merges mesh-delivery announced, for a moved module's registration (ADR 0420 §7), written by the lease holder. + {State: "the merges mesh-delivery announced", Writer: "controller (lease holder)", + KeptIn: "key-value " + DeliveryMergesBucket, Others: "read at a moved module's registration", + Subjects: kvOf(DeliveryMergesBucket), Writes: isController}, + // And the compositions (novox/hq ADR 0539 clauses 2 and 4): each node's in its own bucket, which only that node's + // node-engine reads, and each digest with the judgement of its move, which the assigner reads. + {State: "the compositions and their digests", Writer: "controller (lease holder), compare-and-set", + KeptIn: "key-value composed- for each node, and " + delivery.ControllerComposedBucket, + Others: "a node-engine reads its own node's; the assigner reads the digests", + Subjects: kvOf(delivery.ControllerComposedBucket), Writes: isController}, {State: "conditions", Writer: "controller", KeptIn: "key-value " + ConditionsBucket + " (and its history, " + ConditionHistoryBucket + "); how often each was observed, when last and its newest evidence in the " + "controller's store, so a look that changes nothing writes no bucket", Others: "raise or clear only through observations the controller reads", diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index f1e55584..2860ed10 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -20,7 +20,10 @@ var designRows = []string{ "a judgement of a machine's directories asked of it", "a machine's applied state and its report", "the controller lease", - "plans and their tiers", + // The merges mesh-delivery announced, once part of the plans' row (ADR 0420 §7), and the compositions that replace + // the plans (ADR 0539 clauses 2 and 4). + "the merges mesh-delivery announced", + "the compositions and their digests", "conditions", "calls and their outcomes", "the hand-act log", diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 712735fb..ed17a574 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself. mesh-catalog 2246ccc492e59579a140752e6f574b2121d296af modules/*/module.json, modules/*/files, modules/*/templates, predefined.json mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock -mesh-sdk e64486b0e6ab573ee089aaaae7915406f8a5a745 conformance/events/module-event.json +mesh-sdk 5da87a0dda89f2b48b23fc0cb7765f0312f7eb92 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: