From 05494e7bb933917c19661dcdb1946d7f4f5eaa7e Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 17:25:49 +0100 Subject: [PATCH] W5.1, W5.2, W5.5: the assignment graph computed per branch, kept as objects, and served by mesh-wants' verbs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Design 47 at hq main b229f5c4, lines 1353-1366 (each entry one module version on a node, its edges provisioning and acting; each revision naming its inputs, kept as objects in the old section with its cap, served by the seat's verbs), 1385-1388 (under 256 KiB for 120 modules on four nodes; a 512 MiB cap), 1416-1418 and 522-525 (a node's entries from the catalog graph of the branch its placement names, by default the registration's first), 909-914 (the placement's build is the version), 1835-1836 (an edge from the acting part's version to each version its rule names), and the Reverted table (assignment graph revisions as objects, form 1). internal/assigngraph computes a revision from the halves held per branch, the placements and the capability readings; keeps it numbered after the newest when it differs, each entry once by its digest; and answers revision, entry and difference from the kept objects only. The manifest serves the three verbs and declares the objects assignment-revisions. A placed module whose entry cannot be computed is named in Uncomputed, never left out in silence. Not built here (the design does not say, or held): contribution edges; provisioning to a provider no binding or pin names; tiers; a candidate's revision (W5.9); removal of a revision no record pins (pins are W5.9's). The process does not compute and keep yet: that needs the registrations' first view and a catalog half per branch (handover). Milestone 47/G, issue 1966. Tests: TestTheAssignmentGraphIsComputedFromItsInputsAndKept, TestARevisionOf120ModulesOnFourNodesFitsOneValue, TestTheSeatsVerbs/1953/D4-D6 (failed before), TestManifest's objects row (failed with the old manifest). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ Builds: design 47 G §173, §175, §176, §180, §194, §209, §223, §263, §346 --- .../cmd/mesh-delivery-assigner/graph_tools.go | 82 +++++++ .../mesh-delivery-assigner/seat_tools_test.go | 56 ++++- .../cmd/mesh-delivery-assigner/tools.go | 3 + .../cmd/mesh-delivery-assigner/wants_tools.go | 9 + .../internal/assigngraph/compute.go | 206 ++++++++++++++++++ .../internal/assigngraph/graph_test.go | 176 +++++++++++++++ .../internal/assigngraph/keep.go | 151 +++++++++++++ .../internal/assigngraph/verbs.go | 156 +++++++++++++ .../mesh-delivery-assigner/manifest_test.go | 10 +- modules/mesh-delivery-assigner/module.json | 8 +- 10 files changed, 852 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graph_tools.go create mode 100644 modules/mesh-delivery-assigner/internal/assigngraph/compute.go create mode 100644 modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go create mode 100644 modules/mesh-delivery-assigner/internal/assigngraph/keep.go create mode 100644 modules/mesh-delivery-assigner/internal/assigngraph/verbs.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graph_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graph_tools.go new file mode 100644 index 00000000..5710deb4 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graph_tools.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "encoding/json" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/busstate" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// runtimeObjects is the kept data assignment-revisions as the bundle reaches it through its runtime (the old objects +// section of a foundation part, design 47 lines 1362-1366). +type runtimeObjects struct{ o busstate.Objects } + +func (r runtimeObjects) Put(_ context.Context, name string, data []byte) error { + _, err := r.o.Put(name, data, "") + return err +} + +func (r runtimeObjects) Get(_ context.Context, name string) ([]byte, error) { + d, err := r.o.Get(name) + if err != nil { + return nil, err + } + if d.Info.Name == "" || d.Info.Deleted { + return nil, assigngraph.ErrNoObject + } + return d.Data, nil +} + +func (r runtimeObjects) Names(context.Context) ([]string, error) { + infos, err := r.o.List() + if err != nil { + return nil, err + } + names := []string{} + for _, i := range infos { + if !i.Deleted { + names = append(names, i.Name) + } + } + return names, nil +} + +// graphVerbs are the seat mesh-wants' verbs revision, entry and difference (design 47 lines 1365-1366: an assignment graph +// revision is served by the seat's verbs), each reading only the kept revisions. +func graphVerbs(o assigngraph.Objects) map[string]func(map[string]any) (any, error) { + return map[string]func(map[string]any) (any, error){ + meshwants.VerbRevision: func(a map[string]any) (any, error) { + var in meshwants.RevisionInput + if err := decodeArgs(a, &in); err != nil { + return nil, err + } + return assigngraph.Revision(context.Background(), o, in) + }, + meshwants.VerbEntry: func(a map[string]any) (any, error) { + var in meshwants.EntryInput + if err := decodeArgs(a, &in); err != nil { + return nil, err + } + return assigngraph.Entry(context.Background(), o, in) + }, + meshwants.VerbDifference: func(a map[string]any) (any, error) { + var in meshwants.DifferenceInput + if err := decodeArgs(a, &in); err != nil { + return nil, err + } + return assigngraph.Difference(context.Background(), o, in) + }, + } +} + +// decodeArgs reads a verb's arguments into its input type; a caller's own fields beside them are the runtime's. +func decodeArgs(a map[string]any, into any) error { + raw, err := json.Marshal(a) + if err != nil { + return err + } + return json.Unmarshal(raw, into) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index c884c495..36b9656b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -30,7 +30,9 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" plc "mesh-delivery-assigner/internal/assignment" ) @@ -654,9 +656,61 @@ func wantsVerbs(t *testing.T) { if b.(map[string]any)["module"] != "nats" || len(nodes) != 2 || !nodes[0].Behind || nodes[1].Behind || !slices.Contains(asked, "seat:mesh-hand-acts.hand-acts") { t.Errorf("bus = %+v, asked %v", b, asked) } - if tools := w.tools(); len(tools) != 2 || tools[0].Name != "mesh-wants.upgrade" || tools[1].Run == nil { + // W5.5 (design 47 lines 1365-1366): every verb of the seat is served, revision, entry and difference reading only the + // kept revisions. + objects := memObjects{} + w.revisions = objects + tools := w.tools() + run := map[string]func(map[string]any) (any, error){} + for _, tool := range tools { + if tool.Run == nil { + t.Errorf("the seat's verb %s is not served", tool.Name) + } + run[tool.Name] = tool.Run + } + if len(tools) != 5 || tools[0].Name != "mesh-wants.upgrade" { t.Errorf("the seat's verbs as served: %+v", tools) } + if _, err := run["mesh-wants.revision"](map[string]any{}); !errors.Is(err, assigngraph.ErrNoRevision) { + t.Errorf("revision with none kept: %v", err) + } + c := assigngraph.Computed{Revision: meshwants.AssignmentRevision{V: 1, Kind: "trunk", Placements: "sha256:p", + Catalogs: []meshwants.CatalogOf{{Branch: "main", Revision: 3}}, Entries: []meshwants.NodeEntryRef{{Node: "anchor", Module: "web", Digest: "sha256:e"}}}, + Entries: map[string]meshwants.AssignmentEntry{"sha256:e": {Node: "anchor", Module: "web", Branch: "main", Fp: fpNew, Catalog: "sha256:c"}}} + if _, _, err := assigngraph.Keep(context.Background(), objects, c, "catalog-revision", time.Now()); err != nil { + t.Fatal(err) + } + r, err := run["mesh-wants.revision"](map[string]any{"revision": 1, "caller": "seat:mesh-grants"}) + if err != nil || r.(meshwants.RevisionAnswer).Revision.Catalogs[0].Revision != 3 { + t.Errorf("revision = %+v, %v", r, err) + } + if e, err := run["mesh-wants.entry"](map[string]any{"node": "anchor", "module": "web"}); err != nil || e.(meshwants.EntryAnswer).Entry.Fp != fpNew { + t.Errorf("entry = %+v, %v", e, err) + } + if d, err := run["mesh-wants.difference"](map[string]any{"from": 1, "to": 1}); err != nil || len(d.(meshwants.Difference).Changes) != 0 { + t.Errorf("difference = %+v, %v", d, err) + } +} + +// memObjects is an object bucket in memory. +type memObjects map[string][]byte + +func (m memObjects) Put(_ context.Context, name string, data []byte) error { + m[name] = data + return nil +} +func (m memObjects) Get(_ context.Context, name string) ([]byte, error) { + if d, ok := m[name]; ok { + return d, nil + } + return nil, assigngraph.ErrNoObject +} +func (m memObjects) Names(context.Context) ([]string, error) { + out := []string{} + for n := range m { + out = append(out, n) + } + return out, nil } // readManifest is the module's manifest, read as the controller reads it. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 1c7ec141..c122bd9f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -4,6 +4,8 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/busstate" + "mesh-delivery-assigner/internal/assigngraph" "mesh-delivery-assigner/internal/puts" "os" "regexp" @@ -400,6 +402,7 @@ func serveTools() error { return stdio.SeatPublish(wantsEventSubject(e), b, id) }) wants.watched = stdio.NewWatchedStates() // the large states read whole from one watched copy (ADR 0580 clause 3) + wants.revisions = runtimeObjects{busstate.ObjectsOf(assigngraph.RevisionsObjects)} tools = append(tools, wants.tools()...) tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) // What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index 034e9b15..a2da5bfb 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -22,6 +22,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" ) // The seat mesh-wants' verbs (novox/hq ADR 0565 §7, §8, §9, §15; issue 1953 D4-D6), served through the tool runner beside the @@ -49,6 +50,9 @@ type wantsTools struct { // its changes after (mesh-sdk's WatchedStates, novox/hq ADR 0580 clause 3), never a get per key, which timed out on // a large state (issue 1953 L5f). Nil reads key by key. watched *stdio.WatchedStates + // revisions is the kept data assignment-revisions the verbs revision, entry and difference read (design 47 lines + // 1365-1366); nil serves them not. + revisions assigngraph.Objects } // wantsEventSubject is the subject of one of the seat's events. @@ -313,6 +317,11 @@ func (s *wantsTools) tools() []stdio.Tool { meshwants.VerbUpgrade: func(a map[string]any) (any, error) { return s.doUpgrade(a, stdio.CallerOf(a)) }, meshwants.VerbBus: s.doBus, } + if s.revisions != nil { + for v, f := range graphVerbs(s.revisions) { + run[v] = f + } + } var out []stdio.Tool for _, v := range meshwants.Verbs() { if run[v] == nil { diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/compute.go b/modules/mesh-delivery-assigner/internal/assigngraph/compute.go new file mode 100644 index 00000000..a9667c8f --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assigngraph/compute.go @@ -0,0 +1,206 @@ +// Package assigngraph is the assigner's assignment graph (design 47, *Two graphs, one source of truth*; ADR 0656): the +// module versions placed on nodes, with the edges between versions, computed from the catalog graph of the branch each +// placement names, the placements and the nodes' capability readings, and kept per revision as objects of the seat +// mesh-wants' kept data assignment-revisions. +package assigngraph + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "maps" + "slices" + "strings" + + mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/keys" + + "mesh-delivery-assigner/internal/assignment" +) + +// The kinds of an edge of the assignment graph (ADR 0656; the seat's Edge). +const ( + EdgeProvisioning = "provisioning" + EdgeActing = "acting" +) + +// Inputs are what a revision is computed from, and nothing else (design 47 lines 1353-1356). +type Inputs struct { + // Halves is the catalog graph revision held of each watched branch, by branch. + Halves map[string]meshcatalogclient.Half + // Placements is every node's placement key, by node. + Placements map[string]mma.Assignments + // FirstBranch is the first watched branch of a module's registration, the branch a placement naming none runs + // (design 47 lines 522-523); "" when no registration is known for it. + FirstBranch func(module string) string + // Capabilities is each node's capability reading, by its digest. + Capabilities []meshwants.AssignmentRevisionCapabilitiesItem +} + +// Computed is one assignment graph revision as computed, before it is numbered and kept. +type Computed struct { + Revision meshwants.AssignmentRevision + // Entries is every entry the revision lists, by its digest. + Entries map[string]meshwants.AssignmentEntry + // Uncomputed is each placed module whose entry could not be computed, as "/: why", sorted: it is said, + // never left out in silence. + Uncomputed []string +} + +// placedAt is one placed module with what its entry is computed from. +type placedAt struct { + node string + placed mma.PlacedModule + branch string + entry meshwants.AssignmentEntry + langs map[string]bool + acts []actingRule +} + +type actingRule struct{ seat, step, on, bound string } + +// Compute is the assignment graph revision of in: one entry per module placed on a node whose branch's catalog graph +// holds it, its version the build the placement names or else the catalog entry's fingerprint (design 47 lines 909-914), +// with the edges of provisioning to the providers its placement binds or its node pins, and of acting from each acting +// part's version to each version its rule names (design 47 *An acting part's new version acts first*: "the assigner +// gives an edge from the acting part's version to each version its rule names"). +func Compute(in Inputs) Computed { + out := Computed{Entries: map[string]meshwants.AssignmentEntry{}} + var placed []*placedAt + branches := map[string]bool{} + for _, node := range slices.Sorted(maps.Keys(in.Placements)) { + mods := slices.Clone(in.Placements[node].Modules) + slices.SortFunc(mods, func(a, b mma.PlacedModule) int { return strings.Compare(a.Module, b.Module) }) + for _, m := range mods { + branch := m.Branch + if branch == "" && in.FirstBranch != nil { + branch = in.FirstBranch(m.Module) + } + if branch == "" { + out.Uncomputed = append(out.Uncomputed, fmt.Sprintf("%s/%s: no watched branch: its placement names none and no registration of it is known", node, m.Module)) + continue + } + half, ok := in.Halves[branch] + if !ok { + out.Uncomputed = append(out.Uncomputed, fmt.Sprintf("%s/%s: the catalog graph of the branch %s is not held", node, m.Module, branch)) + continue + } + ce, ok := half.Entries[m.Module] + if !ok { + out.Uncomputed = append(out.Uncomputed, fmt.Sprintf("%s/%s: the catalog graph revision %d of %s holds no entry of it", node, m.Module, half.Revision.Revision, branch)) + continue + } + digest := "" + for _, ref := range half.Revision.Entries { + if ref.Module == m.Module { + digest = ref.Digest + } + } + if digest == "" { + out.Uncomputed = append(out.Uncomputed, fmt.Sprintf("%s/%s: the catalog graph revision %d of %s lists no digest of its entry", node, m.Module, half.Revision.Revision, branch)) + continue + } + fp := ce.Fp + if m.Build != "" { + fp = m.Build + } + p := &placedAt{node: node, placed: m, branch: branch, langs: map[string]bool{}, + entry: meshwants.AssignmentEntry{V: keys.Version, Node: node, Module: m.Module, Branch: branch, Fp: fp, Catalog: digest}} + for lang := range ce.Toolchains { + p.langs[lang] = true + } + for _, r := range ce.Acts { + p.acts = append(p.acts, actingRule{seat: r.Seat, step: r.Step, on: r.On, bound: r.Bound}) + } + for _, r := range ce.Requires { + prov, ok := m.Bindings[r] + if !ok { + prov, ok = in.Placements[node].Pins[r] + } + if ok { + p.entry.Edges = append(p.entry.Edges, meshwants.Edge{Kind: EdgeProvisioning, Node: prov.Node, Module: prov.Module, Provision: r}) + } + } + placed = append(placed, p) + branches[branch] = true + } + } + at := map[string]*placedAt{} + for _, p := range placed { + at[p.node+"/"+p.placed.Module] = p + } + for _, p := range placed { + for i, e := range p.entry.Edges { + if prov, ok := at[e.Node+"/"+e.Module]; ok { + p.entry.Edges[i].Fp = prov.entry.Fp + } + } + } + // Acting: an acting part on the entry's own node acts there (a node's seat is held on every node); else every node it + // is placed on (a seat held once for the mesh is held on one). + byModule := map[string][]*placedAt{} + for _, p := range placed { + byModule[p.placed.Module] = append(byModule[p.placed.Module], p) + } + for _, part := range slices.Sorted(maps.Keys(byModule)) { + holders := byModule[part] + for _, rule := range holders[0].acts { + for _, t := range placed { + if t.placed.Module == part || (rule.on != "" && rule.on != "every" && !t.langs[rule.on]) { + continue + } + from := holders + for _, h := range holders { + if h.node == t.node { + from = []*placedAt{h} + } + } + for _, h := range from { + t.entry.Edges = append(t.entry.Edges, meshwants.Edge{Kind: EdgeActing, Node: h.node, Module: part, Fp: h.entry.Fp, + Seat: rule.seat, Step: rule.step, Bound: rule.bound}) + } + } + } + } + rev := meshwants.AssignmentRevision{V: keys.Version, Kind: meshcatalogclient.KindTrunk, Placements: assignment.ExportDigest(in.Placements), + Capabilities: in.Capabilities, Entries: []meshwants.NodeEntryRef{}} + for _, b := range slices.Sorted(maps.Keys(branches)) { + h := in.Halves[b] + rev.Catalogs = append(rev.Catalogs, meshwants.CatalogOf{Branch: b, Revision: h.Revision.Revision, Source: h.Revision.Source}) + } + for _, p := range placed { + sortEdges(p.entry.Edges) + d := EntryDigest(p.entry) + out.Entries[d] = p.entry + rev.Entries = append(rev.Entries, meshwants.NodeEntryRef{Node: p.node, Module: p.placed.Module, Digest: d}) + } + out.Revision = rev + slices.Sort(out.Uncomputed) + return out +} + +func sortEdges(edges []meshwants.Edge) { + slices.SortFunc(edges, func(a, b meshwants.Edge) int { + return strings.Compare(edgeKey(a), edgeKey(b)) + }) +} + +func edgeKey(e meshwants.Edge) string { + return strings.Join([]string{e.Kind, e.Node, e.Module, e.Provision, e.Seat, e.Step, e.Fp, e.Bound}, "\x00") +} + +// EntryDigest is the digest an entry is stored under: the same entry, the same digest. +func EntryDigest(e meshwants.AssignmentEntry) string { + raw, _ := json.Marshal(e) + sum := sha256.Sum256(raw) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// Same says two revisions list the same entries computed from the same inputs, so the second is no new revision. +func Same(a, b meshwants.AssignmentRevision) bool { + return slices.Equal(a.Entries, b.Entries) && slices.Equal(a.Catalogs, b.Catalogs) && a.Placements == b.Placements && + slices.Equal(a.Capabilities, b.Capabilities) +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go new file mode 100644 index 00000000..e2c92d67 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go @@ -0,0 +1,176 @@ +package assigngraph + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "time" + + mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +type memObjects map[string][]byte + +func (m memObjects) Put(_ context.Context, name string, data []byte) error { + m[name] = data + return nil +} +func (m memObjects) Get(_ context.Context, name string) ([]byte, error) { + d, ok := m[name] + if !ok { + return nil, ErrNoObject + } + return d, nil +} +func (m memObjects) Names(context.Context) ([]string, error) { + out := []string{} + for n := range m { + out = append(out, n) + } + return out, nil +} + +func half(branch string, revision int64, entries ...meshcatalog.CatalogEntry) meshcatalogclient.Half { + h := meshcatalogclient.Half{Revision: meshcatalog.CatalogRevision{Kind: "trunk", Branch: branch, Revision: revision, Source: revision + 10}, + Entries: map[string]meshcatalog.CatalogEntry{}} + for _, e := range entries { + h.Entries[e.Module] = e + h.Revision.Entries = append(h.Revision.Entries, meshcatalog.EntryRef{Module: e.Module, Digest: "sha256:" + e.Module + e.Fp}) + } + return h +} + +// Design 47 *Two graphs, one source of truth* (lines 1353-1366, 1416-1418) and *An acting part's new version acts first* +// (lines 1835-1836): each entry one module version on a node from the catalog graph of the branch its placement names, +// its edges provisioning and acting, a revision computed again from its inputs equal to the one kept. +func TestTheAssignmentGraphIsComputedFromItsInputsAndKept(t *testing.T) { + ctx := context.Background() + in := Inputs{ + Halves: map[string]meshcatalogclient.Half{"main": half("main", 7, + meshcatalog.CatalogEntry{Module: "node-engine", Fp: "src2:e1", Acts: []meshcatalog.ActingRule{{Seat: "node-engine", Step: "compose", On: "every", Bound: "30m"}}}, + meshcatalog.CatalogEntry{Module: "build-go", Fp: "src2:b1", Acts: []meshcatalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}}, + meshcatalog.CatalogEntry{Module: "postgres", Fp: "src2:p1", Provides: []string{"database"}}, + meshcatalog.CatalogEntry{Module: "web", Fp: "src2:w1", Requires: []string{"database", "secrets"}, Toolchains: map[string]meshcatalog.BuildEdge{"go": {Version: "1"}}}, + )}, + Placements: map[string]mma.Assignments{ + "anchor": {Node: "anchor", Modules: []mma.PlacedModule{{Module: "node-engine"}, {Module: "postgres"}, {Module: "build-go"}, + {Module: "web", Bindings: map[string]mma.Provider{"database": {Node: "anchor", Module: "postgres"}}}}}, + "laptop": {Node: "laptop", Modules: []mma.PlacedModule{{Module: "node-engine"}, {Module: "web", Build: "src2:w0"}, + {Module: "photos", Branch: "beta"}, {Module: "ghost"}}}, + }, + FirstBranch: func(module string) string { + if module == "ghost" { + return "" + } + return "main" + }, + } + c := Compute(in) + byNode := map[string]meshwants.AssignmentEntry{} + for _, ref := range c.Revision.Entries { + byNode[ref.Node+"/"+ref.Module] = c.Entries[ref.Digest] + } + webA, webL := byNode["anchor/web"], byNode["laptop/web"] + if len(c.Revision.Entries) != 6 || len(c.Revision.Catalogs) != 1 || c.Revision.Catalogs[0] != (meshwants.CatalogOf{Branch: "main", Revision: 7, Source: 17}) { + t.Fatalf("revision %+v", c.Revision) + } + if webA.Branch != "main" || webA.Fp != "src2:w1" || webA.Catalog != "sha256:websrc2:w1" || webL.Fp != "src2:w0" { + t.Errorf("web's entries: %+v, %+v (the placement's build is the version on laptop)", webA, webL) + } + want := []meshwants.Edge{ + {Kind: EdgeActing, Node: "anchor", Module: "build-go", Fp: "src2:b1", Seat: "node-build-go", Step: "build"}, + {Kind: EdgeActing, Node: "anchor", Module: "node-engine", Fp: "src2:e1", Seat: "node-engine", Step: "compose", Bound: "30m"}, + {Kind: EdgeProvisioning, Node: "anchor", Module: "postgres", Fp: "src2:p1", Provision: "database"}, + } + if fmt.Sprint(webA.Edges) != fmt.Sprint(want) { + t.Errorf("web on anchor's edges:\n got %+v\nwant %+v", webA.Edges, want) + } + // laptop's web: its own node's node-engine composes it, the build module placed once builds it, and no binding or pin + // names a provider there. + if e := webL.Edges; len(e) != 2 || e[0].Module != "build-go" || e[0].Node != "anchor" || e[1].Module != "node-engine" || e[1].Node != "laptop" { + t.Errorf("web on laptop's edges: %+v", e) + } + if e := byNode["anchor/postgres"].Edges; len(e) != 1 || e[0].Module != "node-engine" { + t.Errorf("postgres builds no Go, so only the node-engine acts on it: %+v", e) + } + if len(byNode["anchor/node-engine"].Edges) != 0 { + t.Errorf("an acting part acts on itself: %+v", byNode["anchor/node-engine"].Edges) + } + if strings.Join(c.Uncomputed, "|") != "laptop/ghost: no watched branch: its placement names none and no registration of it is known|"+ + "laptop/photos: the catalog graph of the branch beta is not held" { + t.Errorf("uncomputed %q", c.Uncomputed) + } + + // Kept once; computed again from the same inputs it is the same revision, kept no second time. + o := memObjects{} + at := time.Date(2026, 11, 10, 18, 0, 0, 0, time.UTC) + rev, kept, err := Keep(ctx, o, c, "catalog-revision", at) + if err != nil || !kept || rev.Revision != 1 || rev.Cause != "catalog-revision" || len(o) != 7 { + t.Fatalf("kept %v %+v (%v), objects %d", kept, rev, err, len(o)) + } + if again, kept, err := Keep(ctx, o, Compute(in), "placement-changed", at); err != nil || kept || again.Revision != 1 { + t.Errorf("recomputed from the same inputs: kept %v, %+v, %v", kept, again, err) + } + // A placement change moves laptop's web to the catalog's version: revision 2, one entry more stored, the rest shared. + in.Placements["laptop"].Modules[1].Build = "" + rev2, kept, err := Keep(ctx, o, Compute(in), "placement-changed", at) + if err != nil || !kept || rev2.Revision != 2 || len(o) != 9 || rev2.Placements == rev.Placements { + t.Fatalf("revision 2 %v %+v (%v), objects %d", kept, rev2, err, len(o)) + } + + // The verbs read only what is kept. + r, err := Revision(ctx, o, meshwants.RevisionInput{}) + if err != nil || r.Revision.Revision != 2 { + t.Errorf("revision = %+v, %v", r, err) + } + if _, err := Revision(ctx, o, meshwants.RevisionInput{Group: "g", Heads: "h"}); !errors.Is(err, ErrNoCandidate) { + t.Errorf("a candidate's revision: %v", err) + } + e, err := Entry(ctx, o, meshwants.EntryInput{Revision: 1, Node: "laptop", Module: "web"}) + if err != nil || e.Entry.Fp != "src2:w0" || !strings.HasPrefix(e.Digest, "sha256:") { + t.Errorf("entry = %+v, %v", e, err) + } + d, err := Difference(ctx, o, meshwants.DifferenceInput{From: 1, To: 2}) + if err != nil || len(d.Changes) != 1 || fmt.Sprint(d.Changes[0]) != fmt.Sprint(meshwants.Change{Node: "laptop", Module: "web", From: "src2:w0", To: "src2:w1"}) || + strings.Join(d.Nodes, ",") != "laptop" { + t.Errorf("difference = %+v, %v", d, err) + } + if _, err := Difference(ctx, o, meshwants.DifferenceInput{To: 2}); err == nil { + t.Error("a difference with one revision answered") + } + if ev := Event(rev2); ev.Revision != 2 || len(ev.Catalogs) != 1 || ev.Kind != "trunk" { + t.Errorf("event %+v", ev) + } +} + +// Design 47 line 1385: a whole assignment graph revision of about 120 modules on four nodes is under the bus's 256 KiB +// value limit. +func TestARevisionOf120ModulesOnFourNodesFitsOneValue(t *testing.T) { + var entries []meshcatalog.CatalogEntry + placements := map[string]mma.Assignments{} + for i := range 120 { + entries = append(entries, meshcatalog.CatalogEntry{Module: fmt.Sprintf("module-%03d", i), Fp: "src2:" + strings.Repeat("a", 64)}) + } + for _, node := range []string{"anchor", "home", "desk", "laptop"} { + a := mma.Assignments{Node: node} + for _, e := range entries { + a.Modules = append(a.Modules, mma.PlacedModule{Module: e.Module}) + } + placements[node] = a + } + h := half("main", 1, entries...) + for i := range h.Revision.Entries { + h.Revision.Entries[i].Digest = "sha256:" + strings.Repeat("b", 64) + } + c := Compute(Inputs{Halves: map[string]meshcatalogclient.Half{"main": h}, Placements: placements, FirstBranch: func(string) string { return "main" }}) + raw, _ := json.Marshal(c.Revision) + if len(c.Revision.Entries) != 480 || len(raw) >= 256<<10 { + t.Errorf("%d entries in %d bytes", len(c.Revision.Entries), len(raw)) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/keep.go b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go new file mode 100644 index 00000000..4440c469 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go @@ -0,0 +1,151 @@ +package assigngraph + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// RevisionsObjects is the local name of the kept data in the assigner's manifest's old `objects` section (design 47 +// lines 1362-1366: both foundation parts keep their revisions as objects in the old form; the seat mesh-wants' state +// assignment-revisions names its key forms). +const RevisionsObjects = "assignment-revisions" + +// ErrNoObject is an object the bucket does not hold. +var ErrNoObject = errors.New("no such object") + +// Objects is an object bucket as the assigner keeps its revisions in it. +type Objects interface { + // Put stores data whole under name, replacing one of that name. + Put(ctx context.Context, name string, data []byte) error + // Get is the object's bytes; ErrNoObject when it holds none of that name. + Get(ctx context.Context, name string) ([]byte, error) + // Names is every object's name; none is an empty list. + Names(ctx context.Context) ([]string, error) +} + +// EntryObject is the object name of the entry of that digest (the seat's key form entry.). +func EntryObject(digest string) string { return "entry." + digest } + +// TrunkObject is the object name of a trunk revision (the seat's key form trunk.). +func TrunkObject(revision int64) string { return "trunk." + strconv.FormatInt(revision, 10) } + +// trunkOf is the revision number a trunk object's name holds, and whether it is one. +func trunkOf(name string) (int64, bool) { + n, ok := strings.CutPrefix(name, "trunk.") + if !ok { + return 0, false + } + r, err := strconv.ParseInt(n, 10, 64) + return r, err == nil && r > 0 +} + +// Newest is the newest trunk revision kept, its list only; none kept is false. +func Newest(ctx context.Context, o Objects) (meshwants.AssignmentRevision, bool, error) { + names, err := o.Names(ctx) + if err != nil { + return meshwants.AssignmentRevision{}, false, fmt.Errorf("the objects %s: %w", RevisionsObjects, err) + } + var newest int64 + for _, n := range names { + if r, ok := trunkOf(n); ok && r > newest { + newest = r + } + } + if newest == 0 { + return meshwants.AssignmentRevision{}, false, nil + } + rev, err := ReadRevision(ctx, o, newest) + return rev, err == nil, err +} + +// ReadRevision is the trunk revision of that number, its list only. +func ReadRevision(ctx context.Context, o Objects, revision int64) (meshwants.AssignmentRevision, error) { + var rev meshwants.AssignmentRevision + data, err := o.Get(ctx, TrunkObject(revision)) + if err != nil { + return rev, fmt.Errorf("revision %d: %w", revision, err) + } + if err := json.Unmarshal(data, &rev); err != nil { + return rev, fmt.Errorf("revision %d: %w", revision, err) + } + return rev, nil +} + +// ReadEntry is the entry kept under that digest. +func ReadEntry(ctx context.Context, o Objects, digest string) (meshwants.AssignmentEntry, error) { + var e meshwants.AssignmentEntry + data, err := o.Get(ctx, EntryObject(digest)) + if err != nil { + return e, fmt.Errorf("the entry %s: %w", digest, err) + } + if err := json.Unmarshal(data, &e); err != nil { + return e, fmt.Errorf("the entry %s: %w", digest, err) + } + return e, nil +} + +// Keep keeps c as the next trunk revision, numbered after the newest kept, unless the newest kept lists the same entries +// from the same inputs (Same): then it keeps nothing and answers the newest and false. Each entry object not held yet is +// written before the list, so a list is never kept whose entries are not, and an entry two revisions share is stored +// once (design 47 *An unchanged entry is stored once*). +func Keep(ctx context.Context, o Objects, c Computed, cause string, at time.Time) (meshwants.AssignmentRevision, bool, error) { + newest, has, err := Newest(ctx, o) + if err != nil { + return newest, false, err + } + if has && Same(newest, c.Revision) { + return newest, false, nil + } + names, err := o.Names(ctx) + if err != nil { + return newest, false, fmt.Errorf("the objects %s: %w", RevisionsObjects, err) + } + held := make(map[string]bool, len(names)) + for _, n := range names { + held[n] = true + } + rev := c.Revision + rev.Revision = newest.Revision + 1 + rev.Cause = cause + t := at.UTC() + rev.At = &t + for _, ref := range rev.Entries { + name := EntryObject(ref.Digest) + if held[name] { + continue + } + e, ok := c.Entries[ref.Digest] + if !ok { + return newest, false, fmt.Errorf("revision %d lists the entry of %s/%s (%s), which it does not carry", rev.Revision, ref.Node, ref.Module, ref.Digest) + } + data, err := json.Marshal(e) + if err != nil { + return newest, false, err + } + if err := o.Put(ctx, name, data); err != nil { + return newest, false, fmt.Errorf("the entry of %s/%s in %s: %w", ref.Node, ref.Module, RevisionsObjects, err) + } + held[name] = true + } + list, err := json.Marshal(rev) + if err != nil { + return newest, false, err + } + if err := o.Put(ctx, TrunkObject(rev.Revision), list); err != nil { + return newest, false, fmt.Errorf("revision %d in %s: %w", rev.Revision, RevisionsObjects, err) + } + return rev, true, nil +} + +// Event is the seat's event assignment-revision of a kept revision. +func Event(rev meshwants.AssignmentRevision) meshwants.AssignmentRevisionEvent { + return meshwants.AssignmentRevisionEvent{V: rev.V, Revision: rev.Revision, Kind: rev.Kind, Cause: rev.Cause, Group: rev.Group, + Heads: rev.Heads, Catalogs: rev.Catalogs, At: rev.At} +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go new file mode 100644 index 00000000..69181919 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go @@ -0,0 +1,156 @@ +package assigngraph + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// ErrNoCandidate answers a verb asked of a candidate's revision: the assigner keeps none (the candidate's graph and the +// pins from the gate are milestone 47/G's W5.9, held with research 086's fold). +var ErrNoCandidate = errors.New("the assigner keeps no candidate's assignment graph revision") + +// ErrNoRevision answers a verb asked while no trunk revision is kept yet. +var ErrNoRevision = errors.New("no assignment graph revision is kept yet") + +func revisionOf(ctx context.Context, o Objects, revision int64, group, heads string) (meshwants.AssignmentRevision, error) { + if group != "" || heads != "" { + return meshwants.AssignmentRevision{}, fmt.Errorf("%w (group %q, heads %q)", ErrNoCandidate, group, heads) + } + if revision > 0 { + return ReadRevision(ctx, o, revision) + } + rev, has, err := Newest(ctx, o) + if err == nil && !has { + err = ErrNoRevision + } + return rev, err +} + +// Revision answers the seat's verb revision: one kept revision, the trunk's newest when none is named. +func Revision(ctx context.Context, o Objects, in meshwants.RevisionInput) (meshwants.RevisionAnswer, error) { + rev, err := revisionOf(ctx, o, in.Revision, in.Group, in.Heads) + return meshwants.RevisionAnswer{Revision: rev}, err +} + +// Entry answers the seat's verb entry: one entry whole, by its digest, or by its node and module in a revision (the +// trunk's newest when none is named); an acting edge names the seat whose rule gave it. +func Entry(ctx context.Context, o Objects, in meshwants.EntryInput) (meshwants.EntryAnswer, error) { + digest := in.Digest + if digest == "" { + if in.Node == "" || in.Module == "" { + return meshwants.EntryAnswer{}, errors.New("entry takes a digest, or a node and a module") + } + rev, err := revisionOf(ctx, o, in.Revision, in.Group, in.Heads) + if err != nil { + return meshwants.EntryAnswer{}, err + } + for _, ref := range rev.Entries { + if ref.Node == in.Node && ref.Module == in.Module { + digest = ref.Digest + } + } + if digest == "" { + return meshwants.EntryAnswer{}, fmt.Errorf("revision %d has no entry of %s on %s", rev.Revision, in.Module, in.Node) + } + } + e, err := ReadEntry(ctx, o, digest) + return meshwants.EntryAnswer{Digest: digest, Entry: e}, err +} + +// Difference answers the seat's verb difference: two kept revisions compared, each module on a node whose version +// differs, with the edges that moved, and the nodes it touches. +func Difference(ctx context.Context, o Objects, in meshwants.DifferenceInput) (meshwants.Difference, error) { + if in.Group != "" || in.Heads != "" { + return meshwants.Difference{}, fmt.Errorf("%w (group %q, heads %q)", ErrNoCandidate, in.Group, in.Heads) + } + if in.From <= 0 || in.To <= 0 { + return meshwants.Difference{}, errors.New("difference takes the two kept revisions it compares, from and to") + } + from, err := ReadRevision(ctx, o, in.From) + if err != nil { + return meshwants.Difference{}, err + } + to, err := ReadRevision(ctx, o, in.To) + if err != nil { + return meshwants.Difference{}, err + } + return Compare(ctx, o, from, to) +} + +// Compare is the graph difference of two revisions. +func Compare(ctx context.Context, o Objects, from, to meshwants.AssignmentRevision) (meshwants.Difference, error) { + d := meshwants.Difference{From: from.Revision, To: to.Revision, Changes: []meshwants.Change{}} + was, err := entriesOf(ctx, o, from) + if err != nil { + return d, err + } + now, err := entriesOf(ctx, o, to) + if err != nil { + return d, err + } + nodes := map[string]bool{} + for _, k := range slices.Sorted(maps.Keys(union(was, now))) { + a, hadA := was[k] + b, hasB := now[k] + if hadA && hasB && a.Fp == b.Fp { + continue + } + c := meshwants.Change{} + if hadA { + c.Node, c.Module, c.From = a.Node, a.Module, a.Fp + } + if hasB { + c.Node, c.Module, c.To = b.Node, b.Module, b.Fp + } + c.Edges = moved(a.Edges, b.Edges) + d.Changes = append(d.Changes, c) + nodes[c.Node] = true + } + d.Nodes = slices.Sorted(maps.Keys(nodes)) + return d, nil +} + +func entriesOf(ctx context.Context, o Objects, rev meshwants.AssignmentRevision) (map[string]meshwants.AssignmentEntry, error) { + out := make(map[string]meshwants.AssignmentEntry, len(rev.Entries)) + for _, ref := range rev.Entries { + e, err := ReadEntry(ctx, o, ref.Digest) + if err != nil { + return nil, fmt.Errorf("revision %d: %w", rev.Revision, err) + } + out[ref.Node+"/"+ref.Module] = e + } + return out, nil +} + +func union(a, b map[string]meshwants.AssignmentEntry) map[string]bool { + out := map[string]bool{} + for k := range a { + out[k] = true + } + for k := range b { + out[k] = true + } + return out +} + +// moved are the edges in one list and not the other. +func moved(a, b []meshwants.Edge) []meshwants.Edge { + var out []meshwants.Edge + for _, e := range b { + if !slices.Contains(a, e) { + out = append(out, e) + } + } + for _, e := range a { + if !slices.Contains(b, e) { + out = append(out, e) + } + } + sortEdges(out) + return out +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 0843516b..986d15dd 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -84,11 +84,15 @@ func TestManifest(t *testing.T) { // ADR 0567 §1, §6-§10: it holds the seat mesh-module-assigner, serving its five verbs (adopt and hand-back went with the // controller, ADR 0590); keeps the seat's events on its stream assignments for 30 days; calls the hand-act log; and reads // mesh-config's resolved values to compose every node a proposal touches. - // ADR 0565 §7, §9 (issue 1953 D4): and the seat mesh-wants, serving upgrade and bus. + // ADR 0565 §7, §9 (issue 1953 D4): and the seat mesh-wants, serving upgrade and bus, and (design 47 lines 1362-1366) + // revision, entry and difference from its revisions kept as objects in the old section, form 1, capped at 512 MiB. if len(m.Claims) != 2 || m.Claims[0].Name != meshmoduleassigner.Seat || m.Claims[0].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[0].Serves, []string{"assignments", "proposals", "propose", "withdraw", "again"}) || - m.Claims[1].Name != meshwants.Seat || m.Claims[1].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[1].Serves, []string{"upgrade", "bus"}) { - t.Errorf("the assigner claims %+v, want mesh-module-assigner with its five verbs and mesh-wants with its two", m.Claims) + m.Claims[1].Name != meshwants.Seat || m.Claims[1].At() != catalogue.ScopeMesh || !slices.Equal(m.Claims[1].Serves, []string{"upgrade", "bus", "revision", "entry", "difference"}) { + t.Errorf("the assigner claims %+v, want mesh-module-assigner with its five verbs and mesh-wants with its five", m.Claims) + } + if len(m.Objects) != 1 || m.Objects[0].Name != "assignment-revisions" || m.Objects[0].MaxMiB != 512 || m.Objects[0].Form != 1 { + t.Errorf("objects = %+v, want assignment-revisions, 512 MiB, form 1", m.Objects) } // Whether the seats' definitions let it hold them is the manifest validator's (the judge, at the seat list). // ADR 0621 §1: the first-view tools assignments, seat-holders and wants and design 25 § 12.2's six read tools, of the diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 78ce919a..e7f8e724 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -20,7 +20,10 @@ "scope": "mesh", "serves": [ "upgrade", - "bus" + "bus", + "revision", + "entry", + "difference" ] } ], @@ -115,6 +118,9 @@ { "name": "engine-reads", "history": 1 }, { "name": "written", "history": 5 } ], + "objects": [ + { "name": "assignment-revisions", "max-mib": 512, "form": 1 } + ], "own-secrets": { "broker": "${dir:state}/broker" },