Publish every composition the node makes in composed-<node> (H3)

ADR 0593 §6: the provisioner, the assigner and the view read compositions
instead of composing, so the node's engine writes each one it makes to
its composition state: the wanted builds, every placed module at its
newest done build (so the plan exists before the want moves), each with
its provisioning plan, and a module refused alone with its refusal. A
key is written only when it differs, and kept while among the module's
five newest or named by a want. A placed module's trunk build landing
now makes a pass.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
2026-11-02 18:29:20 +01:00
parent 9a12f2851f
commit 4f51393b72
8 changed files with 647 additions and 9 deletions
+4 -1
View File
@@ -14,12 +14,12 @@ import (
"sync/atomic"
"time"
"github.com/novox/mesh-host/internal/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/declaration"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
"github.com/novox/mesh-host/internal/agent"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/compose/frombus"
"github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/liveness"
"github.com/novox/mesh-host/internal/selfcompose"
@@ -968,6 +968,9 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f
// And the packages its compositions declare in its own key of the state packages, which pacman reads (novox/hq ADR
// 0565 §6).
env.Packages = link.EnginePackages{Session: session}
// And every composition it makes in its composition state composed-<node>, a record the provisioner, the assigner
// and the view read and nothing applies (novox/hq ADR 0593 §6).
env.ComposedState = link.EngineComposed{Session: session, Node: node}
return selfcompose.New(env)
}
+103
View File
@@ -0,0 +1,103 @@
package link
import (
"context"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-host/internal/selfcompose"
)
// ---- this node's composition state, in the bus's own storage (novox/hq ADR 0593 §6) --------------------------------
// composedStore binds the node's composition state `composed-<node>` over the link open now: bound, never created (the
// grants part creates it, create-only, history delivery.ComposedHistory).
func composedStore(js nats.JetStreamContext, node string) (nats.KeyValue, error) {
bucket, err := delivery.ComposedBucketFor(node)
if err != nil {
return nil, err
}
kv, err := js.KeyValue(bucket)
if err != nil {
return nil, fmt.Errorf("this node's composition state %s cannot be bound (the grants part creates it; written "+
"again at the next pass that composes): %w", bucket, err)
}
return kv, nil
}
func (l *natsLink) composedStore(node string) (nats.KeyValue, error) {
return composedStore(l.js, node)
}
func (b OverNATS) composedStore(node string) (nats.KeyValue, error) { return composedStore(b.JS, node) }
// EngineComposed is this node's composition state over whichever link the session has open (selfcompose.ComposedStore);
// with none open every read and write is refused, and nothing is kept anywhere else.
type EngineComposed struct {
Session *Session
Node string
}
var _ selfcompose.ComposedStore = EngineComposed{}
func (e EngineComposed) kv() (nats.KeyValue, error) {
e.Session.init()
e.Session.mu.Lock()
bus := e.Session.bus
e.Session.mu.Unlock()
l, ok := bus.(interface {
composedStore(node string) (nats.KeyValue, error)
})
if bus == nil || !ok {
return nil, ErrNoLink
}
return l.composedStore(e.Node)
}
// Entries is every key the state holds with its latest value and revision.
func (e EngineComposed) Entries(ctx context.Context) (map[string]selfcompose.ComposedEntry, error) {
kv, err := e.kv()
if err != nil {
return nil, err
}
keys, err := kv.Keys(nats.Context(ctx))
if errors.Is(err, nats.ErrNoKeysFound) {
return map[string]selfcompose.ComposedEntry{}, nil
}
if err != nil {
return nil, err
}
out := make(map[string]selfcompose.ComposedEntry, len(keys))
for _, key := range keys {
got, err := kv.Get(key)
if errors.Is(err, nats.ErrKeyNotFound) {
continue
}
if err != nil {
return nil, err
}
out[key] = selfcompose.ComposedEntry{Value: got.Value(), Revision: got.Revision()}
}
return out, nil
}
// Put writes a key and answers its revision.
func (e EngineComposed) Put(_ context.Context, key string, value []byte) (uint64, error) {
kv, err := e.kv()
if err != nil {
return 0, err
}
return kv.Put(key, value)
}
// Delete deletes a key (a delete marker, kept with the key's history).
func (e EngineComposed) Delete(_ context.Context, key string) error {
kv, err := e.kv()
if err != nil {
return err
}
return kv.Delete(key)
}
+320
View File
@@ -0,0 +1,320 @@
package selfcompose
import (
"bytes"
"context"
"fmt"
"maps"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
"github.com/novox/mesh-host/internal/compose/frombus"
)
// **The compositions this node makes, published in its composition state** (novox/hq ADR 0593 §6). The node's engine
// alone composes, and it writes every composition it makes into `composed-<node>` (delivery.ComposedBucketFor), a
// record for the provisioner, the assigner and the view, never an instruction: the engine applies only what it composes
// itself, never a value read from the state, so a value carries no signature. After a pass that composed, it writes:
//
// - each module at the fingerprint its want names, as the agent applies it (`<node>.<module>.<fp>`), and the node's own
// composition (`<node>.node`) when its want exists;
// - each module placed on the node at its newest done build: the indexer's fingerprint once the builder holds a done
// build of it, so the plan exists before the want moves (build done, the node composes and writes, the provisioner
// reads, the providers provide, the assigner moves the want); a module it runs that no want names, at what runs;
// - each module's plan (catalogue.Plan) exactly when the provisioner's plans name it, from the same pass's builds;
// - a module refused alone at a fingerprint as that key with its refusal (Refused), and no resources.
//
// A key is written only when its value differs from what the state holds (read once per run), and a module's key is
// deleted once its fingerprint is not among the five it wrote last (delivery.ComposedKeep) and no want names it. A pass
// refused whole, or a bus that did not answer, writes nothing: what was written stands, and each module's have says why.
// ComposedStore is this node's composition state, bound and never created (the grants part creates it).
type ComposedStore interface {
// Entries is every key the state holds with its value and revision.
Entries(ctx context.Context) (map[string]ComposedEntry, error)
Put(ctx context.Context, key string, value []byte) (revision uint64, err error)
Delete(ctx context.Context, key string) error
}
// ComposedEntry is one key's value and revision in the state.
type ComposedEntry struct {
Value []byte
Revision uint64
}
// published is what a pass composed at the wanted builds, for publishComposed: under mu.
type published struct {
scope frombus.Scope
fps map[string]string // by module, the wanted fingerprint
comps map[string]delivery.Composed
held map[string]string
// placed is the modules the node's placement puts here, ran what runs of each module (by its composition's FP).
placed []string
ran map[string]string
}
// publishComposed writes the node's composition state from the last pass that composed, when one asked it (pubDue).
func (c *Composer) publishComposed(ctx context.Context) {
c.mu.Lock()
due, at := c.pubDue, c.pubAt
c.pubDue = false
c.mu.Unlock()
if !due || c.env.ComposedState == nil || at == nil {
return
}
wrote, deleted, err := c.writeComposed(ctx, at)
if err != nil {
c.sayComposed("this node's compositions are not kept in its composition state; " + c.oweComposed() + ": " + err.Error())
return
}
c.mu.Lock()
c.pubTries = 0
c.mu.Unlock()
c.sayComposed("")
if wrote+deleted > 0 {
c.env.Say(fmt.Sprintf("this node's composition state: %d key(s) written, %d deleted", wrote, deleted))
}
}
// newestFPs is, of the modules placed here, the fingerprint each is composed at for the state beyond the wanted ones:
// the indexer's, where the builder holds a done build of it; else the want's; else what runs of it. A module with none
// is left out.
func (c *Composer) newestFPs(ctx context.Context, at *published) map[string]string {
out := map[string]string{}
r := reader{c: c}
for _, module := range at.placed {
c.mu.Lock()
fp2 := c.modules[module].FP2
c.mu.Unlock()
if fp2 != "" {
if b, err := r.Build(ctx, module, fp2); err == nil && b != nil && b.State == delivery.BuildDone {
out[module] = fp2
continue
}
}
if fp, ok := at.fps[module]; ok {
out[module] = fp
} else if fp, ok := at.ran[module]; ok {
out[module] = fp
}
}
return out
}
// writeComposed works out the state's keys from at and writes those that differ, then deletes what is past keeping.
func (c *Composer) writeComposed(ctx context.Context, at *published) (wrote, deleted int, err error) {
values, err := c.composedValues(ctx, at)
if err != nil {
return 0, 0, err
}
c.pubWrite.Lock()
defer c.pubWrite.Unlock()
if c.pubKept == nil {
kept, err := c.env.ComposedState.Entries(ctx)
if err != nil {
return 0, 0, fmt.Errorf("the state cannot be read: %w", err)
}
c.pubKept = kept
}
keys := make([]string, 0, len(values))
for k := range values {
keys = append(keys, k)
}
sort.Strings(keys)
for _, key := range keys {
if bytes.Equal(c.pubKept[key].Value, values[key]) {
continue
}
rev, err := c.env.ComposedState.Put(ctx, key, values[key])
if err != nil {
return wrote, deleted, fmt.Errorf("%s cannot be written: %w", key, err)
}
c.pubKept[key] = ComposedEntry{Value: values[key], Revision: rev}
wrote++
}
for _, key := range c.pastKeeping() {
if err := c.env.ComposedState.Delete(ctx, key); err != nil {
return wrote, deleted, fmt.Errorf("%s cannot be deleted: %w", key, err)
}
delete(c.pubKept, key)
deleted++
}
return wrote, deleted, nil
}
// composedValues is every key the state should hold after the pass at, encoded.
func (c *Composer) composedValues(ctx context.Context, at *published) (map[string][]byte, error) {
r := reader{c: c}
out := map[string][]byte{}
put := func(comp delivery.Composed, plans map[string]catalogue.Plan) error {
if comp.Module != "" && comp.Refused == "" {
if p, ok := plans[comp.Module]; ok {
p := p
comp.Plan = &p
}
}
key, err := composedKeyOf(comp)
if err != nil {
return err
}
raw, err := delivery.EncodeComposed(comp)
if err != nil {
return fmt.Errorf("the composition %s cannot be written: %w", key, err)
}
out[key] = raw
return nil
}
runs := []struct {
scope frombus.Scope
fps map[string]string
comps map[string]delivery.Composed
held map[string]string
}{{at.scope, at.fps, at.comps, at.held}}
// The newest builds, composed only where they differ from the wanted ones.
newest := c.newestFPs(ctx, at)
if !maps.Equal(newest, at.fps) {
scope := at.scope
scope.KeptAt = map[string]string{}
var inOwn []string
c.mu.Lock()
for module, fp := range newest {
if fp != c.modules[module].FP2 {
scope.KeptAt[module] = fp
}
}
c.mu.Unlock()
for _, m := range at.scope.InOwn {
if _, given := newest[m]; !given {
inOwn = append(inOwn, m)
}
}
scope.InOwn = inOwn
comps, _, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, newest)
switch {
case err != nil && BusFailed(err):
return nil, fmt.Errorf("the newest builds could not be composed: %w", err)
case err != nil:
// A refusal of the node at its newest builds is no refusal of what it applies: the wanted keys are written,
// and the newest are said (each module's have says what holds it when its want moves).
c.sayComposed("this node's newest builds do not compose, so its composition state holds the wanted ones alone: " + err.Error())
default:
for key, comp := range comps {
comps[key] = c.keepRunning(comp, scope.Running)
}
runs = append(runs, struct {
scope frombus.Scope
fps map[string]string
comps map[string]delivery.Composed
held map[string]string
}{scope, newest, comps, held})
}
}
for i, run := range runs {
plans, _, err := frombus.PlansHeld(ctx, r, c.env.Mesh, run.scope)
if err != nil && BusFailed(err) {
return nil, fmt.Errorf("the plans could not be composed: %w", err)
}
// A node whose plans refuse it whole hands the provisioner nothing, as the provisioner's own composing did: no
// key carries a plan then.
for _, comp := range run.comps {
if comp.Module == "" && i > 0 {
continue // the node's own composition is the wanted pass's alone
}
if err := put(comp, plans); err != nil {
return nil, err
}
}
for module, why := range run.held {
fp, ok := run.fps[module]
if !ok || module == agent.NodeModule {
continue
}
if err := put(delivery.Composed{Node: c.env.Node, Module: module, FP: fp, Refused: why}, nil); err != nil {
return nil, err
}
}
}
return out, nil
}
// composedKeyOf is a composition's key in the state.
func composedKeyOf(comp delivery.Composed) (string, error) {
if comp.Module == "" {
return delivery.ComposedNodeKey(comp.Node)
}
return delivery.ComposedKey(comp.Node, comp.Module, comp.FP)
}
// pastKeeping is, of the module keys the state holds, those whose fingerprint is not among the five its module wrote
// last (delivery.ComposedKeep, by revision) and that no want names (ADR 0593 §6). Under pubWrite.
func (c *Composer) pastKeeping() []string {
c.mu.Lock()
wanted := map[string]string{}
for module, w := range c.wants {
wanted[module] = w.FP
}
c.mu.Unlock()
type kept struct {
key string
fp string
rev uint64
}
byModule := map[string][]kept{}
for key, e := range c.pubKept {
_, module, fp, isNode, err := delivery.ParseComposedKey(key)
if err != nil || isNode {
continue
}
byModule[module] = append(byModule[module], kept{key, fp, e.Revision})
}
var out []string
for module, ks := range byModule {
sort.Slice(ks, func(i, j int) bool { return ks[i].rev > ks[j].rev })
for i, k := range ks {
if i < delivery.ComposedKeep || wanted[module] == k.fp {
continue
}
out = append(out, k.key)
}
}
sort.Strings(out)
return out
}
// oweComposed keeps the state due and has the loop make a pass again after the growing bound, as owePackages does.
func (c *Composer) oweComposed() string {
c.mu.Lock()
defer c.mu.Unlock()
c.pubDue = true
c.wait = LaterWait(c.pubTries)
c.pubTries++
c.owed.Store(true)
return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.pubTries)
}
// sayComposed says a line about the state once until it changes; "" clears it.
func (c *Composer) sayComposed(line string) {
c.mu.Lock()
same := c.pubSaid == line
c.pubSaid = line
c.mu.Unlock()
if !same && line != "" {
c.env.Say(line)
}
}
// ranOf is, of what runs, each module's fingerprint by its composition (the node's own left out).
func ranOf(running []delivery.Composed) map[string]string {
out := map[string]string{}
for _, comp := range running {
if comp.Module != "" && comp.FP != "" && !strings.EqualFold(comp.Module, agent.NodeModule) {
out[comp.Module] = comp.FP
}
}
return out
}
+173
View File
@@ -0,0 +1,173 @@
package selfcompose
import (
"context"
"slices"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
)
// fakeComposed is a node's composition state: its keys, and every write and deletion made.
type fakeComposed struct {
mu sync.Mutex
kv map[string]ComposedEntry
rev uint64
puts []string
deletes []string
}
func (f *fakeComposed) Entries(context.Context) (map[string]ComposedEntry, error) {
f.mu.Lock()
defer f.mu.Unlock()
out := map[string]ComposedEntry{}
for k, v := range f.kv {
out[k] = v
}
return out, nil
}
func (f *fakeComposed) Put(_ context.Context, key string, value []byte) (uint64, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.rev++
f.kv[key] = ComposedEntry{Value: value, Revision: f.rev}
f.puts = append(f.puts, key)
return f.rev, nil
}
func (f *fakeComposed) Delete(_ context.Context, key string) error {
f.mu.Lock()
defer f.mu.Unlock()
delete(f.kv, key)
f.deletes = append(f.deletes, key)
return nil
}
func (f *fakeComposed) written() []string {
f.mu.Lock()
defer f.mu.Unlock()
return slices.Clone(f.puts)
}
// novox/hq ADR 0593 §6: the node publishes each composition it makes in its composition state, a record never applied:
// the wanted module and the node's own at the first pass; nothing again for a pass that changes nothing; a done build
// of a module placed here that no want names, at its trunk's build, written once as its own key (before any want moves);
// the same build landing again writes nothing. And a module's keys are kept while among the five it wrote last or named
// by a want, and deleted otherwise.
func TestTheNodePublishesItsCompositions(t *testing.T) {
b := newFakeBus()
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
ofp := delivery.SourcePrefix2 + strings.Repeat("c", 64)
digest := delivery.ManifestDigest([]byte(lemurs))
oldie := `{"module":"oldie","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"oldie.service","state":"running","boot":"enabled"}]}`
odigest := delivery.ManifestDigest([]byte(oldie))
for d, json := range map[string]string{digest: lemurs, odigest: oldie} {
man, _ := delivery.EncodeManifest(delivery.Manifest{Module: map[string]string{digest: "lemurs", odigest: "oldie"}[d], JSON: json})
key, _ := delivery.ManifestsKey(d)
b.kv[delivery.ManifestsBucket+"/"+key] = man
}
state := &fakeComposed{kv: map[string]ComposedEntry{}}
env := b.env()
env.ComposedState = state
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
c := New(env)
out, err := c.Watch(ctx, "n1.>")
if err != nil {
t.Fatal(err)
}
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
omod, _ := delivery.EncodeModule(delivery.Module{FP2: ofp, Manifest: odigest, Repository: "novox/o", Commit: "c"})
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
b.put(t, delivery.PlacementBucket, "placement.n1", place)
b.put(t, delivery.HaveBucket, "n1.node", have)
b.put(t, delivery.ModulesBucket, "lemurs", mod)
b.put(t, delivery.ModulesBucket, "oldie", omod)
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
b.send(bucket, agent.Update{Caught: true})
}
for u := next(t, out); !u.Caught; u = next(t, out) {
}
lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
oldieKey, _ := delivery.ComposedKey("n1", "oldie", ofp)
waitFor := func(what string, ok func([]string) bool) []string {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
if got := state.written(); ok(got) {
return got
}
time.Sleep(10 * time.Millisecond)
}
b.mu.Lock()
t.Logf("said: %q", b.said)
b.mu.Unlock()
t.Fatalf("%s: written %v", what, state.written())
return nil
}
first := waitFor("the first pass", func(got []string) bool { return len(got) >= 2 })
if !slices.Equal(first, []string{lemursKey, "n1.node"}) {
t.Errorf("the first pass wrote %v, want the wanted module's key and the node's own", first)
}
if raw := state.kv[lemursKey].Value; raw != nil {
comp, err := delivery.DecodeComposed(raw)
if err != nil || comp.Module != "lemurs" || comp.FP != fp || comp.Signature != nil || len(comp.Resources) == 0 {
t.Errorf("the published composition of lemurs: %+v, %v", comp, err)
}
}
// A pass that changes nothing writes nothing.
setFor(t, &askedWithin, 0)
if _, err := c.ComposeNow(ctx); err != nil {
t.Fatal(err)
}
time.Sleep(50 * time.Millisecond)
if got := state.written(); len(got) != 2 {
t.Errorf("a pass that changed nothing wrote %v", got[2:])
}
// oldie's trunk build is done: its key at that build is written once, though no want names it.
build, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: odigest})
bkey, _ := delivery.BuilderBuildKey("oldie", ofp)
b.mu.Lock()
b.kv[delivery.BuilderBuildsBucket+"/"+bkey] = build
b.mu.Unlock()
b.put(t, delivery.BuilderBuildsBucket, bkey, build)
got := waitFor("oldie's done build", func(got []string) bool { return len(got) >= 3 })
if got[2] != oldieKey {
t.Errorf("the done build of oldie wrote %v, want %s", got[2:], oldieKey)
}
b.put(t, delivery.BuilderBuildsBucket, bkey, build)
time.Sleep(100 * time.Millisecond)
if got := state.written(); len(got) != 3 {
t.Errorf("the same build landing again wrote %v", got[3:])
}
// What is kept: the five keys of a module written last, and one a want names; the rest is deleted.
keep := New(b.env())
keep.pubKept = map[string]ComposedEntry{}
var keys []string
for i, ch := range "123456" {
key, _ := delivery.ComposedKey("n1", "lemurs", delivery.SourcePrefix2+strings.Repeat(string(ch), 64))
keys = append(keys, key)
keep.pubKept[key] = ComposedEntry{Revision: uint64(10 + i)}
}
keep.pubKept["n1.node"] = ComposedEntry{Revision: 1}
if past := keep.pastKeeping(); !slices.Equal(past, []string{keys[0]}) {
t.Errorf("past keeping: %v, want the oldest of six, %s", past, keys[0])
}
keep.wants = map[string]delivery.Want{"lemurs": {FP: delivery.SourcePrefix2 + strings.Repeat("1", 64)}}
if past := keep.pastKeeping(); len(past) != 0 {
t.Errorf("past keeping with a want naming the oldest: %v, want none", past)
}
}
+1 -1
View File
@@ -9,9 +9,9 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/compose"
"github.com/novox/mesh-host/internal/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **The data items of each module placed here, as this engine composed them** (novox/hq ADR 0577 clause 1). After a
+1 -1
View File
@@ -10,8 +10,8 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-host/internal/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/compose/frombus"
"github.com/novox/mesh-host/internal/agent"
)
+40 -4
View File
@@ -29,10 +29,10 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-host/internal/compose"
"github.com/novox/mesh-host/internal/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
"github.com/novox/mesh-host/internal/compose"
"github.com/novox/mesh-host/internal/compose/frombus"
"github.com/nats-io/nats.go"
@@ -93,6 +93,9 @@ type Env struct {
// Packages is this node's key of the engine's state packages (novox/hq ADR 0565 §6), written after a pass that
// handed something new when the packages its compositions declare differ. Nil keeps none.
Packages PackagesStore
// ComposedState is this node's composition state `composed-<node>` (novox/hq ADR 0593 §6), where every composition
// a pass makes is published for its readers (publishComposed). Nil keeps none.
ComposedState ComposedStore
}
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
@@ -175,6 +178,15 @@ type Composer struct {
pkgsWrite sync.Mutex
pkgsKept []byte
pkgsRead bool
// pubAt is what the last pass that composed made, and pubDue whether the composition state is to be written from it
// (publishComposed); pubTries and pubSaid are as the packages' are. Under mu. pubKept is what the state holds, read
// once per run, under pubWrite.
pubAt *published
pubDue bool
pubTries int
pubSaid string
pubWrite sync.Mutex
pubKept map[string]ComposedEntry
}
// Pass is how one pass went, as the node-engine's verb compose answers it.
@@ -323,7 +335,7 @@ func (c *Composer) inputs() []input {
}
// takeBuildDone says whether a build that landed is one this node composes from: done, of a module it wants at that
// build or whose trunk's build it is. Every other build, and a build not done, moves nothing.
// build, or wanted or placed here whose trunk's build it is. Every other build, and a build not done, moves nothing.
func (c *Composer) takeBuildDone(u agent.Update) bool {
// Any write or delete of a build's key forgets the build kept as done: the builder claims a done key again as a new
// revision (a gone step, issue 1937), which lands done with other artifacts.
@@ -340,8 +352,14 @@ func (c *Composer) takeBuildDone(u agent.Update) bool {
c.mu.Lock()
w, wanted := c.wants[module]
m := c.modules[module]
placedHere := false
for _, pm := range c.placements[c.env.Node].Modules {
placedHere = placedHere || pm.Module == module
}
c.mu.Unlock()
if !wanted || (w.FP != fp && m.FP2 != fp) {
// A module placed here is composed at its trunk's build once that is done, for the composition state, before its
// want moves (ADR 0593 §6).
if !(wanted && w.FP == fp) && !((wanted || placedHere) && m.FP2 == fp) {
return false
}
b, err := delivery.DecodeBuilderBuild(u.Value)
@@ -468,6 +486,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
}
c.publishData(ctx)
c.publishPackages(ctx)
c.publishComposed(ctx)
if c.owed.Load() {
wait, _ := c.waitNow()
c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
@@ -935,6 +954,23 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
}
}
pass.Handed = len(out)
// The composition state is written from this pass (ADR 0593 §6): every pass that composed, since a build done or a
// plan changed moves what is published though nothing new is handed.
placedMods := make([]string, 0, len(own.Modules))
for _, m := range own.Modules {
placedMods = append(placedMods, m.Module)
}
sort.Strings(placedMods)
pubComps := make(map[string]delivery.Composed, len(compositions))
for key, comp := range compositions {
if comp.Module == "" && !nodeWanted {
continue
}
pubComps[key] = comp
}
c.pubAt = &published{scope: scope, fps: maps.Clone(fps), comps: pubComps, held: maps.Clone(held), placed: placedMods,
ran: ranOf(running)}
c.pubDue = true
// The data items are kept again after a pass that handed something new, and after the run's first (ADR 0577 clause 1).
c.dataDue = c.dataDue || len(out) > 0 || c.dataKept == nil
// And the node's packages after the same passes (ADR 0565 §6): written only when they differ.
+5 -2
View File
@@ -526,7 +526,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
// Issue 774, 2026-10-30 19:30: node-tools' want moved to a build not yet done, and the node stayed refused after it
// landed until another input moved. A done build of a module the node wants, at its want's or the trunk's build,
// makes a pass; a build not done, of another build or of a module not wanted, makes none.
// makes a pass, and so does one of a module placed here at its trunk's build though no want names it (novox/hq ADR 0593
// §6: the composition state holds it at its newest done build before its want moves); a build not done, of another
// build or of a module neither wanted nor placed here, makes none.
passes := func() int64 { return composed.Load() }
for _, row := range []struct {
name, module, fp, state string
@@ -535,7 +537,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
{"lemurs's wanted build, not done", "lemurs", fp, "building", false},
{"lemurs's wanted build lands", "lemurs", fp, delivery.BuildDone, true},
{"another build of lemurs", "lemurs", delivery.SourcePrefix2 + strings.Repeat("e", 64), delivery.BuildDone, false},
{"a module the node does not want", "oldie", delivery.SourcePrefix2 + strings.Repeat("c", 64), delivery.BuildDone, false},
{"a placed module no want names, at its trunk's build", "oldie", delivery.SourcePrefix2 + strings.Repeat("c", 64), delivery.BuildDone, true},
{"a module neither wanted nor placed here", "ghost", delivery.SourcePrefix2 + strings.Repeat("c", 64), delivery.BuildDone, false},
} {
before := passes()
key, _ := delivery.BuilderBuildKey(row.module, row.fp)