W5.1, W5.2, W5.5: the assignment graph computed per branch, kept as objects, and served by mesh-wants' verbs

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
This commit is contained in:
2026-11-11 14:43:17 +01:00
parent ac2c4d3e20
commit 05494e7bb9
10 changed files with 852 additions and 5 deletions
@@ -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)
}
@@ -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.
@@ -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.
@@ -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 {
@@ -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 "<node>/<module>: 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)
}
@@ -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))
}
}
@@ -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.<digest>).
func EntryObject(digest string) string { return "entry." + digest }
// TrunkObject is the object name of a trunk revision (the seat's key form trunk.<revision>).
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}
}
@@ -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
}
@@ -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
+7 -1
View File
@@ -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"
},