From d1fabd6dcbef772b4c98535851e3780f90e7992e Mon Sep 17 00:00:00 2001
From: Sung-Kyu Yoo <skyoo2003@gmail.com>
Date: Thu, 1 Oct 2026 19:47:56 +0900
Subject: [PATCH] feat: number writes with store revisions

---
 changes/unreleased/Added-20260930-213500.yaml |   5 +
 internal/cluster/cluster.go                   |  36 ++-
 internal/cluster/cluster_test.go              |  39 +++
 internal/datadir/datadir.go                   |  30 +-
 internal/datadir/datadir_test.go              |  34 +-
 internal/server/resp_cluster_test.go          |   4 +-
 pkg/kvs/api_surface_test.go                   |   2 +-
 pkg/kvs/kvs.go                                | 144 +++++++--
 pkg/kvs/log.go                                |  10 +
 pkg/kvs/log_test.go                           |   9 +-
 pkg/kvs/replication.go                        |  54 +++-
 pkg/kvs/replication_test.go                   |   8 +-
 pkg/kvs/revision_test.go                      | 302 ++++++++++++++++++
 pkg/kvs/speculate_test.go                     |   4 +-
 pkg/kvs/testdata/api-surface.txt              |  11 +-
 website/content/docs/clustering.md            |  10 +-
 website/content/docs/compatibility.md         |   2 +-
 17 files changed, 624 insertions(+), 80 deletions(-)
 create mode 100644 changes/unreleased/Added-20260930-213500.yaml
 create mode 100644 pkg/kvs/revision_test.go

diff --git a/changes/unreleased/Added-20260930-213500.yaml b/changes/unreleased/Added-20260930-213500.yaml
new file mode 100644
index 0000000..0b99236
--- /dev/null
+++ b/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"
diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go
index d900083..5bf26f4 100644
--- a/internal/cluster/cluster.go
+++ b/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) {
diff --git a/internal/cluster/cluster_test.go b/internal/cluster/cluster_test.go
index 57d5bd7..488b1ff 100644
--- a/internal/cluster/cluster_test.go
+++ b/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
diff --git a/internal/datadir/datadir.go b/internal/datadir/datadir.go
index cbbb217..8c871f6 100644
--- a/internal/datadir/datadir.go
+++ b/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
diff --git a/internal/datadir/datadir_test.go b/internal/datadir/datadir_test.go
index 50657cb..eca6c96 100644
--- a/internal/datadir/datadir_test.go
+++ b/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,9 +175,9 @@ 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",
@@ -163,7 +185,7 @@ func TestEnsureRefusesADirectoryItCannotRead(t *testing.T) {
 				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"},
 		},
 	}
 
diff --git a/internal/server/resp_cluster_test.go b/internal/server/resp_cluster_test.go
index d09e8ac..bd55b46 100644
--- a/internal/server/resp_cluster_test.go
+++ b/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)
diff --git a/pkg/kvs/api_surface_test.go b/pkg/kvs/api_surface_test.go
index 03eee1a..087917a 100644
--- a/pkg/kvs/api_surface_test.go
+++ b/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")
diff --git a/pkg/kvs/kvs.go b/pkg/kvs/kvs.go
index dc83f1c..63fe1b1 100644
--- a/pkg/kvs/kvs.go
+++ b/pkg/kvs/kvs.go
@@ -39,6 +39,10 @@ const reapSample = 20
 type Entry struct {
 	Value     interface{}
 	ExpiresAt time.Time
+	// CreateRevision is the revision the key was created at and ModRevision the one it last
+	// changed at. Set fills both in, so whatever a caller puts in them is ignored.
+	CreateRevision int64
+	ModRevision    int64
 }
 
 // Store is a key-value store, holding the keyspace in memory unless Open gave it a log to
@@ -62,7 +66,10 @@ type Store struct {
 	// replicate, when set, is where a write goes instead of straight into the keyspace. A
 	// clustered node points it at consensus, so the HTTP, gRPC, and RESP servers keep calling
 	// Write and never have to know the difference.
-	replicate func(fn func(tx *Tx) error) error
+	replicate func(fn func(tx *Tx) error) (int64, error)
+	// rev is the revision of the last write that changed anything. Like the keyspace, it only
+	// changes under mu held for writing.
+	rev int64
 	// lg is the append log the keyspace survives a restart through. A Store from NewStore
 	// has none and keeps everything in memory, which is what a caller who only wants a cache
 	// gets by default.
@@ -172,14 +179,21 @@ func (s *Store) snapshot() (frame, error) {
 // under would either miss a write or repeat one.
 func (s *Store) snapshotLocked() (frame, error) {
 	now := time.Now()
-	live := make([]record, 0, len(s.data))
+	live := make([]record, 0, len(s.data)+1)
+
+	if s.rev > 0 {
+		live = append(live, record{Op: opRev, Rev: s.rev})
+	}
 
 	for key, entry := range s.data {
 		if !entry.ExpiresAt.IsZero() && !entry.ExpiresAt.After(now) {
 			continue
 		}
 
-		live = append(live, record{Op: opSet, Key: key, value: entry.Value, ExpiresAt: entry.ExpiresAt})
+		live = append(live, record{
+			Op: opSet, Key: key, value: entry.Value, ExpiresAt: entry.ExpiresAt,
+			Rev: entry.ModRevision, CreateRev: entry.CreateRevision,
+		})
 	}
 
 	return s.encodeFrame(live)
@@ -189,17 +203,17 @@ func (s *Store) snapshotLocked() (frame, error) {
 // receive. Encoding here rather than at each Set is what makes a value the codec cannot handle
 // fail the write that stored it.
 func (s *Store) encodeFrame(pending []record) (frame, error) {
-	// The one place a stored value becomes bytes, and so the one place worth checking. Nothing to
-	// encode needs no codec, which is why an empty keyspace snapshots without one and a keyspace
-	// holding a single key does not.
-	if len(pending) > 0 && s.codec == nil {
-		return nil, ErrNoCodec
-	}
-
 	lines := make(frame, 0, len(pending))
 
 	for i := range pending {
 		if pending[i].Op == opSet {
+			// The one place a stored value becomes bytes, and so the one place worth checking.
+			// Nothing to encode needs no codec, which is why an empty keyspace snapshots without
+			// one, revision and all, and a keyspace holding a single key does not.
+			if s.codec == nil {
+				return nil, ErrNoCodec
+			}
+
 			value, err := s.codec.Encode(pending[i].value)
 			if err != nil {
 				return nil, fmt.Errorf("encode %q: %w", pending[i].Key, err)
@@ -268,6 +282,14 @@ func (s *Store) Read(fn func(tx *ReadTx) error) error {
 // A clustered node sends the write through consensus from here, so this one door covers the
 // HTTP, gRPC, and RESP servers at once.
 func (s *Store) Write(fn func(tx *Tx) error) error {
+	_, err := s.WriteRevision(fn)
+
+	return err
+}
+
+// WriteRevision is Write that also reports the revision the write committed at, or the current
+// one when fn changed nothing. A write that fails reports zero.
+func (s *Store) WriteRevision(fn func(tx *Tx) error) (int64, error) {
 	if replicate := s.replicator(); replicate != nil {
 		// No lock: the replicator takes it itself, once to work out what fn would change and
 		// again to apply what the cluster agreed to.
@@ -277,7 +299,23 @@ func (s *Store) Write(fn func(tx *Tx) error) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	return s.writeLocked(fn)
+	if err := s.writeLocked(fn); err != nil {
+		return 0, err
+	}
+
+	return s.rev, nil
+}
+
+// Revision is the revision of the last write that changed anything, and zero for a store nothing
+// has been written to. Every write transaction that changes a key moves it forward once, however
+// many keys it touched; one that changes nothing leaves it where it is. On a clustered node it is
+// the index of the Raft entry that carried the write, so it only ever grows but skips the entries
+// that were not writes.
+func (s *Store) Revision() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	return s.rev
 }
 
 // Speculate runs fn the way Write would, then puts the keyspace back and returns only what fn
@@ -303,7 +341,9 @@ func (s *Store) Speculate(fn func(tx *Tx) error) ([][]byte, error) {
 		return nil, ErrNoCodec
 	}
 
-	tx := &Tx{ReadTx: ReadTx{store: s, now: time.Now()}, speculative: true}
+	// The revision is a guess: the cluster numbers the write when it agrees to it, and applying
+	// it stamps that number over this one.
+	tx := &Tx{ReadTx: ReadTx{store: s, now: time.Now()}, rev: s.rev + 1, speculative: true}
 
 	err := fn(tx)
 	if err != nil {
@@ -332,7 +372,13 @@ func (s *Store) Snapshot() ([][]byte, error) {
 // hold the lock and are the reason the keyspace may change at all: startup replay and the
 // stream from a leader.
 func (s *Store) writeLocked(fn func(tx *Tx) error) error {
-	tx := &Tx{ReadTx: ReadTx{store: s, now: time.Now()}}
+	return s.writeAt(s.rev+1, fn)
+}
+
+// writeAt is writeLocked with the revision the transaction commits at chosen by the caller, which
+// is how a clustered node numbers a write with the Raft entry that carried it.
+func (s *Store) writeAt(rev int64, fn func(tx *Tx) error) error {
+	tx := &Tx{ReadTx: ReadTx{store: s, now: time.Now()}, rev: rev}
 	err := fn(tx)
 	tx.reapExpired()
 
@@ -340,6 +386,11 @@ func (s *Store) writeLocked(fn func(tx *Tx) error) error {
 		err = commitErr
 	}
 
+	// Even when fn failed: what it changed before failing is in memory, and so has happened.
+	if tx.changed && tx.rev > s.rev {
+		s.rev = tx.rev
+	}
+
 	return err
 }
 
@@ -502,6 +553,10 @@ type Tx struct {
 	speculative bool
 	// undoLog is what rollback puts back, and is filled only while speculating.
 	undoLog []undo
+	// rev is the revision the transaction commits at, and changed whether it has made a change a
+	// caller can see, which is what decides whether the store's revision moves at all.
+	rev     int64
+	changed bool
 }
 
 // undo is what one key held before the transaction touched it.
@@ -582,8 +637,21 @@ func (tx *Tx) Get(key string) (Entry, bool) {
 	return entry, true
 }
 
-// Set stores entry under key, replacing whatever was there.
+// Set stores entry under key, replacing whatever was there. The key keeps its CreateRevision
+// when it already held a live value, and takes the transaction's revision otherwise.
 func (tx *Tx) Set(key string, entry Entry) {
+	entry.CreateRevision = tx.rev
+	if old, ok := tx.store.data[key]; ok && !tx.expired(old) {
+		entry.CreateRevision = old.CreateRevision
+	}
+	entry.ModRevision = tx.rev
+
+	tx.put(key, entry)
+}
+
+// put stores entry exactly as given, revisions included, which is what a replay that has them
+// written down needs.
+func (tx *Tx) put(key string, entry Entry) {
 	if tx.store.data == nil {
 		tx.store.data = make(map[string]Entry)
 	}
@@ -605,7 +673,10 @@ func (tx *Tx) Set(key string, entry Entry) {
 	}
 
 	tx.signalChange(key)
-	tx.record(&record{Op: opSet, Key: key, value: entry.Value, ExpiresAt: entry.ExpiresAt})
+	tx.record(&record{
+		Op: opSet, Key: key, value: entry.Value, ExpiresAt: entry.ExpiresAt,
+		Rev: entry.ModRevision, CreateRev: entry.CreateRevision,
+	})
 }
 
 // Delete removes key and reports whether it was there to begin with.
@@ -637,14 +708,14 @@ func (tx *Tx) Flush() {
 	if !tx.speculative {
 		tx.store.signalFlush()
 	}
-	tx.record(&record{Op: opFlush})
+	tx.record(&record{Op: opFlush, Rev: tx.rev})
 }
 
 // remove deletes key as a change callers can see, so it marks the key's watchers.
 func (tx *Tx) remove(key string) {
 	tx.discard(key)
 	tx.signalChange(key)
-	tx.record(&record{Op: opDel, Key: key})
+	tx.record(&record{Op: opDel, Key: key, Rev: tx.rev})
 }
 
 // record queues a change for the log. It sits beside every signalChange call and nowhere else,
@@ -652,6 +723,9 @@ func (tx *Tx) remove(key string) {
 // store tidying up after itself? Reclaiming an expired key is the latter, and needs no record
 // of its own — the expiry that condemned it is already in the entry the log holds.
 func (tx *Tx) record(rec *record) {
+	// Before the check below: a store with no log keeps no records but still counts revisions.
+	tx.changed = true
+
 	if !tx.speculative && !tx.store.recording() {
 		return
 	}
@@ -662,14 +736,24 @@ func (tx *Tx) record(rec *record) {
 // restore applies one replayed record. A key already past its expiry is dropped rather than
 // restored: it would not be visible anyway, and dropping it here keeps it out of the rewrite
 // that follows.
+//
+// A record that carries its revisions keeps them. One that does not, written before kvs kept
+// revisions or stripped of the leader's guess by ApplyReplicated, takes the transaction's.
 func (tx *Tx) restore(rec *record) error {
 	switch rec.Op {
+	case opRev:
+		tx.raise(rec.Rev)
 	case opFlush:
 		tx.Flush()
+		tx.raise(rec.Rev)
 	case opDel:
 		tx.Delete(rec.Key)
+		tx.raise(rec.Rev)
 	case opSet:
 		if !rec.ExpiresAt.IsZero() && !rec.ExpiresAt.After(tx.now) {
+			// Gone, but its revision was handed out all the same.
+			tx.raise(rec.Rev)
+
 			return nil
 		}
 
@@ -677,7 +761,17 @@ func (tx *Tx) restore(rec *record) error {
 		if err != nil {
 			return fmt.Errorf("decode %q: %w", rec.Key, err)
 		}
-		tx.Set(rec.Key, Entry{Value: value, ExpiresAt: rec.ExpiresAt})
+
+		entry := Entry{Value: value, ExpiresAt: rec.ExpiresAt}
+		if rec.Rev == 0 {
+			tx.Set(rec.Key, entry)
+
+			return nil
+		}
+
+		entry.CreateRevision, entry.ModRevision = rec.CreateRev, rec.Rev
+		tx.put(rec.Key, entry)
+		tx.raise(rec.Rev)
 	default:
 		// Refusing to start beats starting with part of the keyspace missing.
 		return fmt.Errorf("unknown log operation %q", rec.Op)
@@ -686,6 +780,20 @@ func (tx *Tx) restore(rec *record) error {
 	return nil
 }
 
+// raise carries the transaction forward to the revision a replayed record was written at, so the
+// store comes back at the revision it stopped at rather than one past whatever survived the
+// rewrite. A record from before revisions carries zero and moves nothing.
+func (tx *Tx) raise(rev int64) {
+	if rev == 0 {
+		return
+	}
+
+	if rev > tx.rev {
+		tx.rev = rev
+	}
+	tx.changed = true
+}
+
 // discard reclaims key and keeps the expiry index in step, without marking its watchers.
 // Reclaiming a key that has already expired is bookkeeping rather than a change: a watch
 // survives its key reaching an expiry it was armed against, and otherwise the sweep's random
diff --git a/pkg/kvs/log.go b/pkg/kvs/log.go
index b468d89..8a7e83d 100644
--- a/pkg/kvs/log.go
+++ b/pkg/kvs/log.go
@@ -57,6 +57,10 @@ const (
 	opSet   = "set"
 	opDel   = "del"
 	opFlush = "flush"
+	// opRev heads a snapshot and every rewrite of the log. The rewrite drops deleted keys, and the
+	// revisions of the writes that deleted them with them, so without it a store would come back
+	// at one past whatever survived rather than at the revision it stopped at.
+	opRev = "rev"
 )
 
 // logName is the file the append log lives in, inside the directory Open was given. The name
@@ -67,11 +71,17 @@ const logName = datadir.LogName
 // record is one durable change. Value holds the encoded form and value the original: encoding
 // is deferred to commit so that a value the codec cannot handle fails the write that stored
 // it, rather than vanishing between here and the next startup.
+//
+// Rev is the revision the record was written at: a set carries the key's ModRevision, with its
+// CreateRevision in CreateRev, a del or a flush the revision of its transaction, and a rev
+// record the store's own. Zero marks a record written before kvs kept revisions.
 type record struct {
 	Op        string    `json:"op"`
 	Key       string    `json:"key,omitempty"`
 	Value     []byte    `json:"value,omitempty"`
 	ExpiresAt time.Time `json:"expires_at,omitzero"`
+	Rev       int64     `json:"rev,omitempty"`
+	CreateRev int64     `json:"create_rev,omitempty"`
 
 	value interface{}
 }
diff --git a/pkg/kvs/log_test.go b/pkg/kvs/log_test.go
index c383c4a..5ea2a44 100644
--- a/pkg/kvs/log_test.go
+++ b/pkg/kvs/log_test.go
@@ -177,8 +177,9 @@ func TestStoreCompactsLogOnOpen(t *testing.T) {
 	}
 
 	reopened := openTestStore(t, dir)
-	if got := logRecords(t, dir); got != 1 {
-		t.Fatalf("records after compaction = %d, want 1", got)
+	// One record for the key, and the revision the store stopped at ahead of it.
+	if got := logRecords(t, dir); got != 2 {
+		t.Fatalf("records after compaction = %d, want 2", got)
 	}
 	if _, err := reopened.Get("counter"); err != nil {
 		t.Fatalf(`Get("counter") after compaction error = %v`, err)
@@ -223,7 +224,7 @@ func TestOpenRefusesADirectoryFromAnotherFormat(t *testing.T) {
 	}
 
 	format := filepath.Join(dir, datadir.FormatName)
-	if err := os.WriteFile(format, []byte("2\n"), 0o600); err != nil {
+	if err := os.WriteFile(format, []byte("3\n"), 0o600); err != nil {
 		t.Fatalf("write format: %v", err)
 	}
 
@@ -236,7 +237,7 @@ func TestOpenRefusesADirectoryFromAnotherFormat(t *testing.T) {
 		t.Errorf("errors.Is(err, datadir.ErrFormat) = false, err = %v", reopenErr)
 	}
 
-	for _, want := range []string{"format 2", "format 1"} {
+	for _, want := range []string{"format 3", "format 2"} {
 		if !strings.Contains(reopenErr.Error(), want) {
 			t.Errorf("message %q does not carry %q", reopenErr.Error(), want)
 		}
diff --git a/pkg/kvs/replication.go b/pkg/kvs/replication.go
index 00518f3..6b9a53e 100644
--- a/pkg/kvs/replication.go
+++ b/pkg/kvs/replication.go
@@ -29,21 +29,22 @@ func (e *NotLeaderError) Is(target error) bool {
 }
 
 // SetReplicator routes writes through replicate instead of straight into the keyspace, which is
-// how a clustered node makes every write pass consensus first. Passing nil puts it back.
+// how a clustered node makes every write pass consensus first, and reports the revision the write
+// committed at. Passing nil puts it back.
 //
 // Set it before anything serves: it is read without the lock a write would take, on the
 // understanding that it is wired up once at startup.
 //
 // Exported for internal/cluster to reach across the package boundary, and outside the v1
 // compatibility promise: see website/content/docs/compatibility.md.
-func (s *Store) SetReplicator(replicate func(fn func(tx *Tx) error) error) {
+func (s *Store) SetReplicator(replicate func(fn func(tx *Tx) error) (int64, error)) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.replicate = replicate
 }
 
-func (s *Store) replicator() func(fn func(tx *Tx) error) error {
+func (s *Store) replicator() func(fn func(tx *Tx) error) (int64, error) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
@@ -62,7 +63,8 @@ func (s *Store) SetCodec(codec Codec) {
 
 // ReplaceWith throws the keyspace away and rebuilds it from snapshot, which is what a node
 // restored from a cluster snapshot needs: the agreed state is the only authority, so whatever the
-// node held before is worth keeping only until it arrives.
+// node held before is worth keeping only until it arrives. That goes for the revision too: the
+// node comes back at the one the snapshot was taken at.
 //
 // Exported for internal/cluster to reach across the package boundary, and outside the v1
 // compatibility promise: see website/content/docs/compatibility.md.
@@ -70,35 +72,59 @@ func (s *Store) ReplaceWith(snapshot [][]byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	return s.writeLocked(func(tx *Tx) error {
+	s.rev = 0
+
+	err := s.writeLocked(func(tx *Tx) error {
 		tx.Flush()
 
-		return applyFrame(tx, snapshot)
+		return applyFrame(tx, snapshot, true)
 	})
+
+	// The flush counts as a change, but an empty snapshot is of a store nothing was written to,
+	// and has no revision record to say so.
+	if len(snapshot) == 0 {
+		s.rev = 0
+	}
+
+	return err
 }
 
-// ApplyReplicated applies one frame the cluster has agreed on. A frame is one transaction on the
-// node that took the write, and applying it inside one transaction here is what keeps a MULTI
-// atomic on every node.
+// ApplyReplicated applies one frame the cluster has agreed on, at the revision the cluster gave
+// it, and reports the store's revision afterwards. A frame is one transaction on the node that
+// took the write, and applying it inside one transaction here is what keeps a MULTI atomic on
+// every node.
+//
+// The revisions written inside the frame are ignored: they are the leader's guess from before the
+// cluster agreed, and only rev is the same on every node.
 //
 // Exported for internal/cluster to reach across the package boundary, and outside the v1
 // compatibility promise: see website/content/docs/compatibility.md.
-func (s *Store) ApplyReplicated(lines [][]byte) error {
+func (s *Store) ApplyReplicated(rev int64, lines [][]byte) (int64, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	return s.writeLocked(func(tx *Tx) error {
-		return applyFrame(tx, lines)
-	})
+	if err := s.writeAt(rev, func(tx *Tx) error {
+		return applyFrame(tx, lines, false)
+	}); err != nil {
+		return 0, err
+	}
+
+	return s.rev, nil
 }
 
-func applyFrame(tx *Tx, lines [][]byte) error {
+// applyFrame restores every record in lines. keepRevisions keeps the revisions the records carry,
+// which is right for a snapshot; without it they take the transaction's.
+func applyFrame(tx *Tx, lines [][]byte, keepRevisions bool) error {
 	for _, line := range lines {
 		rec, err := decodeRecord(line)
 		if err != nil {
 			return err
 		}
 
+		if !keepRevisions {
+			rec.Rev, rec.CreateRev = 0, 0
+		}
+
 		if err := tx.restore(&rec); err != nil {
 			return err
 		}
diff --git a/pkg/kvs/replication_test.go b/pkg/kvs/replication_test.go
index 477d141..0ee2d40 100644
--- a/pkg/kvs/replication_test.go
+++ b/pkg/kvs/replication_test.go
@@ -21,7 +21,7 @@ func TestAgreedChangeAppliesOnAnotherNode(t *testing.T) {
 	}
 
 	follower := newReplicatedStore(t)
-	if err := follower.ApplyReplicated(lines); err != nil {
+	if _, err := follower.ApplyReplicated(1, lines); err != nil {
 		t.Fatalf("ApplyReplicated() error = %v", err)
 	}
 	if got, err := follower.Get("greeting"); err != nil || got != "hello" {
@@ -72,7 +72,7 @@ func TestAppliedFrameKeepsItsOrder(t *testing.T) {
 	}
 
 	replica := newReplicatedStore(t)
-	if err := replica.ApplyReplicated(lines); err != nil {
+	if _, err := replica.ApplyReplicated(1, lines); err != nil {
 		t.Fatalf("ApplyReplicated() error = %v", err)
 	}
 
@@ -106,10 +106,10 @@ func TestReplicatorTakesOverWrites(t *testing.T) {
 	store := newReplicatedStore(t)
 
 	seen := 0
-	store.SetReplicator(func(fn func(tx *Tx) error) error {
+	store.SetReplicator(func(fn func(tx *Tx) error) (int64, error) {
 		seen++
 
-		return ErrNotLeader
+		return 0, ErrNotLeader
 	})
 
 	if err := store.Put("greeting", "hello"); !errors.Is(err, ErrNotLeader) {
diff --git a/pkg/kvs/revision_test.go b/pkg/kvs/revision_test.go
new file mode 100644
index 0000000..47d49af
--- /dev/null
+++ b/pkg/kvs/revision_test.go
@@ -0,0 +1,302 @@
+package kvs
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/skyoo2003/kvs/internal/datadir"
+)
+
+// One transaction is one revision, however many keys it touched.
+func TestRevisionAdvancesOncePerWrite(t *testing.T) {
+	store := NewStore()
+	if got := store.Revision(); got != 0 {
+		t.Fatalf("Revision() of a new store = %d, want 0", got)
+	}
+
+	rev, err := store.WriteRevision(func(tx *Tx) error {
+		tx.Set("a", Entry{Value: "1"})
+		tx.Set("b", Entry{Value: "1"})
+
+		return nil
+	})
+	if err != nil || rev != 1 {
+		t.Fatalf("WriteRevision() = %d, %v, want 1, nil", rev, err)
+	}
+
+	if err := store.Put("a", "2"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+	if got := store.Revision(); got != 2 {
+		t.Fatalf("Revision() = %d, want 2", got)
+	}
+
+	assertRevisions(t, store, "a", 1, 2)
+	assertRevisions(t, store, "b", 1, 1)
+}
+
+// A write that changes nothing has nothing to number.
+func TestWriteThatChangesNothingKeepsTheRevision(t *testing.T) {
+	store := NewStore()
+	if err := store.Put("a", "1"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+
+	rev, err := store.WriteRevision(func(*Tx) error { return nil })
+	if err != nil || rev != 1 {
+		t.Fatalf("WriteRevision() of nothing = %d, %v, want 1, nil", rev, err)
+	}
+	if err := store.Delete("missing"); !errors.Is(err, ErrKeyNotFound) {
+		t.Fatalf("Delete() error = %v, want %v", err, ErrKeyNotFound)
+	}
+	if got := store.Revision(); got != 1 {
+		t.Fatalf("Revision() = %d, want 1", got)
+	}
+}
+
+// A failed write reports no revision, but what it changed before failing is in memory, so the
+// store's revision has moved past it.
+func TestFailedWriteReportsNoRevision(t *testing.T) {
+	store := NewStore()
+	if err := store.Put("a", "1"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+
+	failure := errors.New("halfway")
+	rev, err := store.WriteRevision(func(tx *Tx) error {
+		tx.Set("a", Entry{Value: "2"})
+
+		return failure
+	})
+	if !errors.Is(err, failure) || rev != 0 {
+		t.Fatalf("WriteRevision() = %d, %v, want 0, %v", rev, err, failure)
+	}
+	if got := store.Revision(); got != 2 {
+		t.Fatalf("Revision() = %d, want 2", got)
+	}
+}
+
+// A key deleted and written again is a new key, and says so.
+func TestRecreatedKeyGetsANewCreateRevision(t *testing.T) {
+	store := NewStore()
+	for _, step := range []func() error{
+		func() error { return store.Put("a", "1") },
+		func() error { return store.Delete("a") },
+		func() error { return store.Put("a", "2") },
+	} {
+		if err := step(); err != nil {
+			t.Fatalf("step error = %v", err)
+		}
+	}
+
+	assertRevisions(t, store, "a", 3, 3)
+}
+
+// An expired key is gone, so overwriting it creates the key rather than changing it.
+func TestOverwrittenExpiredKeyCountsAsCreated(t *testing.T) {
+	store := NewStore()
+	if err := store.Write(func(tx *Tx) error {
+		tx.Set("a", Entry{Value: "1", ExpiresAt: time.Now().Add(-time.Second)})
+
+		return nil
+	}); err != nil {
+		t.Fatalf("Write() error = %v", err)
+	}
+	if err := store.Put("a", "2"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+
+	assertRevisions(t, store, "a", 2, 2)
+}
+
+// The revision has to come back after a restart even when the last write was a delete, whose key
+// the startup rewrite drops. Opening twice covers both the replayed log and the rewritten one.
+func TestRevisionSurvivesRestart(t *testing.T) {
+	dir := t.TempDir()
+
+	store := openTestStore(t, dir)
+	for _, step := range []func() error{
+		func() error { return store.Put("a", "1") },
+		func() error { return store.Put("b", "1") },
+		func() error { return store.Delete("b") },
+	} {
+		if err := step(); err != nil {
+			t.Fatalf("step error = %v", err)
+		}
+	}
+
+	for range 2 {
+		if err := store.Close(); err != nil {
+			t.Fatalf("Close() error = %v", err)
+		}
+
+		store = openTestStore(t, dir)
+		if got := store.Revision(); got != 3 {
+			t.Fatalf("Revision() after restart = %d, want 3", got)
+		}
+		assertRevisions(t, store, "a", 1, 1)
+	}
+
+	rev, err := store.WriteRevision(func(tx *Tx) error {
+		tx.Set("c", Entry{Value: "1"})
+
+		return nil
+	})
+	if err != nil || rev != 4 {
+		t.Fatalf("WriteRevision() after restart = %d, %v, want 4, nil", rev, err)
+	}
+}
+
+// A directory written before revisions opens with everything it held at revision 1.
+func TestLegacyLogOpensAtRevisionOne(t *testing.T) {
+	dir := t.TempDir()
+	writeTestFile(t, filepath.Join(dir, datadir.FormatName), "1\n")
+	// "aGVsbG8=" is "hello": the log stores values as base64.
+	writeTestFile(t, filepath.Join(dir, logName), `{"op":"set","key":"a","value":"aGVsbG8="}`+"\n")
+
+	store := openTestStore(t, dir)
+	if got, err := store.Get("a"); err != nil || got != "hello" {
+		t.Fatalf(`Get("a") = %v, %v, want %v, nil`, got, err, "hello")
+	}
+	if got := store.Revision(); got != 1 {
+		t.Fatalf("Revision() = %d, want 1", got)
+	}
+	assertRevisions(t, store, "a", 1, 1)
+}
+
+// The leader's guess at a revision is only a guess: what the cluster agreed to number the write
+// is what every node records.
+func TestReplicatedWriteTakesTheAgreedRevision(t *testing.T) {
+	leader := newReplicatedStore(t)
+	follower := newReplicatedStore(t)
+
+	set, err := leader.Speculate(func(tx *Tx) error {
+		tx.Set("a", Entry{Value: "1"})
+
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("Speculate() error = %v", err)
+	}
+	if rev, applyErr := follower.ApplyReplicated(7, set); applyErr != nil || rev != 7 {
+		t.Fatalf("ApplyReplicated(7) = %d, %v, want 7, nil", rev, applyErr)
+	}
+	assertRevisions(t, follower, "a", 7, 7)
+
+	del, err := follower.Speculate(func(tx *Tx) error {
+		tx.Delete("a")
+
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("Speculate() error = %v", err)
+	}
+	if rev, applyErr := follower.ApplyReplicated(9, del); applyErr != nil || rev != 9 {
+		t.Fatalf("ApplyReplicated(9) = %d, %v, want 9, nil", rev, applyErr)
+	}
+}
+
+// A snapshot is the agreed state, revision included, even when the node held a later one.
+func TestReplaceWithTakesTheSnapshotRevision(t *testing.T) {
+	leader := newReplicatedStore(t)
+	replica := newReplicatedStore(t)
+
+	for range 3 {
+		if err := leader.Put("a", "leader"); err != nil {
+			t.Fatalf("Put() error = %v", err)
+		}
+	}
+	for range 5 {
+		if err := replica.Put("b", "replica"); err != nil {
+			t.Fatalf("Put() error = %v", err)
+		}
+	}
+
+	snapshot, err := leader.Snapshot()
+	if err != nil {
+		t.Fatalf("Snapshot() error = %v", err)
+	}
+	if err := replica.ReplaceWith(snapshot); err != nil {
+		t.Fatalf("ReplaceWith() error = %v", err)
+	}
+
+	if got := replica.Revision(); got != 3 {
+		t.Fatalf("Revision() after ReplaceWith = %d, want 3", got)
+	}
+	assertRevisions(t, replica, "a", 1, 3)
+}
+
+// A snapshot of a store nothing was ever written to carries revision zero, and so does the node
+// restored from it, whatever it held before.
+func TestReplaceWithAnEmptySnapshotKeepsRevisionZero(t *testing.T) {
+	snapshot, err := newReplicatedStore(t).Snapshot()
+	if err != nil {
+		t.Fatalf("Snapshot() error = %v", err)
+	}
+
+	replica := newReplicatedStore(t)
+	if err := replica.Put("b", "replica"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+	if err := replica.ReplaceWith(snapshot); err != nil {
+		t.Fatalf("ReplaceWith() error = %v", err)
+	}
+
+	if got := replica.Revision(); got != 0 {
+		t.Fatalf("Revision() after ReplaceWith = %d, want 0", got)
+	}
+}
+
+// A store emptied by a flush still has a revision to snapshot, and nothing that needs a codec.
+func TestSnapshotOfAnEmptiedStoreNeedsNoCodec(t *testing.T) {
+	store := NewStore()
+	if err := store.Put("a", "1"); err != nil {
+		t.Fatalf("Put() error = %v", err)
+	}
+	if err := store.Write(func(tx *Tx) error {
+		tx.Flush()
+
+		return nil
+	}); err != nil {
+		t.Fatalf("Write() error = %v", err)
+	}
+
+	if _, err := store.Snapshot(); err != nil {
+		t.Fatalf("Snapshot() error = %v", err)
+	}
+	if got := store.Revision(); got != 2 {
+		t.Fatalf("Revision() = %d, want 2", got)
+	}
+}
+
+func assertRevisions(t *testing.T, store *Store, key string, create, mod int64) {
+	t.Helper()
+
+	var entry Entry
+	if err := store.Read(func(tx *ReadTx) error {
+		var ok bool
+		if entry, ok = tx.Get(key); !ok {
+			return ErrKeyNotFound
+		}
+
+		return nil
+	}); err != nil {
+		t.Fatalf("Get(%q) error = %v", key, err)
+	}
+
+	if entry.CreateRevision != create || entry.ModRevision != mod {
+		t.Fatalf("%q revisions = create %d, mod %d, want create %d, mod %d",
+			key, entry.CreateRevision, entry.ModRevision, create, mod)
+	}
+}
+
+func writeTestFile(t *testing.T, path, text string) {
+	t.Helper()
+
+	if err := os.WriteFile(path, []byte(text), 0o600); err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+}
diff --git a/pkg/kvs/speculate_test.go b/pkg/kvs/speculate_test.go
index 75e0c6b..aea05ee 100644
--- a/pkg/kvs/speculate_test.go
+++ b/pkg/kvs/speculate_test.go
@@ -244,7 +244,7 @@ func TestSpeculateFrameAppliesElsewhere(t *testing.T) {
 	}
 
 	follower := newSpeculativeStore(t)
-	if err := follower.ApplyReplicated(lines); err != nil {
+	if _, err := follower.ApplyReplicated(1, lines); err != nil {
 		t.Fatalf("ApplyReplicated() error = %v", err)
 	}
 	if got := stored(t, follower, "langs"); !slices.Equal(got, []string{"go", "rust"}) {
@@ -252,7 +252,7 @@ func TestSpeculateFrameAppliesElsewhere(t *testing.T) {
 	}
 
 	// Applying it on the leader too gets both to the same place.
-	if err := leader.ApplyReplicated(lines); err != nil {
+	if _, err := leader.ApplyReplicated(2, lines); err != nil {
 		t.Fatalf("ApplyReplicated() on leader error = %v", err)
 	}
 	if got := stored(t, leader, "langs"); !slices.Equal(got, []string{"go", "rust"}) {
diff --git a/pkg/kvs/testdata/api-surface.txt b/pkg/kvs/testdata/api-surface.txt
index 1d5b793..3edeed6 100644
--- a/pkg/kvs/testdata/api-surface.txt
+++ b/pkg/kvs/testdata/api-surface.txt
@@ -16,6 +16,9 @@ var (
 type Entry struct {
 	Value		interface{}
 	ExpiresAt	time.Time
+
+	CreateRevision	int64
+	ModRevision	int64
 }
 
 type Store struct {
@@ -32,6 +35,10 @@ func (s *Store) Read(fn func(tx *ReadTx) error) error
 
 func (s *Store) Write(fn func(tx *Tx) error) error
 
+func (s *Store) WriteRevision(fn func(tx *Tx) error) (int64, error)
+
+func (s *Store) Revision() int64
+
 func (s *Store) Speculate(fn func(tx *Tx) error) ([][]byte, error)
 
 func (s *Store) Snapshot() ([][]byte, error)
@@ -116,10 +123,10 @@ func (e *NotLeaderError) Error() string
 
 func (e *NotLeaderError) Is(target error) bool
 
-func (s *Store) SetReplicator(replicate func(fn func(tx *Tx) error) error)
+func (s *Store) SetReplicator(replicate func(fn func(tx *Tx) error) (int64, error))
 
 func (s *Store) SetCodec(codec Codec)
 
 func (s *Store) ReplaceWith(snapshot [][]byte) error
 
-func (s *Store) ApplyReplicated(lines [][]byte) error
+func (s *Store) ApplyReplicated(rev int64, lines [][]byte) (int64, error)
diff --git a/website/content/docs/clustering.md b/website/content/docs/clustering.md
index 3985f5e..0ba864d 100644
--- a/website/content/docs/clustering.md
+++ b/website/content/docs/clustering.md
@@ -174,9 +174,13 @@ two things to keep in step.
 the first time it uses one, and refuses to start on a directory whose version it does not
 recognize — including one written before that file existed. The alternative is a replay reading
 bytes laid out by another version, which surfaces much later and looks like corruption rather
-than a version mismatch. There is no conversion: move the directory aside and load the data
-again. The version covers the Raft store too, so it moves when the consensus library changes
-its own layout.
+than a version mismatch. A build reads the formats before its own that it lists as readable and
+restamps them with its own, so upgrading happens in place; a newer format is refused, so
+downgrading does not, and means moving the directory aside and loading the data again. Format 2,
+which added revisions, reads format 1. During a rolling upgrade an older node cannot install a
+snapshot from an upgraded one, so replace the nodes one after another rather than leaving a
+cluster mixed. The version covers the Raft store too, so it moves when the consensus library
+changes its own layout.
 
 **`--node-id` is required when there is no RESP listener.** It defaults to `--resp-addr`, so
 `--resp-addr none` leaves nothing to borrow and `serve` refuses to start rather than join a
diff --git a/website/content/docs/compatibility.md b/website/content/docs/compatibility.md
index e67a4e6..35877c1 100644
--- a/website/content/docs/compatibility.md
+++ b/website/content/docs/compatibility.md
@@ -51,7 +51,7 @@ Importing `github.com/skyoo2003/kvs/pkg/kvs` gets you:
 The former module-root library path is no longer provided; update imports to this package path.
 
 - `Store` and its constructors `NewStore` and `Open`, with `Get`, `Put`, `Delete`, `Read`,
-  `Write`, `Snapshot`, `Speculate`, `Watch`, `SetCodec`, and `Close`
+  `Write`, `WriteRevision`, `Revision`, `Snapshot`, `Speculate`, `Watch`, `SetCodec`, and `Close`
 - The transaction types `ReadTx` and `Tx`, and `Watch`
 - `Entry`, and the `Codec` interface with `StringCodec`
 - The sentinel errors `ErrKeyNotFound`, `ErrNoCodec`, `ErrUnsupportedValue`, and `ErrNotLeader`