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/history_integration_test.go b/internal/link/history_integration_test.go new file mode 100644 index 0000000..f6cc0b3 --- /dev/null +++ b/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) + } +} diff --git a/internal/link/watch.go b/internal/link/watch.go index 0a78e6e..a1f5687 100644 --- a/internal/link/watch.go +++ b/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") diff --git a/internal/witness/draining_test.go b/internal/witness/draining_test.go index 6488631..4a7d32d 100644 --- a/internal/witness/draining_test.go +++ b/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 }