From 089ac780079c2ffea91f3bf62b4a7c22fc302b0e Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 08:56:36 +0100 Subject: [PATCH 1/6] Pin mesh-sdk at the bus-want group's head 959b157, so the node agent reads the warrant, the server's fingerprint and the server's list (issue 774) --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 7ec0e0a..d56efbd 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.20261102191058-f02f142af128 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103075323-959b1575a8c1 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 43cc4ec..fd53779 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 h1:aI1DhZAuqVBXbybSqhLNOGX5rUOrn9T0cHNCDUQF7IQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103075323-959b1575a8c1 h1:lW1NzV8w3pLjH9ACAASYgtYVF3XdCQqIvK/BU81K34Y= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103075323-959b1575a8c1/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= -- 2.54.0 From ffcd8f6d61e3d83be7699d885d916b9f7536923f Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 08:56:36 +0100 Subject: [PATCH 2/6] =?UTF-8?q?An=20apply=20leaves=20the=20resources=20its?= =?UTF-8?q?=20scope=20keeps=20as=20they=20run,=20so=20the=20bus=20server?= =?UTF-8?q?=20is=20not=20replaced=20by=20an=20unwarranted=20want=20(ADR=20?= =?UTF-8?q?0594=20=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/apply/apply.go | 68 +++++++++++++++++++++++-- internal/apply/module_scope_test.go | 78 +++++++++++++++++++++++++++++ 2 files changed, 142 insertions(+), 4 deletions(-) diff --git a/internal/apply/apply.go b/internal/apply/apply.go index fd1432e..ff94eda 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -143,8 +143,9 @@ func (r Report) Changed() bool { // (novox/hq issue 224). // waits busy is a container left running because it said busy (novox/hq ADR 0335), which nothing // moved either. + // Kept as it runs is a bus server a want without the warrant's mark left alone (novox/hq ADR 0594 §2). if o.Action != "unchanged" && o.Action != "held" && o.Action != "waiting" && o.Action != heldStill && - o.Action != waitsBusy { + o.Action != waitsBusy && o.Action != keptAsItRuns { return true } } @@ -321,6 +322,11 @@ type Scope struct { // in its own composition has no want, so Placed does not name it (issue 774, review of the re-land, round 1: nats on // the bus node has none, and its nats.conf was guessed as nats's and removed). Nil guards by Placed alone. RemovedByWant map[string]bool + // Keep are the resources, by id, the apply leaves as they run when they exist (novox/hq ADR 0594 §2): the bus + // server's container and what restarts it, on a want of the bus module without the warrant's mark. A kept resource + // that exists is neither recreated nor rewritten and is recorded as the module's with what it was last recorded + // reading; one that does not exist is made. Nil keeps nothing. + Keep map[string]bool } // placedOwner is the module a want still places on the node that the record belongs to, other than the scope's own: @@ -1065,7 +1071,11 @@ func applyScoped( if from, handed := handedFiles[resource.Identity()]; handed && was.ID == "" { previous.Wrote = from.Wrote } - outcome, err = applyOne(ctx, sys, resource, run, changed, in, previous, unseal, keepFound) + if o, kept := keepAsItRuns(ctx, scope, resource, run, previous); kept { + outcome = o + } else { + outcome, err = applyOne(ctx, sys, resource, run, changed, in, previous, unseal, keepFound) + } } if err != nil { // **What was found is kept whatever the apply then did** (novox/hq ADR 0118). The @@ -1272,7 +1282,7 @@ func applyScoped( if outcome.busy != nil { report.Busy = append(report.Busy, *outcome.busy) } - if outcome.Action == heldStill || outcome.Action == waitsBusy { + if outcome.Action == heldStill || outcome.Action == waitsBusy || outcome.Action == keptAsItRuns { // Not changed: nothing about it moved, so nothing that restarts on it restarts. log(fmt.Sprintf(" %s %s (%s): %s", outcome.Action, outcome.ID, outcome.Target, outcome.Detail)) } else if moved(outcome) { @@ -1429,7 +1439,8 @@ func recordApplied(known *store.State, origin string, resource declaration.Resou // moved reports whether applying a resource moved it, so what restarts or reloads on it does: anything but // unchanged, and not a container held still or waiting for its busy one to say idle. func moved(outcome Outcome) bool { - return outcome.Action != "unchanged" && outcome.Action != heldStill && outcome.Action != waitsBusy + return outcome.Action != "unchanged" && outcome.Action != heldStill && outcome.Action != waitsBusy && + outcome.Action != keptAsItRuns } // afterWhatTheyRead is the resources in declared order, except that **a service comes after every @@ -3019,6 +3030,55 @@ func applyNetwork(ctx context.Context, r *declaration.Network, run Runner) (Outc return out, nil } +// keptAsItRuns is the action of a resource the scope keeps (Scope.Keep, novox/hq ADR 0594 §2), found on the machine +// and left as it runs: the bus server on a want of the bus module without the warrant's mark. Its own word, beside +// heldStill: nothing about it moved, so nothing that restarts on it restarts, and a person reading the apply sees why +// a changed server was not replaced. +const keptAsItRuns = "kept-as-it-runs" + +// keepAsItRuns is the outcome of a resource the scope keeps that exists on the machine, true; false for any other, +// which is applied as always. A container exists when the runtime knows it by its name; an archive when it was +// unpacked at its path before. The outcome carries what the resource was last recorded reading and holding, so its +// record is the module's from here on (taken over from the node's own composition at the bus module's first want) and +// a later warranted apply still sees what changed under it. +func keepAsItRuns(ctx context.Context, scope *Scope, r declaration.Resource, run Runner, previous store.Applied) (Outcome, bool) { + if scope == nil || !scope.Keep[r.Identity()] { + return Outcome{}, false + } + out := begin(r) + out.Action = keptAsItRuns + why := "the bus server is left as it runs: only a want carrying the operator's warrant replaces it (novox/hq ADR 0594 §2)" + switch res := r.(type) { + case *declaration.Container: + if res.Schedule != "" || res.RunOnce { + return Outcome{}, false + } + if _, err := containerState(ctx, res.Name, run); err != nil { + return Outcome{}, false + } + out.reads = previous.Reads + out.busySince = previous.BusySince + out.Detail = why + return out, true + case *declaration.Archive: + if previous.Wrote == "" || previous.Target != res.Path { + return Outcome{}, false + } + if _, err := os.Stat(res.Path); err != nil { + return Outcome{}, false + } + out.wrote = previous.Wrote + out.unpacked = previous.Unpacked + if previous.Wrote != res.Digest { + out.Detail = why + "; the declared " + res.Digest + " waits for the warrant" + } else { + out.Detail = why + } + return out, true + } + return Outcome{}, false +} + // heldStill is the action of a container left alone because a maintenance window holds it stopped // (novox/hq issue 224). Its own word rather than "unchanged": a person reading a machine that looks // half-stopped needs to see why, and "unchanged" beside a stopped server says the opposite. diff --git a/internal/apply/module_scope_test.go b/internal/apply/module_scope_test.go index 1bdb768..4ac1aee 100644 --- a/internal/apply/module_scope_test.go +++ b/internal/apply/module_scope_test.go @@ -2,6 +2,8 @@ package apply import ( "context" + "fmt" + "os" "path/filepath" "strings" "testing" @@ -219,6 +221,82 @@ func TestAResourceThatMovedToItsModulesCompositionIsLeftByTheNodesOwnApply(t *te } }) } + // Issue 774, ADR 0594 §2: the bus module's first own apply takes over the server the node's own composition made + // while the bus was folded. Kept (a want without the warrant's mark), the server's container and its program archive + // are left as they run, whatever the build changed in them, and become nats's; its files and the tools' account file + // are written. Not kept, the same apply replaces the container. + for _, keep := range []bool{true, false} { + t.Run(fmt.Sprintf("the bus server kept %v", keep), func(t *testing.T) { + dir := t.TempDir() + conf, account, program := filepath.Join(dir, "nats.conf"), filepath.Join(dir, "tools", "account"), filepath.Join(dir, "program") + bus := func(env, confBody, programBody, module string) *declaration.Declaration { + archive, _ := archiveDecl(t, "nats.program-snapshot", program, map[string]string{"snapshot": programBody}, "") + if module != "" { + archive = strings.Replace(archive, `"id":"nats.program-snapshot"`, `"id":"nats.program-snapshot","module":"nats"`, 1) + module = `,"module":"nats"` + } + return declare(t, archive+`, + {"id":"nats.server-conf","type":"file","path":"`+conf+`","content":"`+confBody+`"`+module+`}, + {"id":"nats.server","type":"container","name":"mesh-broker-nats","image":"`+pinned+`","env":{"V":"`+env+ + `"},"restart-on":["nats.program-snapshot","nats.server-conf"]`+module+`}`) + } + m := &machine{containers: map[string]*fakeContainer{}} + _, state, err := ApplyModule(context.Background(), archHost(t), bus("1", "a", "one", ""), Scope{Module: store.NodeModule}, + state0(), store.OriginDeclared, m.run, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + made := m.containers["mesh-broker-nats"] + if made == nil { + t.Fatalf("the node's own made no server: %v", m.asked) + } + m.asked = nil + next := bus("2", "b", "two", "nats") + next.Resources = append(next.Resources, declare(t, `{"id":"nats.tools-account","type":"file","path":"`+account+ + `","content":"tools","module":"nats"}`).Resources...) + scope := Scope{Module: "nats", Placed: map[string]bool{"nats": true}} + if keep { + scope.Keep = map[string]bool{"nats.server": true, "nats.program-snapshot": true} + } + report, after, err := ApplyModule(context.Background(), archHost(t), next, scope, state, + store.OriginDeclared, m.run, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + var moved []string + for _, a := range m.asked { + for _, verb := range []string{"docker rm", "docker run", "docker create", "docker stop", "docker restart"} { + if strings.HasPrefix(a, verb+" ") { + moved = append(moved, a) + } + } + } + if keep == (len(moved) > 0) { + t.Fatalf("kept %v, and the server was moved by %v", keep, moved) + } + if got, _ := os.ReadFile(conf); string(got) != "b" { + t.Errorf("the server's file holds %q", got) + } + if got, _ := os.ReadFile(account); string(got) != "tools" { + t.Errorf("the tools' account file holds %q", got) + } + if got, _ := os.ReadFile(filepath.Join(program, "snapshot")); keep != (string(got) == "one") { + t.Errorf("kept %v, and the server's program is %q", keep, got) + } + for _, id := range []string{"nats.server", "nats.program-snapshot"} { + rec, ok := after.Find(id) + if !ok || rec.Module != "nats" { + t.Errorf("%s is recorded %+v, not as nats's", id, rec) + } + if o := outcomeOf(report, id); keep != (o.Action == keptAsItRuns) { + t.Errorf("kept %v, and %s was %s: %s", keep, id, o.Action, o.Detail) + } + } + if keep && report.Changed() == false { + t.Error("an apply that wrote the server's files says nothing changed") + } + }) + } } func state0() store.State { return store.State{} } -- 2.54.0 From 9fd542985376716ef92ddd09893c4fae753a106b Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 09:03:24 +0100 Subject: [PATCH 3/6] =?UTF-8?q?The=20node=20agent=20keeps=20the=20bus=20se?= =?UTF-8?q?rver=20unless=20the=20want=20carries=20the=20warrant's=20mark,?= =?UTF-8?q?=20and=20the=20bus's=20have=20says=20what=20the=20server=20runs?= =?UTF-8?q?=20(ADR=200594=20=C2=A72-=C2=A74,=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/nodeagent.go | 15 +-- internal/agent/agent.go | 169 ++++++++++++++++++++++++++-- internal/agent/agent_test.go | 122 +++++++++++++++++++- internal/link/have.go | 1 + internal/link/messages.go | 2 + internal/selfcompose/selfcompose.go | 24 ++++ 6 files changed, 318 insertions(+), 15 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 8001ee3..a62df63 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -85,7 +85,7 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool return agent.Have{}, false } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.SealedAt}, true + Sealed: got.SealedAt, Server: got.Server, Build: got.Build}, true } // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). @@ -103,7 +103,7 @@ func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, b return agent.Have{}, false, err } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.SealedAt}, true, nil + Sealed: got.SealedAt, Server: got.Server, Build: got.Build}, true, nil } func (h sessionHaves) Write(ctx context.Context, x agent.Have) { @@ -120,7 +120,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) { } h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase, Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits, - Sealed: x.Sealed}) + Sealed: x.Sealed, Server: x.Server}) } func (h sessionHaves) Removed(ctx context.Context, module string) { @@ -538,7 +538,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus 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 { + Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed, keep map[string]bool) agent.Result { actx, done := applyContext(session.WithLink(ctx)) // While the grants part's list is followed, the controller's list in the bus module's composition is not // written: the apply keeps the list on the machine (ADR 0547 clause 8, never both). @@ -547,7 +547,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus if herr != nil { res = failedWith(herr.Error()) } else { - res = applyComposition(actx, opts, held, others, ag.Placed(scopeName(c)), ag.RemovedByWant(), sched, say) + res = applyComposition(actx, opts, held, others, ag.Placed(scopeName(c)), ag.RemovedByWant(), keep, sched, say) } done() // The mesh's own certificate it carries is kept for this machine's record (ADR 0555 §3). @@ -632,6 +632,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone env.Composed, env.SelfComposed, env.Policy = composer, true, composer.Policy env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does + env.BusServer = composer.BusServer // the bus server an unwarranted want leaves as it runs (ADR 0594 §2) session.NoteComposes(linked, delivery.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere") ag = agent.New(env) own.set(ag) @@ -692,7 +693,7 @@ func (c *agentNow) get() *agent.Agent { // // removed is the modules whose want says removed: in the node's own apply, a resource of any other module goes only when // its module is among them (issue 774). -func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed, placed, removed map[string]bool, +func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed, placed, removed, keep map[string]bool, sched *apply.Scheduler, say link.Announce) agent.Result { name := scopeName(c) if c.Module == "" { @@ -708,7 +709,7 @@ func applyComposition(ctx context.Context, opts options, c delivery.Composed, ot if err := refuseEmptyWrites(d); err != nil { return failedWith(err.Error()) } - scope := apply.Scope{Module: name, Others: resourcesOf(others), Placed: placed} + scope := apply.Scope{Module: name, Others: resourcesOf(others), Placed: placed, Keep: keep} if name == agent.NodeModule { scope.RemovedByWant = removed if scope.RemovedByWant == nil { diff --git a/internal/agent/agent.go b/internal/agent/agent.go index c392947..a6d9892 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -62,6 +62,13 @@ type Have struct { // slot (delivery.Have.SealedAt): kept in the have so the composer reuses them after a restart (ADR 0547 §3), each at // its own slot alone (issue 1126, C1). Sealed delivery.Seals + // Server is, in the bus module's have, what the bus server runs (delivery.Have.Server, novox/hq ADR 0594 §4): the + // fingerprint whose server this node agent last made, or the old-form build of the server it took over. Empty on + // every other have. + Server string + // Build is the old-form build a have from before the fingerprints names (delivery.Have.Build): read, never written + // by the agent; the server the bus module's first want takes over runs it. + Build string } // Haves is the node's have bucket as the agent uses it. @@ -103,8 +110,13 @@ type Env struct { Haves Haves // Apply applies one module's composition (scope NodeModule for the node's own), given the compositions the node's // other modules run so a path, unit or group they hold is not removed or taken. It is called with one module at a - // time. - Apply func(ctx context.Context, c delivery.Composed, others []delivery.Composed) Result + // time. keep is the resources, by id, the apply leaves as they run when they exist (apply.Scope.Keep): the bus + // server's, on a want of the bus module without the warrant's mark (novox/hq ADR 0594 §2); nil keeps nothing. + Apply func(ctx context.Context, c delivery.Composed, others []delivery.Composed, keep map[string]bool) Result + // BusServer is, for a module that provides the mesh's bus, the ids its manifest declares of the bus server's + // resources (catalogue.BusServerOf, novox/hq ADR 0594 §7), empty for any other module; known says whether the + // module's manifest could be read. Nil names no bus module. + BusServer func(ctx context.Context, module string) (ids []string, known bool) // Remove removes the module's own resources and nothing of another's. Remove func(ctx context.Context, module string, others []delivery.Composed) Result // Backup backs a module up before a new composition of it is applied; nil backs up nothing. @@ -221,6 +233,9 @@ type Agent struct { again chan againAsk transient chan struct{} stopped chan struct{} + // server is, by bus module, what its have says the bus server runs (Have.Server): read once from the have, then + // kept as this agent writes it, so every have written of the module carries it (novox/hq ADR 0594 §4). + server map[string]string } type againAsk struct { @@ -261,7 +276,7 @@ func New(env Env) *Agent { 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), failedAgain: make(chan string, 8), tries: map[string]int{}, again: make(chan againAsk, 64), - transient: make(chan struct{}, 1), stopped: make(chan struct{})} + transient: make(chan struct{}, 1), stopped: make(chan struct{}), server: map[string]string{}} } // Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition @@ -572,6 +587,13 @@ func (a *Agent) NotComposedEach(ctx context.Context, why string, held map[string // said runs, its fingerprint, digest and sealed values, rather than a have naming nothing: what runs is not lost, and // the boxes the composer reuses after a restart stay (review of L2, rounds 2 and 3). func (a *Agent) writeHave(ctx context.Context, h Have) { + if h.Server == "" && h.Module != NodeModule { + h.Server = a.serverOf(ctx, h.Module) + } else if h.Server != "" { + a.mu.Lock() + a.server[h.Module] = h.Server + a.mu.Unlock() + } if h.Composed == "" && h.Module != NodeModule { if last, ok := a.env.Haves.Read(ctx, h.Module); ok && last.Composed != "" { h.FP, h.Composed = last.FP, last.Composed @@ -588,6 +610,56 @@ func (a *Agent) writeHave(ctx context.Context, h Have) { a.env.Haves.Write(ctx, h) } +// serverOf is what the module's have says the bus server runs (Have.Server): kept since this agent read or wrote it, +// else read from the have once. Empty for every module but the bus. +func (a *Agent) serverOf(ctx context.Context, module string) string { + a.mu.Lock() + s, known := a.server[module] + a.mu.Unlock() + if known { + return s + } + if h, ok := a.env.Haves.Read(ctx, module); ok { + s = h.Server + } + a.mu.Lock() + if _, written := a.server[module]; !written { + a.server[module] = s + } + s = a.server[module] + a.mu.Unlock() + return s +} + +// busKeep is, for a module that provides the mesh's bus, the ids of its server's resources (novox/hq ADR 0594 §7) as +// its composition names them (`.`), and true; nil and true for any other module. known is false when the +// module's manifest could not be read: whether it is the bus cannot be told, so nothing of it is applied. +func (a *Agent) busKeep(ctx context.Context, module string) (server map[string]bool, known bool) { + if module == NodeModule || a.env.BusServer == nil { + return nil, true + } + ids, known := a.env.BusServer(ctx, module) + if !known || len(ids) == 0 { + return nil, known + } + server = map[string]bool{} + for _, id := range ids { + server[module+"."+id] = true + } + return server, true +} + +// serverOwed says a want of the bus module carries the warrant's mark at a fingerprint whose server does not run yet +// (novox/hq ADR 0594 §4): applied once more though its fingerprint and composition are what runs, so the server is +// replaced at the fingerprint already applied. +func (a *Agent) serverOwed(ctx context.Context, module string, w wantNow) bool { + if w.Want.Warrant == "" { + return false + } + server, _ := a.busKeep(ctx, module) + return len(server) > 0 && a.serverOf(ctx, module) != w.Want.FP +} + func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) delivery.Seals { a.mu.Lock() if c, ok := a.applied[module]; ok && c.Err == nil && c.Dig == composed { @@ -1213,6 +1285,16 @@ func (a *Agent) settleOne(ctx context.Context, module string) { a.remove(ctx, module, w) return } + if _, known := a.busKeep(ctx, module); !known { + // Whether it is the bus, whose server an unwarranted want never replaces (novox/hq ADR 0594 §2), cannot be told: + // nothing of it is applied until its manifest reads. + a.mu.Lock() + a.sayOnce("bus-unknown:"+module, fmt.Sprintf("%s at %s is not applied: its manifest cannot be read, so whether it "+ + "is the bus, whose server only a warranted want replaces, cannot be told", module, w.Want.FP)) + a.mu.Unlock() + return + } + owed := a.serverOwed(ctx, module, w) a.mu.Lock() comp, have := a.compositionFor(module, w.Want.FP) a.mu.Unlock() @@ -1247,14 +1329,15 @@ func (a *Agent) settleOne(ctx context.Context, module string) { if !a.checked[module] { a.checked[module] = true a.mu.Unlock() - if a.baseline(ctx, module, w, comp) { + if !owed && a.baseline(ctx, module, w, comp) { return } } else { a.mu.Unlock() } a.mu.Lock() - if r, ok := a.running[module]; ok && r.FP == w.Want.FP && r.Composed == w.Want.Composed && r.Phase != delivery.PhaseRemoved { + // What runs is what the want names, and no warranted want owes the bus server's replacement (ADR 0594 §4). + if r, ok := a.running[module]; ok && !owed && r.FP == w.Want.FP && r.Composed == w.Want.Composed && r.Phase != delivery.PhaseRemoved { a.mu.Unlock() return } @@ -1405,6 +1488,35 @@ func (a *Agent) inactiveAtStart(ctx context.Context, module string, comp compNow func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNow, others []delivery.Composed) { fp, dig := w.Want.FP, w.Want.Composed + // The bus module's server is kept as it runs unless the want carries the warrant's mark (novox/hq ADR 0594 §2, §6): + // the mark alone decides. Its have says what the server runs (§4): before the apply what it ran, the fingerprint + // applied when the apply kept it a fingerprint or a build ago, and the want's fingerprint once a warranted apply + // replaced it. + keep, _ := a.busKeep(ctx, module) + ranServer, madeServer := "", "" + if len(keep) > 0 { + ranServer = a.serverOf(ctx, module) + if ranServer == "" { + if h, ok := a.env.Haves.Read(ctx, module); ok { + ranServer = h.FP + if ranServer == "" { + ranServer = h.Build + } + } + } + if ranServer == "" { + ranServer = fp // a bus module this node never ran: its server is made at this fingerprint + } + madeServer = ranServer + a.mu.Lock() + a.server[module] = ranServer + a.mu.Unlock() + if w.Want.Warrant != "" { + keep, madeServer = nil, fp + a.env.Say(fmt.Sprintf("%s at %s carries the warrant %q: its server, running %s, is replaced", module, fp, + w.Want.Warrant, ranServer)) + } + } a.mu.Lock() _, ranBefore := a.running[module] ranBefore = ranBefore || a.prior[module] @@ -1444,7 +1556,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo } } a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)}) - res := a.env.Apply(ctx, comp.C, others) + res := a.env.Apply(ctx, comp.C, others, keep) a.mu.Lock() what := fmt.Sprintf("%s at %s", module, fp) if module == NodeModule { @@ -1494,7 +1606,8 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo if a.env.Settled != nil { a.env.Settled(module) } - a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail}) + a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail, + Server: madeServer}) } func (a *Agent) remove(ctx context.Context, module string, w wantNow) { @@ -1599,6 +1712,13 @@ 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 } + // For the bus module the judgement covers its server alone (novox/hq ADR 0594 §3): a want with the warrant's mark is + // the operator's warrant, the put's ask after the streams' snapshot or the hand write at the terminal, and stands as + // its yes (ADR 0546 §5); one without it is judged with the server's resources as they run, which it leaves. + server, _ := a.busKeep(ctx, module) + if len(server) > 0 && w.Want.Warrant != "" { + 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] @@ -1632,6 +1752,11 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp } var j judge.Judged switch { + case len(server) > 0 && !known: + // What the bus runs cannot be read, and its server is left as it runs whatever it is: nothing to hold. + return false + case len(server) > 0: + j = judge.JudgeMove(&from, withServerAsItRuns(from, comp.C, server)) case known: j = judge.JudgeMove(&from, comp.C) case prior: @@ -1653,6 +1778,36 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp return true } +// withServerAsItRuns is c with each resource of server taken as from runs it (novox/hq ADR 0594 §3): what an apply that +// keeps the bus server makes of the machine, so its judgement covers the rest of the module alone. A server resource +// from does not run stays as c declares it: an apply makes one that does not exist. +func withServerAsItRuns(from, c delivery.Composed, server map[string]bool) delivery.Composed { + running := map[string]json.RawMessage{} + for _, raw := range from.Resources { + if id := resourceID(raw); server[id] { + running[id] = raw + } + } + out := c + out.Resources = make([]json.RawMessage, 0, len(c.Resources)) + for _, raw := range c.Resources { + if was, ok := running[resourceID(raw)]; ok { + raw = was + } + out.Resources = append(out.Resources, raw) + } + return out +} + +// resourceID is a composed resource's id, empty when it names none. +func resourceID(raw json.RawMessage) string { + var r struct { + ID string `json:"id"` + } + _ = json.Unmarshal(raw, &r) + return r.ID +} + // 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. // diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index f2b3306..d65e75e 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -17,6 +17,15 @@ import ( const node = "n1" +func mustComposedKey(t *testing.T, module, fp string) string { + t.Helper() + key, err := delivery.ComposedKey(node, module, fp) + if err != nil { + t.Fatal(err) + } + return key +} + func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } // A watcher the test writes to; it says Caught once the keys written before the watch began are out. @@ -117,6 +126,7 @@ type bed struct { backup func(module string) Backup kept map[string][]json.RawMessage others [][]delivery.Composed + keeps []map[string]bool // what each apply was told to keep as it runs (ADR 0594 §2) said []string // settled is each module Settled was told, and settledRan whether what runs held it then. settled []string @@ -128,10 +138,11 @@ type bed struct { func newBed(t *testing.T) *bed { b := &bed{t: t, wants: newFeed(), comps: newFeed(), haves: &haves{}} b.env = Env{Node: node, Wants: b.wants, Composed: b.comps, Haves: b.haves, - Apply: func(_ context.Context, c delivery.Composed, others []delivery.Composed) Result { + Apply: func(_ context.Context, c delivery.Composed, others []delivery.Composed, keep map[string]bool) Result { b.mu.Lock() defer b.mu.Unlock() b.others = append(b.others, others) + b.keeps = append(b.keeps, keep) name := c.Module if name == "" { name = NodeModule @@ -1529,6 +1540,115 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( return len(b.appliedList()) == applied+2 && h.Phase == string(delivery.PhaseRunning) && h.FP == fpB && r.Phase == string(delivery.PhaseRunning) && r.FP == fpB && r.Composed == rdig }) + // Issue 774, ADR 0594 §2-§4: the bus module's server is kept as it runs on every want without the warrant's mark (a + // new fingerprint whose server recreates, a revision of its user list), never held for a yes, and its have names the + // fingerprint applied and, beside it, what the server runs: at the take-over the old-form build. A warranted want at + // the fingerprint already applied replaces the server once (the warrant is its yes), and the same want again applies + // nothing. A module that is not the bus keeps nothing, marked or not. + t.Run("the bus module", func(t *testing.T) { + b := newBed(t) + b.env.SelfComposed = true + b.env.Policy = func(context.Context, string) string { return catalogue.PolicyRecord } + b.env.BusServer = func(_ context.Context, m string) ([]string, bool) { + if m == "nats" { + return []string{"program-entrypoint", "program-snapshot", "server"}, true + } + return nil, true + } + const took = "build-1792674595610176744" + b.haves.Write(context.Background(), Have{Module: "nats", Build: took, Phase: string(delivery.PhaseRunning)}) + stop := b.run() + defer stop() + bus := func(fp, image, users string) (string, []byte, string) { + server, _ := json.Marshal(map[string]any{"id": "nats.server", "type": "container", "module": "nats", + "name": "mesh-broker-nats", "image": "nats@sha256:" + strings.Repeat(image, 64)}) + c := delivery.Composed{Node: node, Module: "nats", FP: fp, Resources: []json.RawMessage{server, + file("nats", "server-conf", "listen"), file("nats", "bus-users", users)}} + key, _ := delivery.ComposedKey(node, "nats", fp) + return encodedAs(t, c, key) + } + want := func(module, fp, warrant string) { + key, _ := delivery.WantKey(node, module) + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Self: true, Warrant: warrant}) + if err != nil { + t.Fatal(err) + } + b.wants.put(key, raw) + } + server := map[string]bool{"nats.server": true, "nats.program-snapshot": true, "nats.program-entrypoint": true} + lastKeep := func() map[string]bool { + b.mu.Lock() + defer b.mu.Unlock() + return b.keeps[len(b.keeps)-1] + } + fpA, fpB := fpOf("a"), fpOf("b") + for i, row := range []struct { + name, fp, image, users, warrant string + applies bool + keep map[string]bool + server string + }{ + {name: "the take-over of the folded server", fp: fpA, image: "1", users: "u1", applies: true, keep: server, server: took}, + {name: "a new fingerprint whose server recreates", fp: fpB, image: "2", users: "u1", applies: true, keep: server, server: took}, + {name: "a revision of the user list", fp: fpB, image: "2", users: "u2", applies: true, keep: server, server: took}, + {name: "the operator's warrant at the fingerprint applied", fp: fpB, image: "2", users: "u2", warrant: delivery.WarrantOperator, + applies: true, server: fpB}, + {name: "the same warrant again", fp: fpB, image: "2", users: "u2", warrant: delivery.WarrantOperator, server: fpB}, + {name: "a build after the warrant", fp: fpA, image: "3", users: "u2", applies: true, keep: server, server: fpB}, + } { + before := len(b.appliedList()) + key, raw, dig := bus(row.fp, row.image, row.users) + b.comps.put(key, raw) + want("nats", row.fp, row.warrant) + if row.applies { + eventually(t, row.name+": not applied", func() bool { + h := b.haves.last("nats") + return len(b.appliedList()) == before+1 && h.Composed == dig && h.Phase == string(delivery.PhaseRunning) + }) + if got := lastKeep(); !maps.Equal(got, row.keep) { + t.Errorf("%s: the apply kept %v, want %v", row.name, got, row.keep) + } + } else { + time.Sleep(100 * time.Millisecond) + if got := b.appliedList(); len(got) != before { + t.Errorf("%s: applied again: %v", row.name, got) + } + } + if h := b.haves.last("nats"); h.FP != row.fp || h.Server != row.server || h.Waits != "" { + t.Errorf("row %d, %s: the bus's have is %+v, its server %s", i, row.name, h, row.server) + } + } + // Every have written of the bus says what its server runs, applying included: one that did not would lose it. + b.haves.mu.Lock() + for _, h := range b.haves.log[1:] { + if h.Module == "nats" && h.Server == "" { + t.Errorf("a have of the bus names no server: %+v", h) + } + } + b.haves.mu.Unlock() + for _, warrant := range []string{"", delivery.WarrantOperator} { + before := len(b.appliedList()) + fp := fpOf("c") + if warrant != "" { + fp = fpOf("d") + } + key, raw, _ := encodedAs(t, delivery.Composed{Node: node, Module: "web", FP: fp, + Resources: []json.RawMessage{file("web", "x", warrant)}}, mustComposedKey(t, "web", fp)) + b.comps.put(key, raw) + want("web", fp, warrant) + eventually(t, "web was not applied", func() bool { return len(b.appliedList()) == before+1 }) + if got := lastKeep(); got != nil { + t.Errorf("a module that is not the bus kept %v", got) + } + if h := b.haves.last("web"); h.Server != "" { + t.Errorf("web's have names a server: %+v", h) + } + } + time.Sleep(100 * time.Millisecond) + if got := b.appliedList(); len(got) != 7 { + t.Errorf("applied %v, not once for each change", got) + } + }) } // 1957/AC1 and AC4 (unit; ADR 0568 §1, §4): an engine start takes a module from what the machine shows, not only from the diff --git a/internal/link/have.go b/internal/link/have.go index 6f8040a..37cb7ed 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -384,6 +384,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { h.Phase = delivery.PhaseFailed } h.SealedAt = f.Sealed + h.Server = f.Server h.Health = s.healthOf(module, f) if h.Detail == "" { h.Detail = s.wordsOf(module, h.Health) diff --git a/internal/link/messages.go b/internal/link/messages.go index 531a1ab..97e3060 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -641,4 +641,6 @@ type ModuleFact struct { Waits string // Sealed is the values the composition it names carries sealed to this node's key, by slot (delivery.Have.SealedAt). Sealed delivery.Seals + // Server is, for the bus module, what its server runs (delivery.Have.Server, novox/hq ADR 0594 §4). + Server string } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 3019186..fb1426e 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -1163,6 +1163,30 @@ func (c *Composer) Policy(ctx context.Context, module string) string { return policy } +// BusServer is, for a module whose manifest provides the mesh's bus, the ids it declares of the bus server's resources +// (catalogue.BusServerOf, novox/hq ADR 0594 §7), none for any other; known says whether the manifest could be read. +func (c *Composer) BusServer(ctx context.Context, module string) (ids []string, known bool) { + c.mu.Lock() + m, ok := c.modules[module] + c.mu.Unlock() + if !ok || m.Manifest == "" { + return nil, false + } + raw, err := reader{c: c}.Manifest(ctx, m.Manifest) + if err != nil { + return nil, false + } + parsed, held, err := c.taken(module, m, raw) + if err != nil || held != "" { + return nil, false + } + for id := range catalogue.BusServerOf(parsed) { + ids = append(ids, id) + } + sort.Strings(ids) + return ids, true +} + // 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). -- 2.54.0 From 052004d1cf173c640906f542645e37292f63ffbd Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 09:03:30 +0100 Subject: [PATCH 4/6] =?UTF-8?q?The=20node's=20own=20key=20says=20its=20age?= =?UTF-8?q?nt=20keeps=20the=20bus=20server,=20so=20the=20assigner=20writes?= =?UTF-8?q?=20the=20bus=20want=20only=20to=20such=20a=20node=20(ADR=200594?= =?UTF-8?q?=20=C2=A78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/nodeagent.go | 4 +++- internal/link/have.go | 16 ++++++++++++---- internal/link/have_test.go | 12 ++++++++++++ 3 files changed, 27 insertions(+), 5 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index a62df63..08cbecb 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -633,7 +633,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus env.Composed, env.SelfComposed, env.Policy = composer, true, composer.Policy env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does env.BusServer = composer.BusServer // the bus server an unwarranted want leaves as it runs (ADR 0594 §2) - session.NoteComposes(linked, delivery.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere") + // And its agent keeps the bus server as it runs unless a want carries the warrant's mark (ADR 0594 §8). + session.NoteComposes(linked, delivery.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere", + delivery.KeepsBusServer) ag = agent.New(env) own.set(ag) current.set(ag) diff --git a/internal/link/have.go b/internal/link/have.go index 37cb7ed..46b017c 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" "sync" @@ -55,6 +56,9 @@ type haveState struct { // nodeComposes and nodeComposesWhy are whose compositions the node applies and why (HaveNode.Composes), written into // its key. nodeComposes, nodeComposesWhy string + // nodeKeeps is what the node agent keeps as it runs on a want without the warrant's mark (HaveNode.Keeps, novox/hq + // ADR 0594 §8), written into its key. + nodeKeeps []string // written is the build of each module this node-engine wrote a have key for and that the declaration it applied // still holds; removed is each such module the latest declaration no longer holds, with the build its key last // said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted). @@ -428,6 +432,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { facts.SealedAt = s.nodeSealed facts.Candidates = s.nodeCandidates facts.Composes, facts.ComposesWhy = s.nodeComposes, s.nodeComposesWhy + facts.Keeps = s.nodeKeeps if raw, err = delivery.EncodeHaveNode(facts); err == nil { want := facts if !q.putHave(write, bucket, key, raw, func(current []byte) bool { @@ -757,14 +762,17 @@ func (q *Session) NoteCandidates(ctx context.Context, answers map[string]deliver q.writeHaves(ctx, bus) } -// NoteComposes says, in the node's own key, whose compositions the node applies (delivery.ComposesOwn) and why: written -// at each start of its agent, where every have is read (issue 1126). -func (q *Session) NoteComposes(ctx context.Context, composes, why string) { +// NoteComposes says, in the node's own key, whose compositions the node applies (delivery.ComposesOwn) and why, and what +// its agent keeps as it runs on a want without the warrant's mark (delivery.KeepsBusServer, novox/hq ADR 0594 §8): written +// at each start of its agent, where every have is read (issue 1126). The assigner writes the bus module's want only to +// a node whose key says it keeps the bus server. +func (q *Session) NoteComposes(ctx context.Context, composes, why string, keeps ...string) { q.init() s := &q.haves s.mu.Lock() - changed := s.nodeComposes != composes || s.nodeComposesWhy != why + changed := s.nodeComposes != composes || s.nodeComposesWhy != why || !slices.Equal(s.nodeKeeps, keeps) s.nodeComposes, s.nodeComposesWhy = composes, why + s.nodeKeeps = slices.Clone(keeps) known := s.node != nil s.mu.Unlock() if !known || !changed { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index f092ed2..f6fc2c7 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "sync" "testing" @@ -579,6 +580,17 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if n, _ := delivery.DecodeHaveNode(l.kv["n1.node"].value); n.Candidates != nil { t.Fatalf("an answer left after its group went: %+v", n.Candidates) } + // Issue 774, ADR 0594 §4, §8: the bus module's have names what its server runs beside the fingerprint applied, and the + // node's key says its agent keeps the bus server. + q.NoteModule(context.Background(), ModuleFact{Module: "nats", FP: fp, Composed: dig, Server: "build-1792674595610176744"}) + if got := l.have(t, "n1.nats"); got.FP != fp || got.Server != "build-1792674595610176744" { + t.Fatalf("the bus's have: %+v", got) + } + q.NoteComposes(context.Background(), delivery.ComposesOwn, "it composes its own", delivery.KeepsBusServer) + if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || !slices.Equal(n.Keeps, []string{delivery.KeepsBusServer}) || + n.Composes != delivery.ComposesOwn { + t.Fatalf("the node key does not say it keeps the bus server: %+v %v", n, err) + } } // A module applied and running that has nothing long-running to judge is ok once the engine has looked: there is nothing -- 2.54.0 From a096abac92653621faa28cc39077c67330187c3d Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 09:06:19 +0100 Subject: [PATCH 5/6] =?UTF-8?q?The=20node=20agent's=20applies=20run=20thro?= =?UTF-8?q?ugh=20a=20runner=20its=20options=20name,=20and=20one=20integrat?= =?UTF-8?q?ion=20test=20proves=20an=20unwarranted=20bus=20want=20keeps=20t?= =?UTF-8?q?he=20server=20and=20a=20warranted=20one=20replaces=20it=20(ADR?= =?UTF-8?q?=200594=20=C2=A72-=C2=A74,=20=C2=A78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/main.go | 15 +- cmd/mesh-host/nodeagent.go | 2 +- cmd/mesh-host/nodeagent_integration_test.go | 216 ++++++++++++++++++++ 3 files changed, 230 insertions(+), 3 deletions(-) diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index b352d48..bff6c87 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -156,6 +156,17 @@ type options struct { file string // out is where what a command says goes. Stdout, and a buffer under test. out io.Writer + // runner is what the node agent's applies run commands through (issue 774, ADR 0594): nil is the machine's own, + // apply.ExecRunner; a test gives one that records what an apply asked of the container runtime. + runner apply.Runner +} + +// run is the runner the node agent's applies use: the options' own, else the machine's. +func (o options) run() apply.Runner { + if o.runner != nil { + return o.runner + } + return apply.ExecRunner } // parseArgs takes the subcommand first, then its flags. @@ -1637,11 +1648,11 @@ func applyModuleHeld(ctx context.Context, opts options, declared *declaration.De func(known store.State) (apply.Report, store.State, error) { if scope == nil { return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared, - apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), + opts.run(), announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) } return apply.ApplyModule(ctx, built, declared, *scope, known, store.OriginDeclared, - apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), + opts.run(), announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) }, announceOr(say)) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 08cbecb..09e9b31 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -508,7 +508,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // What this engine applied, kept on the machine and read at the start, so what runs is known before the node // composes its own (issue 774). Applied: appliedStoreFor(opts.state), - Inactive: inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), + Inactive: inactiveIn(&liveness.Exec{Run: liveness.Runner(opts.run()), System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), Revisions: func(ctx context.Context, key string) ([]agent.Update, error) { got, err := session.KVHistory(ctx, composedBucket, key) if err != nil { diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 0c9f677..a611c86 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -6,6 +6,7 @@ import ( "context" "crypto/ed25519" "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -14,13 +15,17 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/nodegrant" "github.com/novox/mesh-host/internal/selfcompose" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" ) // The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition @@ -619,3 +624,214 @@ func mustComposedKey(t *testing.T, module, fp string) string { } return key } + +// recordingRuntime is a container runtime that keeps its containers in memory and records every command an apply asks +// of it (issue 774): what the bus server's container met is read from it, never from a real runtime. +type recordingRuntime struct { + mu sync.Mutex + containers map[string]string // name: the spec label it was made with + asked []string +} + +func (r *recordingRuntime) run(_ context.Context, name string, args ...string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.asked = append(r.asked, name+" "+strings.Join(args, " ")) + if name != "docker" || len(args) == 0 { + return "", nil + } + switch args[0] { + case "info": + return "27.0\n", nil + case "container": + if len(args) > 1 && args[1] == "inspect" { + spec, ok := r.containers[args[len(args)-1]] + if !ok { + return "", errors.New("no such container") + } + return "true\t" + spec + "\n", nil + } + case "image": + if len(args) > 1 && args[1] == "inspect" { + return "2026-01-01T00:00:00Z\n", nil + } + case "rm": + delete(r.containers, args[len(args)-1]) + case "run", "create": + var made, spec string + for i, a := range args { + if a == "--name" && i+1 < len(args) { + made = args[i+1] + } + if a == "--label" && i+1 < len(args) && strings.HasPrefix(args[i+1], "mesh-host.spec=") { + spec = strings.TrimPrefix(args[i+1], "mesh-host.spec=") + } + } + r.containers[made] = spec + return "made\n", nil + } + return "", nil +} + +// moved is each command that removed, made, stopped or restarted the container named. +func (r *recordingRuntime) moved(container string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, a := range r.asked { + for _, verb := range []string{"docker rm ", "docker run ", "docker create ", "docker stop ", "docker restart ", "docker kill ", + "docker rename ", "docker start ", "docker container rm ", "docker container rename ", "docker container stop ", + "docker container restart ", "docker container start ", "docker container kill "} { + if strings.HasPrefix(a, verb) && strings.Contains(a, container) { + out = append(out, a) + } + } + } + return out +} + +// Issue 774, ADR 0594 §2-§4: the bus server the node's own composition made while the bus module was folded is taken +// over by the bus module's first want, at a new-form fingerprint and without the warrant's mark, and left as it runs: +// the node agent started whole against a real bus composes the bus module, writes its file and records the server as +// the bus module's, asks the container runtime to remove, make, stop or restart nothing, and its have names the +// fingerprint applied beside the old-form build the server still runs; its own key says it keeps the bus server. The +// operator's warranted want at the same fingerprint then replaces the server once, and the have says it runs that +// fingerprint. +func TestNatsTheBusServerIsKeptUntilAWarrantedWantReplacesIt(t *testing.T) { + b := newItBed(t) + runtime := &recordingRuntime{containers: map[string]string{}} + b.opts.runner = runtime.run + const server = "mesh-broker-it" + conf := filepath.Join(b.dir, "nats.conf") + image := "registry.example/nats@sha256:" + strings.Repeat("a", 64) + + // What the node's own composition made of the folded bus: the server and its file, recorded as the node's own. + built, err := system.For(builtFor) + if err != nil { + t.Fatal(err) + } + folded, err := declaration.Parse([]byte(`{"declaration":1,"resources":[` + + `{"id":"nats.server-conf","type":"file","path":"` + conf + `","content":"listen: old\n"},` + + `{"id":"nats.server","type":"container","name":"` + server + `","image":"` + image + `","env":{"FOLDED":"1"}}]}`)) + if err != nil { + t.Fatal(err) + } + _, known, err := apply.ApplyModule(context.Background(), built, folded, apply.Scope{Module: store.NodeModule}, store.State{}, + store.OriginDeclared, runtime.run, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if err := store.Save(b.opts.state, known); err != nil { + t.Fatal(err) + } + if moved := runtime.moved(server); len(moved) != 1 { + t.Fatalf("the folded server was not made once: %v", moved) + } + made := runtime.containers[server] + runtime.asked = nil + + kvOf := func(bucket string) nats.KeyValue { + kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5}) + if err != nil { + t.Fatal(err) + } + return kv + } + placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket) + kvOf(selfcompose.ResolvedBucket) + kvOf(delivery.BuilderBuildsBucket) + kvOf(delivery.PlacementCandidatesBucket) + kvOf(link.PackagesBucket) + man := `{"module":"nats","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],"resources":[` + + `{"id":"server-conf","type":"file","path":"` + conf + `","mode":"0644","content":"listen: new\n"},` + + `{"id":"server","type":"container","name":"` + server + `","image":"` + image + `"}]}` + digest := delivery.ManifestDigest([]byte(man)) + mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "nats", JSON: man}) + mkey, _ := delivery.ManifestsKey(digest) + fp := fpIt("e") + modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("3", 40)}) + rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, + Capabilities: []string{"service-manager", "package-manager", "container-runtime"}}}) + hkey, _ := delivery.HaveNodeKey(itNode) + places, _ := delivery.PlacementValues(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "nats"}}}) + // The have the engine before the keeper wrote of the folded bus: the old-form build, no fingerprint. + const took = "build-1792674595610176744" + natsHave, _ := delivery.HaveKey(itNode, "nats") + oldHave, _ := delivery.EncodeHave(delivery.Have{Build: took, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + for _, w := range []struct { + kv nats.KeyValue + key string + value []byte + }{{manifests, mkey, mraw}, {modules, "nats", modv}, {b.haves, hkey, rec}, {b.haves, natsHave, oldHave}} { + if _, err := w.kv.Put(w.key, w.value); err != nil { + t.Fatal(err) + } + } + busWant := func(warrant string) { + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Self: true, Warrant: warrant}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.WantKey(itNode, "nats") + if _, err := b.wants.Put(key, raw); err != nil { + t.Fatal(err) + } + } + busWant("") + stop := b.start() + defer stop() + for key, value := range places { + if _, err := placement.Put(key, value); err != nil { + t.Fatal(err) + } + } + + within(t, 10*time.Second, "the bus module's want was not applied", func() bool { + h := b.have("nats") + got, _ := os.ReadFile(conf) + return h.FP == fp && h.Phase == delivery.PhaseRunning && string(got) == "listen: new\n" + }) + if h := b.have("nats"); h.Server != took { + t.Errorf("the bus's have says its server runs %q, not the build it took over", h.Server) + } + if moved := runtime.moved(server); len(moved) != 0 || runtime.containers[server] != made { + t.Fatalf("an unwarranted want moved the bus server: %v", moved) + } + state, err := store.Load(b.opts.state) + if err != nil { + t.Fatal(err) + } + if rec, ok := state.Find("nats.server"); !ok || rec.Module != "nats" { + t.Errorf("the server is recorded %+v, not as the bus module's", rec) + } + // Said with the node's facts, as every start of its agent says them. + b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{ + map[string]any{"name": "container-runtime", "present": true}}}}) + within(t, 5*time.Second, "the node's key does not say it keeps the bus server", func() bool { + e, err := b.haves.Get(hkey) + if err != nil { + return false + } + n, err := delivery.DecodeHaveNode(e.Value()) + return err == nil && len(n.Keeps) == 1 && n.Keeps[0] == delivery.KeepsBusServer + }) + + // The operator's warrant at the fingerprint applied: the server is replaced once. + busWant(delivery.WarrantOperator) + within(t, 10*time.Second, "the warranted want did not replace the bus server", func() bool { + return b.have("nats").Server == fp + }) + var runs int + for _, a := range runtime.moved(server) { + if strings.HasPrefix(a, "docker run ") && strings.Contains(a, " --name "+server+" ") { + runs++ + } + } + if runs != 1 || runtime.containers[server] == made { + t.Errorf("the warranted want made the server %d times, its spec still the folded one %v: %v", runs, + runtime.containers[server] == made, runtime.moved(server)) + } + if h := b.have("nats"); h.FP != fp || h.Phase != delivery.PhaseRunning { + t.Errorf("the bus's have after the warrant: %+v", h) + } +} -- 2.54.0 From 2c5f818d152201fef9b86ade5ea5feeee3859a12 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 09:18:31 +0100 Subject: [PATCH 6/6] =?UTF-8?q?Review=20round=201=20of=20the=20bus=20want:?= =?UTF-8?q?=20a=20removal=20keeps=20the=20bus=20server,=20the=20have=20say?= =?UTF-8?q?s=20a=20server=20made=20anew=20runs=20the=20want's=20fingerprin?= =?UTF-8?q?t,=20the=20old=20build=20is=20read=20before=20the=20first=20wri?= =?UTF-8?q?te,=20and=20an=20unanswered=20inspect=20fails=20the=20apply=20r?= =?UTF-8?q?ather=20than=20replacing=20the=20server=20(ADR=200594=20=C2=A72?= =?UTF-8?q?,=20=C2=A74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/nodeagent.go | 16 ++--- internal/agent/agent.go | 58 ++++++++++++---- internal/agent/agent_test.go | 75 +++++++++++++++++--- internal/apply/apply.go | 60 ++++++++++++++-- internal/apply/module_scope_test.go | 102 +++++++++++++++++++++++----- internal/link/have.go | 17 ++++- internal/link/have_test.go | 6 ++ 7 files changed, 279 insertions(+), 55 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 09e9b31..b30421e 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -123,8 +123,8 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) { Sealed: x.Sealed, Server: x.Server}) } -func (h sessionHaves) Removed(ctx context.Context, module string) { - h.session.NoteRemoved(ctx, module, "") +func (h sessionHaves) Removed(ctx context.Context, module string, words ...string) { + h.session.NoteRemoved(ctx, module, "", words...) } // shorten keeps a have's detail within what a have value holds: its head, which says what failed, and its tail, which is @@ -564,9 +564,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus sayFacts(ctx, ag) return res }, - Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result { + Remove: func(ctx context.Context, module string, others []delivery.Composed, keep map[string]bool) agent.Result { actx, done := applyContext(session.WithLink(ctx)) - res := removeComposition(actx, opts, module, others, ag.Placed(module), sched, say) + res := removeComposition(actx, opts, module, others, ag.Placed(module), keep, sched, say) done() afterApply() sayFacts(ctx, ag) @@ -732,13 +732,13 @@ func applyComposition(ctx context.Context, opts options, c delivery.Composed, ot if got.Probe != nil { return failedWith(got.Probe.partial().Error()) } - return agent.Result{} + return agent.Result{Kept: got.Outcome.Kept} } // removeComposition removes one module's own resources and nothing of another's. -func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed, placed map[string]bool, +func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed, placed, keep map[string]bool, sched *apply.Scheduler, say link.Announce) agent.Result { - scope := apply.Scope{Module: module, Others: resourcesOf(others), Placed: placed} + scope := apply.Scope{Module: module, Others: resourcesOf(others), Placed: placed, Keep: keep} got := applyModuleHeld(ctx, opts, &declaration.Declaration{Version: declaration.Version}, &scope, sched, say) switch { case got.Refused != "": @@ -746,7 +746,7 @@ func removeComposition(ctx context.Context, opts options, module string, others case got.Err != nil: return agent.Result{Failed: true, Detail: shorten(got.Err.Error())} } - return agent.Result{} + return agent.Result{Kept: got.Outcome.Kept} } // busyModules are the modules whose last apply left a container running because it said busy. diff --git a/internal/agent/agent.go b/internal/agent/agent.go index a6d9892..aebc25d 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -77,8 +77,9 @@ type Haves interface { Read(ctx context.Context, module string) (Have, bool) // Write writes a module's have where it differs from the bucket's. Write(ctx context.Context, h Have) - // Removed writes the module's have with the phase removed, once, if a key of it exists. - Removed(ctx context.Context, module string) + // Removed writes the module's have with the phase removed, once, if a key of it exists; words, when given, are its + // detail. + Removed(ctx context.Context, module string, words ...string) } // Result is how one apply or removal came out. @@ -88,6 +89,8 @@ type Result struct { // JetStream API error while the bus reloads), not the module's: the apply is tried again without a new revision. Transient bool Detail string + // Kept is the resources of the keep set the apply left as they ran, by id (apply.Report.Kept, novox/hq ADR 0594 §4). + Kept []string } // Backup is the answer of the backup asked before a module moves. @@ -117,8 +120,9 @@ type Env struct { // resources (catalogue.BusServerOf, novox/hq ADR 0594 §7), empty for any other module; known says whether the // module's manifest could be read. Nil names no bus module. BusServer func(ctx context.Context, module string) (ids []string, known bool) - // Remove removes the module's own resources and nothing of another's. - Remove func(ctx context.Context, module string, others []delivery.Composed) Result + // Remove removes the module's own resources and nothing of another's, and leaves those of keep as they run (the bus + // server's, ADR 0594 §2). + Remove func(ctx context.Context, module string, others []delivery.Composed, keep map[string]bool) Result // Backup backs a module up before a new composition of it is applied; nil backs up nothing. Backup func(ctx context.Context, module string) Backup // Kept is the resources the node last applied, by module, from the declaration it kept before the switch (ADR 0539 @@ -612,6 +616,10 @@ func (a *Agent) writeHave(ctx context.Context, h Have) { // serverOf is what the module's have says the bus server runs (Have.Server): kept since this agent read or wrote it, // else read from the have once. Empty for every module but the bus. +// +// A have written before this node agent kept the server names none: the server then runs what that have says the module +// runs, its fingerprint, else its old-form build (the take-over of a server the node's own composition made). Read so on +// the first read, before this agent writes the have, so every have it writes carries it. func (a *Agent) serverOf(ctx context.Context, module string) string { a.mu.Lock() s, known := a.server[module] @@ -619,8 +627,17 @@ func (a *Agent) serverOf(ctx context.Context, module string) string { if known { return s } + if server, _ := a.busKeep(ctx, module); len(server) == 0 { + return "" // not the bus, or not known to be yet: nothing kept, so a later read asks again + } if h, ok := a.env.Haves.Read(ctx, module); ok { s = h.Server + if s == "" { + s = h.FP + } + if s == "" { + s = h.Build + } } a.mu.Lock() if _, written := a.server[module]; !written { @@ -1496,14 +1513,6 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo ranServer, madeServer := "", "" if len(keep) > 0 { ranServer = a.serverOf(ctx, module) - if ranServer == "" { - if h, ok := a.env.Haves.Read(ctx, module); ok { - ranServer = h.FP - if ranServer == "" { - ranServer = h.Build - } - } - } if ranServer == "" { ranServer = fp // a bus module this node never ran: its server is made at this fingerprint } @@ -1606,6 +1615,10 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo if a.env.Settled != nil { a.env.Settled(module) } + if len(keep) > 0 && len(res.Kept) == 0 { + // Nothing of the server was there to keep: the apply made it at this fingerprint (ADR 0594 §4). + madeServer = fp + } a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail, Server: madeServer}) } @@ -1624,7 +1637,16 @@ func (a *Agent) remove(ctx context.Context, module string, w wantNow) { } others := a.heldOut(module) a.mu.Unlock() - res := a.env.Remove(ctx, module, others) + // A removed want of the bus module never removes its server (novox/hq ADR 0594 §2): a removal carries no warrant, and + // removing the server is not built. One whose manifest cannot be read keeps what would be the server's, said. + keep, known := a.busKeep(ctx, module) + if !known { + keep = map[string]bool{} + for _, id := range catalogue.BusServer { + keep[module+"."+id] = true + } + } + res := a.env.Remove(ctx, module, others, keep) a.mu.Lock() if res.Failed { a.attempt[module] = attempt{FP: "removed", WantRev: w.Rev} @@ -1644,6 +1666,16 @@ func (a *Agent) remove(ctx context.Context, module string, w wantNow) { if a.env.Settled != nil { a.env.Settled(module) } + if len(res.Kept) > 0 { + words := fmt.Sprintf("removed, but for %s: the bus server stays as it runs until a want carrying the operator's "+ + "warrant removes it (novox/hq ADR 0594 §2)", strings.Join(res.Kept, ", ")) + if !known { + words += "; its manifest could not be read, so what would be its server was left" + } + a.env.Say(module + " is " + words) + a.env.Haves.Removed(ctx, module, words) + return + } a.env.Haves.Removed(ctx, module) } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index d65e75e..34846be 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -65,10 +65,11 @@ func (f *feed) put(key string, value []byte) { } type haves struct { - mu sync.Mutex - stored map[string]Have - log []Have - removed []string + mu sync.Mutex + stored map[string]Have + log []Have + removed []string + removedWords []string // unreadable is each module whose have the bus cannot read now (ReadErr). unreadable map[string]bool } @@ -99,10 +100,11 @@ func (h *haves) Write(_ context.Context, x Have) { h.stored[x.Module] = x h.log = append(h.log, x) } -func (h *haves) Removed(_ context.Context, m string) { +func (h *haves) Removed(_ context.Context, m string, words ...string) { h.mu.Lock() defer h.mu.Unlock() h.removed = append(h.removed, m) + h.removedWords = append(h.removedWords, strings.Join(words, "")) } func (h *haves) last(module string) Have { h.mu.Lock() @@ -127,7 +129,9 @@ type bed struct { kept map[string][]json.RawMessage others [][]delivery.Composed keeps []map[string]bool // what each apply was told to keep as it runs (ADR 0594 §2) - said []string + // serverGone is a bus server not on the machine: an apply keeping it keeps nothing and makes it. + serverGone bool + said []string // settled is each module Settled was told, and settledRan whether what runs held it then. settled []string settledRan []bool @@ -143,6 +147,13 @@ func newBed(t *testing.T) *bed { defer b.mu.Unlock() b.others = append(b.others, others) b.keeps = append(b.keeps, keep) + var kept []string // what of keep was there to keep: all of it, unless the test took the server away + for id := range keep { + if !b.serverGone { + kept = append(kept, id) + } + } + slices.Sort(kept) name := c.Module if name == "" { name = NodeModule @@ -151,13 +162,19 @@ func newBed(t *testing.T) *bed { if b.failWith != "" && (b.failTimes == 0 || len(b.applied) <= b.failTimes) { return Result{Failed: true, Transient: b.transient, Detail: b.failWith} } - return Result{} + return Result{Kept: kept} }, - Remove: func(_ context.Context, m string, _ []delivery.Composed) Result { + Remove: func(_ context.Context, m string, _ []delivery.Composed, keep map[string]bool) Result { b.mu.Lock() defer b.mu.Unlock() b.removed = append(b.removed, m) - return Result{} + b.keeps = append(b.keeps, keep) + var kept []string + for id := range keep { + kept = append(kept, id) + } + slices.Sort(kept) + return Result{Kept: kept} }, Backup: func(_ context.Context, m string) Backup { b.mu.Lock() @@ -1582,8 +1599,22 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( return b.keeps[len(b.keeps)-1] } fpA, fpB := fpOf("a"), fpOf("b") + // A have written before the first apply (the node cannot compose yet) keeps the old build's name as what the + // server runs: the writer names no build, so it is read before the first write. + want("nats", fpA, "") + eventually(t, "the bus's want was not taken", func() bool { + b.ag.mu.Lock() + defer b.ag.mu.Unlock() + return b.ag.wants["nats"].Present + }) + b.ag.NotComposed(context.Background(), "a stand-in for the bus's secret") + if h := b.haves.last("nats"); h.Build != "" || h.Server != took { + t.Fatalf("a have written before the first apply: %+v", h) + } + b.ag.NotComposed(context.Background(), "") for i, row := range []struct { name, fp, image, users, warrant string + gone bool // the server is not on the machine: made, at the want's fingerprint applies bool keep map[string]bool server string @@ -1595,7 +1626,12 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( applies: true, server: fpB}, {name: "the same warrant again", fp: fpB, image: "2", users: "u2", warrant: delivery.WarrantOperator, server: fpB}, {name: "a build after the warrant", fp: fpA, image: "3", users: "u2", applies: true, keep: server, server: fpB}, + {name: "a build whose server is not there", fp: fpOf("e"), image: "4", users: "u2", gone: true, applies: true, + keep: server, server: fpOf("e")}, } { + b.mu.Lock() + b.serverGone = row.gone + b.mu.Unlock() before := len(b.appliedList()) key, raw, dig := bus(row.fp, row.image, row.users) b.comps.put(key, raw) @@ -1645,9 +1681,28 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( } } time.Sleep(100 * time.Millisecond) - if got := b.appliedList(); len(got) != 7 { + if got := b.appliedList(); len(got) != 8 { t.Errorf("applied %v, not once for each change", got) } + // A removed want removes the rest of the bus and never its server: it is kept, said in the journal and in the + // removed have's words. + rkey, _ := delivery.WantKey(node, "nats") + rraw, _ := delivery.EncodeWant(delivery.Want{Removed: true}) + b.wants.put(rkey, rraw) + eventually(t, "the bus's removal was not asked", func() bool { + b.haves.mu.Lock() + defer b.haves.mu.Unlock() + return slices.Contains(b.haves.removed, "nats") + }) + if got := lastKeep(); !maps.Equal(got, server) { + t.Errorf("the bus's removal kept %v, not its server %v", got, server) + } + b.haves.mu.Lock() + words := b.haves.removedWords[len(b.haves.removedWords)-1] + b.haves.mu.Unlock() + if !strings.Contains(words, "the bus server stays") || !strings.Contains(words, "nats.server") { + t.Errorf("the removed have does not say the server stays: %q", words) + } }) } diff --git a/internal/apply/apply.go b/internal/apply/apply.go index ff94eda..c1027f0 100644 --- a/internal/apply/apply.go +++ b/internal/apply/apply.go @@ -120,6 +120,9 @@ type Report struct { MovedOut []MovedOut `json:"moved-out,omitempty"` // Left is every socket, pipe or device a move-out directory holds and did not move (ADR 0422 §4). Left []LeftInPlace `json:"left,omitempty"` + // Kept is every resource the scope keeps (Scope.Keep) that this apply left as it runs, by id (novox/hq ADR 0594 + // §2, §4): what says whether the bus server still runs what it ran, or was made by this apply. + Kept []string `json:"kept,omitempty"` } // Waiting says whether a container of this apply waits to be replaced because it is busy: what makes the @@ -416,6 +419,11 @@ func applyScoped( if err := refuseRootIdentities(d); err != nil { return report, known, &Error{Resource: "the declaration's identities", Err: err, Done: report} } + // What of the bus server is kept as it runs, asked of the runtime before anything moves (ADR 0594 §2). + keepNow, err := keptServer(ctx, d, scope, run) + if err != nil { + return report, known, &Error{Resource: "the bus server", Err: err, Done: report} + } declared := map[string]bool{} for _, r := range d.Resources { @@ -544,6 +552,17 @@ func applyScoped( removeOrphan := func(orphan store.Applied) error { var action, detail string var err error + if keepNow[orphan.ID] { + // A removed want of the bus module removes the rest of it and never its server (ADR 0594 §2): only a + // warranted want replaces or removes the server. Its record stays, the module's. + detail := "no longer declared; the bus server is left as it runs: only a want carrying the operator's " + + "warrant removes it (novox/hq ADR 0594 §2)" + report.Outcomes = append(report.Outcomes, Outcome{ID: orphan.ID, Type: orphan.Type, Target: orphan.Target, + Action: keptAsItRuns, Detail: detail}) + report.Kept = append(report.Kept, orphan.ID) + log(fmt.Sprintf(" %s %s (%s): %s", keptAsItRuns, orphan.ID, orphan.Target, detail)) + return nil + } if module, held := heldByOthers[orphan.ID]; held && !declared[orphan.ID] { if module == "" { module = "another module" @@ -1071,8 +1090,9 @@ func applyScoped( if from, handed := handedFiles[resource.Identity()]; handed && was.ID == "" { previous.Wrote = from.Wrote } - if o, kept := keepAsItRuns(ctx, scope, resource, run, previous); kept { + if o, kept := keepAsItRuns(keepNow, resource, previous); kept { outcome = o + report.Kept = append(report.Kept, o.ID) } else { outcome, err = applyOne(ctx, sys, resource, run, changed, in, previous, unseal, keepFound) } @@ -3041,8 +3061,8 @@ const keptAsItRuns = "kept-as-it-runs" // unpacked at its path before. The outcome carries what the resource was last recorded reading and holding, so its // record is the module's from here on (taken over from the node's own composition at the bus module's first want) and // a later warranted apply still sees what changed under it. -func keepAsItRuns(ctx context.Context, scope *Scope, r declaration.Resource, run Runner, previous store.Applied) (Outcome, bool) { - if scope == nil || !scope.Keep[r.Identity()] { +func keepAsItRuns(keep map[string]bool, r declaration.Resource, previous store.Applied) (Outcome, bool) { + if !keep[r.Identity()] { return Outcome{}, false } out := begin(r) @@ -3053,9 +3073,7 @@ func keepAsItRuns(ctx context.Context, scope *Scope, r declaration.Resource, run if res.Schedule != "" || res.RunOnce { return Outcome{}, false } - if _, err := containerState(ctx, res.Name, run); err != nil { - return Outcome{}, false - } + // Found running by keptServer before the loop: it is left whatever its declaration says now. out.reads = previous.Reads out.busySince = previous.BusySince out.Detail = why @@ -3079,6 +3097,36 @@ func keepAsItRuns(ctx context.Context, scope *Scope, r declaration.Resource, run return Outcome{}, false } +// keptServer is what of the scope's Keep this apply keeps (novox/hq ADR 0594 §2), decided once before anything is +// applied, as the archives that restart the server come before it: every kept container the declaration names is +// asked of the runtime once. All found: the Keep stands. One the runtime says is not there: nothing is kept, so the +// server is made at what the declaration says now and its archives are applied with it. A runtime that does not +// answer, or answers anything else, is an error, never taken as an absent server: replacing a running bus server +// because one inspect failed is the harm the Keep exists to prevent. +func keptServer(ctx context.Context, d *declaration.Declaration, scope *Scope, run Runner) (map[string]bool, error) { + if scope == nil || len(scope.Keep) == 0 { + return nil, nil + } + for _, r := range d.Resources { + c, ok := r.(*declaration.Container) + if !ok || !scope.Keep[r.Identity()] || c.Schedule != "" || c.RunOnce { + continue + } + cri, err := containerRuntime(ctx, run) + if err != nil { + return nil, fmt.Errorf("%w, so whether the bus server %s runs cannot be told; nothing was applied", err, c.Name) + } + if _, err := run(ctx, cri, "container", "inspect", "--format", "{{.State.Running}}", c.Name); err != nil { + if absent(err) { + return nil, nil // made, at what the declaration says now + } + return nil, fmt.Errorf("the container runtime did not say whether the bus server %s runs (%v); nothing was "+ + "applied, so it is not replaced on an unanswered look", c.Name, err) + } + } + return scope.Keep, nil +} + // heldStill is the action of a container left alone because a maintenance window holds it stopped // (novox/hq issue 224). Its own word rather than "unchanged": a person reading a machine that looks // half-stopped needs to see why, and "unchanged" beside a stopped server says the opposite. diff --git a/internal/apply/module_scope_test.go b/internal/apply/module_scope_test.go index 4ac1aee..9f60560 100644 --- a/internal/apply/module_scope_test.go +++ b/internal/apply/module_scope_test.go @@ -2,9 +2,10 @@ package apply import ( "context" - "fmt" + "errors" "os" "path/filepath" + "slices" "strings" "testing" @@ -225,8 +226,24 @@ func TestAResourceThatMovedToItsModulesCompositionIsLeftByTheNodesOwnApply(t *te // while the bus was folded. Kept (a want without the warrant's mark), the server's container and its program archive // are left as they run, whatever the build changed in them, and become nats's; its files and the tools' account file // are written. Not kept, the same apply replaces the container. - for _, keep := range []bool{true, false} { - t.Run(fmt.Sprintf("the bus server kept %v", keep), func(t *testing.T) { + // Review of 774, round 1: the keep is decided once before anything moves, by the runtime's answer for the server. A + // server not there is made and its archives applied with it, none said kept; a runtime that does not answer fails the + // apply, nothing written and nothing replaced; a removal of the bus removes the rest and leaves the server. + for _, tc := range []struct { + name string + keep bool + gone bool // the server is not on the machine + mute bool // the runtime does not answer the server's inspect + remove bool // the bus module's want is removed + }{ + {name: "kept", keep: true}, + {name: "not kept"}, + {name: "kept, the server not there", keep: true, gone: true}, + {name: "kept, the runtime does not answer", keep: true, mute: true}, + {name: "kept, the bus removed", keep: true, remove: true}, + } { + t.Run("the bus server "+tc.name, func(t *testing.T) { + keep := tc.keep && !tc.gone dir := t.TempDir() conf, account, program := filepath.Join(dir, "nats.conf"), filepath.Join(dir, "tools", "account"), filepath.Join(dir, "program") bus := func(env, confBody, programBody, module string) *declaration.Declaration { @@ -250,29 +267,54 @@ func TestAResourceThatMovedToItsModulesCompositionIsLeftByTheNodesOwnApply(t *te if made == nil { t.Fatalf("the node's own made no server: %v", m.asked) } + if tc.gone { + delete(m.containers, "mesh-broker-nats") + } m.asked = nil + run := m.run + if tc.mute { + run = func(ctx context.Context, name string, args ...string) (string, error) { + if name == "docker" && len(args) > 1 && args[0] == "container" && args[1] == "inspect" { + m.asked = append(m.asked, "docker "+strings.Join(args, " ")) + return "", errors.New("Cannot connect to the Docker daemon at unix:///var/run/docker.sock") + } + return m.run(ctx, name, args...) + } + } next := bus("2", "b", "two", "nats") next.Resources = append(next.Resources, declare(t, `{"id":"nats.tools-account","type":"file","path":"`+account+ `","content":"tools","module":"nats"}`).Resources...) scope := Scope{Module: "nats", Placed: map[string]bool{"nats": true}} - if keep { + if tc.keep { scope.Keep = map[string]bool{"nats.server": true, "nats.program-snapshot": true} } + moved := func() []string { + var out []string + for _, a := range m.asked { + for _, verb := range []string{"docker rm", "docker run", "docker create", "docker stop", "docker restart", "docker rename"} { + if strings.HasPrefix(a, verb+" ") { + out = append(out, a) + } + } + } + return out + } report, after, err := ApplyModule(context.Background(), archHost(t), next, scope, state, - store.OriginDeclared, m.run, nil, nil, nil, nil) + store.OriginDeclared, run, nil, nil, nil, nil) + if tc.mute { + if err == nil || !strings.Contains(err.Error(), "did not say whether the bus server") { + t.Fatalf("an unanswered inspect did not fail the apply: %v", err) + } + if got, _ := os.ReadFile(conf); string(got) != "a" || len(moved()) > 0 { + t.Errorf("an apply that could not tell whether the server runs wrote %q and moved %v", got, moved()) + } + return + } if err != nil { t.Fatal(err) } - var moved []string - for _, a := range m.asked { - for _, verb := range []string{"docker rm", "docker run", "docker create", "docker stop", "docker restart"} { - if strings.HasPrefix(a, verb+" ") { - moved = append(moved, a) - } - } - } - if keep == (len(moved) > 0) { - t.Fatalf("kept %v, and the server was moved by %v", keep, moved) + if keep == (len(moved()) > 0) { + t.Fatalf("kept %v, and the server was moved by %v", keep, moved()) } if got, _ := os.ReadFile(conf); string(got) != "b" { t.Errorf("the server's file holds %q", got) @@ -283,7 +325,8 @@ func TestAResourceThatMovedToItsModulesCompositionIsLeftByTheNodesOwnApply(t *te if got, _ := os.ReadFile(filepath.Join(program, "snapshot")); keep != (string(got) == "one") { t.Errorf("kept %v, and the server's program is %q", keep, got) } - for _, id := range []string{"nats.server", "nats.program-snapshot"} { + var kept []string + for _, id := range []string{"nats.program-snapshot", "nats.server"} { rec, ok := after.Find(id) if !ok || rec.Module != "nats" { t.Errorf("%s is recorded %+v, not as nats's", id, rec) @@ -291,10 +334,35 @@ func TestAResourceThatMovedToItsModulesCompositionIsLeftByTheNodesOwnApply(t *te if o := outcomeOf(report, id); keep != (o.Action == keptAsItRuns) { t.Errorf("kept %v, and %s was %s: %s", keep, id, o.Action, o.Detail) } + if keep { + kept = append(kept, id) + } } - if keep && report.Changed() == false { + if !slices.Equal(report.Kept, kept) && !(len(kept) == 0 && len(report.Kept) == 0) { + t.Errorf("the report says it kept %v, want %v", report.Kept, kept) + } + if tc.keep && report.Changed() == false { t.Error("an apply that wrote the server's files says nothing changed") } + if !tc.remove { + return + } + m.asked = nil + gone, last, err := ApplyModule(context.Background(), archHost(t), &declaration.Declaration{}, scope, after, + store.OriginDeclared, run, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if len(moved()) > 0 || m.containers["mesh-broker-nats"] == nil || !exists(filepath.Join(program, "snapshot")) { + t.Errorf("the bus's removal removed its server: %v", moved()) + } + if exists(conf) || exists(account) { + t.Errorf("the bus's removal left the rest of it: conf %v, account %v", exists(conf), exists(account)) + } + if _, ok := last.Find("nats.server"); !ok || !slices.Equal(gone.Kept, []string{"nats.program-snapshot", "nats.server"}) && + !slices.Equal(gone.Kept, []string{"nats.server", "nats.program-snapshot"}) { + t.Errorf("the removal kept %v, the server recorded %v", gone.Kept, ok) + } }) } } diff --git a/internal/link/have.go b/internal/link/have.go index 46b017c..79c3820 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -59,6 +59,8 @@ type haveState struct { // nodeKeeps is what the node agent keeps as it runs on a want without the warrant's mark (HaveNode.Keeps, novox/hq // ADR 0594 §8), written into its key. nodeKeeps []string + // removedWords is, by module in removed, what its removed have says when it is more than that it was removed. + removedWords map[string]string // written is the build of each module this node-engine wrote a have key for and that the declaration it applied // still holds; removed is each such module the latest declaration no longer holds, with the build its key last // said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted). @@ -508,6 +510,9 @@ func (q *Session) writeRemoved(ctx context.Context, b HaveBucket, node string) b } h := delivery.Have{V: delivery.Version, Build: s.removed[module], Phase: delivery.PhaseRemoved, Health: delivery.HealthUnknown, Detail: "the applied declaration no longer holds the module"} + if words := s.removedWords[module]; words != "" { + h.Detail = words + } raw, err := delivery.EncodeHave(h) if err != nil { q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err)) @@ -699,13 +704,23 @@ func (q *Session) NoteUserList(ctx context.Context, module string, w delivery.Us // NoteRemoved takes that a module's own resources were removed, and writes its have with the phase removed once (issue // 1369): the key is never deleted. -func (q *Session) NoteRemoved(ctx context.Context, module, build string) { +// +// words, when given, are the removed have's detail (the bus server left as it runs, novox/hq ADR 0594 §2). +func (q *Session) NoteRemoved(ctx context.Context, module, build string, words ...string) { q.init() s := &q.haves s.mu.Lock() if s.removed == nil { s.removed = map[string]string{} } + if said := strings.Join(words, ""); said != "" { + if s.removedWords == nil { + s.removedWords = map[string]string{} + } + s.removedWords[module] = said + } else { + delete(s.removedWords, module) + } if f, ok := s.modules[module]; ok && build == "" { build = f.Build } diff --git a/internal/link/have_test.go b/internal/link/have_test.go index f6fc2c7..e681bf9 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -586,6 +586,12 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if got := l.have(t, "n1.nats"); got.FP != fp || got.Server != "build-1792674595610176744" { t.Fatalf("the bus's have: %+v", got) } + // Its removal leaves the server, and its removed have says so (review of 774, round 1). + q.NoteRemoved(context.Background(), "nats", "", "removed, but for nats.server: the bus server stays as it runs") + if got := l.have(t, "n1.nats"); got.Phase != delivery.PhaseRemoved || !strings.Contains(got.Detail, "the bus server stays") { + t.Fatalf("the bus's removed have: %+v", got) + } + q.NoteModule(context.Background(), ModuleFact{Module: "nats", FP: fp, Composed: dig, Server: "build-1792674595610176744"}) q.NoteComposes(context.Background(), delivery.ComposesOwn, "it composes its own", delivery.KeepsBusServer) if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || !slices.Equal(n.Keeps, []string{delivery.KeepsBusServer}) || n.Composes != delivery.ComposesOwn { -- 2.54.0