diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index ca55fd3..f33b5af 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -294,9 +294,20 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess Haves: sessionHaves{session}, Kept: keptByModule(opts, signer), Recorded: recordedIn(opts.state, say), - Trusted: trustedKeys(opts, signer), - Backup: backupNode(session), - Say: say, + Revisions: func(ctx context.Context, key string) ([]agent.Update, error) { + got, err := session.KVHistory(ctx, composedBucket, key) + if err != nil { + return nil, err + } + out := make([]agent.Update, 0, len(got)) + for _, u := range got { + out = append(out, agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted}) + } + return out, nil + }, + Trusted: trustedKeys(opts, signer), + Backup: backupNode(session), + Say: say, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) res := applyComposition(actx, opts, c, others, sched, say) diff --git a/internal/agent/agent.go b/internal/agent/agent.go index b5350e8..c71c99b 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -97,7 +97,10 @@ type Env struct { // Recorded says the engine holds a record of every resource of the composition, so it was applied and not removed // since; nil says yes. A composition the kept declaration or a have says runs is applied again when it says no. Recorded func(ctx context.Context, c delivery.Composed) bool - Say func(string) + // Revisions are a composition key's revisions the bucket keeps, oldest first; nil reads none. The composition a want + // names is taken from them when a newer one replaced it under the same key (ADR 0539 clause 1). + Revisions func(ctx context.Context, key string) ([]Update, error) + Say func(string) } // Agent follows the wants. @@ -108,6 +111,9 @@ type Agent struct { wants map[string]wantNow // by module comps map[string]compNow // by composed key running map[string]delivery.Have + // applied is the composition each module was last applied from: a revision a want named from the key's history is + // not the key's latest value, and what it holds is still held out of the others' applies. + applied map[string]compNow // attempt is the last apply of a module that failed or is held, with the revisions it was made from. attempt map[string]attempt checked map[string]bool @@ -144,7 +150,7 @@ func New(env Env) *Agent { if env.Say == nil { env.Say = func(string) {} } - return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, + 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{}, retry: make(chan []string, 8)} } @@ -235,6 +241,8 @@ func (a *Agent) Running() []delivery.Composed { } if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed { out = append(out, c.C) + } else if c, ok := a.applied[m]; ok && c.Dig == h.Composed { + out = append(out, c.C) } } return out @@ -355,6 +363,51 @@ func (a *Agent) compositionFor(module, fp string) (compNow, bool) { return c, ok } +// 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) + } + if err != nil { + return compNow{}, false + } + revs, err := a.env.Revisions(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 + } + c, err := delivery.DecodeComposed(u.Value) + if err != nil { + continue + } + 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 +} + // others are what the node's other modules run now, so what they hold is not taken: the compositions the agent runs and, // for every module the agent does not hold yet, the resources the kept declaration has for it (ADR 0429 ยง4). func (a *Agent) others(except string) []delivery.Composed { @@ -373,6 +426,8 @@ func (a *Agent) others(except string) []delivery.Composed { } if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed { out = append(out, c.C) + } else if c, ok := a.applied[m]; ok && c.Dig == h.Composed { + out = append(out, c.C) } } if a.env.Kept != nil { @@ -511,6 +566,13 @@ func (a *Agent) settleOne(ctx context.Context, module string) { a.mu.Lock() comp, have := a.compositionFor(module, w.Want.FP) a.mu.Unlock() + if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" { + // The composition the want names may be an older revision of the key, replaced by a newer one at the same + // fingerprint (a want held there keeps its digest): taken from the revisions the bucket keeps (ADR 0539 clause 1). + if old, found := a.revisionNamed(ctx, module, w.Want.FP, w.Want.Composed); found { + comp, have = old, true + } + } if !have || comp.Err != nil { return // waits for the composition; a change of its key settles the module again } @@ -666,6 +728,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo } delete(a.attempt, module) a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning} + a.applied[module] = comp a.mu.Unlock() a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail}) } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 320cb19..ef014b0 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -809,3 +809,29 @@ func TestACompositionWhoseResourcesWereRemovedIsAppliedAgain(t *testing.T) { }) } } + +// The composition a want names is applied from the key's history when a newer one has replaced it under the same key: +// a want held at a fingerprint keeps naming the digest it was written with, and the controller writes every new +// composition at that fingerprint over the same key (ADR 0539 clause 1, the sweep's kept revisions; issue 1945, live on +// 2026-10-25: shanks' node-tools composition had digest 575a5c96, its held want named a36e63c8, and nothing was applied). +func TestTheRevisionAWantNamesIsAppliedFromHistory(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + ck, oldRaw, oldDig := composition(t, "node-tools", fp, false, file("node-tools", "x", "old")) + _, newRaw, _ := composition(t, "node-tools", fp, false, file("node-tools", "x", "new")) + b.comps.before(ck, newRaw) // the key's latest value + wk, wraw := wantFor(t, "node-tools", fp, oldDig, "") + b.wants.before(wk, wraw) + b.env.Revisions = func(_ context.Context, key string) ([]Update, error) { + if key != ck { + return nil, nil + } + return []Update{{Key: ck, Value: oldRaw, Revision: 3}, {Key: ck, Value: newRaw, Revision: 7}}, nil + } + stop := b.run() + defer stop() + eventually(t, "the revision the want names was not applied", func() bool { return len(b.appliedList()) == 1 }) + if h := b.haves.last("node-tools"); h.Composed != oldDig { + t.Fatalf("the have names %s, want %s", h.Composed, oldDig) + } +} diff --git a/internal/link/watch.go b/internal/link/watch.go index 0a78e6e..2a0ec1e 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -83,6 +83,64 @@ var ( _ KVBus = (*natsLink)(nil) ) +// KVHistoryBus is a link that can read the revisions a bucket keeps of one key: the node agent takes the composition its +// want names from them when a newer one replaced it (novox/hq ADR 0539 clause 1). +type KVHistoryBus interface { + // KVHistory is the key's revisions, oldest first; none when the key has none. + KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) +} + +// KVHistory reads the key's revisions on the link open now. +func (b OverNATS) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) { + js := b.JS + if b.Conn != nil { + bound, err := b.Conn.JetStream() + if err != nil { + return nil, err + } + js = bound + } + kv, err := js.KeyValue(bucket) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", bucket, err) + } + entries, err := kv.History(key, nats.Context(ctx)) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, err) + } + out := make([]KVUpdate, 0, len(entries)) + for _, e := range entries { + out = append(out, KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(), + Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}) + } + return out, nil +} + +func (l *natsLink) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) { + return OverNATS{Conn: l.conn, JS: l.js}.KVHistory(ctx, bucket, key) +} + +var ( + _ KVHistoryBus = OverNATS{} + _ KVHistoryBus = (*natsLink)(nil) +) + +// KVHistory reads a key's revisions on the link open now. +func (q *Session) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + h, ok := bus.(KVHistoryBus) + if bus == nil || !ok { + return nil, ErrNoLink + } + return h.KVHistory(ctx, bucket, key) +} + // ErrNoLink is what a watch asks for with no link open. var ErrNoLink = errors.New("no link to the bus is open")