Bound a key's history read to 10 s and test it on a real bus; the witness test waits for its process to run from its image (issue 1945)

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"
 )
@@ -90,6 +91,9 @@ type KVHistoryBus interface {
 	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
@@ -104,6 +108,9 @@ func (b OverNATS) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate
 	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

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
 }