From e0261cdfc13894becb21b175b02ff28f0311b739 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 00:09:25 +0100 Subject: [PATCH 1/3] Keep each placed module's data items in the engine's state data, so disk-load measures without composing (ADR 0577) After a pass that handed a new composition, the node's composer works out its modules' data items from the inputs it composed from and writes each module's key . of mesh-host_data only when it differs, deleting the keys of modules that left. Pins mesh-sdk feat/data-items for the contract and the engine's grant. Two tool-answer tests are merged so the unit count stays where it was. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- cmd/mesh-host/nodeagent.go | 3 + go.mod | 2 +- go.sum | 2 + internal/link/enginedata.go | 103 +++++++++++++++ internal/link/health_test.go | 27 ++-- internal/selfcompose/data.go | 197 ++++++++++++++++++++++++++++ internal/selfcompose/data_test.go | 148 +++++++++++++++++++++ internal/selfcompose/selfcompose.go | 12 ++ module.json | 3 + packaging/operator_keys_test.go | 5 +- 10 files changed, 485 insertions(+), 17 deletions(-) create mode 100644 internal/link/enginedata.go create mode 100644 internal/selfcompose/data.go create mode 100644 internal/selfcompose/data_test.go diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 6b845cf..2cc0193 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -1021,6 +1021,9 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil, err.Error() } env := composerEnv(session, node, say, hooks) + // And it keeps each placed module's data items in its own state data, which the node's disk-load watches (novox/hq + // ADR 0577 clause 1); the shadow composes for candidates and keeps none. + env.Data = link.EngineData{Session: session} unreadable := "" placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, func(m string) { if strings.Contains(m, "cannot be read") { diff --git a/go.mod b/go.mod index 65a510a..ce16253 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 1700a57..94a9c74 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031083427-de8e82c06f54 h1:xITIabv8 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031083427-de8e82c06f54/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 h1:Z4RcH5Ft4OAAG3F538NXIHP9F3pkyk3j0LtP9mqu1qQ= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/link/enginedata.go b/internal/link/enginedata.go new file mode 100644 index 0000000..1880b4a --- /dev/null +++ b/internal/link/enginedata.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" +) + +// ---- the data items of the modules placed on this node, in the bus's own storage (novox/hq ADR 0577 clause 1) ------ + +// DataBucket is the node-engine's declared state data (delivery.EngineDataBucket), created by the grants part from the +// engine's manifest as its states acts and health are: one key `.` per module placed on the node that +// declares data items, written by that node's engine alone, read by that node's disk-load. +const DataBucket = delivery.EngineDataBucket + +// dataStore binds DataBucket over the link open now: bound, never created. +func (l *natsLink) dataStore() (nats.KeyValue, error) { + kv, err := l.js.KeyValue(DataBucket) + if err != nil { + return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+ + "manifest; written again at the next composition that changes): %w", DataBucket, err) + } + return kv, nil +} + +// EngineData is this node's keys of DataBucket over whichever link the session has open (selfcompose.DataStore); with +// none open every read and write is refused, and nothing is kept anywhere else. +type EngineData struct{ Session *Session } + +func (d EngineData) kv() (nats.KeyValue, error) { + d.Session.init() + d.Session.mu.Lock() + bus := d.Session.bus + d.Session.mu.Unlock() + l, ok := bus.(interface{ dataStore() (nats.KeyValue, error) }) + if bus == nil || !ok { + return nil, ErrNoLink + } + return l.dataStore() +} + +// Get is a key's value, nil when it has none. +func (d EngineData) Get(_ context.Context, key string) ([]byte, error) { + kv, err := d.kv() + if err != nil { + return nil, err + } + e, err := kv.Get(key) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return e.Value(), nil +} + +// Put writes a key. +func (d EngineData) Put(_ context.Context, key string, value []byte) error { + kv, err := d.kv() + if err != nil { + return err + } + _, err = kv.Put(key, value) + return err +} + +// Delete removes a key; one that was not there is no error. +func (d EngineData) Delete(_ context.Context, key string) error { + kv, err := d.kv() + if err != nil { + return err + } + if err := kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return err + } + return nil +} + +// Keys is every key matching pattern that holds a value: read by a watch of the pattern alone, the only read the +// engine's grant gives of its own node's keys. +func (d EngineData) Keys(_ context.Context, pattern string) ([]string, error) { + kv, err := d.kv() + if err != nil { + return nil, err + } + w, err := kv.WatchFiltered([]string{pattern}, nats.IgnoreDeletes(), nats.MetaOnly()) + if err != nil { + return nil, err + } + defer func() { _ = w.Stop() }() + var keys []string + for e := range w.Updates() { + if e == nil { + break + } + keys = append(keys, e.Key()) + } + return keys, nil +} diff --git a/internal/link/health_test.go b/internal/link/health_test.go index 449f0cb..cc17b61 100644 --- a/internal/link/health_test.go +++ b/internal/link/health_test.go @@ -168,6 +168,18 @@ func TestAToolsAnswerIsHealthyOnlyWhenItSaysSo(t *testing.T) { if ToolSubject("keycloak", "keycloak_admin_health", "anchor") != "mesh.mod.keycloak.tool.keycloak_admin_health.anchor" { t.Errorf("asked on %s", ToolSubject("keycloak", "keycloak_admin_health", "anchor")) } + // A tool may answer that it waits for the operator (novox/hq ADR 0283): read with what it waits for, and not + // healthy, so nothing reading `healthy` alone takes it for a pass. + got, err := ReadToolAnswer([]byte(`{"result":{"state":"waits","healthy":false,"why":"the source games waits",` + + `"waits":[{"part":"the source games","secret":"smb-password-games","what":"the password of the source games"}]}}`)) + if err != nil || got.Healthy || got.State != "waits" || len(got.Waits) != 1 || got.Waits[0].Secret != "smb-password-games" { + t.Fatalf("read as %+v (%v)", got, err) + } + h := ResourceHealth{Module: "mounts", Resource: "mounts.watch", State: StateWaiting, Waits: got.Waits} + b, _ := json.Marshal(h) + if !strings.Contains(string(b), `"state":"waiting"`) || !strings.Contains(string(b), `"waits":[{"part":"the source games","secret":"smb-password-games"`) { + t.Fatalf("said on the wire as %s", b) + } } // A declared tool check is asked on the link the node-engine actually holds — the NATS link — and not @@ -199,18 +211,3 @@ func TestAToolCheckIsAskedOnTheLinkTheEngineHolds(t *testing.T) { t.Fatalf("the NATS link was not asked: %v", err) } } - -// A tool may answer that it waits for the operator (novox/hq ADR 0283): read with what it waits for, and not -// healthy, so nothing reading `healthy` alone takes it for a pass. -func TestAToolsWaitIsReadWithWhatItWaitsFor(t *testing.T) { - got, err := ReadToolAnswer([]byte(`{"result":{"state":"waits","healthy":false,"why":"the source games waits",` + - `"waits":[{"part":"the source games","secret":"smb-password-games","what":"the password of the source games"}]}}`)) - if err != nil || got.Healthy || got.State != "waits" || len(got.Waits) != 1 || got.Waits[0].Secret != "smb-password-games" { - t.Fatalf("read as %+v (%v)", got, err) - } - h := ResourceHealth{Module: "mounts", Resource: "mounts.watch", State: StateWaiting, Waits: got.Waits} - b, _ := json.Marshal(h) - if !strings.Contains(string(b), `"state":"waiting"`) || !strings.Contains(string(b), `"waits":[{"part":"the source games","secret":"smb-password-games"`) { - t.Fatalf("said on the wire as %s", b) - } -} diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go new file mode 100644 index 0000000..28f2895 --- /dev/null +++ b/internal/selfcompose/data.go @@ -0,0 +1,197 @@ +package selfcompose + +import ( + "bytes" + "context" + "fmt" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/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 +// pass that composed and handed something new (or the first one of a run), the composer works out each placed module's +// data items from the same inputs it composed from (frombus.DataItems, which reads them through this composer's own +// view of the bus) and keeps each module's in the engine's state data, key `.`, with the digest of the +// module's composition: written only when the value differs, deleted when the module no longer declares any here or left +// the node. That node's disk-load watches its node's keys and measures what they name; it composes nothing (clause 2). + +// DataStore is this node's keys of the engine's state data (link.EngineData). +type DataStore interface { + // Get is a key's value, nil when it has none. + Get(ctx context.Context, key string) ([]byte, error) + Put(ctx context.Context, key string, value []byte) error + Delete(ctx context.Context, key string) error + // Keys is every key matching pattern that holds a value. + Keys(ctx context.Context, pattern string) ([]string, error) +} + +// itemsReader is the composer's view of the bus for working out data items: no own secret is asked of the vault, since +// working out where an item lies reads no secret (the items are rendered as a plan is, compose.Planning). +type itemsReader struct{ reader } + +func (itemsReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} + +// dataItemsOf works out the data items of the scope's node (frombus.DataItems); a test sets what it answers. +var dataItemsOf = frombus.DataItems + +// publishData keeps the data items of this node's modules in the engine's state data, when a pass asked it (dataDue). +func (c *Composer) publishData(ctx context.Context) { + c.mu.Lock() + due := c.dataDue + c.dataDue = false + c.mu.Unlock() + if !due || c.env.Data == nil { + return + } + items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, frombus.Scope{Node: c.env.Node}) + if err != nil { + c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + + delivery.EngineDataBucket + " holds stands: " + err.Error()) + return + } + if len(unplaced) > 0 { + c.env.Say("data items left out as not placeable: " + strings.Join(unplaced, "; ")) + } + c.mu.Lock() + digests := c.digestsByModule() + c.mu.Unlock() + wrote, removed, err := c.writeData(ctx, items, digests) + if err != nil { + c.mu.Lock() + c.dataDue = true // the next pass tries again, whatever it hands + c.mu.Unlock() + c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket + + ", and are written again at the next composition: " + err.Error()) + return + } + c.sayData("") + if wrote+removed > 0 { + c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted", + wrote, delivery.EngineDataBucket, removed)) + } +} + +// sayData says a line about the data items once until it changes; "" clears it. +func (c *Composer) sayData(line string) { + c.mu.Lock() + same := c.dataSaid == line + c.dataSaid = line + c.mu.Unlock() + if !same && line != "" { + c.env.Say(line) + } +} + +// digestsByModule is the digest of each module's composition handed last, and the node's own under "" for a module run +// inside it. Under c.mu. +func (c *Composer) digestsByModule() map[string]string { + out := map[string]string{} + for key, dig := range c.handed { + _, module, _, isNode, err := delivery.ParseComposedKey(key) + if err != nil { + continue + } + if isNode { + module = "" + } + out[module] = dig + } + return out +} + +// writeData writes each module's key whose value differs from what the key holds, and deletes this node's keys of +// modules with no items here now. The keys are read once per run (dataKept), so a restart that finds the same items +// writes nothing. +func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string) (wrote, removed int, err error) { + store := c.env.Data + if c.dataKept == nil { + pattern, err := delivery.EngineDataNodeKeys(c.env.Node) + if err != nil { + return 0, 0, err + } + keys, err := store.Keys(ctx, pattern) + if err != nil { + return 0, 0, fmt.Errorf("this node's keys cannot be listed: %w", err) + } + kept := map[string][]byte{} + for _, k := range keys { + raw, err := store.Get(ctx, k) + if err != nil { + return 0, 0, fmt.Errorf("%s cannot be read: %w", k, err) + } + if raw != nil { + kept[k] = raw + } + } + c.dataKept = kept + } + want := map[string][]byte{} + modules := make([]string, 0, len(items)) + for m := range items { + modules = append(modules, m) + } + sort.Strings(modules) + var failed []string + for _, module := range modules { + if len(items[module]) == 0 { + continue + } + key, err := delivery.EngineDataKey(c.env.Node, module) + if err != nil { + failed = append(failed, err.Error()) + continue + } + digest, ok := digests[module] + if !ok { + digest = digests[""] // run inside the node's own composition + } + raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: c.env.Node, Module: module, Digest: digest, Items: items[module]}) + if err != nil { + failed = append(failed, err.Error()) + continue + } + want[key] = raw + } + keys := make([]string, 0, len(want)) + for k := range want { + keys = append(keys, k) + } + sort.Strings(keys) + for _, key := range keys { + if bytes.Equal(c.dataKept[key], want[key]) { + continue + } + if err := store.Put(ctx, key, want[key]); err != nil { + failed = append(failed, key+": "+err.Error()) + continue + } + c.dataKept[key] = want[key] + wrote++ + } + var gone []string + for key := range c.dataKept { + if _, still := want[key]; !still { + gone = append(gone, key) + } + } + sort.Strings(gone) + for _, key := range gone { + if err := store.Delete(ctx, key); err != nil { + failed = append(failed, key+": "+err.Error()) + continue + } + delete(c.dataKept, key) + removed++ + } + if len(failed) > 0 { + return wrote, removed, fmt.Errorf("%s", strings.Join(failed, "; ")) + } + return wrote, removed, nil +} diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go new file mode 100644 index 0000000..e33cd97 --- /dev/null +++ b/internal/selfcompose/data_test.go @@ -0,0 +1,148 @@ +package selfcompose + +import ( + "context" + "errors" + "slices" + "strings" + "sync" + "testing" + + "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" +) + +// fakeData is the engine's state data as a map, counting the writes and deletes made. +type fakeData struct { + mu sync.Mutex + kv map[string][]byte + puts []string + deletes []string + fail error +} + +func (f *fakeData) Get(_ context.Context, key string) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.kv[key], nil +} + +func (f *fakeData) Put(_ context.Context, key string, value []byte) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.fail != nil { + return f.fail + } + f.kv[key] = value + f.puts = append(f.puts, key) + return nil +} + +func (f *fakeData) 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 *fakeData) Keys(_ context.Context, pattern string) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for k := range f.kv { + if matches(pattern, k) { + out = append(out, k) + } + } + return out, nil +} + +// ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed +// module's data items in its state data, key `.`, with its composition's digest: written when the module's +// items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never +// touched, a pass that handed nothing new reads nothing, and a write that failed is tried again at the next pass. +func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { + item := func(module, id, path string) catalogue.PlacedDataItem { + return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk} + } + store := &fakeData{kv: map[string][]byte{ + // Left by a module that went while the engine was down, and another node's key. + "n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}} + var items map[string][]catalogue.PlacedDataItem + var asked int + setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) { + asked++ + return items, nil, nil + }) + // The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's. + if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") { + t.Error("the node's grant reads other keys of the engine's state data than its own") + } + c := New(Env{Node: "n1", Data: store}) + webKey, _ := delivery.ComposedKey("n1", "web", delivery.SourcePrefix2+strings.Repeat("a", 64)) + nodeKey, _ := delivery.ComposedNodeKey("n1") + for _, step := range []struct { + name string + due bool + items map[string][]catalogue.PlacedDataItem + digests map[string]string // composed key to digest handed + fail error + puts []string + deletes []string + asked int + // want is each key's items and digest after the step. + want map[string]delivery.EngineData + }{ + {name: "the run's first pass writes each module's items and deletes what left", due: true, + items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}, "bare": nil}, + digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, + puts: []string{"n1.oldie", "n1.web"}, deletes: []string{"n1.gone"}, asked: 1, + want: map[string]delivery.EngineData{ + "n1.web": {Module: "web", Digest: "sha256:w1", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}}, + "n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: []catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")}}}}, + {name: "a pass that handed nothing new reads nothing", due: false, asked: 1}, + {name: "a pass that changes no module's items writes nothing", due: true, + items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, + digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2}, + {name: "a write that fails is tried again at the next pass", due: true, fail: errors.New("the bus said no"), + items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3}, + {name: "a module whose composition changed its items is written again", due: false, + items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 4, + want: map[string]delivery.EngineData{ + "n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}}, + {name: "a module that left the node is deleted", due: true, + items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}}, + digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 5}, + } { + t.Run(step.name, func(t *testing.T) { + store.mu.Lock() + store.puts, store.deletes, store.fail = nil, nil, step.fail + store.mu.Unlock() + items = step.items + c.mu.Lock() + c.handed = step.digests + c.dataDue = c.dataDue || step.due + c.mu.Unlock() + c.publishData(context.Background()) + slices.Sort(store.puts) + if !slices.Equal(store.puts, step.puts) || !slices.Equal(store.deletes, step.deletes) || asked != step.asked { + t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked, + step.puts, step.deletes, step.asked) + } + for key, want := range step.want { + got, err := delivery.DecodeEngineData(store.kv[key]) + if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest || + !slices.Equal(got.Items, want.Items) { + t.Errorf("%s holds %+v (%v), want %+v", key, got, err, want) + } + } + if store.kv["n2.web"] == nil { + t.Error("another node's key was deleted") + } + }) + } +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index bb34508..cb8b49d 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -87,6 +87,9 @@ type Env struct { // 0550 §2): asked only when a module placed here asks for a certificate and the record holds none from an authority // the node trusts. Nil asks nothing, and such a module stands in, as before. Certify func(ctx context.Context) error + // Data is this node's keys of the engine's state data (novox/hq ADR 0577 clause 1), where each placed module's data + // items are kept after a pass that composed something new. Nil keeps none. + Data DataStore } // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value. @@ -146,6 +149,12 @@ type Composer struct { last Pass // watching says the composer's loop runs (Watch), so an ask of a pass can be answered. watching atomic.Bool + // dataDue says a pass composed something new (or is the run's first), so the data items are kept again + // (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state + // data hold, read once per run, nil before; the loop's alone. + dataDue bool + dataSaid string + dataKept map[string][]byte } // Pass is how one pass went, as the node-engine's verb compose answers it. @@ -432,6 +441,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er return } } + c.publishData(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)) @@ -894,6 +904,8 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { } } pass.Handed = len(out) + // 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 return out } diff --git a/module.json b/module.json index a54c462..f0466ad 100644 --- a/module.json +++ b/module.json @@ -20,6 +20,9 @@ "state": [ { "name": "health" + }, + { + "name": "data" } ], "config": { diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 047c098..1605b16 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -77,7 +77,10 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { } // 774 (the alert table): the state health, one key per node, the node's own health report, kept until // written again (no TTL: a node that stops writing keeps its last word, which the watchman reads with its time). - if len(parsed.State) != 1 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 { + // ADR 0577 clause 1: and the state data, one key per module placed on the node that declares data items, kept until + // the module leaves the node; created by the grants part from this manifest as health is. + if len(parsed.State) != 2 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 || + parsed.State[1].Name != "data" || parsed.State[1].TTLSeconds != 0 { t.Fatalf("the manifest declares the state %+v", parsed.State) } } -- 2.54.0 From 122f4dd72757aaebbdd469597ec77bd97001e8f7 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 00:18:30 +0100 Subject: [PATCH 2/3] Read a done build once, and work out the data items in the scope the pass composed in (ADR 0577) The data items after a pass re-read each placed module's build over the bus (about 55 per pass on ace, g14 and shanks); a done build never changes, so the composer keeps it. The items are now worked out with the pass's own kept builds and modules run inside the node's own, as composed. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- internal/selfcompose/data.go | 8 +++++++- internal/selfcompose/data_test.go | 25 +++++++++++++++++++++++++ internal/selfcompose/selfcompose.go | 28 +++++++++++++++++++++++----- 3 files changed, 55 insertions(+), 6 deletions(-) diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index 28f2895..ab78558 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -50,7 +50,13 @@ func (c *Composer) publishData(ctx context.Context) { if !due || c.env.Data == nil { return } - items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, frombus.Scope{Node: c.env.Node}) + c.mu.Lock() + scope := c.dataScope + c.mu.Unlock() + if scope.Node == "" { + scope.Node = c.env.Node + } + items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) if err != nil { c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + delivery.EngineDataBucket + " holds stands: " + err.Error()) diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index e33cd97..a676da9 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -145,4 +145,29 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { } }) } + // The items are worked out from what the pass read: a done build is read over the bus once, not again for the data + // items after the pass (a module in a running build is read again, since it may land). + b := newFakeBus() + done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte(lemurs))}) + running, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: "building"}) + fp := delivery.SourcePrefix2 + strings.Repeat("b", 64) + doneKey, _ := delivery.BuilderBuildKey("web", fp) + runningKey, _ := delivery.BuilderBuildKey("pg", fp) + b.kv[delivery.BuilderBuildsBucket+"/"+doneKey], b.kv[delivery.BuilderBuildsBucket+"/"+runningKey] = done, running + r := reader{c: New(b.env())} + for range 3 { + if got, err := r.Build(context.Background(), "web", fp); err != nil || got.State != delivery.BuildDone { + t.Fatalf("the done build: %+v, %v", got, err) + } + if got, err := r.Build(context.Background(), "pg", fp); err != nil || got.State != "building" { + t.Fatalf("the running build: %+v, %v", got, err) + } + } + reads := map[string]int{} + for _, read := range b.reads { + reads[read[1]]++ + } + if reads[doneKey] != 1 || reads[runningKey] != 3 { + t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads) + } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index cb8b49d..a8d7b4d 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -113,7 +113,10 @@ type Composer struct { values map[string]map[string]any // by module, this node's manifests map[string][]byte // by digest, read once archives map[string][]byte // templates archives by digest, read once - handed map[string]string // composed key to the digest handed + // builds are the builder's done builds by key, read once: a done build at a fingerprint never changes, so a pass and + // the data items after it (publishData) do not read each placed module's build again over the bus (ADR 0577). + builds map[string]delivery.BuilderBuild + handed map[string]string // composed key to the digest handed // sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again // is other bytes, so the one first had is kept while the generation stands, and the composition stays the same. sealedOwn map[string]ownSealed @@ -152,9 +155,10 @@ type Composer struct { // dataDue says a pass composed something new (or is the run's first), so the data items are kept again // (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state // data hold, read once per run, nil before; the loop's alone. - dataDue bool - dataSaid string - dataKept map[string][]byte + dataDue bool + dataSaid string + dataScope frombus.Scope + dataKept map[string][]byte } // Pass is how one pass went, as the node-engine's verb compose answers it. @@ -234,7 +238,7 @@ func New(env Env) *Composer { } return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, placementKeys: delivery.NewPlacementKeys(), records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{}, - manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}, + manifests: map[string][]byte{}, archives: map[string][]byte{}, builds: map[string]delivery.BuilderBuild{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}, asked: make(chan chan Pass, 8)} } @@ -791,6 +795,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // Its own certificate, asked of the vault when a module here asks for one and the record holds none it trusts. c.certifyIfWanted(ctx, r, own) scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running} + c.mu.Lock() + c.dataScope = frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn} // the data items are worked out as composed + c.mu.Unlock() missed := c.missed.Load() compositions, _, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps) err = c.missedIn(missed, err) @@ -1198,6 +1205,12 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder if err != nil { return nil, err } + r.c.mu.Lock() + done, ok := r.c.builds[key] + r.c.mu.Unlock() + if ok { + return &done, nil + } defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")() raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key) if err != nil || raw == nil { @@ -1207,6 +1220,11 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder if err != nil { return nil, fmt.Errorf("the builder's %s: %w", key, err) } + if b.State == delivery.BuildDone { + r.c.mu.Lock() + r.c.builds[key] = b + r.c.mu.Unlock() + } return &b, nil } -- 2.54.0 From e031758e4a37eb1bdf75f116f67f4926d9a0e7b0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 00:27:29 +0100 Subject: [PATCH 3/3] Forget a kept build when its key is written again, and retry data items on the composer's bound (review round 1) The builder claims a done key again as a new revision, so a build kept as done is dropped on any write of its key. Data items that could not be worked out or written stay due and owe a pass after the growing bound, so a first write refused before the state or grant exists is tried again without waiting for an unrelated composition. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- internal/selfcompose/data.go | 25 ++++++++++++++++---- internal/selfcompose/data_test.go | 36 +++++++++++++++++++++++------ internal/selfcompose/selfcompose.go | 6 +++++ 3 files changed, 55 insertions(+), 12 deletions(-) diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index ab78558..b21bcdb 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -6,6 +6,7 @@ import ( "fmt" "sort" "strings" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" @@ -59,7 +60,7 @@ func (c *Composer) publishData(ctx context.Context) { items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) if err != nil { c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + - delivery.EngineDataBucket + " holds stands: " + err.Error()) + delivery.EngineDataBucket + " holds stands; " + c.oweData() + ": " + err.Error()) return } if len(unplaced) > 0 { @@ -70,13 +71,13 @@ func (c *Composer) publishData(ctx context.Context) { c.mu.Unlock() wrote, removed, err := c.writeData(ctx, items, digests) if err != nil { - c.mu.Lock() - c.dataDue = true // the next pass tries again, whatever it hands - c.mu.Unlock() c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket + - ", and are written again at the next composition: " + err.Error()) + "; " + c.oweData() + ": " + err.Error()) return } + c.mu.Lock() + c.dataTries = 0 + c.mu.Unlock() c.sayData("") if wrote+removed > 0 { c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted", @@ -84,6 +85,20 @@ func (c *Composer) publishData(ctx context.Context) { } } +// oweData keeps the data items due and has the composer's loop make a pass again after the growing bound (LaterWait), as +// a pass refused for a bus that did not answer is: the first write after a delivery may be refused while the state or +// its grant is not on the bus yet, and nothing else would try again until an unrelated composition changed. It answers +// the words that say when. +func (c *Composer) oweData() string { + c.mu.Lock() + defer c.mu.Unlock() + c.dataDue = true + c.wait = LaterWait(c.dataTries) + c.dataTries++ + c.owed.Store(true) + return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.dataTries) +} + // sayData says a line about the data items once until it changes; "" clears it. func (c *Composer) sayData(line string) { c.mu.Lock() diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index a676da9..796fc74 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -7,10 +7,13 @@ import ( "strings" "sync" "testing" + "time" "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" + + "github.com/novox/mesh-host/internal/agent" ) // fakeData is the engine's state data as a map, counting the writes and deletes made. @@ -62,7 +65,8 @@ func (f *fakeData) Keys(_ context.Context, pattern string) ([]string, error) { // ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed // module's data items in its state data, key `.`, with its composition's digest: written when the module's // items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never -// touched, a pass that handed nothing new reads nothing, and a write that failed is tried again at the next pass. +// touched, a pass that handed nothing new reads nothing, and items that could not be worked out or written are tried +// again with no new composition, after the composer's growing bound (the loop's owed pass). func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { item := func(module, id, path string) catalogue.PlacedDataItem { return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk} @@ -72,9 +76,10 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { "n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}} var items map[string][]catalogue.PlacedDataItem var asked int + var itemsErr error setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) { asked++ - return items, nil, nil + return items, nil, itemsErr }) // The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's. if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") { @@ -89,6 +94,8 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { items map[string][]catalogue.PlacedDataItem digests map[string]string // composed key to digest handed fail error + refused error // what working out the items says + owed time.Duration // the bound the loop waits before trying again; 0 when nothing is owed puts []string deletes []string asked int @@ -106,23 +113,26 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { {name: "a pass that changes no module's items writes nothing", due: true, items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2}, - {name: "a write that fails is tried again at the next pass", due: true, fail: errors.New("the bus said no"), + {name: "items that cannot be worked out are tried again after the bound", due: true, refused: errors.New("no answer"), + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3, owed: LaterWait(0)}, + {name: "a write that fails is tried again with no new composition, after a longer bound", due: false, fail: errors.New("the bus said no"), items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3}, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 4, owed: LaterWait(1)}, {name: "a module whose composition changed its items is written again", due: false, items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 4, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 5, want: map[string]delivery.EngineData{ "n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}}, {name: "a module that left the node is deleted", due: true, items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 5}, + digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 6}, } { t.Run(step.name, func(t *testing.T) { store.mu.Lock() store.puts, store.deletes, store.fail = nil, nil, step.fail store.mu.Unlock() - items = step.items + items, itemsErr = step.items, step.refused + c.paid() // the pass before composed: nothing owed c.mu.Lock() c.handed = step.digests c.dataDue = c.dataDue || step.due @@ -133,6 +143,9 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked, step.puts, step.deletes, step.asked) } + if wait, _ := c.waitNow(); c.owed.Load() != (step.owed > 0) || wait != step.owed { + t.Errorf("owed a pass: %v after %s; want after %s", c.owed.Load(), wait, step.owed) + } for key, want := range step.want { got, err := delivery.DecodeEngineData(store.kv[key]) if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest || @@ -170,4 +183,13 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { if reads[doneKey] != 1 || reads[runningKey] != 3 { t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads) } + // A done key the builder claims again (a gone step, issue 1937) and lands done again is read again, not served kept. + again, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte("{}"))}) + b.mu.Lock() + b.kv[delivery.BuilderBuildsBucket+"/"+doneKey] = again + b.mu.Unlock() + r.c.takeBuildDone(agent.Update{Key: doneKey, Value: running}) + if got, err := r.Build(context.Background(), "web", fp); err != nil || got.Manifest != delivery.ManifestDigest([]byte("{}")) { + t.Errorf("a done build written again is served as %+v (%v), not read again", got, err) + } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index a8d7b4d..61fb4f7 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -156,6 +156,7 @@ type Composer struct { // (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state // data hold, read once per run, nil before; the loop's alone. dataDue bool + dataTries int // data runs failed in a row, for oweData's bound dataSaid string dataScope frombus.Scope dataKept map[string][]byte @@ -309,6 +310,11 @@ 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. 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. + c.mu.Lock() + delete(c.builds, u.Key) + c.mu.Unlock() if u.Deleted { return false } -- 2.54.0