diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 4c11ece..76727ef 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -275,3 +275,20 @@ func reconcileActs(ctx context.Context, h *link.HandActs) { } } } + +// compositionReadFor bounds one read of the verb composition: the revisions of one key from the bucket. +const compositionReadFor = 20 * time.Second + +// compositionFrom answers the verb composition from the node agent that runs now (issue 774); refused in words while +// none does. +func compositionFrom(current *agentNow) func(module, digest string, diff bool) (any, error) { + return func(module, digest string, diff bool) (any, error) { + ag := current.get() + if ag == nil { + return nil, errors.New("this engine's node agent has not started, so it holds no composition yet") + } + ctx, cancel := context.WithTimeout(context.Background(), compositionReadFor) + defer cancel() + return ag.Composition(ctx, module, digest, diff) + } +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index c689a92..278fa82 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1141,6 +1141,7 @@ func runLink(ctx context.Context, opts options) error { queue.Acts = engineActs(opts, mine.Node, queue, setuidSearch) go reconcileActs(ctx, queue.Acts) current := &agentNow{} + queue.Acts.Composition = compositionFrom(current) setRunsHere(runsFromAgent(current)) factsOfNow := func(ctx context.Context, ag *agent.Agent) { sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag) diff --git a/go.mod b/go.mod index 87e742d..f89f63a 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.20261030201806-eff81f1220e2 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030233444-8ed3b872e0ba 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 156a748..a698341 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2 h1:RF73nVI1dTrrmlFosVDTMJmpUQ+rEeB7++BHH/RvZFg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030201806-eff81f1220e2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030233444-8ed3b872e0ba h1:vRjCeYiVsX8Q8FrT8IHa3z+zpzh2WVaDCnu7l/XYATI= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030233444-8ed3b872e0ba/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 7b6494c..45bf7d8 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -742,49 +742,21 @@ func (a *Agent) compositionFor(module, fp string) (compNow, bool) { // revisionNamed is the revision of a module's composition key whose digest is the one given, signed by a trusted key, // from the revisions the bucket keeps; false when none is. func (a *Agent) revisionNamed(ctx context.Context, module, fp, digest string) (compNow, bool) { - if a.env.Revisions == nil { - return compNow{}, false - } - var key string - var err error - if module == NodeModule { - key, err = delivery.ComposedNodeKey(a.env.Node) - } else { - key, err = delivery.ComposedKey(a.env.Node, module, fp) - } + key, err := a.compositionKey(module, fp) if err != nil { return compNow{}, false } - revs, err := a.env.Revisions(ctx, key) + revs, err := a.revisionsOf(ctx, key) if err != nil { a.mu.Lock() a.sayOnce("history:"+key, fmt.Sprintf("the revisions of %s cannot be read: %v", key, err)) a.mu.Unlock() return compNow{}, false } - var trusted []ed25519.PublicKey - if a.env.Trusted != nil { - trusted = a.env.Trusted() - } for i := len(revs) - 1; i >= 0; i-- { - u := revs[i] - if u.Deleted || u.Key != key { - continue + if revs[i].Err == nil && revs[i].Dig == digest { + return revs[i], true } - c, err := delivery.DecodeComposed(u.Value) - if err != nil { - continue - } - 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 - } - return compNow{Raw: u.Value, C: c, Dig: dig, Rev: u.Revision}, true } return compNow{}, false } @@ -1577,6 +1549,16 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp r, ran := a.running[module] if ran && r.Composed == comp.Dig && r.Phase != delivery.PhaseRemoved { a.mu.Unlock() + // Composed back to what runs: a have still waiting for a yes for a move since gone says running again, or it waits + // for a yes nobody needs (issue 774, 2026-10-31: four providers stayed waiting-yes once their consumers composed). + if h, ok := a.env.Haves.Read(ctx, module); ok && h.Phase == string(delivery.PhaseWaitingYes) { + // Its sealed values only when they are of what runs; else writeHave reads them (sealedOf). + var sealed delivery.Seals + if h.Composed == r.Composed { + sealed = h.Sealed + } + a.writeHave(ctx, Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(delivery.PhaseRunning), Sealed: sealed}) + } return false } from, known := a.runningComposition(module) diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 9ec0741..4423a9f 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -1294,6 +1294,19 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( } return catalogue.PolicyRoll } + // The bucket's revisions of each key, as the verb composition reads them (issue 774). + var historyMu sync.Mutex + history := map[string][]Update{} + b.env.Revisions = func(_ context.Context, key string) ([]Update, error) { + historyMu.Lock() + defer historyMu.Unlock() + return append([]Update(nil), history[key]...), nil + } + keep := func(key string, raw []byte) { + historyMu.Lock() + defer historyMu.Unlock() + history[key] = append(history[key], Update{Key: key, Value: raw, Revision: uint64(len(history[key]) + 1)}) + } stop := b.run() defer stop() unsigned := func(module, fp string, rs ...json.RawMessage) (string, []byte, string) { @@ -1432,10 +1445,45 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( raw, _ := json.Marshal(map[string]any{"id": module + ".server", "type": "container", "module": module, "name": module, "image": image}) return raw } - dk, draw, _ := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64))) + dk, draw, ddig := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64))) b.comps.put(dk, draw) want("db", fp, "") eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 4 }) + // A revision at the same build that recreates the container waits for a yes; composed back to what runs, the have + // says running again, not waiting-yes for a move nobody needs (issue 774, 2026-10-31: ace.postgres, novox.postgres, + // novox.keycloak and novox.mongodb stayed waiting-yes after their providers' received files went back). + _, drawMoved, ddigMoved := unsigned("db", fp, container("db@sha256:"+strings.Repeat("7", 64))) + b.comps.put(dk, drawMoved) + eventually(t, "db's revision did not wait for a yes", func() bool { + h := b.haves.last("db") + return h.Phase == string(delivery.PhaseWaitingYes) && h.Waits == ddigMoved + }) + // The verb composition says why it waits: the revision that waits, what runs, the judgement, and with diff the + // container it recreates; the revision that runs, asked by its digest, is no move (issue 774). + keep(dk, draw) + keep(dk, drawMoved) + view, err := b.ag.Composition(context.Background(), "db", "", true) + if err != nil || view.Have == nil || view.Have.Phase != string(delivery.PhaseWaitingYes) || view.Composition.Digest != ddigMoved || + view.Runs == nil || view.Runs.Composed != ddig || view.Judged.Judgement != string(delivery.JudgedRecreates) || + !view.Judged.WaitsForYes || len(view.Diff) != 1 || view.Diff[0].ID != "db.server" || view.Diff[0].Change != "changed" || + len(view.Revisions) != 2 { + t.Errorf("the composition of db waiting for a yes reads %+v (%v)", view, err) + } + if view, err := b.ag.Composition(context.Background(), "db", ddig, false); err != nil || view.Composition.Digest != ddig || + view.Judged.Judgement != string(delivery.JudgedUnchanged) || view.Judged.WaitsForYes || view.Diff != nil { + t.Errorf("db's running revision, asked by its digest, reads %+v (%v)", view, err) + } + if _, err := b.ag.Composition(context.Background(), "db", "sha256:"+strings.Repeat("0", 64), false); err == nil { + t.Error("a digest no revision has was answered") + } + b.comps.put(dk, draw) + eventually(t, "db's have stays waiting-yes after its composition went back to what runs", func() bool { + h := b.haves.last("db") + return h.Phase == string(delivery.PhaseRunning) && h.Composed == ddig && h.Waits == "" + }) + if got := len(b.appliedList()); got != 4 { + t.Fatalf("a composition equal to what runs was applied again: %d applies", got) + } fpB := fpOf("b") dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64))) // rec is a record module this agent has run nothing of since its start (the switch and every delivery of the diff --git a/internal/agent/composition.go b/internal/agent/composition.go new file mode 100644 index 0000000..71ea0e7 --- /dev/null +++ b/internal/agent/composition.go @@ -0,0 +1,230 @@ +package agent + +import ( + "context" + "crypto/ed25519" + "errors" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The node-engine seat's read verb composition (issue 774): one module's composition beside what runs of it, so why a +// record module waits for a yes, or what its next apply would change, is read from the engine and not guessed from its +// have's words. It changes nothing. + +// CompositionView is what the verb composition answers. +type CompositionView struct { + Node string `json:"node"` + Module string `json:"module"` + // Want is the module's want as the agent holds it; nil when it holds none. + Want *WantView `json:"want,omitempty"` + // Have is what the module's have key says now; nil when it says nothing. + Have *HaveView `json:"have,omitempty"` + // Composition is the revision read: the latest of the want's build, or the one whose digest was asked. + Composition RevisionView `json:"composition"` + // Revisions are the revisions of its key the bucket keeps, oldest first, each with its digest. + Revisions []RevisionView `json:"revisions,omitempty"` + // Runs is what runs of the module, and where that is known from; nil when nothing of it runs. + Runs *RunsView `json:"runs,omitempty"` + // Judged is the move from what runs to the composition read, as the agent judges it before an apply. + Judged JudgedView `json:"judged"` + // Diff is every resource that differs from what runs, when asked. + Diff []delivery.ResourceChange `json:"diff,omitempty"` +} + +// WantView is a want as the verb shows it. +type WantView struct { + FP string `json:"fp"` + Composed string `json:"composed,omitempty"` + Yes string `json:"yes,omitempty"` + Removed bool `json:"removed,omitempty"` +} + +// HaveView is a have as the verb shows it. +type HaveView struct { + FP string `json:"fp,omitempty"` + Composed string `json:"composed,omitempty"` + Phase string `json:"phase"` + Detail string `json:"detail,omitempty"` + Judgement string `json:"judgement,omitempty"` + Waits string `json:"waits,omitempty"` +} + +// RevisionView is one revision of a composition key. +type RevisionView struct { + Key string `json:"key"` + Revision uint64 `json:"revision,omitempty"` + Digest string `json:"digest,omitempty"` + Resources int `json:"resources,omitempty"` + Latest bool `json:"latest,omitempty"` + // Refused is why the revision cannot be read or trusted. + Refused string `json:"refused,omitempty"` +} + +// RunsView is what runs of a module. +type RunsView struct { + FP string `json:"fp,omitempty"` + Composed string `json:"composed,omitempty"` + Phase string `json:"phase,omitempty"` + // From is applied (what this engine applied), have (what its have named at the start) or kept (the declaration the + // node kept before the switch). + From string `json:"from"` + Resources int `json:"resources"` +} + +// JudgedView is a move's judgement. +type JudgedView struct { + Judgement string `json:"judgement"` + Words string `json:"words"` + // WaitsForYes says the agent would wait for a person's yes before applying it: a module of policy record, a move that + // recreates a container, moves data or cannot be judged, and no yes in its want for this digest. + WaitsForYes bool `json:"waits_for_yes"` +} + +// Composition reads one module's composition beside what runs of it: the want's build (else the have's), its latest +// revision or the one whose digest is given, and the move's judgement; with diff, the resources that differ. +func (a *Agent) Composition(ctx context.Context, module, digest string, diff bool) (CompositionView, error) { + if module == "" { + return CompositionView{}, errors.New("name the module, or node for the node's own composition") + } + view := CompositionView{Node: a.env.Node, Module: module} + var have *Have + if a.env.Haves != nil { + if h, ok := a.env.Haves.Read(ctx, module); ok { + have = &h + phase := h.Phase + if phase == "" { + phase = string(delivery.PhaseRunning) + } + view.Have = &HaveView{FP: h.FP, Composed: h.Composed, Phase: phase, Detail: h.Detail, Judgement: h.Judgement, Waits: h.Waits} + } + } + a.mu.Lock() + w := a.wants[module] + fp := "" + if w.Present { + fp = w.Want.FP + view.Want = &WantView{FP: w.Want.FP, Composed: w.Want.Composed, Yes: w.Want.Yes, Removed: w.Want.Removed} + } else if have != nil { + fp = have.FP + } + latest, haveLatest := a.compositionFor(module, fp) + r, ran := a.running[module] + from, known := a.runningComposition(module) + source := "applied" + if _, ok := a.applied[module]; !ok { + if _, ok := a.had[module]; ok { + source = "have" + } else { + source = "kept" + } + } + a.mu.Unlock() + if module != NodeModule && fp == "" { + return view, fmt.Errorf("%s has no want and no have on %s, so no build of it is composed here", module, a.env.Node) + } + key, err := a.compositionKey(module, fp) + if err != nil { + return view, err + } + revs, err := a.revisionsOf(ctx, key) + if err != nil { + return view, fmt.Errorf("the revisions of %s cannot be read: %w", key, err) + } + var read *compNow + for i := range revs { + c := revs[i] + rv := RevisionView{Key: key, Revision: c.Rev, Digest: c.Dig, Resources: len(c.C.Resources), + Latest: haveLatest && c.Rev == latest.Rev} + if c.Err != nil { + rv.Refused = c.Err.Error() + } + view.Revisions = append(view.Revisions, rv) + if digest != "" && c.Err == nil && c.Dig == digest { + read = &revs[i] + } + } + switch { + case digest != "" && read == nil && haveLatest && latest.Err == nil && latest.Dig == digest: + read = &latest + case digest != "" && read == nil: + return view, fmt.Errorf("no revision of %s the bucket keeps has the digest %s; revisions lists those it keeps", key, digest) + case read == nil && !haveLatest: + return view, fmt.Errorf("the bucket holds no composition of %s at %s", module, fp) + case read == nil: + read = &latest + } + view.Composition = RevisionView{Key: key, Revision: read.Rev, Digest: read.Dig, Resources: len(read.C.Resources), + Latest: haveLatest && read.Rev == latest.Rev} + if read.Err != nil { + view.Composition.Refused = read.Err.Error() + return view, nil + } + var running *delivery.Composed + if known { + running = &from + rv := &RunsView{From: source, Resources: len(from.Resources), FP: from.FP} + if ran { + rv.FP, rv.Composed, rv.Phase = r.FP, r.Composed, string(r.Phase) + } + view.Runs = rv + } + j := delivery.JudgeMove(running, read.C) + view.Judged = JudgedView{Judgement: string(j.Judgement), Words: j.Words} + if a.env.SelfComposed && a.env.Policy != nil && module != NodeModule && (!w.Present || w.Want.Yes != read.Dig) { + switch j.Judgement { + case delivery.JudgedRecreates, delivery.JudgedMovesData, delivery.JudgedCannot: + view.Judged.WaitsForYes = a.env.Policy(ctx, module) == catalogue.PolicyRecord + } + } + if diff { + view.Diff = delivery.ChangedResources(running, read.C) + } + return view, nil +} + +// compositionKey is a module's composition key at a build, or the node's own. +func (a *Agent) compositionKey(module, fp string) (string, error) { + if module == NodeModule { + return delivery.ComposedNodeKey(a.env.Node) + } + return delivery.ComposedKey(a.env.Node, module, fp) +} + +// revisionsOf is every revision of a composition key the bucket keeps, oldest first, each with its digest as this node +// trusts it (its own unsigned with SelfComposed, else signed by a trusted key), or why it is refused. Nil Revisions +// reads none. +func (a *Agent) revisionsOf(ctx context.Context, key string) ([]compNow, error) { + if a.env.Revisions == nil { + return nil, nil + } + revs, err := a.env.Revisions(ctx, key) + if err != nil { + return nil, err + } + var trusted []ed25519.PublicKey + if a.env.Trusted != nil { + trusted = a.env.Trusted() + } + var out []compNow + for _, u := range revs { + if u.Deleted || u.Key != key { + continue + } + c, err := delivery.DecodeComposed(u.Value) + if err != nil { + out = append(out, compNow{Raw: u.Value, Rev: u.Revision, Err: err}) + continue + } + 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) + } + out = append(out, compNow{Raw: u.Value, C: c, Dig: dig, Rev: u.Revision, Err: err}) + } + return out, nil +} diff --git a/internal/link/handacts.go b/internal/link/handacts.go index a411516..f77f0f7 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -38,6 +38,8 @@ const ( ActAccess = "access-act" ActJudgeDirectories = "judge-directories" ActSetuidSearch = "setuid-search" + // ActComposition reads a module's composition beside what runs of it; read-only, no warrant (issue 774). + ActComposition = "composition" ) // The states of an act. @@ -127,6 +129,9 @@ type HandActs struct { Searching func() bool // RootFree is this machine's root-free verdict now. RootFree func() RootVerdict + // Composition reads one module's composition beside what runs of it, read-only (the verb composition, issue 774): + // the latest of its want's build or the one whose digest is given, and with diff the resources that differ. + Composition func(module, digest string, diff bool) (any, error) // Say says one of the seat's events, asked, done or refused; nil says nothing. Say func(event string, body any) Now func() time.Time @@ -176,6 +181,15 @@ func (h *HandActs) Call(c EngineCall) (any, error) { return nil, errors.New("this engine judges no machine root-free") } return map[string]any{"machines": []RootVerdict{h.RootFree()}}, nil + case ActComposition: + if h.Composition == nil { + return nil, errors.New("this engine reads no composition yet: its node agent has not started") + } + diff := arg("diff") == "true" + if b, ok := c.Args["diff"].(bool); ok { + diff = b + } + return h.Composition(arg("module"), arg("digest"), diff) case "acts": if id := arg("id"); id != "" { return h.Act(id) diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index 54bceb6..a4b8688 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -236,15 +236,25 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { t.Run("a call of the node seat reaches its verb with its proven caller", func(t *testing.T) { h := &HandActs{Node: "n1", State: &MemState{}, RootFree: func() RootVerdict { return RootVerdict{Machine: "n1", Free: true, Why: "judged"} }, - JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }} + JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }, + Composition: func(module, digest string, diff bool) (any, error) { + return map[string]any{"module": module, "digest": digest, "diff": diff}, nil + }} for _, row := range []struct{ verb, subject, want string }{ {"judge-directories", "mesh.seat.node-engine.call.judge-directories.n1.shanks~node-tools", `"said":"judged for shanks.node-tools"`}, {"judge-directories", EngineToolSubject("judge-directories", "n1"), `"said":"judged for unproven"`}, {"judge-directories", "mesh.seat.node-engine.call.judge-directories.n2.shanks~node-tools", `"said":"judged for unproven"`}, {"root-free", "mesh.seat.node-engine.call.root-free.n1.ace~messenger", `"free":true`}, + // The read verb composition (issue 774) takes its module, digest and diff, as a word or a boolean. + {"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","digest":"sha256:a","diff":true}`, `"diff":true,"digest":"sha256:a","module":"postgres"`}, + {"composition", EngineToolSubject("composition", "n1") + " " + `{"module":"postgres","diff":"true"}`, `"diff":true,"digest":"","module":"postgres"`}, {"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`}, } { - if got := string(AnswerEngineCall(h.Call, row.verb, "n1", row.subject, []byte(`{}`))); !strings.Contains(got, row.want) || + subject, body, ok := strings.Cut(row.subject, " ") + if !ok { + body = `{}` + } + if got := string(AnswerEngineCall(h.Call, row.verb, "n1", subject, []byte(body))); !strings.Contains(got, row.want) || !strings.Contains(got, `"node":"n1"`) { t.Errorf("%s on %s: %s", row.verb, row.subject, got) }