diff --git a/cmd/mesh-host/file_references_test.go b/cmd/mesh-host/file_references_test.go index a8303fb..0be46f6 100644 --- a/cmd/mesh-host/file_references_test.go +++ b/cmd/mesh-host/file_references_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "fmt" "os" "path/filepath" @@ -10,6 +11,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/link" ) @@ -82,6 +84,77 @@ func TestADeclarationWithAFileByDigestAppliesFromTheObject(t *testing.T) { if err != nil || string(got) != string(body) { t.Errorf("1695/AC1: the file holds %d bytes (%v), want %d", len(got), err, len(body)) } + + // Issue 924 (novox, 2026-10-29 01:16): the controller's sweep removed an object a composition the node ran still named, + // and the module's apply failed while it ran. Content the node once had is its own keep's, and the bus is not asked. + gone := aFilesLink{err: fmt.Errorf("the object %s was deleted from the bucket declaration-files-novox", name)} + never := []byte(strings.Repeat("never had here\n", 6000)) + for _, row := range []struct { + name string + raw []byte + onDisk []byte // nil removes the file + refused string // "" applies + }{ + {"the file here drifted, its object deleted", raw, []byte("the module wrote it since"), ""}, + {"the file here is gone, its object deleted", raw, nil, ""}, + {"never had here, its object deleted", referenceDeclaration(target, never), nil, "never held it"}, + } { + if row.onDisk == nil { + _ = os.Remove(target) + } else if err := os.WriteFile(target, row.onDisk, 0o600); err != nil { + t.Fatal(err) + } + report = applyAndKeep(link.WithFiles(context.Background(), gone), opts, row.raw, nil, nil, nil) + if row.refused == "" { + got, _ := os.ReadFile(target) + if report.Refused != "" || string(got) != string(body) { + t.Errorf("924: %s: refused %q, the file holds %d bytes; want it applied from the node's own keep", + row.name, report.Refused, len(got)) + } + } else if !strings.Contains(report.Refused, row.refused) || !strings.Contains(report.Refused, "was deleted") { + t.Errorf("924: %s: refused %q; want a refusal saying the bucket and the keep both lack it", row.name, report.Refused) + } + } + + // Issue 924, the live case: what runs came from the kept declaration, whose export names a digest swept long ago, + // while the node wrote a newer export since. Composed as it runs, the file names what the node wrote, and applies with + // no object; a file someone else changed since is left as named. + newer := []byte(strings.Repeat("the export the node wrote since\n", 3000)) + newerName, _ := declaration.ObjectName(declaration.DigestOf(newer)) + ctx = link.WithFiles(context.Background(), aFilesLink{objects: map[string][]byte{newerName: newer}}) + if report = applyAndKeep(ctx, opts, referenceDeclaration(target, newer), nil, nil, nil); report.Refused != "" { + t.Fatalf("924: the newer export: %s", report.Refused) + } + stale := json.RawMessage(`{"id":"m.big","type":"file","path":"` + target + `","digest":"` + declaration.DigestOf(body) + + `","size":` + strconv.Itoa(len(body)) + `}`) + for _, row := range []struct { + name string + edit []byte // what someone else wrote there since; nil for nobody + digest string + }{ + {"the file is what the node wrote", nil, declaration.DigestOf(newer)}, + {"someone changed the file since", []byte("a hand edit"), declaration.DigestOf(body)}, + } { + if row.edit != nil { + if err := os.WriteFile(target, row.edit, 0o600); err != nil { + t.Fatal(err) + } + } + runs := runsAsWritten([]delivery.Composed{{Node: "n", Module: "m", Resources: []json.RawMessage{stale}}}, opts.state) + var got struct { + Digest string `json:"digest"` + } + _ = json.Unmarshal(runs[0].Resources[0], &got) + if got.Digest != row.digest { + t.Errorf("924: %s: what runs names %s, want %s", row.name, got.Digest, row.digest) + } + if row.edit == nil { + raw := []byte(`{"declaration":1,"resources":[` + string(runs[0].Resources[0]) + `]}`) + if report = applyAndKeep(link.WithFiles(context.Background(), gone), opts, raw, nil, nil, nil); report.Refused != "" { + t.Errorf("924: %s: composed as it runs, refused %q; want applied with no object", row.name, report.Refused) + } + } + } } // 1695/AC8: the hand-run `apply ` and `reconcile` have no bus: a reference whose file here verifies is applied diff --git a/cmd/mesh-host/filling.go b/cmd/mesh-host/filling.go index d77d2e9..5257e3e 100644 --- a/cmd/mesh-host/filling.go +++ b/cmd/mesh-host/filling.go @@ -30,9 +30,10 @@ func serveFilling(ctx context.Context, opts options, signer ed25519.PublicKey, s var fillingNow atomic.Pointer[filling.Engine] // knockFilling tells the provisioner's client to fill again what the filling refused or failed and the declaration -// applied now may let through (filling.Engine.Knock), once a module's apply or removal settled (agent.Env.Settled). -// What it told, or why it could not, is said. -func knockFilling(module string, say func(string)) { +// applied now may let through (filling.Engine.Knock), after what runs changed: a module's apply or removal settled +// (agent.Env.Settled, " settled"), the haves were read at the start, or a module was taken as running +// (agent.Env.RunsChanged, issue 924). What it told, or why it could not, is said. +func knockFilling(after string, say func(string)) { e := fillingNow.Load() if e == nil { return @@ -40,9 +41,9 @@ func knockFilling(module string, say func(string)) { told, err := e.Knock(filling.Socket) switch { case err != nil: - say(fmt.Sprintf("after %s settled: %v", module, err)) + say(fmt.Sprintf("after %s: %v", after, err)) case told != "": - say(fmt.Sprintf("after %s settled, the provisioner's client is told to fill again: %s", module, told)) + say(fmt.Sprintf("after %s, the provisioner's client is told to fill again: %s", after, told)) } } diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 1b6289a..bcff6e6 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -516,7 +516,7 @@ func runApply(ctx context.Context, opts options, d *declaration.Declaration, raw // A file given by digest is had before the plan is drawn from it (novox/hq ADR 0486 §1, issue 1695). This command has // no link to the bus, so a file already here that verifies is used, and any other is refused: never written empty. - if err := apply.ResolveFileReferences(ctx, d, link.FilesFrom(ctx)); err != nil { + if err := apply.ResolveFileReferencesKept(ctx, d, link.FilesFrom(ctx), objectKeep(opts)); err != nil { return fmt.Errorf("a file of this declaration could not be had by its digest: %w", err) } @@ -1664,6 +1664,15 @@ func announceOr(say link.Announce) func(string) { // run by hand beside this service — the lock beside the state does the same (store.Lock). var applying sync.Mutex +// objectKeep is the node's own keep of the objects its compositions name by digest, beside its state (apply.ObjectKeep, +// issue 924): a composition the node once applied never again needs the controller's bucket to hold its files. +func objectKeep(opts options) *apply.ObjectKeep { + if opts.state == "" { + return nil + } + return apply.KeepObjectsIn(filepath.Join(filepath.Dir(opts.state), "objects")) +} + // appliedModule is how the apply of one module's declaration came out. type appliedModule struct { // Refused is why nothing was applied, when nothing was. @@ -1703,7 +1712,7 @@ func applyModuleHeld(ctx context.Context, opts options, declared *declaration.De // **A file given by digest is fetched and verified before anything of the composition is applied** (novox/hq ADR // 0486 §1, issue 1695): one object missing or differing refuses the module whole. - if err := apply.ResolveFileReferences(ctx, declared, link.FilesFrom(ctx)); err != nil { + if err := apply.ResolveFileReferencesKept(ctx, declared, link.FilesFrom(ctx), objectKeep(opts)); err != nil { return appliedModule{Refused: "a file of this composition could not be had by its digest: " + err.Error()} } diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 667d18e..31b2e59 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -510,8 +510,11 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus Backup: backupNode(session), // What the filling refused before this apply may be named now: the provisioner's client is told to fill // again (ace, 2026-10-25 21:17). - Settled: func(module string) { knockFilling(module, say) }, - Say: say, + Settled: func(module string) { knockFilling(module+" settled", say) }, + // And once the haves are read at the start, and whenever a module is taken as running without an apply, so + // a refusal from the engine's start does not wait for some module to settle (issue 924). + RunsChanged: func(why string) { knockFilling(why, say) }, + Say: say, // A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min. Backoff: link.Backoff, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { @@ -560,7 +563,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus hooks := composerHooks{ running: func() []delivery.Composed { if now := own.get(); now != nil { - return now.Everything() + return runsAsWritten(now.Everything(), opts.state) } return nil }, @@ -615,6 +618,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus if composer != nil { composes.set(delivery.ComposesOwn) env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy + env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does session.NoteComposes(linked, delivery.ComposesOwn, why) // And once the catalogue no longer places it, the agent begins again on the controller's compositions: the // placement key deleted ends this run, the way back (issue 1126, the seed's revert). diff --git a/cmd/mesh-host/runs_as_written.go b/cmd/mesh-host/runs_as_written.go new file mode 100644 index 0000000..a2bca2a --- /dev/null +++ b/cmd/mesh-host/runs_as_written.go @@ -0,0 +1,88 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/store" +) + +// runsAsWritten is what the node runs as the composer reads it (selfcompose.Env.Running), with every file given by digest +// named as this node last wrote it, where the two differ (issue 924, novox 2026-10-29 01:16). What runs of a module the +// agent has not applied since its start can come from the kept declaration, older than the compositions applied since: +// its export (mesh-vault.kept) then named a digest the controller's sweep had removed, while the file on disk was the +// newer export the node wrote. A node composing its own carries such a file as it runs (mesh-sdk frombus keptOf), so it +// names what is on disk: **only when the node's record says it wrote exactly that content there** (store Applied.Wrote, +// Target), never a file someone else changed. A file that cannot be read, or that the record does not vouch for, is +// left as the composition names it. +func runsAsWritten(running []delivery.Composed, statePath string) []delivery.Composed { + if len(running) == 0 || statePath == "" { + return running + } + known, err := store.Load(statePath) + if err != nil { + return running + } + out := make([]delivery.Composed, len(running)) + for i, c := range running { + out[i] = c + var changed []json.RawMessage + for j, raw := range c.Resources { + if fixed, ok := asWritten(raw, known); ok { + if changed == nil { + changed = append([]json.RawMessage(nil), c.Resources...) + } + changed[j] = fixed + } + } + if changed != nil { + out[i].Resources = changed + } + } + return out +} + +// asWritten is the file resource raw named by the digest and size of what this node last wrote at its path, when raw is a +// file given by another digest and the file there is what the record says the node wrote. +func asWritten(raw json.RawMessage, known store.State) (json.RawMessage, bool) { + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil { + return nil, false + } + var kind, id, path, digest string + _ = json.Unmarshal(fields["type"], &kind) + _ = json.Unmarshal(fields["id"], &id) + _ = json.Unmarshal(fields["path"], &path) + _ = json.Unmarshal(fields["digest"], &digest) + if kind != string(declaration.TypeFile) || digest == "" || path == "" { + return nil, false + } + rec, ok := known.Find(id) + if !ok || rec.Wrote == "" || rec.Target != path { + return nil, false + } + body, err := os.ReadFile(path) + if err != nil { + return nil, false + } + sum := sha256.Sum256(body) + if hex.EncodeToString(sum[:]) != rec.Wrote { + return nil, false // changed since the node wrote it: not the node's to vouch for + } + if now := declaration.DigestOf(body); now == digest { + return nil, false + } else { + fields["digest"], _ = json.Marshal(now) + } + fields["size"], _ = json.Marshal(len(body)) + fixed, err := json.Marshal(fields) + if err != nil { + return nil, false + } + return fixed, true +} diff --git a/go.mod b/go.mod index f6dc073..9127519 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.20261028234900-c2999a4e13cc + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029003747-575cb82fba69 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 0af9259..7fd559b 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028234900-c2999a4e13cc h1:mOk2tAKTiuvnap/BUIYz91lLVbHi/EB6JPAD6KxTPvc= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028234900-c2999a4e13cc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029003747-575cb82fba69 h1:0uFXqo13BukGqe1fltCwk9O/nAuDWx5paHUsmdfgar4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029003747-575cb82fba69/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/agent/agent.go b/internal/agent/agent.go index 5bf7222..acb91db 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -109,7 +109,9 @@ type Env struct { Kept func() map[string][]json.RawMessage // KeptFilled is the paths the kept declaration names as filled by the provisioner, by the module whose file is at // each (the node's own under NodeModule). A module that still runs from the kept declaration keeps its filled files - // filled: without them the filling refused each one as "not named as filled" (issue 1945). Nil names none. + // filled: without them the filling refused each one as "not named as filled" (issue 1945). Nil names none. A module + // whose have names a composition the bucket keeps runs from that composition, never from the kept declaration + // (issue 924): the kept declaration is older than every composition applied since the switch. KeptFilled func() map[string][]string // Trusted is the public keys a composition may be signed by: the controller's and the operator's (ADR 0539 §3). A // composition unsigned, signed by another key, or changed after it was signed is refused and nothing of it is @@ -131,6 +133,12 @@ type Env struct { // them. The node-engine's filling tells the provisioner's client then to fill again what it refused (mesh-host // internal/filling Knock; ace, 2026-10-25 21:17). Nil tells nothing. Settled func(module string) + // RunsChanged is told, outside the agent's lock, that what runs (Everything) changed other than by an apply or a + // removal: once at the start, when the haves and the compositions they name are read, and each time a module is + // taken as running from its have or the kept declaration. Like Settled, the filling then tells the provisioner's + // client to fill again what it refused (issue 924, the second meeting: ace, g14 and shanks on 2026-10-29 at 00:09 + // UTC refused power's three new drop-ins from an engine start on, as no module settled after it). Nil tells nothing. + RunsChanged func(why string) // Backoff is how long to wait before the try-th retry (1, 2, …) of an apply that failed transiently. Nil retries a // transient failure only on RetryTransient or a change of either key. Backoff func(try int) time.Duration @@ -143,6 +151,11 @@ type Env struct { // move of a module of policy record that recreates a container or moves its data waits for a person's yes in its // want (ADR 0547 §3). Nil judges no move. Policy func(ctx context.Context, module string) string + // HoldsData says whether a module's manifest declares data (catalogue.Manifest.DataItems), and whether its manifest + // could be read: what a composition's Data is, for what runs of a module from the kept declaration, which carries no + // Data of its own. A composition says Data, so without it a data module from the kept declaration is judged to move + // its data against its first composition after a restart (mongodb and mssql on novox, issue 924). Nil reads none. + HoldsData func(ctx context.Context, module string) (data, known bool) } // transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has @@ -160,6 +173,13 @@ type Agent struct { // applied is the composition each module was last applied from: a revision a want named from the key's history is // not the key's latest value, and what it holds is still held out of the others' applies. applied map[string]compNow + // had is, for each module whose have names a composition the bucket keeps (its latest value or a revision), that + // composition, read once at the start (loadHad): what runs of a module the agent has not taken yet, ahead of the kept + // declaration (issue 924). + had map[string]compNow + // keptData is, for each module of the kept declaration whose manifest could be read, whether it declares data + // (Env.HoldsData): the Data of what runs of it from the kept declaration. + keptData map[string]bool // attempt is the last apply of a module that failed or is held, with the revisions it was made from. attempt map[string]attempt checked map[string]bool @@ -218,7 +238,7 @@ func New(env Env) *Agent { if env.Say == nil { env.Say = func(string) {} } - return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{}, + return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{}, had: map[string]compNow{}, keptData: map[string]bool{}, attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{}, retry: make(chan []string, 8), reconcile: make(chan []string, 64), tries: map[string]int{}, again: make(chan againAsk, 64), transient: make(chan struct{}, 1), stopped: make(chan struct{})} @@ -236,7 +256,7 @@ func (a *Agent) Run(ctx context.Context) error { if err != nil { return fmt.Errorf("the agent cannot watch its compositions: %w", err) } - var wantsCaught, compsCaught bool + var wantsCaught, compsCaught, hadRead bool dirty := map[string]bool{} for { select { @@ -303,7 +323,17 @@ func (a *Agent) Run(ctx context.Context) error { } // Nothing is applied until what was there at the start is all in: a module's composition may arrive after its // want, and the node's own composition comes before any module's. - if !(wantsCaught && compsCaught) || len(dirty) == 0 { + if !(wantsCaught && compsCaught) { + continue + } + if !hadRead { + // Before the first settle: what runs of each module with a have, from the composition it names, so the + // filling judges by it at once and not by the kept declaration until some module settles (issue 924). + hadRead = true + a.loadHad(ctx) + a.runsChanged("the haves were read at the start") + } + if len(dirty) == 0 { continue } a.settle(ctx, dirty) @@ -726,7 +756,12 @@ func (a *Agent) revisionNamed(ctx context.Context, module, fp, digest string) (c if err != nil { continue } - dig, err := delivery.VerifyComposed(c, trusted) + var dig string + if a.env.SelfComposed { + dig, err = delivery.ComposedDigestOf(c) // the node's own: unsigned (ADR 0547 §2) + } else { + dig, err = delivery.VerifyComposed(c, trusted) + } if err != nil || dig != digest { continue } @@ -755,22 +790,33 @@ func (a *Agent) others(except string) []delivery.Composed { out = append(out, c.C) } else if c, ok := a.applied[m]; ok && c.Dig == h.Composed { out = append(out, c.C) + } else if c, ok := a.had[m]; ok && c.Dig == h.Composed { + out = append(out, c.C) // a revision its have named, replaced since under the same key } } + hadNames := make([]string, 0, len(a.had)) + for m := range a.had { + if _, held := a.running[m]; !held && m != except { + hadNames = append(hadNames, m) + } + } + sort.Strings(hadNames) + for _, m := range hadNames { + out = append(out, a.had[m].C) + } if a.env.Kept != nil { kept := a.env.Kept() names := make([]string, 0, len(kept)) for m := range kept { - if _, held := a.running[m]; !held && m != except { + _, held := a.running[m] + _, had := a.had[m] + if !held && !had && m != except { names = append(names, m) } } sort.Strings(names) for _, m := range names { - c := delivery.Composed{Node: a.env.Node, Module: m, Resources: kept[m]} - if a.env.KeptFilled != nil { - c.Filled = a.env.KeptFilled()[m] - } + c, _ := a.keptComposed(m, kept[m]) if m == NodeModule { c.Module = "" } @@ -780,6 +826,85 @@ func (a *Agent) others(except string) []delivery.Composed { return out } +// loadHad reads, once at the start, the have of every module with a want or in the kept declaration and, where it names +// a composition, that composition: the bucket's value at its key when its digest is the have's, else the revision the +// bucket keeps with that digest. A have naming none, or one the bucket no longer keeps, leaves the module to the kept +// declaration, said once (issue 924). +func (a *Agent) loadHad(ctx context.Context) { + a.mu.Lock() + seen := map[string]bool{} + for m, w := range a.wants { + if m != NodeModule && w.Present && !w.Want.Removed { + seen[m] = true + } + } + if a.env.Kept != nil { + for m := range a.env.Kept() { + if m != NodeModule { + seen[m] = true + } + } + } + a.mu.Unlock() + names := make([]string, 0, len(seen)) + for m := range seen { + names = append(names, m) + } + sort.Strings(names) + a.readKeptData(ctx, names...) + for _, m := range names { + if ctx.Err() != nil { + return + } + h, ok := a.env.Haves.Read(ctx, m) + if !ok || h.Composed == "" || !haveRuns(h) { + continue + } + a.mu.Lock() + c, found := a.compositionFor(m, h.FP) + a.mu.Unlock() + if !found || c.Err != nil || c.Dig != h.Composed { + c, found = a.revisionNamed(ctx, m, h.FP, h.Composed) + } + a.mu.Lock() + if found { + a.had[m] = c + } else if _, kept := a.keptHas(m); kept { + a.sayOnce("had:"+m, fmt.Sprintf("%s's have names the composition %s, which the bucket no longer keeps: it runs "+ + "from the kept declaration until it is applied or taken", m, short(h.Composed))) + } + a.mu.Unlock() + } +} + +// haveRuns says a have names what runs, in the phases baseline takes as running: running (or empty), held for a failed +// backup, or a composer's mark over what still runs. A have backing up, applying, checking, failed or waiting names a +// composition not applied, or not yet (issue 924, review of 031e106). +func haveRuns(h Have) bool { + switch delivery.Phase(h.Phase) { + case "", delivery.PhaseRunning, delivery.PhaseHeld: + return true + } + return composerMark(h) +} + +// keptHas says whether the kept declaration has resources for the module. Called holding a.mu or not: Kept takes no lock +// of the agent's. +func (a *Agent) keptHas(module string) ([]json.RawMessage, bool) { + if a.env.Kept == nil { + return nil, false + } + rs, ok := a.env.Kept()[module] + return rs, ok +} + +// runsChanged tells RunsChanged, outside the agent's lock. +func (a *Agent) runsChanged(why string) { + if a.env.RunsChanged != nil { + a.env.RunsChanged(why) + } +} + // unreadableModules are the keys of the module compositions in the node's bucket that cannot be read or are refused, of a // module the assigner has not removed: for each module only the composition its want names, else its newest, since a // stale one nobody applies (written before a rule the SDK now holds) must not hold the node's own back. Called holding @@ -901,7 +1026,8 @@ func (a *Agent) heldOut(except string) []delivery.Composed { } } if len(rs) > 0 { - out = append(out, delivery.Composed{Node: a.env.Node, Module: m, Resources: rs}) + // Held out, never judged: its Data from its manifest where known, as every other composition says it. + out = append(out, delivery.Composed{Node: a.env.Node, Module: m, Resources: rs, Data: a.keptData[m]}) } } return out @@ -1048,6 +1174,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning} } a.mu.Unlock() + a.runsChanged(module + " was taken as held") return true } if h.Composed != "" { @@ -1068,6 +1195,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com a.mu.Lock() a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning} a.mu.Unlock() + a.runsChanged(module + " was taken from its have") // Said again, so the session holds the module and its key carries the health this engine judges, // not the one an older engine left there (xorg's unknown on g14 and shanks, 2026-10-25). a.writeHave(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed, @@ -1096,6 +1224,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com a.mu.Lock() a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning} a.mu.Unlock() + a.runsChanged(module + " was taken from the kept declaration") a.writeHave(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)}) return true } @@ -1286,6 +1415,7 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp if module == NodeModule || a.env.Policy == nil || a.env.Policy(ctx, module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig { return false } + a.readKeptData(ctx, module) // the manifest may read now though it did not at the start a.mu.Lock() r, ran := a.running[module] if ran && r.Composed == comp.Dig && r.Phase != delivery.PhaseRemoved { @@ -1331,14 +1461,61 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp // runningComposition is what the module runs now: the composition last applied, else its resources in the declaration // the node kept from before the switch. Called with a.mu held. +// +// The composition its have named at the start comes before the kept declaration (issue 924). From the kept declaration, +// what runs carries Data as its manifest declares it; a module whose manifest could not be read cannot be judged against +// it (false), and that is said, rather than taking it to hold no data and judging a false move of its data. func (a *Agent) runningComposition(module string) (delivery.Composed, bool) { if c, ok := a.applied[module]; ok && c.Err == nil { return c.C, true } - if a.env.Kept != nil { - if rs, ok := a.env.Kept()[module]; ok { - return delivery.Composed{Node: a.env.Node, Module: module, Resources: rs}, true + if c, ok := a.had[module]; ok { + return c.C, true + } + if rs, ok := a.keptHas(module); ok { + c, known := a.keptComposed(module, rs) + if !known { + a.sayOnce("kept-data:"+module, fmt.Sprintf("%s runs from the kept declaration, and whether it holds data cannot be "+ + "read from its manifest: a move of it cannot be judged", module)) } + return c, known } return delivery.Composed{}, false } + +// keptComposed is what runs of a module from the kept declaration: its resources there, the paths it names as filled, +// and Data as the module's manifest declares it; false when the manifest could not be read, so Data is not known. Called +// with a.mu held. +func (a *Agent) keptComposed(module string, rs []json.RawMessage) (delivery.Composed, bool) { + c := delivery.Composed{Node: a.env.Node, Module: module, Resources: rs} + if a.env.KeptFilled != nil { + c.Filled = a.env.KeptFilled()[module] + } + data, known := a.keptData[module] + c.Data = data + return c, known || module == NodeModule +} + +// readKeptData reads, outside the agent's lock, whether each module of the kept declaration not known yet declares data +// (Env.HoldsData), and keeps the answers it could read. +func (a *Agent) readKeptData(ctx context.Context, modules ...string) { + if a.env.HoldsData == nil { + return + } + for _, m := range modules { + if m == NodeModule { + continue + } + a.mu.Lock() + _, done := a.keptData[m] + a.mu.Unlock() + if _, kept := a.keptHas(m); done || !kept { + continue + } + if data, known := a.env.HoldsData(ctx, m); known { + a.mu.Lock() + a.keptData[m] = data + a.mu.Unlock() + } + } +} diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 7c256bb..a73faa2 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -1026,18 +1026,102 @@ func TestTheRevisionAWantNamesIsAppliedFromHistory(t *testing.T) { } } -// 1945/AC18 (unit): a module that still runs from the kept declaration carries the kept declaration's filled paths of -// its files, so the filling writes them ("not named as filled" for claude-code and screen-lock, 2026-10-25). -func TestAModuleRunningFromTheKeptDeclarationKeepsItsFilledPaths(t *testing.T) { - b := newBed(t) - b.kept = map[string][]json.RawMessage{"claude-code": {file("claude-code", "settings", "")}} - b.env.KeptFilled = func() map[string][]string { - return map[string][]string{"claude-code": {"/tmp/r6/claude-code/settings"}} +// 1945/AC18 and issue 924 (unit): what runs of a module the agent has not taken yet carries the filled paths the +// filling judges by. From the kept declaration while the module has no have naming a composition ("not named as filled" +// for claude-code and screen-lock, 2026-10-25); from the composition its have names once it has one, the kept +// declaration being older, read at the start before any module settles and told to the filling then, so a file the +// newer composition fills is accepted on the first fill (power's three drop-ins on ace, g14 and shanks, 2026-10-29). +// Only a have in a phase baseline takes as running names what runs; and what runs carries Data, from the composition +// or, from the kept declaration, from the module's manifest, so a data module is not judged to move its data against +// its first composition after a restart (mongodb and mssql on novox). +func TestWhatRunsBeforeTheAgentTakesAModuleNamesItsFilledPaths(t *testing.T) { + const old, added = "/tmp/r6/power/settings", "/tmp/r6/power/bound" + filledFile := func(id string) json.RawMessage { + return json.RawMessage(`{"id":"power.` + id + `","type":"file","module":"power","path":"/tmp/r6/power/` + id + + `","mode":"0644","owner-at-runtime":"owned"}`) } - a := New(b.env) - got := a.Everything() - if len(got) != 1 || got[0].Module != "claude-code" || strings.Join(got[0].Filled, ",") != "/tmp/r6/claude-code/settings" { - t.Fatalf("what runs from the kept declaration: %+v", got) + fp := fpOf("a") + ck, _ := delivery.ComposedKey(node, "power", fp) + applied := delivery.Composed{Node: node, Module: "power", FP: fp, Filled: []string{old, added}, Data: true, + Resources: []json.RawMessage{filledFile("settings"), filledFile("bound")}} + _, appliedRaw, appliedDig := signedBy(t, applied, ck, controllerKey) + newer := applied + newer.Resources = append([]json.RawMessage{}, applied.Resources...) + newer.Resources = append(newer.Resources, file("power", "later", "1")) + _, newerRaw, _ := signedBy(t, newer, ck, controllerKey) + for _, tc := range []struct { + name string + // have is the module's have; bucket the value its composition key holds now, history the revisions kept. + have *Have + bucket []byte + history []Update + filled, told string + // data is what runs' Data; holdsData the manifest's answer (nil: none read). + data bool + holdsData func(context.Context, string) (bool, bool) + }{ + {name: "no have: the kept declaration's", filled: old}, + {name: "a have naming the bucket's value", have: &Have{Module: "power", FP: fp, Composed: appliedDig}, + bucket: appliedRaw, filled: old + "," + added, told: "the haves were read at the start", data: true}, + {name: "a have applying names nothing that runs yet", have: &Have{Module: "power", FP: fp, Composed: appliedDig, + Phase: string(delivery.PhaseApplying)}, bucket: appliedRaw, filled: old, told: "the haves were read at the start"}, + {name: "a data module from the kept declaration says Data as its manifest does", filled: old, + told: "the haves were read at the start", data: true, + holdsData: func(context.Context, string) (bool, bool) { return true, true }}, + {name: "a have naming a revision replaced since", have: &Have{Module: "power", FP: fp, Composed: appliedDig}, + bucket: newerRaw, history: []Update{{Key: ck, Value: appliedRaw, Revision: 3}, {Key: ck, Value: newerRaw, Revision: 7}}, + filled: old + "," + added, told: "the haves were read at the start", data: true}, + {name: "a have naming a composition the bucket no longer keeps", have: &Have{Module: "power", FP: fp, Composed: appliedDig}, + bucket: newerRaw, filled: old, told: "the haves were read at the start"}, + } { + t.Run(tc.name, func(t *testing.T) { + b := newBed(t) + b.kept = map[string][]json.RawMessage{"power": {filledFile("settings")}} + b.env.KeptFilled = func() map[string][]string { return map[string][]string{"power": {old}} } + if tc.have != nil { + b.haves.stored = map[string]Have{"power": *tc.have} + } + if tc.bucket != nil { + b.comps.before(ck, tc.bucket) + } + b.env.Revisions = func(_ context.Context, key string) ([]Update, error) { return tc.history, nil } + b.env.HoldsData = tc.holdsData + var told []string + var mu sync.Mutex + b.env.RunsChanged = func(why string) { mu.Lock(); told = append(told, why); mu.Unlock() } + filledNow := func() string { + for _, c := range b.ag.Everything() { + if c.Module == "power" { + if c.Data != tc.data { + t.Errorf("what runs says Data %v, want %v", c.Data, tc.data) + } + return strings.Join(c.Filled, ",") + } + } + return "" + } + if tc.told == "" { // no want, no have: nothing to start, and the kept declaration is what runs + a := New(b.env) + b.ag = a + if got := filledNow(); got != tc.filled { + t.Fatalf("filled %q, want %q", got, tc.filled) + } + return + } + stop := b.run() + defer stop() + eventually(t, "the filling was not told what runs at the start", func() bool { + mu.Lock() + defer mu.Unlock() + return len(told) > 0 && told[0] == tc.told + }) + if got := filledNow(); got != tc.filled { + t.Fatalf("filled %q, want %q", got, tc.filled) + } + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("applied %v: nothing has a want", got) + } + }) } } diff --git a/internal/apply/objectkeep.go b/internal/apply/objectkeep.go new file mode 100644 index 0000000..d5be5df --- /dev/null +++ b/internal/apply/objectkeep.go @@ -0,0 +1,111 @@ +package apply + +import ( + "fmt" + "os" + "path/filepath" + "time" + + "git.novox.be/novox/mesh-sdk/go/declaration" +) + +// ObjectKeep is the node-engine's own keep of the content of files given by digest (novox/hq ADR 0486 §1, issue 924): one +// file per object, named as the object is, beside the node's state. **What a node once had it never asks the bus for +// again**, so neither the controller's sweep of the bucket (an object is removed a day after no composition of the +// controller names it) nor a bus that does not answer can refuse a composition the node already ran. The node's account +// may only read its bucket (ADR 0486 §2), so the keep is the node's, not the bucket's: nothing is written back. +// +// An object unused for KeptObjectsFor is let go when another is kept, so the keep holds what the node applied lately and +// not every export it ever ran. Every failure to keep is silent and costs only a fetch: the keep is a copy. +type ObjectKeep struct { + Dir string + // Now is the clock; nil is time.Now. + Now func() time.Time +} + +// KeptObjectsFor is how long an object of the keep is held after it was last used. +const KeptObjectsFor = 14 * 24 * time.Hour + +// KeepObjectsIn is the keep at dir; an empty dir keeps nothing. +func KeepObjectsIn(dir string) *ObjectKeep { + if dir == "" { + return nil + } + return &ObjectKeep{Dir: dir} +} + +func (k *ObjectKeep) now() time.Time { + if k.Now != nil { + return k.Now() + } + return time.Now() +} + +// get is the kept content of the object name when it is what f names, and marks it used. +func (k *ObjectKeep) get(name string, f *declaration.File) ([]byte, bool) { + if k == nil { + return nil, false + } + path := filepath.Join(k.Dir, name) + body, err := os.ReadFile(path) + if err != nil || int64(len(body)) != f.Size || f.Verify(body) != nil { + return nil, false + } + now := k.now() + _ = os.Chtimes(path, now, now) + return body, true +} + +// put keeps body under name, unless it is kept already (then it is marked used), and lets go of what was unused too long. +func (k *ObjectKeep) put(name string, body []byte) { + if k == nil { + return + } + path := filepath.Join(k.Dir, name) + now := k.now() + if info, err := os.Stat(path); err == nil && info.Size() == int64(len(body)) { + _ = os.Chtimes(path, now, now) + return + } + if err := os.MkdirAll(k.Dir, 0o700); err != nil { + return + } + tmp, err := os.CreateTemp(k.Dir, ".put-*") + if err != nil { + return + } + _, werr := tmp.Write(body) + cerr := tmp.Close() + if werr != nil || cerr != nil || os.Chmod(tmp.Name(), 0o600) != nil || os.Rename(tmp.Name(), path) != nil { + _ = os.Remove(tmp.Name()) + return + } + _ = os.Chtimes(path, now, now) + k.prune(now) +} + +// prune lets go of every object unused for KeptObjectsFor. +func (k *ObjectKeep) prune(now time.Time) { + entries, err := os.ReadDir(k.Dir) + if err != nil { + return + } + for _, e := range entries { + info, err := e.Info() + if err != nil || !info.Mode().IsRegular() { + continue + } + if now.Sub(info.ModTime()) >= KeptObjectsFor { + _ = os.Remove(filepath.Join(k.Dir, e.Name())) + } + } +} + +// never says, in a failed fetch's error before its cause (a transient one's detail keeps the tail), that the node's own +// keep did not hold the object either. +func (k *ObjectKeep) never() string { + if k == nil { + return "" + } + return fmt.Sprintf(" (this node's own keep of objects, %s, never held it)", k.Dir) +} diff --git a/internal/apply/references.go b/internal/apply/references.go index ad6bed5..64ce30c 100644 --- a/internal/apply/references.go +++ b/internal/apply/references.go @@ -27,13 +27,23 @@ type FetchObject func(ctx context.Context, name string, limit int64) ([]byte, er // unchanged declaration, and a node that is asleep and has no bus, apply as they always did. One object is fetched // once however many files name its digest. func ResolveFileReferences(ctx context.Context, d *declaration.Declaration, fetch FetchObject) error { + return ResolveFileReferencesKept(ctx, d, fetch, nil) +} + +// ResolveFileReferencesKept is ResolveFileReferences with the node's own keep of objects (ObjectKeep): every content had +// (a file here that verifies, or an object of the bus) is kept, and a kept one is used before the bus is asked. **A node +// depends on its bucket only for content it never had** (issue 924, 2026-10-29: novox's own composition named the +// export mesh-vault runs by a digest whose object the controller's sweep had removed, so the vault's apply failed while +// it ran; the bucket's objects are kept only while a composition of the controller names them, and a node composing its +// own names objects the controller no longer knows of). A nil keep keeps nothing. +func ResolveFileReferencesKept(ctx context.Context, d *declaration.Declaration, fetch FetchObject, keep *ObjectKeep) error { fetched := map[string][]byte{} for i, r := range d.Resources { f, ok := r.(*declaration.File) if !ok || !f.IsReference() { continue } - body, err := referencedContent(ctx, f, fetch, fetched) + body, err := referencedContent(ctx, f, fetch, fetched, keep) if err != nil { return err } @@ -50,31 +60,37 @@ func ResolveFileReferences(ctx context.Context, d *declaration.Declaration, fetc } func referencedContent(ctx context.Context, f *declaration.File, fetch FetchObject, - fetched map[string][]byte) ([]byte, error) { + fetched map[string][]byte, keep *ObjectKeep) ([]byte, error) { if body, ok := fetched[f.Digest]; ok { return body, nil } - if body, ok := alreadyThere(f); ok { - fetched[f.Digest] = body - return body, nil - } name, err := declaration.ObjectName(f.Digest) if err != nil { return nil, fmt.Errorf("file %s (%s): %w", f.ID, f.Path, err) } + if body, ok := alreadyThere(f); ok { + fetched[f.Digest] = body + keep.put(name, body) + return body, nil + } + if body, ok := keep.get(name, f); ok { + fetched[f.Digest] = body + return body, nil + } if fetch == nil { return nil, fmt.Errorf("file %s (%s) is given by digest %s and is not on this machine, and this node-engine "+ "has no link to the bus to fetch it from", f.ID, f.Path, f.Digest) } body, err := fetch(ctx, name, f.Size) if err != nil { - return nil, fmt.Errorf("file %s (%s): fetching the object %s of this node's own object bucket: %w", - f.ID, f.Path, name, err) + return nil, fmt.Errorf("file %s (%s): fetching the object %s of this node's own object bucket%s: %w", + f.ID, f.Path, name, keep.never(), err) } if err := f.Verify(body); err != nil { return nil, fmt.Errorf("%s (%s): %w; nothing was applied", f.ID, f.Path, err) } fetched[f.Digest] = body + keep.put(name, body) return body, nil } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 8a596e8..a49734c 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -757,6 +757,27 @@ func (c *Composer) Policy(ctx context.Context, module string) string { return policy } +// HoldsData says whether a module's manifest declares data (catalogue.Manifest.DataItems), as a composition's Data says +// it, and whether the manifest could be read: the agent's Data for what runs of a module from the kept declaration +// (agent.Env.HoldsData, issue 924). +func (c *Composer) HoldsData(ctx context.Context, module string) (data, known bool) { + c.mu.Lock() + m, ok := c.modules[module] + c.mu.Unlock() + if !ok || m.Manifest == "" { + return false, false + } + raw, err := reader{c: c}.Manifest(ctx, m.Manifest) + if err != nil { + return false, false + } + parsed, err := catalogue.ParseManifestForInputs(raw) + if err != nil { + return false, false + } + return len(parsed.DataItems()) > 0, true +} + // reader is the composer's view of the bus as frombus reads it. type reader struct{ c *Composer }