Merge pull request 'Apply the revision of a composition its want names, read from the key's history, when the latest has another digest (issue 1945)' (#139) from fix/r6-want-names-revision into main

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)

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})
 }

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)
+	}
+}

internal/link/history_integration_test.go

@@ -0,0 +1,68 @@
+//go:build integration
+
+package link
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+)
+
+// The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and
+// none for a key that never was (issue 1945: the node agent takes the composition its want names from them).
+func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
+	srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	conn, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(conn.Close)
+	js, err := conn.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "composed-it", History: 10})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, v := range []string{"one", "two", "three"} {
+		if _, err := kv.PutString("it.node-tools.x", v); err != nil {
+			t.Fatal(err)
+		}
+	}
+	bus := OverNATS{Conn: conn, JS: js}
+	ctx := context.Background()
+	got, err := bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
+	if err != nil || len(got) != 3 || string(got[0].Value) != "one" || string(got[2].Value) != "three" ||
+		got[0].Revision >= got[2].Revision {
+		t.Fatalf("three revisions read back as %+v (%v)", got, err)
+	}
+	if err := kv.Delete("it.node-tools.x"); err != nil {
+		t.Fatal(err)
+	}
+	got, err = bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
+	if err != nil || len(got) != 4 || !got[3].Deleted {
+		t.Fatalf("after a delete: %+v (%v)", got, err)
+	}
+	if err := kv.Purge("it.node-tools.x"); err != nil {
+		t.Fatal(err)
+	}
+	got, err = bus.KVHistory(ctx, "composed-it", "it.node-tools.x")
+	if err != nil || len(got) != 1 || !got[0].Deleted {
+		t.Fatalf("after a purge: %+v (%v)", got, err)
+	}
+	if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 {
+		t.Fatalf("a key that never was: %+v (%v)", got, err)
+	}
+}

internal/link/watch.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/nats-io/nats.go"
 )
@@ -83,6 +84,70 @@ 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)
+}
+
+// historyWithin bounds one read of a key's revisions.
+const historyWithin = 10 * time.Second
+
+// 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)
+	}
+	// Bounded on its own: a read that does not answer must not hold the node agent (the review of issue 1945).
+	ctx, cancel := context.WithTimeout(ctx, historyWithin)
+	defer cancel()
+	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")
 

internal/witness/draining_test.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"slices"
+	"strconv"
 	"testing"
 	"time"
 )
@@ -49,6 +50,20 @@ func runFrom(t *testing.T, dir string) *exec.Cmd {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() })
+	// The process is known to run from the image only once /proc says so: waited for, never slept on (a check that
+	// read /proc before the kernel showed the new image found nothing, issue 1945's try of 2026-10-25 at 12:47).
+	exe := filepath.Join("/proc", strconv.Itoa(cmd.Process.Pid), "exe")
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		if got, err := os.Readlink(exe); err == nil && got == image {
+			break
+		}
+		if time.Now().After(deadline) {
+			got, err := os.Readlink(exe)
+			t.Fatalf("PID %d does not run from %s within 5s: %s reads %q (%v)", cmd.Process.Pid, image, exe, got, err)
+		}
+		runtime.Gosched()
+	}
 	return cmd
 }