feat: number writes with store revisions

changes/unreleased/Added-20260930-213500.yaml

@@ -0,0 +1,5 @@
+kind: Added
+body: 'Writes now carry a revision: `Store.Revision` and `Store.WriteRevision` report it, and every `Entry` records the revisions it was created and last changed at. Data directories move to format 2, and format 1 is upgraded in place.'
+time: 2026-09-30T21:35:00.000000+09:00
+custom:
+    Issue: "272"

internal/cluster/cluster.go

@@ -190,43 +190,47 @@ func (n *Node) bootstrapIfAsked(
 //
 // The order is the reason Speculate exists: applying first and replicating after would leave a
 // leader that lost an election holding writes nobody else has.
-func (n *Node) Write(fn func(tx *kvs.Tx) error) error {
+func (n *Node) Write(fn func(tx *kvs.Tx) error) (int64, error) {
 	n.applyMu.Lock()
 	defer n.applyMu.Unlock()
 
 	if n.raft.State() != raft.Leader {
-		return &kvs.NotLeaderError{Leader: n.LeaderID()}
+		return 0, &kvs.NotLeaderError{Leader: n.LeaderID()}
 	}
 
 	lines, err := n.store.Speculate(fn)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	// A transaction that changed nothing has nothing to agree on.
 	if len(lines) == 0 {
-		return nil
+		return n.store.Revision(), nil
 	}
 
 	payload, err := json.Marshal(lines)
 	if err != nil {
-		return fmt.Errorf("encode frame: %w", err)
+		return 0, fmt.Errorf("encode frame: %w", err)
 	}
 
 	future := n.raft.Apply(payload, applyTimeout)
 	if err := future.Error(); err != nil {
 		if errors.Is(err, raft.ErrLeadershipLost) || errors.Is(err, raft.ErrNotLeader) {
-			return &kvs.NotLeaderError{Leader: n.LeaderID()}
+			return 0, &kvs.NotLeaderError{Leader: n.LeaderID()}
 		}
 
-		return fmt.Errorf("replicate write: %w", err)
+		return 0, fmt.Errorf("replicate write: %w", err)
 	}
 
-	// The FSM reports what applying the frame did, which is where a decode failure surfaces.
-	if applyErr, ok := future.Response().(error); ok && applyErr != nil {
-		return applyErr
+	// The FSM reports what applying the frame did: the revision it landed at, or the error that
+	// stopped it, which is where a decode failure surfaces.
+	switch result := future.Response().(type) {
+	case error:
+		return 0, result
+	case int64:
+		return result, nil
+	default:
+		return 0, fmt.Errorf("replicate write: unexpected FSM response %T", result)
 	}
-
-	return nil
 }
 
 // Join adds another node as a voting member. Only the leader can, so a node that is asked and is
@@ -308,11 +312,15 @@ func (f *fsm) Apply(entry *raft.Log) interface{} {
 		return fmt.Errorf("decode frame: %w", err)
 	}
 
-	if err := f.store.ApplyReplicated(lines); err != nil {
+	// The entry's index is the write's revision: every node applies the same entry at the same
+	// index, however far behind it was or whichever snapshot it started from.
+	//nolint:gosec // A Raft index counts log entries and stays far below 2^63.
+	rev, err := f.store.ApplyReplicated(int64(entry.Index), lines)
+	if err != nil {
 		return err
 	}
 
-	return nil
+	return rev
 }
 
 func (f *fsm) Snapshot() (raft.FSMSnapshot, error) {

internal/cluster/cluster_test.go

@@ -111,6 +111,45 @@ func TestFSMSnapshotRoundTrip(t *testing.T) {
 	if got, err := target.Get("greeting"); err != nil || got != "hello" {
 		t.Fatalf("Get() after restore = %v, %v, want %v, nil", got, err, "hello")
 	}
+	if got, want := target.Revision(), source.Revision(); got != want {
+		t.Fatalf("Revision() after restore = %d, want %d", got, want)
+	}
+}
+
+// Every node numbers a write the same way, because the number is the Raft entry that carried it.
+func TestClusterRevisionsAgree(t *testing.T) {
+	nodes := startCluster(t, 3)
+	leader := waitForLeader(t, nodes)
+
+	var revs [2]int64
+	for i := range revs {
+		rev, err := leader.store.WriteRevision(func(tx *kvs.Tx) error {
+			tx.Set("greeting", kvs.Entry{Value: "hello " + strconv.Itoa(i)})
+
+			return nil
+		})
+		if err != nil {
+			t.Fatalf("WriteRevision() error = %v", err)
+		}
+		revs[i] = rev
+	}
+	if revs[0] <= 0 || revs[1] <= revs[0] {
+		t.Fatalf("revisions = %v, want positive and increasing", revs)
+	}
+
+	for _, node := range nodes {
+		eventually(t, node.id+" to reach revision "+strconv.FormatInt(revs[1], 10), func() bool {
+			var mod int64
+			_ = node.store.Read(func(tx *kvs.ReadTx) error {
+				entry, _ := tx.Get("greeting")
+				mod = entry.ModRevision
+
+				return nil
+			})
+
+			return node.store.Revision() == revs[1] && mod == revs[1]
+		})
+	}
 }
 
 // A cluster of one is still a cluster, and is what the first node of a new deployment is until

internal/datadir/datadir.go

@@ -14,11 +14,18 @@ import (
 	"strings"
 )
 
-// Version is what this build writes and the only one it reads. Raise it whenever the bytes in
-// a data directory change shape — including when a dependency that owns part of the directory,
-// the Raft log store among them, changes its own format under us. There is no conversion
-// between versions, which is why the check has to be loud.
-const Version = 1
+// Version is what this build writes. Raise it whenever the bytes in a data directory change
+// shape — including when a dependency that owns part of the directory, the Raft log store among
+// them, changes its own format under us. A build reads its own version and those back to
+// oldestReadable, and nothing newer: there is no conversion code, which is why the check has to
+// be loud.
+const Version = 2
+
+// oldestReadable is the oldest version this build still reads as it lies. Format 2 is format 1
+// with revisions added, and a record without one is read as revision 1, so a format 1 directory
+// needs nothing converted. It is restamped all the same, so that the older build it came from
+// refuses it rather than meeting records it cannot read.
+const oldestReadable = 1
 
 const (
 	// FormatName is the file holding Version. It sits beside the data rather than inside it
@@ -69,9 +76,9 @@ func (e *FormatError) Is(target error) bool {
 }
 
 // Ensure makes dir usable by this build, creating it and stamping the current version when it
-// is new. A directory belonging to another version comes back as a FormatError and nothing is
-// touched: refusing to start is the whole point, because the alternative is a replay that
-// half-works.
+// is new, and restamping it when it holds an older version this build still reads. A directory
+// belonging to any other version comes back as a FormatError and nothing is touched: refusing to
+// start is the whole point, because the alternative is a replay that half-works.
 func Ensure(dir string) error {
 	if err := os.MkdirAll(dir, 0o750); err != nil {
 		return fmt.Errorf("create data dir: %w", err)
@@ -110,8 +117,13 @@ func Ensure(dir string) error {
 
 	text := strings.TrimSpace(string(raw))
 
-	if found, convErr := strconv.Atoi(text); convErr != nil || found != Version {
+	found, convErr := strconv.Atoi(text)
+
+	switch {
+	case convErr != nil || found < oldestReadable || found > Version:
 		return &FormatError{Dir: dir, Raw: text}
+	case found < Version:
+		return stamp(path)
 	}
 
 	return nil

internal/datadir/datadir_test.go

@@ -23,7 +23,7 @@ func TestEnsureStampsANewDirectory(t *testing.T) {
 		t.Fatalf("read format: %v", err)
 	}
 
-	if got, want := strings.TrimSpace(string(raw)), "1"; got != want {
+	if got, want := strings.TrimSpace(string(raw)), "2"; got != want {
 		t.Errorf("format = %q, want %q", got, want)
 	}
 
@@ -46,6 +46,28 @@ func TestEnsureStampsANewDirectory(t *testing.T) {
 	}
 }
 
+// Format 2 only added to format 1, so a format 1 directory opens as it is and is restamped, which
+// is what keeps the build that wrote it from reading records it does not know.
+func TestEnsureUpgradesAnOlderFormat(t *testing.T) {
+	dir := t.TempDir()
+	writeFormat(t, dir, "1")
+	writeFile(t, filepath.Join(dir, datadir.LogName), "")
+
+	if err := datadir.Ensure(dir); err != nil {
+		t.Fatalf("Ensure() on format 1 error = %v", err)
+	}
+
+	//nolint:gosec // The path is the temporary directory this test just made.
+	raw, err := os.ReadFile(filepath.Join(dir, datadir.FormatName))
+	if err != nil {
+		t.Fatalf("read format: %v", err)
+	}
+
+	if got, want := strings.TrimSpace(string(raw)), "2"; got != want {
+		t.Errorf("format = %q, want %q", got, want)
+	}
+}
+
 // A stat that fails for any reason other than absence leaves the question open, and answering
 // it with "nothing here" would stamp this build's version onto another one's keyspace.
 func TestEnsureRefusesWhenItCannotTellWhetherDataIsThere(t *testing.T) {
@@ -153,17 +175,17 @@ func TestEnsureRefusesADirectoryItCannotRead(t *testing.T) {
 			name: "a later format",
 			arrange: func(t *testing.T, dir string) {
 				t.Helper()
-				writeFormat(t, dir, "2")
+				writeFormat(t, dir, "3")
 			},
-			wants: []string{"format 2", "format 1"},
+			wants: []string{"format 3", "format 2"},
 		},
 		{
 			name: "an earlier format",
 			arrange: func(t *testing.T, dir string) {
 				t.Helper()
 				writeFormat(t, dir, "0")
 			},
-			wants: []string{"format 0", "format 1"},
+			wants: []string{"format 0", "format 2"},
 		},
 		{
 			name: "something that is not a version",
@@ -199,7 +221,7 @@ func TestEnsureRefusesADirectoryItCannotRead(t *testing.T) {
 				t.Helper()
 				writeFile(t, filepath.Join(dir, datadir.LogName), "")
 			},
-			wants: []string{"no format file", "format 1"},
+			wants: []string{"no format file", "format 2"},
 		},
 		{
 			name: "a raft directory written before versioning",
@@ -210,7 +232,7 @@ func TestEnsureRefusesADirectoryItCannotRead(t *testing.T) {
 					t.Fatalf("create raft dir: %v", err)
 				}
 			},
-			wants: []string{"no format file", "format 1"},
+			wants: []string{"no format file", "format 2"},
 		},
 	}
 

internal/server/resp_cluster_test.go

@@ -143,8 +143,8 @@ func TestWriteFailureRedirectsRatherThanBlamingTheClient(t *testing.T) {
 	store := kvs.NewStore()
 	store.SetCodec(kvs.StringCodec{})
 	// What a follower does with a write: hand it to consensus, which says take it elsewhere.
-	store.SetReplicator(func(func(*kvs.Tx) error) error {
-		return &kvs.NotLeaderError{Leader: "127.0.0.1:6381"}
+	store.SetReplicator(func(func(*kvs.Tx) error) (int64, error) {
+		return 0, &kvs.NotLeaderError{Leader: "127.0.0.1:6381"}
 	})
 
 	client := newRESPClient(t, store)

pkg/kvs/api_surface_test.go

@@ -25,7 +25,7 @@ const surfaceHeader = `# Exported API surface of this package - see website/cont
 # for how much of it v1 promises, and for the cluster plumbing it exempts by name.
 # A line changed or removed below is a breaking change and needs a major version, unless the
 # page exempts it. A line added is a new promise: it cannot be taken back within v1.
-# Regenerate deliberately: go test -run TestPublicAPISurface . -update
+# Regenerate deliberately: go test ./pkg/kvs -run TestPublicAPISurface -update
 ` + headerSep
 
 var updateSurface = flag.Bool("update", false, "rewrite "+goldenPath+" from the current source")