Merge branch 'feat/compose-once' of /home/jochen/projects/novox/mesh-catalog-compose into feat/node-packages

This commit is contained in:
2026-11-01 16:19:54 +01:00
5 changed files with 389 additions and 44 deletions
@@ -0,0 +1,225 @@
package main
// BenchmarkComposeNode times ComposeNode of an unchanged node over a mesh captured from the bus (issue 1954): the
// placement, wants, haves, the indexer's modules and manifests, the builder's builds, mesh-config's resolved values and
// the templates archives, read with the read tools (mesh-placement.placement, mesh-delivery-view.assigner_wants,
// node_haves, indexer_modules, builder_build, mesh-config.config) and the manifests and archives by digest. The capture
// is not kept in this repository (a few megabytes, and the mesh's own addresses and keys): the benchmark reads the
// directory MESH_COMPOSE_CAPTURE names and is skipped without it. MESH_COMPOSE_NODE names the node (novox when unset).
//
// The capture's layout: placement.json, wants.json, haves.json, modules.json and config.json as those tools answer,
// builds/<module>.<hex>.json as builder_build answers, manifests/<digest> and archives/<hex> as their bytes.
import (
"context"
"encoding/json"
"os"
"path/filepath"
"strings"
"sync/atomic"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// countingStore counts the holder's reads of its own states (and the providers' entries), each a bus request live.
type countingStore struct {
Store
gets, keys atomic.Int64
}
func (c *countingStore) Get(state, key string) (*Entry, error) {
c.gets.Add(1)
return c.Store.Get(state, key)
}
func (c *countingStore) Keys(state string) ([]string, error) {
c.keys.Add(1)
return c.Store.Keys(state)
}
// captured is the captured mesh as the states the holder reads, and the archives by digest.
func captured(b *testing.B, dir string) (map[string]map[string][]byte, map[string][]byte) {
b.Helper()
read := func(name string, into any) {
raw, err := os.ReadFile(filepath.Join(dir, name))
if err != nil {
b.Fatal(err)
}
if err := json.Unmarshal(raw, into); err != nil {
b.Fatalf("%s: %v", name, err)
}
}
states := map[string]map[string][]byte{}
put := func(state, key string, raw []byte) {
if states[state] == nil {
states[state] = map[string][]byte{}
}
states[state][key] = raw
}
var placement struct {
Keys map[string]struct {
Placement json.RawMessage `json:"placement"`
} `json:"keys"`
}
read("placement.json", &placement)
for node, k := range placement.Keys {
var p delivery.Placement
if err := json.Unmarshal(k.Placement, &p); err != nil {
b.Fatal(err)
}
p.Node = node
raw, err := delivery.EncodePlacement(p)
if err != nil {
b.Fatal(err)
}
put(PlacementRead, "placement."+node, raw)
}
var wants struct {
Wants map[string]json.RawMessage `json:"wants"`
}
read("wants.json", &wants)
for k, v := range wants.Wants {
put(WantsRead, k, v)
}
var haves struct {
Keys map[string]json.RawMessage `json:"keys"`
}
read("haves.json", &haves)
for k, v := range haves.Keys {
put(HaveRead, k, v)
}
var modules struct {
Modules map[string]json.RawMessage `json:"modules"`
}
read("modules.json", &modules)
for m, v := range modules.Modules {
key, err := delivery.ModulesKey(m)
if err != nil {
b.Fatal(err)
}
put(ModulesRead, key, v)
}
var config struct {
Resolved []struct {
Key, Module, Node, Source string
Value any
} `json:"resolved"`
}
read("config.json", &config)
for _, r := range config.Resolved {
raw, _ := json.Marshal(map[string]any{"value": r.Value, "source": r.Source})
put(ResolvedRead, r.Node+"."+r.Module+"."+r.Key, raw)
}
manifests, _ := filepath.Glob(filepath.Join(dir, "manifests", "*"))
for _, f := range manifests {
raw, err := os.ReadFile(f)
if err != nil {
b.Fatal(err)
}
digest := filepath.Base(f)
v, err := delivery.EncodeManifest(delivery.Manifest{Module: "captured", JSON: string(raw)})
if err != nil {
b.Fatal(err)
}
key, err := delivery.ManifestsKey(digest)
if err != nil {
b.Fatal(err)
}
put(ManifestsRead, key, v)
}
builds, _ := filepath.Glob(filepath.Join(dir, "builds", "*.json"))
for _, f := range builds {
var bb struct {
Module string `json:"module"`
Fingerprint string `json:"fingerprint"`
Build json.RawMessage `json:"build"`
}
read(strings.TrimPrefix(f, dir+"/"), &bb)
key, err := delivery.BuilderBuildKey(bb.Module, bb.Fingerprint)
if err != nil {
b.Fatal(err)
}
put(BuildsRead, key, bb.Build)
}
archives := map[string][]byte{}
files, _ := filepath.Glob(filepath.Join(dir, "archives", "*"))
for _, f := range files {
raw, err := os.ReadFile(f)
if err != nil {
b.Fatal(err)
}
archives[filepath.Base(f)] = raw
}
return states, archives
}
func BenchmarkComposeNode(b *testing.B) {
dir := os.Getenv("MESH_COMPOSE_CAPTURE")
if dir == "" {
b.Skip("MESH_COMPOSE_CAPTURE names no captured mesh")
}
node := os.Getenv("MESH_COMPOSE_NODE")
if node == "" {
node = "novox"
}
states, archives := captured(b, dir)
ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
for k, v := range states[state] {
if err := handle(stdio.StateChange{State: state, Key: k, Op: stdio.OpPut, Value: v, Current: true}); err != nil {
return err
}
}
return nil
}, func(state, key string) ([]byte, bool, error) {
v, ok := states[state][key]
return v, ok, nil
})
fetch := fromruntime.KeptFetch(func(_ context.Context, url string) ([]byte, error) {
_, hex, _ := strings.Cut(url, "sha256:")
raw, ok := archives[hex]
if !ok {
return nil, os.ErrNotExist
}
return raw, nil
})
own := &countingStore{Store: newStore()}
h := &Holder{Store: Watching(own, ws), Compose: composerOver(frombus.StatesReader(fromruntime.Watched{WatchedStates: ws}, fetch)),
Now: time.Now, Logf: func(string, ...any) {}}
// Composed twice before it is timed: a provision's state follows the states of those it depends on, which the first
// composition writes, so the second may still change a key; from then on nothing changes.
first, err := h.ComposeNode(context.Background(), node)
if err != nil {
b.Fatal(err)
}
if second, err := h.ComposeNode(context.Background(), node); err != nil {
b.Fatal(err)
} else if len(second.Changed) > 0 {
b.Logf("the second composition of %s changed %v", node, second.Changed)
}
b.Logf("first composition of %s: %d plans written, %d held, %d problems, %d keys changed", node, len(first.Plans),
len(first.Held), len(first.Problems), len(first.Changed))
for _, p := range first.Problems {
b.Logf("problem: %s", p)
}
own.gets.Store(0)
own.keys.Store(0)
b.ResetTimer()
var c Composed
for b.Loop() {
c, err = h.ComposeNode(context.Background(), node)
if err != nil {
b.Fatal(err)
}
if len(c.Plans)+len(c.Gone)+len(c.Changed) > 0 {
b.Fatalf("an unchanged node composed changes: %+v", c)
}
}
b.Logf("the last composition of %s: %s", node, TookSaid(c.Took))
b.ReportMetric(float64(own.gets.Load())/float64(b.N), "own-gets/op")
b.ReportMetric(float64(own.keys.Load())/float64(b.N), "own-keys/op")
}
@@ -75,12 +75,18 @@ var BoundReads = []string{"gitea", "keycloak", "postgres", "route-adapter", "rou
// boundRead is a provider's entries as this module reads them.
func boundRead(module string) string { return module + ".bound" }
// Composer is each module's provisioning plan on a node, composed from the bus: at the builds keptAt names
// (module → fingerprint), the newest build of every other placed module. A module whose build or plan refuses is held
// (frombus.PlansHeld; issue 774, one bad module must not drop its node): in held with why, none of its plan in plans.
// err is a refusal that is no single module's.
type Composer func(ctx context.Context, node string, keptAt map[string]string) (plans map[string]catalogue.Plan,
held map[string]string, err error)
// Composer is a node's provisioning plans, composed from the bus over one read: at the builds keptAt names (module →
// fingerprint, the newest build of every other placed module), and at the newest build of each placed module (issue
// 1954: once, not twice). A module whose build or plan refuses is held (frombus.PlansHeld; issue 774, one bad module
// must not drop its node): in held with why, none of its plan in plans; a scope's Err is a refusal that is no single
// module's.
type Composer func(ctx context.Context, node string, keptAt map[string]string) Composition
// Composition is a node's plans at its wanted builds and at its newest, and where the time went.
type Composition struct {
AtWants, Newest frombus.HeldPlans
Took []frombus.Phase
}
// Asker is the operator's channel as this module asks on it: publish an ask under its own name, and read the router's
// record of it.
@@ -111,13 +117,40 @@ type Composed struct {
// stay as they were, and every other module is composed (issue 774).
Held map[string]string `json:"modules-held,omitempty"`
Problems []string `json:"problems,omitempty"`
// Took is where the composition's time went, phase by phase in milliseconds, its whole last (issue 1954).
Took []Took `json:"took"`
}
// Took is one phase of a composition and its milliseconds.
type Took struct {
Phase string `json:"phase"`
MS int64 `json:"ms"`
}
// TookSaid says a composition's phases in one line.
func TookSaid(took []Took) string {
var said []string
for _, t := range took {
said = append(said, fmt.Sprintf("%s %d ms", t.Phase, t.MS))
}
return strings.Join(said, ", ")
}
// ComposeNode composes one node: its plans and its provisions.
func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error) {
func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, err error) {
h.mu.Lock()
defer h.mu.Unlock()
out := Composed{Node: node}
out = Composed{Node: node}
began, start := time.Now(), time.Now()
mark := func(phase string, took time.Duration) {
out.Took = append(out.Took, Took{Phase: phase, MS: took.Milliseconds()})
}
lap := func(phase string) {
now := time.Now()
mark(phase, now.Sub(start))
start = now
}
defer func() { mark("whole", time.Since(began)) }()
wanted, err := h.wantedOn(node)
if err != nil {
return out, err
@@ -129,14 +162,19 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error)
wanted[module] = fp
}
}
atWants, heldAtWants, err := h.Compose(ctx, node, wanted)
if err != nil {
return out, fmt.Errorf("the plans of %s cannot be composed: %w", node, err)
lap("wants")
c := h.Compose(ctx, node, wanted)
for _, p := range c.Took {
mark(p.Name, p.Took)
}
newest, heldNewest, err := h.Compose(ctx, node, nil)
if err != nil {
return out, fmt.Errorf("the plans of %s at the newest builds cannot be composed: %w", node, err)
start = time.Now()
if c.AtWants.Err != nil {
return out, fmt.Errorf("the plans of %s cannot be composed: %w", node, c.AtWants.Err)
}
if c.Newest.Err != nil {
return out, fmt.Errorf("the plans of %s at the newest builds cannot be composed: %w", node, c.Newest.Err)
}
atWants, heldAtWants, newest, heldNewest := c.AtWants.Plans, c.AtWants.Held, c.Newest.Plans, c.Newest.Held
// A module held at its wanted build has no plan composed: its last plan and its provisions are kept, said in
// `refused`. One held at its newest build alone keeps its plan at its want; what its newest build needs ahead waits.
if len(heldAtWants) > 0 {
@@ -153,6 +191,7 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error)
if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil {
out.Problems = append(out.Problems, err.Error())
}
lap("plans written")
needed := map[string]provisions.Provision{}
elsewhere := map[string]provisions.Provision{}
vaultNode := h.vaultNode()
@@ -199,6 +238,7 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error)
}
h.judgeElsewhere(elsewhere, &out)
h.judgeNode(node, needed, heldAtWants, &out)
lap("provisions judged")
return out, nil
}
@@ -1046,21 +1086,40 @@ func orSaid(words ...string) string {
}
// FromWatched is the holder's composer over one watched copy of its reads (novox/hq ADR 0580 clause 2): frombus's plans
// of a node, at the builds keptAt names, read from the copy (placement, haves, modules, resolved values), a build by key
// and a manifest once per digest; no state is read key by key per composition. A plan names no value, so no own secret
// and no credential is read (compose.Planning).
// of a node, at the builds keptAt names and at the newest over one read of the copy (placement, haves, modules,
// resolved values), a build by key and a manifest once per digest; no state is read key by key per composition. A plan
// names no value, so no own secret and no credential is read (compose.Planning).
func FromWatched(ws *stdio.WatchedStates) Composer {
r := fromruntime.WatchedReader(ws)
return func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
return frombus.PlansHeld(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
return composerOver(fromruntime.WatchedReader(ws))
}
// composerOver is the holder's composer over a reader: both scopes over one read (frombus.PlansHeldTwice).
func composerOver(r frombus.Reader) Composer {
return func(ctx context.Context, node string, keptAt map[string]string) Composition {
at, newest, took := frombus.PlansHeldTwice(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
return Composition{AtWants: at, Newest: newest, Took: took}
}
}
// WatchedReads are the states the holder reads whole, each from the watched copy: what the composer reads and the
// wants (ADR 0580 clause 2). Its own states and the providers' entries stay read from the bus, after its own writes.
var WatchedReads = []string{PlacementRead, WantsRead, ModulesRead, HaveRead, ResolvedRead}
// wants (ADR 0580 clause 2), and the providers' entries, whose watch is also what rejudges a key.
var WatchedReads = append([]string{PlacementRead, WantsRead, ModulesRead, HaveRead, ResolvedRead}, boundReads()...)
// Watching is a Store whose WatchedReads come from the copy; every other state, and every write, from s.
// OwnStates are the states the holder alone writes (ADR 0570 clause 1), read from the watched copy too (issue 1954: some
// 250 to 400 reads by key per composition): each of its own writes is taken into the copy at once (WatchedStates.Wrote),
// so it reads back what it wrote before the watch hands it back.
var OwnStates = []string{PlansState, ProvisionsState, RefusedState}
func boundReads() []string {
out := make([]string, 0, len(BoundReads))
for _, p := range BoundReads {
out = append(out, boundRead(p))
}
return out
}
// Watching is a Store whose WatchedReads and OwnStates come from the copy, its writes of OwnStates taken into the copy as
// they land, and a manifest read once by its digest; every other state, and every write, from s.
func Watching(s Store, ws *stdio.WatchedStates) Store { return watchingStore{Store: s, ws: ws} }
type watchingStore struct {
@@ -1068,11 +1127,24 @@ type watchingStore struct {
ws *stdio.WatchedStates
}
// RevisionPutter is a Store that says the revision a put landed at (the bus does), so the copy keeps it exactly.
type RevisionPutter interface {
PutRevision(state, key string, value any) (uint64, error)
}
func (w watchingStore) copied(state string) bool {
return slices.Contains(WatchedReads, state) || slices.Contains(OwnStates, state)
}
func (w watchingStore) Get(state, key string) (*Entry, error) {
if !slices.Contains(WatchedReads, state) {
get := w.ws.Get
switch {
case state == ManifestsRead:
get = w.ws.Once // a manifest by its digest never changes: read once, kept (ADR 0580 clause 1)
case !w.copied(state):
return w.Store.Get(state, key)
}
v, found, err := w.ws.Get(state, key)
v, found, err := get(state, key)
if err != nil || !found {
return nil, err
}
@@ -1080,12 +1152,39 @@ func (w watchingStore) Get(state, key string) (*Entry, error) {
}
func (w watchingStore) Keys(state string) ([]string, error) {
if !slices.Contains(WatchedReads, state) {
if !w.copied(state) {
return w.Store.Keys(state)
}
return w.ws.Keys(state)
}
func (w watchingStore) Put(state, key string, value any) error {
if !slices.Contains(OwnStates, state) {
return w.Store.Put(state, key, value)
}
raw, err := json.Marshal(value)
if err != nil {
return err
}
var revision uint64
if rp, ok := w.Store.(RevisionPutter); ok {
revision, err = rp.PutRevision(state, key, json.RawMessage(raw))
} else {
err = w.Store.Put(state, key, json.RawMessage(raw))
}
if err != nil {
return err
}
return w.ws.Wrote(state, key, raw, revision)
}
func (w watchingStore) Delete(state, key string) error {
if err := w.Store.Delete(state, key); err != nil || !slices.Contains(OwnStates, state) {
return err
}
return w.ws.Removed(state, key)
}
// NodesOfNodeKey is the node a key that starts with it names: a resolved value's `<node>.<module>.<key>`, a have's
// `<node>.<module>` or `<node>.node`. A change of either recomposes that node (ADR 0580 clause 2).
func NodesOfNodeKey(key string) []string {
@@ -22,7 +22,6 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/compose/frombus/frombustest"
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
@@ -262,9 +261,7 @@ func TestTheHolderComposesProvidesAndAsksOnTheBus(t *testing.T) {
reader := frombus.StatesReader(fromruntime.Watched{WatchedStates: ws}, fetch)
a := &asked{}
h := &Holder{Store: Watching(s, ws), Asks: a, Now: time.Now, Logf: t.Logf,
Compose: func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
return frombus.PlansHeld(ctx, reader, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
}}
Compose: composerOver(reader)}
state := func(key string) provisions.Provision {
e, _ := s.Get(ProvisionsState, key)
if e == nil {
@@ -15,6 +15,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provisions"
)
@@ -68,12 +69,10 @@ func newMesh() *mesh {
now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC)
m.h = &Holder{Store: m.s, Asks: m.r, Logf: func(string, ...any) {},
Now: func() time.Time { now = now.Add(time.Second); return now },
Compose: func(_ context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
Compose: func(_ context.Context, node string, keptAt map[string]string) Composition {
m.composed = append(m.composed, node)
if keptAt != nil {
return m.atWant[node], m.heldAt[node], nil
}
return m.atNewest[node], nil, nil
return Composition{AtWants: frombus.HeldPlans{Plans: m.atWant[node], Held: m.heldAt[node]},
Newest: frombus.HeldPlans{Plans: m.atNewest[node]}}
}}
// What each provider's manifest declares: postgres, keycloak and the vault keep entries, the others report by their
// have.
@@ -385,12 +384,9 @@ func TestAModuleHeldWithoutAWantIsComposedAtTheBuildItRuns(t *testing.T) {
m.s.write(t, HaveRead, "novox.nats", `{"v":1,"fp":"`+fp+`","phase":"running","health":"ok"}`)
m.s.write(t, HaveRead, "g14.dunst", `{"v":1,"fp":"src2:`+strings.Repeat("ee", 32)+`","phase":"running","health":"ok"}`)
var keptAt map[string]string
m.h.Compose = func(_ context.Context, node string, at map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
if at != nil {
keptAt = at
return map[string]catalogue.Plan{"nats": {}}, nil, nil
}
return nil, nil, nil
m.h.Compose = func(_ context.Context, node string, at map[string]string) Composition {
keptAt = at
return Composition{AtWants: frombus.HeldPlans{Plans: map[string]catalogue.Plan{"nats": {}}}}
}
// ADR 0580 clause 2: the reads come from one watched copy, each state watched once and none read key by key from the
// bus, and a change the copy takes wakes the node it touches (a resolved value its node).
@@ -419,8 +415,16 @@ func TestAModuleHeldWithoutAWantIsComposedAtTheBuildItRuns(t *testing.T) {
if i == 0 && !slices.Contains(c.Plans, "novox.nats") {
t.Fatalf("no plan written for a module held without a want: %+v", c)
}
// Issue 1954: its own states come from the copy too, its own writes read back at once (nothing is written
// again), and the result says where the time went, its whole last.
if i == 1 && len(c.Plans)+len(c.Changed) > 0 {
t.Fatalf("composed again unchanged, it wrote %+v: its own writes were not read back", c)
}
if n := len(c.Took); n == 0 || c.Took[n-1].Phase != "whole" || c.Took[0].Phase != "wants" {
t.Fatalf("the composition says its phases as %v", c.Took)
}
}
for _, r := range WatchedReads {
for _, r := range append(slices.Clone(WatchedReads), OwnStates...) {
if bus.read[r] != 0 || watches[r] > 1 {
t.Errorf("%s was read %d times from the bus and watched %d times", r, bus.read[r], watches[r])
}
@@ -43,6 +43,11 @@ func (runtimeStore) Put(state, key string, value any) error {
return err
}
// PutRevision is Put, saying the revision it landed at.
func (runtimeStore) PutRevision(state, key string, value any) (uint64, error) {
return stdio.State(state).Put(key, value)
}
func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }
func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
@@ -95,14 +100,28 @@ func (w *work) run(h *Holder) {
c, err := h.ComposeNode(context.Background(), n)
switch {
case err != nil:
logf("[%s] %s is not composed: %v", Module, n, err)
logf("[%s] %s is not composed (%s): %v", Module, n, TookSaid(c.Took), err)
case len(c.Plans)+len(c.Gone)+len(c.Changed)+len(c.Asked)+len(c.Problems) > 0:
logf("[%s] composed %s: %+v", Module, n, c)
case slowest(c.Took) >= slowAt:
// Nothing changed and it took long: where the time went is said, so a slow composition is seen (issue 1954).
logf("[%s] composed %s, nothing changed: %s", Module, n, TookSaid(c.Took))
}
}
}
}
// slowAt is how long a composition that changed nothing may take before its phases are said.
const slowAt = time.Second
// slowest is a composition's whole time: its last phase, "whole".
func slowest(took []Took) time.Duration {
if len(took) == 0 {
return 0
}
return time.Duration(took[len(took)-1].MS) * time.Millisecond
}
func main() {
s := runtimeStore{}
ws := stdio.NewWatchedStates() // one copy of what it reads, each state by its watch (ADR 0580 clause 2)
@@ -122,7 +141,8 @@ func main() {
return nil
})
for _, provider := range BoundReads {
go stdio.State(boundRead(provider)).Follow("", func(c stdio.StateChange) error {
// Into the copy the holder reads them from, its handler run after the copy took the change.
go ws.Follow(boundRead(provider), func(c stdio.StateChange) error {
if said, err := h.Rejudge(c.Key); err != nil {
logf("[%s] %s is not judged again: %v", Module, c.Key, err)
} else if said != "" {