Serve the node-engine verb composition: a module's composition beside what runs, its judgement and its diff (issue 774)
Four providers waited for a yes on 2026-10-31 and nobody could read what waited: the have said only "where its data is kept". The read verb composition answers it from the engine: the latest revision of the want's build or the one asked by digest, every revision the bucket keeps, what runs and where that is known from, the want and the have, the move's judgement and whether it waits for a yes, and with diff the resources that differ, a received file naming its consumers. revisionNamed now reads the same revisions.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
+4
-32
@@ -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
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
@@ -1445,6 +1458,24 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
||||
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")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user