diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 31ebb60..d6b2c40 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -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-, 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) } diff --git a/internal/link/composedstate.go b/internal/link/composedstate.go new file mode 100644 index 0000000..b069208 --- /dev/null +++ b/internal/link/composedstate.go @@ -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-` 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) +} diff --git a/internal/selfcompose/composedstate.go b/internal/selfcompose/composedstate.go new file mode 100644 index 0000000..8da3ea6 --- /dev/null +++ b/internal/selfcompose/composedstate.go @@ -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-` (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 (`..`), and the node's own +// composition (`.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 +} diff --git a/internal/selfcompose/composedstate_test.go b/internal/selfcompose/composedstate_test.go new file mode 100644 index 0000000..665e370 --- /dev/null +++ b/internal/selfcompose/composedstate_test.go @@ -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) + } +} diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index bd06c96..6c1e86d 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -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 diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index 7e08855..d88f0a1 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -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" ) diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index f1d2dff..0adeae2 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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-` (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. diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 7b3a698..a9e663a 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -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)