diff --git a/cmd/mesh-controller/composed_judge.go b/cmd/mesh-controller/composed_judge.go new file mode 100644 index 00000000..742c1070 --- /dev/null +++ b/cmd/mesh-controller/composed_judge.go @@ -0,0 +1,256 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "path" + "sort" + "strings" + + sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" +) + +// The judgement of a move (novox/hq ADR 0539 clauses 2 and 14, by ADR 0377 §3's tests): beside each composition's digest +// the controller writes whether moving the node from the composition it runs, as its have names it, to this one +// recreates a container, moves the module's data, or neither, or that it cannot be judged. The assigner holds a move of +// a module of policy record from it until a person allows it. +// +// ADR 0377 §3 over two compositions of one module: **recreates** when a container resource is changed, added or +// removed; **moves data** when, with no container changed, a directory or file resource is changed, added or removed +// that a container of the module mounts or that holds or lies within one of its directories (where a module keeps its +// data), when its `data` declaration changes, or, for a module that declares data, when one of its long-running +// processes changes; **neither** otherwise, and for a module the node has never run. **Cannot judge** when the +// composition the node runs is gone or cannot be read. A held composition never becomes what is judged against: with A +// running, B held and C composed, C is judged against A, the one the have names. + +// judgeMove is the value written beside a composition whose digest is digest: its judgement against what the node's have +// names for the module, and the module's policy. +func judgeMove(ctx context.Context, at composedPlace, key string, to sdkdelivery.Composed, digest string) (sdkdelivery.ComposedDigest, error) { + out := sdkdelivery.ComposedDigest{Digest: digest} + if at.policy != nil && to.Module != "" { + out.Policy, out.PolicyFrom, out.PolicyWhy = at.policy(to.Module) + } + if to.Module == "" { + // The node's own composition: no module's containers or data are judged; its want is written as any. + out.Judgement = sdkdelivery.JudgedNeither + return out, nil + } + have, found, err := at.ran(ctx, to.Module) + if err != nil { + return out, err + } + if !found || have.Phase == sdkdelivery.PhaseRemoved || (have.Composed == "" && have.FP == "") { + out.Judgement, out.Words = sdkdelivery.JudgedNeither, "the node runs no composition of "+to.Module + return out, nil + } + if have.Composed == "" { + out.Judgement = sdkdelivery.JudgedCannot + out.Words = fmt.Sprintf("%s's have on %s names no composition, only the fingerprint %s", to.Module, at.node, have.FP) + return out, nil + } + out.Against = have.Composed + if have.Composed == digest { + out.Judgement, out.Words = sdkdelivery.JudgedNeither, "the node runs this composition" + return out, nil + } + from, found, err := composedByDigest(ctx, at, to.Module, have.FP, have.Composed) + if err != nil { + return out, err + } + if !found { + out.Judgement = sdkdelivery.JudgedCannot + out.Words = fmt.Sprintf("the composition %s runs on %s (%s) is gone", to.Module, at.node, short(strings.TrimPrefix(have.Composed, "sha256:"))) + return out, nil + } + out.Judgement, out.Words = judged(from, to) + return out, nil +} + +// composedByDigest finds, in the node's bucket, the composition of module at fingerprint fp whose digest is digest: the +// key's current value or one of the revisions it keeps. +func composedByDigest(ctx context.Context, at composedPlace, module, fp, digest string) (sdkdelivery.Composed, bool, error) { + if fp == "" { + return sdkdelivery.Composed{}, false, nil + } + key, err := sdkdelivery.ComposedKey(at.node, module, fp) + if err != nil { + return sdkdelivery.Composed{}, false, nil + } + values, err := historyOf(ctx, at.bucket, key) + if err != nil { + return sdkdelivery.Composed{}, false, err + } + for _, v := range values { + c, err := sdkdelivery.DecodeComposed(v) + if err != nil { + continue + } + if d, err := sdkdelivery.ComposedDigestOf(c); err == nil && d == digest { + return c, true, nil + } + } + return sdkdelivery.Composed{}, false, nil +} + +// historyOf is every value a key keeps, the newest first; none when there is no such key. +func historyOf(ctx context.Context, kv composedBucketsKV, key string) ([][]byte, error) { + if h, ok := kv.(interface { + History(ctx context.Context, key string, opts ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) + }); ok { + entries, err := h.History(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + out := make([][]byte, 0, len(entries)) + for i := len(entries) - 1; i >= 0; i-- { + if entries[i].Operation() == jetstream.KeyValuePut { + out = append(out, entries[i].Value()) + } + } + return out, nil + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return [][]byte{e.Value()}, nil +} + +// judged applies ADR 0377 §3 to two compositions of one module. +func judged(from, to sdkdelivery.Composed) (sdkdelivery.Judgement, string) { + before, after := resourcesByID(from), resourcesByID(to) + if ids := changedOf(before, after, func(kind string) bool { return kind == string(sdkdecl.TypeContainer) }); len(ids) > 0 { + return sdkdelivery.JudgedRecreates, "it recreates " + strings.Join(ids, ", ") + } + if from.Data != to.Data { + return sdkdelivery.JudgedMovesData, "its data declaration changes" + } + places := dataPlaces(before, after) + held := func(kind string) bool { + return kind == string(sdkdecl.TypeDirectory) || kind == string(sdkdecl.TypeFile) + } + var moved []string + for _, id := range changedOf(before, after, held) { + r := before[id] + if r == nil { + r = after[id] + } + if withinPlaces(r.path, places) { + moved = append(moved, id) + } + } + if to.Data || from.Data { + moved = append(moved, changedOf(before, after, func(kind string) bool { return kind == string(sdkdecl.TypeProcess) })...) + } + if len(moved) > 0 { + sort.Strings(moved) + return sdkdelivery.JudgedMovesData, "it changes where its data is kept: " + strings.Join(moved, ", ") + } + return sdkdelivery.JudgedNeither, "it recreates no container and moves no data" +} + +// judgedResource is a resource as the judgement reads it: its kind, the path it acts on, what its containers mount, +// and its canonical bytes. +type judgedResource struct { + kind string + path string + mounts []string + raw []byte +} + +func resourcesByID(c sdkdelivery.Composed) map[string]*judgedResource { + out := map[string]*judgedResource{} + for _, raw := range c.Resources { + var head struct { + ID string `json:"id"` + Type string `json:"type"` + Path string `json:"path"` + // A container's volumes, `:[:]`. + Volumes []string `json:"volumes"` + } + if err := json.Unmarshal(raw, &head); err != nil { + continue + } + r := &judgedResource{kind: head.Type, path: head.Path, raw: compactComposed(raw)} + for _, v := range head.Volumes { + source, _, _ := strings.Cut(v, ":") + r.mounts = append(r.mounts, source) + } + out[head.ID] = r + } + return out +} + +func compactComposed(raw []byte) []byte { + var b bytes.Buffer + if json.Compact(&b, raw) != nil { + return raw + } + return b.Bytes() +} + +// changedOf are the ids of the resources of the kinds counted that are changed, added or removed, sorted. +func changedOf(before, after map[string]*judgedResource, counted func(kind string) bool) []string { + var out []string + for id, now := range after { + if !counted(now.kind) { + continue + } + if was, ok := before[id]; !ok || !bytes.Equal(was.raw, now.raw) { + out = append(out, id) + } + } + for id, was := range before { + if _, ok := after[id]; !ok && counted(was.kind) { + out = append(out, id) + } + } + sort.Strings(out) + return out +} + +// dataPlaces are the paths where the module keeps what a move must not lose: what its containers mount, and its +// directories, before and after. +func dataPlaces(before, after map[string]*judgedResource) []string { + var out []string + for _, rs := range []map[string]*judgedResource{before, after} { + for _, r := range rs { + if r.kind == string(sdkdecl.TypeContainer) { + out = append(out, r.mounts...) + } + if r.kind == string(sdkdecl.TypeDirectory) && r.path != "" { + out = append(out, r.path) + } + } + } + return out +} + +// withinPlaces says p is one of places, or lies within one, or holds one. +func withinPlaces(p string, places []string) bool { + if p == "" { + return false + } + p = path.Clean(p) + for _, place := range places { + if place == "" || !strings.HasPrefix(place, "/") { + continue + } + place = path.Clean(place) + if p == place || strings.HasPrefix(p, place+"/") || strings.HasPrefix(place, p+"/") { + return true + } + } + return false +} diff --git a/cmd/mesh-controller/composed_live.go b/cmd/mesh-controller/composed_live.go new file mode 100644 index 00000000..5ed9c1ce --- /dev/null +++ b/cmd/mesh-controller/composed_live.go @@ -0,0 +1,198 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// composeModules composes each named node and writes its compositions that differ (ADR 0539 clause 5): what each act +// that once sent a declaration calls instead. A node that cannot compose is that node's: the others are written, and +// every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that +// does not hold it writes nothing and says so. +func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) { + if _, err := theLease.epoch(ctx); err != nil { + return nil, fmt.Errorf("nothing is composed: %w", err) + } + bus, err := aBus() + if err != nil { + return nil, fmt.Errorf("nothing is composed: the bus: %w", err) + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return nil, err + } + sign, err := controllerSigner(ctx, open) + if err != nil { + return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err) + } + fps, err := currentFingerprints(ctx, open.inventory) + if err != nil { + return nil, err + } + entries, err := open.inventory.Catalogued(ctx) + if err != nil { + return nil, err + } + manifests := make(map[string]catalogue.Manifest, len(entries)) + for _, e := range entries { + manifests[e.Manifest.Module] = e.Manifest + } + upgrades, err := open.inventory.Upgrades(ctx) + if err != nil { + return nil, err + } + policy := func(module string) (string, string, string) { + u, ok := upgrades[module] + if !ok { + return "", "", "" + } + return u.Policy(), u.From, u.Why + } + sort.Strings(nodes) + out := map[string][]string{} + var errs []error + for _, node := range nodes { + at, err := openComposedPlace(ctx, js, bus.Conn(), node) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + continue + } + at.sign, at.policy = sign, policy + declared, err := composeSend(ctx, open, node) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + continue + } + body, err := declared.Body() + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + continue + } + comps, moved, err := compositionsOf(node, body, fps, manifests) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + continue + } + if len(moved) > 0 { + fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n", + node, strings.Join(moved, ", ")) + } + wrote, err := writeNodeCompositions(ctx, at, comps) + if len(wrote) > 0 { + out[node] = wrote + fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", ")) + } + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + } + } + return out, errors.Join(errs...) +} + +// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send +// of a verb or an event calls. +func composeAndSay(ctx context.Context, open *stores, nodes []string) error { + if len(nodes) == 0 { + return nil + } + _, err := composeModules(ctx, open, nodes) + if err != nil { + fmt.Printf("compositions: %v\n", err) + } + return err +} + +// openComposedPlace opens a node's buckets: its compositions, the digests, its files, and reads its want and have. +func openComposedPlace(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, node string) (composedPlace, error) { + if err := broker.EnsureComposed(ctx, js, node); err != nil { + return composedPlace{}, err + } + name, err := sdkdelivery.ComposedBucketFor(node) + if err != nil { + return composedPlace{}, err + } + bucket, err := js.KeyValue(ctx, name) + if err != nil { + return composedPlace{}, fmt.Errorf("the bucket %s: %w", name, err) + } + digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket) + if err != nil { + return composedPlace{}, fmt.Errorf("the bucket %s: %w", sdkdelivery.ControllerComposedBucket, err) + } + objects, err := openBucket(ctx, conn, node) + if err != nil { + return composedPlace{}, err + } + return composedPlace{node: node, bucket: bucket, digests: digests, objects: objects, + wanted: func(ctx context.Context, module string) (string, error) { return wantedOn(ctx, js, node, module) }, + ran: func(ctx context.Context, module string) (sdkdelivery.Have, bool, error) { + return haveOn(ctx, js, node, module) + }, + }, nil +} + +// wantedOn is the fingerprint the assigner's want names for a module on a node; "" when there is none or the bucket is +// not there yet. +func wantedOn(ctx context.Context, js jetstream.JetStream, node, module string) (string, error) { + kv, err := js.KeyValue(ctx, sdkdelivery.AssignerWantsBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return "", nil + } + if err != nil { + return "", err + } + key, err := sdkdelivery.WantKey(node, module) + if err != nil { + return "", err + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return "", nil + } + if err != nil { + return "", err + } + w, err := sdkdelivery.DecodeWant(e.Value()) + if err != nil { + return "", nil // a want of release 5, naming no composition: no fingerprint a composition must be kept for + } + return w.FP, nil +} + +// haveOn is what a node's node-engine says it runs of a module. +func haveOn(ctx context.Context, js jetstream.JetStream, node, module string) (sdkdelivery.Have, bool, error) { + kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return sdkdelivery.Have{}, false, nil + } + if err != nil { + return sdkdelivery.Have{}, false, err + } + key, err := sdkdelivery.HaveKey(node, module) + if err != nil { + return sdkdelivery.Have{}, false, err + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return sdkdelivery.Have{}, false, nil + } + if err != nil { + return sdkdelivery.Have{}, false, err + } + h, err := sdkdelivery.DecodeHave(e.Value()) + if err != nil { + return sdkdelivery.Have{}, false, nil + } + return h, true, nil +} diff --git a/cmd/mesh-controller/composed_writer.go b/cmd/mesh-controller/composed_writer.go new file mode 100644 index 00000000..e9389d1d --- /dev/null +++ b/cmd/mesh-controller/composed_writer.go @@ -0,0 +1,347 @@ +package main + +import ( + "context" + "crypto/ed25519" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" +) + +// The compositions (novox/hq ADR 0539 clauses 1 to 5). Every act that sent a declaration writes compositions instead: +// the controller composes a node as it composed a send (composeSend: the same composition, the same makes), splits the +// declaration by module into one composition per module at the fingerprint the module is registered at, and the node's +// own resources, signs each with its key, and writes those that differ, by compare-and-set while it holds the lease: +// the composition into the node's bucket `composed-`, which only that node reads, and its digest with the +// judgement of the move (composed_judge.go) into `mesh-controller_composed`, which the assigner reads. A file over 64 +// KiB goes by digest into the node's object bucket first. Nothing is published on a node's declaration subject. +// +// **A module with no fingerprint of the new form goes with the node's own composition.** The mesh composes resources +// of modules it provides itself, built by nobody (the private network's mesh-wireguard), and a module registered +// before fingerprints of the new form has none either: no want can ever name such a module (the assigner's first rule +// asks for its done build), so its resources ride in `.node`, which the node's own want names, their `module` +// taken off (a composition holds only its own module's resources); their ids still name the module. +// +// **Derived state, pruned** (clause 4): a module's compositions on a node are kept for its five newest fingerprints and +// for any a want or the node's have names; the rest are deleted with their digests. + +// composedBucketsKV is what the writer needs of a key-value bucket: jetstream.KeyValue, or a test's. +type composedBucketsKV interface { + Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) + Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) + Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) + Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error + ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error) +} + +// composedPlace is where the writer writes a node's compositions and reads what decides them: the node's bucket, the +// digests, the node's object bucket, the wants and the haves. +type composedPlace struct { + node string + bucket composedBucketsKV + digests composedBucketsKV + objects objectStore + // wanted is the fingerprint the node's want names for a module, "" when none; ran the composition digest its have + // names (the move's judgement reads it) and the fingerprint it runs. + wanted func(ctx context.Context, module string) (string, error) + ran func(ctx context.Context, module string) (sdkdelivery.Have, bool, error) + // policy is a module's upgrade policy, where it comes from and why (catalogue.UpgradeOf), carried beside the + // judgement for the assigner's hold (ADR 0539 clause 14). + policy func(module string) (policy, from, why string) + // sign signs a composition with the controller's key. + sign func(c sdkdelivery.Composed) (sdkdelivery.Composed, error) +} + +// compositionsOf splits a node's declaration body, as a send composed it, into its compositions: each module's at the +// fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names +// no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest +// declares data. +func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest) ( + out map[string]sdkdelivery.Composed, moved []string, err error) { + + var envelope struct { + Resources []map[string]json.RawMessage `json:"resources"` + Adoption *sdkdecl.Adoption `json:"adoption"` + Filled []string `json:"filled_files"` + Provisioner string `json:"provisioner_account"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + return nil, nil, fmt.Errorf("%s's declaration: %w", node, err) + } + inModule := map[string]bool{} + resources := make([]json.RawMessage, 0, len(envelope.Resources)) + for _, r := range envelope.Resources { + var module string + if raw, ok := r["module"]; ok { + _ = json.Unmarshal(raw, &module) + } + if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) { + delete(r, "module") + if !inModule[module] { + inModule[module] = true + moved = append(moved, module) + } + } + raw, err := json.Marshal(r) + if err != nil { + return nil, nil, err + } + resources = append(resources, raw) + } + // A filled file of a module moved into the node's own composition is no module's file any more: the provisioner + // fills it all the same, but a composition carries only its own module's filled paths. + var filled []string + for _, p := range envelope.Filled { + if !filledByMoved(envelope.Resources, p, inModule) { + filled = append(filled, p) + } + } + sort.Strings(moved) + out, err = sdkdelivery.SplitComposed(node, resources, fps, filled, + sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner}) + if err != nil { + return nil, nil, err + } + for key, c := range out { + if m, ok := manifests[c.Module]; ok && c.Module != "" { + c.Data = len(m.DataItems()) > 0 + out[key] = c + } + } + return out, moved, nil +} + +// filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition. +func filledByMoved(resources []map[string]json.RawMessage, path string, moved map[string]bool) bool { + for _, r := range resources { + var kind, at, id string + _ = json.Unmarshal(r["type"], &kind) + _ = json.Unmarshal(r["path"], &at) + _ = json.Unmarshal(r["id"], &id) + if kind == string(sdkdecl.TypeFile) && at == path { + if module, ok := sdkdecl.ModuleGuess(id); ok && moved[module] { + return true + } + } + } + return false +} + +// writeComposed writes one composition and its digest where they differ from what the buckets hold: the files by +// digest first, then the composition, then its digest and judgement, so a reader that finds the digest finds the +// composition and its files. It answers whether it wrote anything. +func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdelivery.Composed) (bool, error) { + c, objects, err := c.ByDigest() + if err != nil { + return false, err + } + if len(objects) > 0 { + if at.objects == nil { + return false, fmt.Errorf("%s holds a file over 64 KiB and %s's object bucket is not open", key, at.node) + } + named := make(map[string][]byte, len(objects)) + for digest, body := range objects { + name, err := sdkdecl.ObjectName(digest) + if err != nil { + return false, err + } + named[name] = body + } + if err := putObjects(ctx, at.objects, named); err != nil { + return false, err + } + } + digest, err := sdkdelivery.ComposedDigestOf(c) + if err != nil { + return false, err + } + wrote := false + held, revision, err := readComposed(ctx, at.bucket, key) + if err != nil { + return false, err + } + if heldDigest, _ := sdkdelivery.ComposedDigestOf(held); revision == 0 || heldDigest != digest { + signed, err := at.sign(c) + if err != nil { + return false, err + } + value, err := sdkdelivery.EncodeComposed(signed) + if err != nil { + return false, err + } + if err := casWrite(ctx, at.bucket, key, value, revision); err != nil { + return false, err + } + wrote = true + } + judged, err := judgeMove(ctx, at, key, c, digest) + if err != nil { + return wrote, err + } + written, err := keepDigest(ctx, at.digests, key, judged) + return wrote || written, err +} + +// readComposed is the composition a bucket holds under key and its revision; revision 0 when there is none. +func readComposed(ctx context.Context, kv composedBucketsKV, key string) (sdkdelivery.Composed, uint64, error) { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return sdkdelivery.Composed{}, 0, nil + } + if err != nil { + return sdkdelivery.Composed{}, 0, err + } + c, err := sdkdelivery.DecodeComposed(e.Value()) + if err != nil { + // A value this controller cannot read is written over: it is derived, and composed again here. + return sdkdelivery.Composed{}, e.Revision(), nil + } + return c, e.Revision(), nil +} + +// keepDigest writes the digest and judgement beside a composition when it differs from what the bucket holds. +func keepDigest(ctx context.Context, kv composedBucketsKV, key string, v sdkdelivery.ComposedDigest) (bool, error) { + value, err := sdkdelivery.EncodeComposedDigest(v) + if err != nil { + return false, err + } + v.V = sdkdelivery.Version + var revision uint64 + e, err := kv.Get(ctx, key) + switch { + case errors.Is(err, jetstream.ErrKeyNotFound): + case err != nil: + return false, err + default: + revision = e.Revision() + if held, derr := sdkdelivery.DecodeComposedDigest(e.Value()); derr == nil && sdkdelivery.SameComposedDigest(held, v) { + return false, nil + } + } + return true, casWrite(ctx, kv, key, value, revision) +} + +// casWrite creates key when revision is 0 and otherwise replaces the revision read: a second writer that wrote in +// between makes this one fail, said, and the next event composes again. +func casWrite(ctx context.Context, kv composedBucketsKV, key string, value []byte, revision uint64) error { + var err error + if revision == 0 { + _, err = kv.Create(ctx, key, value) + } else { + _, err = kv.Update(ctx, key, value, revision) + } + if err != nil { + return fmt.Errorf("writing %s: %w", key, err) + } + return nil +} + +// writeNodeCompositions writes every composition of a node that differs, and prunes what is no longer kept. It answers +// the keys it wrote. +func writeNodeCompositions(ctx context.Context, at composedPlace, comps map[string]sdkdelivery.Composed) ([]string, error) { + keys := make([]string, 0, len(comps)) + for k := range comps { + keys = append(keys, k) + } + sort.Strings(keys) + var wrote []string + var errs []error + for _, key := range keys { + w, err := writeComposed(ctx, at, key, comps[key]) + if err != nil { + errs = append(errs, err) + continue + } + if w { + wrote = append(wrote, key) + } + } + if err := pruneCompositions(ctx, at); err != nil { + errs = append(errs, err) + } + return wrote, errors.Join(errs...) +} + +// pruneCompositions deletes, for each module of the node, the compositions outside its ComposedKeep newest fingerprints +// that neither its want nor its have names, each with its digest (ADR 0539 clause 4). Newest is by when each was last +// written: the controller writes only the fingerprint a module is registered at, so an older one is never written again. +func pruneCompositions(ctx context.Context, at composedPlace) error { + lister, err := at.bucket.ListKeys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil + } + return err + } + type kept struct { + key, fp string + at time.Time + } + byModule := map[string][]kept{} + for key := range lister.Keys() { + node, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key) + if err != nil || isNode || node != at.node { + continue + } + e, err := at.bucket.Get(ctx, key) + if err != nil { + continue + } + byModule[module] = append(byModule[module], kept{key, fp, e.Created()}) + } + var errs []error + for module, ks := range byModule { + if len(ks) <= sdkdelivery.ComposedKeep { + continue + } + sort.Slice(ks, func(i, j int) bool { return ks[i].at.After(ks[j].at) }) + wanted, err := at.wanted(ctx, module) + if err != nil { + errs = append(errs, err) + continue + } + have, _, err := at.ran(ctx, module) + if err != nil { + errs = append(errs, err) + continue + } + for _, k := range ks[sdkdelivery.ComposedKeep:] { + if k.fp == wanted || k.fp == have.FP { + continue + } + if err := at.bucket.Delete(ctx, k.key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + errs = append(errs, err) + continue + } + if err := at.digests.Delete(ctx, k.key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + errs = append(errs, err) + } + } + } + return errors.Join(errs...) +} + +// controllerSigner signs a composition with the controller's active key, which its identity store keeps and never hands +// out (as it signed declarations). +func controllerSigner(ctx context.Context, open *stores) (func(sdkdelivery.Composed) (sdkdelivery.Composed, error), error) { + ident, err := open.Identity(ctx) + if err != nil { + return nil, err + } + key, err := ident.Active(ctx) + if err != nil { + return nil, err + } + public := ed25519.PublicKey(key.Public) + return func(c sdkdelivery.Composed) (sdkdelivery.Composed, error) { + return sdkdelivery.SignComposedBy(c, public, func(m []byte) ([]byte, error) { return ident.Sign(ctx, m) }) + }, nil +} diff --git a/cmd/mesh-controller/composed_writer_test.go b/cmd/mesh-controller/composed_writer_test.go new file mode 100644 index 00000000..8ffd4552 --- /dev/null +++ b/cmd/mesh-controller/composed_writer_test.go @@ -0,0 +1,337 @@ +package main + +import ( + "compress/gzip" + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/json" + "fmt" + "io" + "os" + "sort" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" +) + +// fakeComposedKV is a key-value bucket in memory: revisions, a creation time per write, history, compare-and-set. +type fakeComposedKV struct { + name string + rev uint64 + clock time.Time + history map[string][]fakeComposedEntry + writes int + deletes int +} + +type fakeComposedEntry struct { + bucket, key string + value []byte + rev uint64 + at time.Time + op jetstream.KeyValueOp +} + +func (e fakeComposedEntry) Bucket() string { return e.bucket } +func (e fakeComposedEntry) Key() string { return e.key } +func (e fakeComposedEntry) Value() []byte { return e.value } +func (e fakeComposedEntry) Revision() uint64 { return e.rev } +func (e fakeComposedEntry) Created() time.Time { return e.at } +func (e fakeComposedEntry) Delta() uint64 { return 0 } +func (e fakeComposedEntry) Operation() jetstream.KeyValueOp { return e.op } + +func newFakeComposedKV(name string) *fakeComposedKV { + return &fakeComposedKV{name: name, clock: time.Date(2026, 10, 24, 17, 0, 0, 0, time.UTC), history: map[string][]fakeComposedEntry{}} +} + +func (f *fakeComposedKV) last(key string) (fakeComposedEntry, bool) { + h := f.history[key] + if len(h) == 0 || h[len(h)-1].op != jetstream.KeyValuePut { + return fakeComposedEntry{}, false + } + return h[len(h)-1], true +} + +func (f *fakeComposedKV) put(key string, value []byte, op jetstream.KeyValueOp) uint64 { + f.rev++ + f.clock = f.clock.Add(time.Second) + f.history[key] = append(f.history[key], fakeComposedEntry{f.name, key, value, f.rev, f.clock, op}) + return f.rev +} + +func (f *fakeComposedKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + e, ok := f.last(key) + if !ok { + return nil, jetstream.ErrKeyNotFound + } + return e, nil +} + +func (f *fakeComposedKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) { + if _, ok := f.last(key); ok { + return 0, jetstream.ErrKeyExists + } + f.writes++ + return f.put(key, value, jetstream.KeyValuePut), nil +} + +func (f *fakeComposedKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) { + e, ok := f.last(key) + if !ok || e.rev != revision { + return 0, fmt.Errorf("wrong last sequence") + } + f.writes++ + return f.put(key, value, jetstream.KeyValuePut), nil +} + +func (f *fakeComposedKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + if _, ok := f.last(key); !ok { + return jetstream.ErrKeyNotFound + } + f.deletes++ + f.put(key, nil, jetstream.KeyValueDelete) + return nil +} + +func (f *fakeComposedKV) History(_ context.Context, key string, _ ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) { + h := f.history[key] + if len(h) == 0 { + return nil, jetstream.ErrKeyNotFound + } + out := make([]jetstream.KeyValueEntry, len(h)) + for i, e := range h { + out[i] = e + } + return out, nil +} + +type fakeLister struct{ keys chan string } + +func (l fakeLister) Keys() <-chan string { return l.keys } +func (l fakeLister) Stop() error { return nil } + +func (f *fakeComposedKV) ListKeys(_ context.Context, _ ...jetstream.WatchOpt) (jetstream.KeyLister, error) { + var keys []string + for k := range f.history { + if _, ok := f.last(k); ok { + keys = append(keys, k) + } + } + sort.Strings(keys) + ch := make(chan string, len(keys)) + for _, k := range keys { + ch <- k + } + close(ch) + return fakeLister{ch}, nil +} + +// fakeObjects is a node's object bucket in memory. +type fakeObjects map[string][]byte + +func (o fakeObjects) Has(_ context.Context, name string) (bool, bool, error) { + _, ok := o[name] + return ok, false, nil +} +func (o fakeObjects) Put(_ context.Context, name string, content []byte) error { + o[name] = content + return nil +} + +// aPlace is a node's buckets in memory, signed with a key of the test's, the want and the have given. +func aPlace(t *testing.T, node string) (composedPlace, ed25519.PublicKey, map[string]string, map[string]sdkdelivery.Have) { + t.Helper() + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + wants, haves := map[string]string{}, map[string]sdkdelivery.Have{} + at := composedPlace{node: node, bucket: newFakeComposedKV("composed-" + node), digests: newFakeComposedKV("mesh-controller_composed"), + objects: fakeObjects{}, + wanted: func(_ context.Context, module string) (string, error) { return wants[module], nil }, + ran: func(_ context.Context, module string) (sdkdelivery.Have, bool, error) { + h, ok := haves[module] + return h, ok, nil + }, + policy: func(string) (string, string, string) { return catalogue.PolicyRoll, catalogue.FromDefault, "" }, + sign: func(c sdkdelivery.Composed) (sdkdelivery.Composed, error) { return sdkdelivery.SignComposed(c, priv) }, + } + return at, pub, wants, haves +} + +// capturedBody is a node's declaration as the controller's `plan` verb answered it on 2026-10-24 at 16:19, the body a +// send of declarationFor carries (testdata/declaration-.json.gz, copied from mesh-sdk's; the resources carrying +// a sealed value or a secret taken out), and a fingerprint standing in for each module's build. +func capturedBody(t *testing.T, node string) ([]byte, map[string]string) { + t.Helper() + f, err := os.Open("testdata/declaration-" + node + ".json.gz") + if err != nil { + t.Fatal(err) + } + defer f.Close() + z, err := gzip.NewReader(f) + if err != nil { + t.Fatal(err) + } + raw, err := io.ReadAll(z) + if err != nil { + t.Fatal(err) + } + var d struct { + Builds map[string]string `json:"builds"` + } + if err := json.Unmarshal(raw, &d); err != nil { + t.Fatal(err) + } + fps := map[string]string{} + for m, b := range d.Builds { + fps[m] = sdkdelivery.SourcePrefix2 + fmt.Sprintf("%064x", []byte(b))[:64] + } + return raw, fps +} + +func resourceIDs(t *testing.T, raws []json.RawMessage) []string { + t.Helper() + var out []string + for _, r := range raws { + var head struct{ ID string } + if err := json.Unmarshal(r, &head); err != nil { + t.Fatal(err) + } + out = append(out, head.ID) + } + return out +} + +// 1945/AC1: for a captured node's declaration, the union of its module compositions and its own is the declaration's +// resources, each once; each module's at its fingerprint; a module with no fingerprint of the new form (the private +// network's mesh-wireguard, which nobody builds) rides in the node's own. +func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) { + for _, node := range []string{"ace", "g14"} { + body, fps := capturedBody(t, node) + var d struct { + Resources []json.RawMessage `json:"resources"` + } + _ = json.Unmarshal(body, &d) + comps, moved, err := compositionsOf(node, body, fps, nil) + if err != nil { + t.Fatalf("%s: %v", node, err) + } + if strings.Join(moved, ",") != "mesh-wireguard" { + t.Errorf("%s: moved into the node's own %v, want mesh-wireguard", node, moved) + } + var union []string + for key, c := range comps { + _, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key) + if err != nil || (!isNode && fp != fps[module]) { + t.Errorf("%s: %s is at %s (%v)", node, key, fp, err) + } + if _, err := sdkdelivery.EncodeComposed(c); err != nil { + t.Errorf("%s: %s: %v", node, key, err) + } + union = append(union, resourceIDs(t, c.Resources)...) + } + want := resourceIDs(t, d.Resources) + sort.Strings(union) + sort.Strings(want) + if strings.Join(union, " ") != strings.Join(want, " ") { + t.Errorf("%s: the compositions hold %d resources, the declaration %d, or they differ", node, len(union), len(want)) + } + own, _ := sdkdelivery.ComposedNodeKey(node) + if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); !strings.Contains(ids, "mesh-wireguard.") { + t.Errorf("%s: the node's own composition does not hold mesh-wireguard's resources: %s", node, ids) + } + } +} + +// 1945/AC1: a composition is written signed by the controller's key with its digest beside it; written again +// unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set. +func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) { + body, fps := capturedBody(t, "g14") + comps, _, err := compositionsOf("g14", body, fps, nil) + if err != nil { + t.Fatal(err) + } + at, pub, _, _ := aPlace(t, "g14") + wrote, err := writeNodeCompositions(context.Background(), at, comps) + if err != nil || len(wrote) != len(comps) { + t.Fatalf("first write: %d of %d (%v)", len(wrote), len(comps), err) + } + for key := range comps { + e, err := at.bucket.Get(context.Background(), key) + if err != nil { + t.Fatal(err) + } + c, err := sdkdelivery.DecodeComposed(e.Value()) + if err != nil { + t.Fatal(err) + } + digest, err := sdkdelivery.VerifyComposed(c, []ed25519.PublicKey{pub}) + if err != nil { + t.Fatalf("%s: %v", key, err) + } + d, err := at.digests.Get(context.Background(), key) + if err != nil { + t.Fatal(err) + } + v, err := sdkdelivery.DecodeComposedDigest(d.Value()) + if err != nil || v.Digest != digest { + t.Errorf("%s: the digest beside it is %q, the composition's %s (%v)", key, v.Digest, digest, err) + } + } + writes := at.bucket.(*fakeComposedKV).writes + at.digests.(*fakeComposedKV).writes + if again, err := writeNodeCompositions(context.Background(), at, comps); err != nil || len(again) != 0 { + t.Errorf("written again unchanged: %v (%v)", again, err) + } + if now := at.bucket.(*fakeComposedKV).writes + at.digests.(*fakeComposedKV).writes; now != writes { + t.Errorf("an unchanged write wrote %d values", now-writes) + } + + key, _ := sdkdelivery.ComposedKey("g14", "fail2ban", fps["fail2ban"]) + changed := comps[key] + changed.Data = true + comps[key] = changed + wrote, err = writeNodeCompositions(context.Background(), at, comps) + if err != nil || len(wrote) != 1 || wrote[0] != key { + t.Errorf("a changed composition wrote %v (%v), want %s alone", wrote, err, key) + } +} + +// 1945/AC1: a module's compositions outside its five newest fingerprints are deleted with their digests, unless its +// want or its have names one. +func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) { + at, _, wants, haves := aPlace(t, "ace") + file := json.RawMessage(`{"id":"m.f","type":"file","module":"m","path":"/etc/m","content":"x"}`) + fp := func(i int) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(fmt.Sprintf("%x", i), 64) } + for i := 1; i <= 8; i++ { + c := sdkdelivery.Composed{Node: "ace", Module: "m", FP: fp(i), Resources: []json.RawMessage{file}} + key, _ := sdkdelivery.ComposedKey("ace", "m", fp(i)) + if _, err := writeNodeCompositions(context.Background(), at, map[string]sdkdelivery.Composed{key: c}); err != nil { + t.Fatal(err) + } + if i == 1 { + wants["m"] = fp(1) + } + if i == 2 { + haves["m"] = sdkdelivery.Have{FP: fp(2), Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK} + } + } + if err := pruneCompositions(context.Background(), at); err != nil { + t.Fatal(err) + } + for i := 1; i <= 8; i++ { + key, _ := sdkdelivery.ComposedKey("ace", "m", fp(i)) + _, err := at.bucket.Get(context.Background(), key) + _, derr := at.digests.Get(context.Background(), key) + kept := err == nil && derr == nil + wantKept := i >= 4 || i == 1 || i == 2 + if kept != wantKept { + t.Errorf("fingerprint %d kept %v, want %v (the wanted 1, the run 2, the five newest 4 to 8)", i, kept, wantKept) + } + } +} diff --git a/cmd/mesh-controller/testdata/declaration-ace.json.gz b/cmd/mesh-controller/testdata/declaration-ace.json.gz new file mode 100644 index 00000000..11f4a417 Binary files /dev/null and b/cmd/mesh-controller/testdata/declaration-ace.json.gz differ diff --git a/cmd/mesh-controller/testdata/declaration-g14.json.gz b/cmd/mesh-controller/testdata/declaration-g14.json.gz new file mode 100644 index 00000000..1629385a Binary files /dev/null and b/cmd/mesh-controller/testdata/declaration-g14.json.gz differ diff --git a/go.mod b/go.mod index fa3e43b8..792c08ac 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.20261024144045-64a502d16aac + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144241-e64486b0e6ab 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 78cf12a8..5aa4866c 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024143642-de5b33843f25 h1:jxqWvEKo git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024143642-de5b33843f25/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024144045-64a502d16aac h1:jInLKz6gUCzs94z0egG7f05V7x+0t4BXt9cCa87ePLA= 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= 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/composed.go b/internal/broker/composed.go new file mode 100644 index 00000000..4791d87a --- /dev/null +++ b/internal/broker/composed.go @@ -0,0 +1,98 @@ +package broker + +import ( + "context" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" +) + +// The compositions' buckets (novox/hq ADR 0539 clauses 2 and 4): one per node, `composed-`, holding that node's +// compositions, which only that node's node-engine reads, since a composition can carry a sealed value (ADR 0488 §2's +// reason); and `mesh-controller_composed`, holding each composition's digest and the judgement of its move, which the +// 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). + +// 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 + +// ComposedBucketBytes is a node's composition bucket's size: five revisions of every module's compositions of the node. +const ComposedBucketBytes = 256 << 20 + +// ComposedDigestsBytes is the size of the bucket of digests: a value of a few hundred bytes per node, module and +// fingerprint kept. +const ComposedDigestsBytes = 64 << 20 + +// ComposedConfig is a node's composition bucket as the controller asserts it. +func ComposedConfig(node string) (jetstream.KeyValueConfig, error) { + bucket, err := delivery.ComposedBucketFor(node) + if err != nil { + return jetstream.KeyValueConfig{}, err + } + return jetstream.KeyValueConfig{ + Bucket: bucket, + Description: "the compositions of " + node + " (novox/hq ADR 0539): each module's resources for this node at one " + + "fingerprint, and the node's own, signed; written by the controller alone, read by this node's node-engine " + + "alone; deleted when outside the module's five newest fingerprints and named by no want", + History: delivery.ComposedHistory, + MaxValueSize: ComposedValueBytes, + MaxBytes: ComposedBucketBytes, + Storage: jetstream.FileStorage, + }, nil +} + +// ComposedDigestsConfig is the bucket of the compositions' digests as the controller asserts it. +func ComposedDigestsConfig() jetstream.KeyValueConfig { + return jetstream.KeyValueConfig{ + Bucket: delivery.ControllerComposedBucket, + Description: "each composition's digest and the judgement of its move from what the node runs (novox/hq ADR 0539 " + + "clauses 2 and 14): written by the controller alone, read by the assigner", + History: delivery.ControllerComposedHistory, + MaxValueSize: 4 << 10, + MaxBytes: ComposedDigestsBytes, + Storage: jetstream.FileStorage, + } +} + +// EnsureComposed creates the bucket of digests and each named node's composition bucket if absent and brings their +// options to match. An update, never a delete. +func EnsureComposed(ctx context.Context, js jetstream.JetStream, nodes ...string) error { + if _, err := js.CreateOrUpdateKeyValue(ctx, ComposedDigestsConfig()); err != nil { + return fmt.Errorf("asserting bucket %s: %w", delivery.ControllerComposedBucket, err) + } + for _, node := range nodes { + cfg, err := ComposedConfig(node) + if err != nil { + return err + } + if _, err := js.CreateOrUpdateKeyValue(ctx, cfg); err != nil { + return fmt.Errorf("asserting bucket %s: %w", cfg.Bucket, err) + } + } + return nil +} + +// ComposedWrites is what the controller publishes to write a node's compositions: the bucket's own subjects. +func ComposedWrites(node string) (string, error) { + bucket, err := delivery.ComposedBucketFor(node) + if err != nil { + return "", err + } + return "$KV." + bucket + ".>", nil +} + +// ComposedReads is what a node's node-engine publishes to follow its wants and compositions (ADR 0539 clauses 1, 2 and +// 9): every key of its own composition bucket, and its own wants (`want..` and `want..node`) in the +// assigner's bucket; nothing of another node's, nor the bucket of digests. +func ComposedReads(node string) ([]string, error) { + bucket, err := delivery.ComposedBucketFor(node) + if err != nil { + return nil, err + } + if !safeSubject.MatchString(node) { + return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node) + } + return append(readGrants(bucket, ""), readGrants(delivery.AssignerWantsBucket, "want."+node+".>")...), nil +} diff --git a/internal/broker/composed_test.go b/internal/broker/composed_test.go new file mode 100644 index 00000000..95489842 --- /dev/null +++ b/internal/broker/composed_test.go @@ -0,0 +1,49 @@ +package broker + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1945/AC1: a node's composition bucket keeps five revisions of a key and is named for its node; the bucket of digests +// likewise; a node name a subject would widen is refused. +func TestTheCompositionBuckets(t *testing.T) { + cfg, err := ComposedConfig("ace") + if err != nil || cfg.Bucket != "composed-ace" || cfg.History != 5 { + t.Fatalf("ComposedConfig(ace) = %+v, %v", cfg, err) + } + if d := ComposedDigestsConfig(); d.Bucket != delivery.ControllerComposedBucket || d.History != 5 { + t.Errorf("ComposedDigestsConfig = %+v", d) + } + if _, err := ComposedConfig("a.b"); err == nil { + t.Errorf("a dotted node got a bucket") + } +} + +// 1945/AC1: a node-engine reads its own compositions and its own wants and nothing of another node's, nor the digests; +// the controller writes a node's bucket by its own subjects. +func TestANodeReadsOnlyItsOwnCompositions(t *testing.T) { + reads, err := ComposedReads("ace") + if err != nil { + t.Fatal(err) + } + all := strings.Join(reads, " ") + for _, want := range []string{"KV_composed-ace", "$KV.mesh-delivery-assigner_wants.want.ace.>"} { + if !strings.Contains(all, want) { + t.Errorf("a node-engine of ace cannot reach %s: %v", want, reads) + } + } + for _, never := range []string{"composed-g14", delivery.ControllerComposedBucket, "$KV.composed-ace", "want.*", "want.>"} { + if strings.Contains(all, never) { + t.Errorf("a node-engine of ace is granted %s: %v", never, reads) + } + } + if w, err := ComposedWrites("ace"); err != nil || w != "$KV.composed-ace.>" { + t.Errorf("ComposedWrites(ace) = %q, %v", w, err) + } + if _, err := ComposedReads("a*"); err == nil { + t.Errorf("a node name with a wildcard was granted") + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index e55ba60f..29c48bc0 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -427,6 +427,11 @@ 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." + delivery.ControllerComposedBucket + ".>", + Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " + + "and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)", + }, { Subject: "$KV." + GrantsBucket + ".>", Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + @@ -738,7 +743,15 @@ func PermissionsFor(p Principal) (Permissions, error) { return Permissions{}, err } pub = append(pub, writes) + // And each node's compositions, which it alone writes (novox/hq ADR 0539 clause 2). + composed, err := ComposedWrites(node) + if err != nil { + return Permissions{}, err + } + 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. @@ -845,6 +858,13 @@ func PermissionsFor(p Principal) (Permissions, error) { return Permissions{}, err } pub = append(pub, fileReads...) + // And its own compositions and wants, which it follows (novox/hq ADR 0539 clauses 1, 2 and 9), and nothing of another + // node's. + composedReads, err := ComposedReads(p.Node) + if err != nil { + return Permissions{}, err + } + pub = append(pub, composedReads...) // 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 a6a80775..5e67cf01 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_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_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"] } 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" } } } @@ -38,7 +38,7 @@ accounts { subscribe: { allow: ["_INBOX.enrol.one.>"] } } } { user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: { - publish: { allow: ["$JS.ACK.NODES.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_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files-one.*", "$JS.API.CONSUMER.INFO.NODES.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_node-engine_have", "$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.>"] } + 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.>"] } 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"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 63b17263..712735fb 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 64a502d16aacaaf87c0acde4ae548f011a7020b1 conformance/events/module-event.json +mesh-sdk e64486b0e6ab573ee089aaaae7915406f8a5a745 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: