Review of L2, round 2: a refusal at the first compose after a start keeps what the module's have said runs (1); each have of a self-composing node carries the values its composition holds sealed, and the composer reuses them after a restart, so a rotated secret moves no composition at a restart (2); pin mesh-sdk dd9d9d5 (issue 1126, ADR 0547 §3)
mesh/delivery closed: a newer head of the same pull request
mesh/check checking 4476e5808047 on ff9386dcccc2 | novox/mesh-host#149 in group feat/1126-nodes-compose

This commit is contained in:
2026-10-26 20:23:16 +01:00
parent 58de6aa10a
commit 4476e58080
9 changed files with 255 additions and 72 deletions
+122 -35
View File
@@ -57,6 +57,9 @@ type Have struct {
// waits for a person's yes (ADR 0547 §3).
Judgement string
Waits string
// Sealed is, on a node composing its own, each value the composition it names carries sealed to the node's key
// (delivery.Have.Sealed): kept in the have so the composer reuses them after a restart (ADR 0547 §3).
Sealed []string
}
// Haves is the node's have bucket as the agent uses it.
@@ -156,8 +159,8 @@ type Agent struct {
// prior is a module the node ran before this agent started, so it has something to back up.
prior map[string]bool
said map[string]string
// notComposed is the modules whose have says the node could not compose them, and why (NotComposed).
notComposed map[string]string
// notComposed is the modules whose have says the node could not compose them: why, and the have before (NotComposed).
notComposed map[string]notComposedMark
retry chan []string
// tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and
// transient asks for every transiently failed module at once. stopped is closed when Run returns.
@@ -204,7 +207,7 @@ func New(env Env) *Agent {
env.Say = func(string) {}
}
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{},
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]string{},
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), tries: map[string]int{}, again: make(chan againAsk, 64),
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
}
@@ -305,44 +308,128 @@ func (a *Agent) RetryTransient() {
}
}
// notComposedMark is why a module's have says the node could not compose it, and the have it said before.
type notComposedMark struct {
why string
prior Have
}
// NotComposed says the node composing its own could not compose its modules (ADR 0547 §2: a stand-in refused, an
// input that cannot be read): each module with a want gets its have failed, naming why, while it keeps running what it
// ran, so the failure is seen where every have is read and not only in the journal. Called with "" once the node
// composes again: each module so marked has its have back as what it runs, unless an apply wrote it since.
// ran, so the failure is seen where every have is read and not only in the journal. The have keeps what runs: the
// fingerprint, the digest and the sealed values the agent runs, or, before it ran anything since its start, those its
// last have said (a refusal at the first compose after a start). Called with "" once the node composes again: each
// module so marked has its have back as it was, unless an apply wrote it since.
func (a *Agent) NotComposed(ctx context.Context, why string) {
a.mu.Lock()
var write []Have
var modules []string
if why != "" {
for module, w := range a.wants {
if !w.Present || w.Want.Removed || a.notComposed[module] == why {
continue
if w.Present && !w.Want.Removed && module != NodeModule && a.notComposed[module].why != why {
modules = append(modules, module)
}
a.notComposed[module] = why
r := a.running[module]
write = append(write, Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(delivery.PhaseFailed),
Detail: "the node cannot compose it, and keeps running what it ran: " + why})
}
} else {
for module := range a.notComposed {
delete(a.notComposed, module)
r, ok := a.running[module]
if !ok {
continue
}
phase := r.Phase
if phase == "" {
phase = delivery.PhaseRunning
}
write = append(write, Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(phase)})
modules = append(modules, module)
}
}
a.mu.Unlock()
sort.Slice(write, func(i, j int) bool { return write[i].Module < write[j].Module })
for _, h := range write {
a.env.Haves.Write(ctx, h)
sort.Strings(modules)
for _, module := range modules {
a.mu.Lock()
mark, marked := a.notComposed[module]
r, ran := a.running[module]
a.mu.Unlock()
if why == "" {
a.mu.Lock()
delete(a.notComposed, module)
a.mu.Unlock()
back := mark.prior
if ran {
back = Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(delivery.PhaseRunning)}
if r.Phase != "" {
back.Phase = string(r.Phase)
}
}
if back.Composed == "" {
continue // nothing said what ran: nothing to say back
}
a.writeHave(ctx, back)
continue
}
if !marked {
if ran {
mark.prior = Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(r.Phase)}
if mark.prior.Phase == "" {
mark.prior.Phase = string(delivery.PhaseRunning)
}
} else if h, ok := a.env.Haves.Read(ctx, module); ok {
mark.prior = h // what its last have said runs: this agent has run nothing of it since its start
}
}
mark.why = why
a.mu.Lock()
a.notComposed[module] = mark
a.mu.Unlock()
a.writeHave(ctx, Have{Module: module, FP: mark.prior.FP, Composed: mark.prior.Composed, Sealed: mark.prior.Sealed,
Phase: string(delivery.PhaseFailed), Detail: "the node cannot compose it, and keeps running what it ran: " + why})
}
}
// writeHave writes a module's have; on a node composing its own, with the sealed values of the composition it names
// when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else
// those the module's last have said at that digest.
func (a *Agent) writeHave(ctx context.Context, h Have) {
if a.env.SelfComposed && h.Sealed == nil && h.Module != NodeModule && h.Composed != "" {
h.Sealed = a.sealedOf(ctx, h.Module, h.FP, h.Composed)
}
a.env.Haves.Write(ctx, h)
}
func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) []string {
a.mu.Lock()
if c, ok := a.applied[module]; ok && c.Err == nil && c.Dig == composed {
a.mu.Unlock()
return SealedIn(c.C)
}
if c, ok := a.compositionFor(module, fp); ok && c.Err == nil && c.Dig == composed {
a.mu.Unlock()
return SealedIn(c.C)
}
a.mu.Unlock()
if h, ok := a.env.Haves.Read(ctx, module); ok && h.Composed == composed {
return h.Sealed
}
return nil
}
// SealedIn is each value a composition carries sealed to its node's key: a file's `sealed` and a vault box
// (catalogue.VaultSealedField), as their JSON text, in the order of its resources.
func SealedIn(c delivery.Composed) []string {
var out []string
for _, raw := range c.Resources {
var r map[string]json.RawMessage
if json.Unmarshal(raw, &r) != nil {
continue
}
for _, field := range []string{"sealed", catalogue.VaultSealedField} {
f, ok := r[field]
if !ok {
continue
}
var text string
if json.Unmarshal(f, &text) != nil {
text = string(f) // a vault box is carried as its JSON
}
if text != "" {
out = append(out, text)
}
}
}
return out
}
// Running is the compositions the node runs now: each module's that the agent applied or found already running, and
// the node's own. What the node judges its health by and says of itself is built from them.
func (a *Agent) Running() []delivery.Composed {
@@ -805,8 +892,8 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
a.mu.Unlock()
// Said again, so the session holds the module and its key carries the health this engine judges,
// not the one an older engine left there (xorg's unknown on g14 and shanks, 2026-10-25).
a.env.Haves.Write(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed,
Phase: string(delivery.PhaseRunning), Detail: h.Detail})
a.writeHave(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed,
Phase: string(delivery.PhaseRunning), Detail: h.Detail, Sealed: h.Sealed})
return true
}
return false
@@ -828,7 +915,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
a.mu.Lock()
a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning}
a.mu.Unlock()
a.env.Haves.Write(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
a.writeHave(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
return true
}
@@ -842,7 +929,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
// A data module is backed up before a new composition of it moves (ADR 0429 §14), unless the want says to go on
// without one (ADR 0539 §16).
if comp.C.Data && ranBefore && a.env.Backup != nil && w.Want.Backup != delivery.BackupSkip {
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseBackingUp)})
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseBackingUp)})
if got := a.env.Backup(ctx, module); !got.Done {
a.env.Say(fmt.Sprintf("%s is held: its backup before %s failed: %s", module, fp, got.Words))
a.mu.Lock()
@@ -850,12 +937,12 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
prev := a.running[module]
a.mu.Unlock()
// It keeps running what it has: the have says what it holds and why it waits.
a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
a.writeHave(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
Detail: got.Words})
return
}
}
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
res := a.env.Apply(ctx, comp.C, others)
a.mu.Lock()
what := fmt.Sprintf("%s at %s", module, fp)
@@ -885,7 +972,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
phase = delivery.PhaseFailed // still retried, and said as failed: no longer a moment
}
a.env.Say(fmt.Sprintf("applying %s failed because the bus did not answer (try %d), retried %s: %s", what, try, when, res.Detail))
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(phase),
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(phase),
Detail: fmt.Sprintf("retrying %s (try %d): %s", when, try, res.Detail)})
return
}
@@ -894,7 +981,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
a.mu.Unlock()
a.env.Say(fmt.Sprintf("applying %s failed: %s", what, res.Detail))
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
return
}
delete(a.attempt, module)
@@ -905,7 +992,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
if a.env.Settled != nil {
a.env.Settled(module)
}
a.env.Haves.Write(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})
}
func (a *Agent) remove(ctx context.Context, module string, w wantNow) {
@@ -1031,7 +1118,7 @@ func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp
a.sayOnce("yes:"+module+":"+comp.Dig, fmt.Sprintf("%s at %s waits for a person's yes: %s", module, w.Want.FP, j.Words))
prev := a.running[module]
a.mu.Unlock()
a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseWaitingYes),
a.writeHave(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseWaitingYes),
Detail: j.Words, Judgement: string(j.Judgement), Waits: comp.Dig})
return true
}
+29 -7
View File
@@ -992,7 +992,8 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
}
fp := fpOf("a")
ck, craw, dig := unsigned("web", fp, file("web", "x", "1"))
pw, _ := json.Marshal(map[string]any{"id": "web.pw", "type": "file", "path": "/tmp/r6/web/pw", "sealed": "1", "module": "web"})
ck, craw, dig := unsigned("web", fp, file("web", "x", "1"), pw)
b.comps.put(ck, craw)
want("web", fp, "")
eventually(t, "the node's own composition was not applied", func() bool { return len(b.appliedList()) == 1 })
@@ -1000,15 +1001,36 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
// The same composition handed again is no move.
b.comps.put(ck, craw)
time.Sleep(50 * time.Millisecond)
// The node cannot compose (a stand-in refused): web's have says so, failed, naming why, while it runs what it ran;
// composed again, the have is what it runs.
// The node cannot compose (a stand-in refused): each wanted module's have says so, failed, naming why, and still
// names what runs; composed again, the have is what it runs. web was applied by this agent; idle was not since the
// agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said.
idleFP, idleDig := fpOf("c"), "sha256:"+strings.Repeat("9", 64)
b.haves.Write(context.Background(), Have{Module: "idle", FP: idleFP, Composed: idleDig, Phase: string(delivery.PhaseRunning),
Sealed: []string{`{"box":"idle"}`}})
want("idle", idleFP, "")
eventually(t, "idle's want was not taken", func() bool {
b.ag.mu.Lock()
defer b.ag.mu.Unlock()
return b.ag.wants["idle"].Present
})
if h := b.haves.last("web"); len(h.Sealed) != 1 || h.Sealed[0] != "1" {
t.Errorf("web's running have does not carry its sealed value: %+v", h)
}
b.ag.NotComposed(context.Background(), "a stand-in for web's password")
if h := b.haves.last("web"); h.Phase != string(delivery.PhaseFailed) || h.Composed != dig || !strings.Contains(h.Detail, "web's password") {
t.Errorf("web's have while the node cannot compose: %+v", h)
for _, row := range []struct {
module, fp, composed, sealed string
}{{"web", fp, dig, "1"}, {"idle", idleFP, idleDig, `{"box":"idle"}`}} {
h := b.haves.last(row.module)
if h.Phase != string(delivery.PhaseFailed) || h.FP != row.fp || h.Composed != row.composed ||
len(h.Sealed) != 1 || h.Sealed[0] != row.sealed || !strings.Contains(h.Detail, "web's password") {
t.Errorf("%s's have while the node cannot compose: %+v", row.module, h)
}
}
b.ag.NotComposed(context.Background(), "")
if h := b.haves.last("web"); h.Phase != string(delivery.PhaseRunning) || h.Composed != dig {
t.Errorf("web's have once the node composes again: %+v", h)
for _, row := range []struct{ module, composed string }{{"web", dig}, {"idle", idleDig}} {
if h := b.haves.last(row.module); h.Phase != string(delivery.PhaseRunning) || h.Composed != row.composed || len(h.Sealed) != 1 {
t.Errorf("%s's have once the node composes again: %+v", row.module, h)
}
}
// A revision at the same build (a config value changed) is applied.
_, craw2, dig2 := unsigned("web", fp, file("web", "x", "2"))
+1
View File
@@ -333,6 +333,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
case f.Failed:
h.Phase = delivery.PhaseFailed
}
h.Sealed = f.Sealed
h.Health = s.healthOf(module, f)
if w, ok := s.userList[module]; ok {
h.UserList = &w
+2
View File
@@ -746,4 +746,6 @@ type ModuleFact struct {
// Judgement and Waits are, in phase waiting-yes, the judgement of the move and the composition waiting (ADR 0547 §3).
Judgement string
Waits string
// Sealed is the values the composition it names carries sealed to this node's key (delivery.Have.Sealed).
Sealed []string
}
+32 -24
View File
@@ -55,6 +55,9 @@ type Env struct {
// keeps the vault's box.
Running func() []delivery.Composed
Unseal func(sealed string) ([]byte, error)
// Held is the sealed values a module's have says it runs (delivery.Have.Sealed), kept by the agent on the bus: what
// runs after a restart, before the agent applied anything, and after the switch once the kept declaration is old.
Held func(module string) []string
}
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
@@ -431,39 +434,44 @@ func (c *Composer) refused(why string) {
}
}
// runningBox is the box the node already runs that opens to the same value as box, or box: read before c.mu is taken,
// since the agent's lock is taken to read what runs.
// runningBox is the box the node already runs that opens to the same value as box, or box: what the agent runs, and
// what each wanted module's have says it runs sealed (Held). Read before c.mu is taken, since the agent's lock is taken
// to read what runs.
func (c *Composer) runningBox(box string) string {
if c.env.Running == nil || c.env.Unseal == nil {
if c.env.Unseal == nil || (c.env.Running == nil && c.env.Held == nil) {
return box
}
value, err := c.env.Unseal(box)
if err != nil {
return box
}
for _, comp := range c.env.Running() {
for _, raw := range comp.Resources {
var r map[string]json.RawMessage
if json.Unmarshal(raw, &r) != nil {
continue
}
for _, field := range []string{"sealed", catalogue.VaultSealedField} {
f, ok := r[field]
if !ok {
continue
}
var held string
if json.Unmarshal(f, &held) != nil {
held = string(f) // a vault box is carried as its JSON
}
if held == "" || held == box {
continue
}
if v, err := c.env.Unseal(held); err == nil && bytes.Equal(v, value) {
return held
}
var held []string
if c.env.Running != nil {
for _, comp := range c.env.Running() {
held = append(held, agent.SealedIn(comp)...)
}
}
if c.env.Held != nil {
c.mu.Lock()
modules := make([]string, 0, len(c.wants))
for m := range c.wants {
if m != agent.NodeModule {
modules = append(modules, m)
}
}
c.mu.Unlock()
sort.Strings(modules)
for _, m := range modules {
held = append(held, c.env.Held(m)...)
}
}
for _, h := range held {
if h == box {
continue
}
if v, err := c.env.Unseal(h); err == nil && bytes.Equal(v, value) {
return h
}
}
return box
}
+52 -2
View File
@@ -90,10 +90,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
env := b.env()
// The vault's own: keeper's password sealed to n1's key, sealed again (other bytes) at every ask.
var asks, generation int
sealedAt := map[string]int{} // each box handed, by the generation it opens to
env.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
b.mu.Lock()
defer b.mu.Unlock()
asks++
sealedAt[fmt.Sprintf(`{"ask":%d}`, asks)] = generation + 1
return json.RawMessage(fmt.Sprintf(`{"node":"n1","key":"sk","own":[{"module":"keeper","name":"password","origin":"made",`+
`"generation":%d,"sealed":{"ask":%d}}]}`, generation+1, asks)), nil
}
@@ -207,6 +209,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
if !strings.Contains(handed, `{\"ask\":`) {
t.Fatalf("keeper's composition does not carry the sealed value: %s", handed)
}
before := handed
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2}`))
quiet(t, out)
b.mu.Lock()
@@ -214,8 +217,55 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
b.mu.Unlock()
// The vault says the secret changed: the node composes again on that word, with no other input moving.
secretChanged <- []byte(`{"name":"own/n1/keeper/password","generation":2}`)
if u := next(t, out); u.Key != keeperKey {
t.Fatalf("a new generation handed %s, want keeper again", u.Key)
rotated := next(t, out)
if rotated.Key != keeperKey {
t.Fatalf("a new generation handed %s, want keeper again", rotated.Key)
}
// Review of L2 (round 2): rotated, then the node-engine restarts. The agent has applied nothing yet, so the box the
// node runs comes from keeper's have (Held); the vault seals the value again, other bytes, and the box the have holds
// is kept, since it opens to the same value: the composition does not move. The box from before the rotation opens
// to the old value, so the vault's is taken.
boxOf := func(raw string) string {
c, err := delivery.DecodeComposed([]byte(raw))
if err != nil {
t.Fatal(err)
}
sealed := agent.SealedIn(c)
if len(sealed) != 1 {
t.Fatalf("keeper's composition carries %v", sealed)
}
return sealed[0]
}
for _, row := range []struct {
name, held string
kept bool
}{{"the have holds the rotated value", boxOf(string(rotated.Value)), true}, {"the have holds the value before the rotation", boxOf(before), false}} {
restarted := env
restarted.Running = func() []delivery.Composed { return nil }
restarted.Held = func(module string) []string {
if module == "keeper" {
return []string{row.held}
}
return nil
}
restarted.Unseal = func(box string) ([]byte, error) {
b.mu.Lock()
defer b.mu.Unlock()
if g, ok := sealedAt[box]; ok {
return []byte(fmt.Sprintf("password-%d", g)), nil
}
return nil, fmt.Errorf("not sealed to n1: %s", box)
}
again := New(restarted)
again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}}
rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
if err != nil || len(rows) != 1 {
t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
}
if kept := rows[0].Sealed == row.held; kept != row.kept {
t.Errorf("%s: composed with %s (held %s), kept %v, want %v", row.name, rows[0].Sealed, row.held, kept, row.kept)
}
}
// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as