replicate: fix Replicator incremental off-by-one (sinceVersion=maxVersion)

Stream.SinceTs is EXCLUSIVE (entries with version <= SinceTs are ignored), but
the library Replicator advanced its push cursor as sinceVersion = maxVersion+1
in Incremental and Snapshot, so the write landing at exactly maxVersion+1 was
skipped by every incremental until the next full snapshot re-captured it via
Backup(0). Durability gap (RPO), and for a merge-restored follower a
revoked-secret-resurrects hazard when a delete tombstone lands on the boundary.

Fix: resume AT maxVersion (re-including it is a no-op under the exclusive bound,
so no duplication) — matching the native cmd/replicate tool, which already
resumes at LastVersion=until (replica.go). Same one-off removed from the
post-Restore cursor (inc.version, not inc.version+1) so a node promoted after a
Restore does not skip its first boundary write.

Also seed snapVersion from db.MaxVersion() after loading the snapshot so Restore
skips incrementals already contained in the snapshot instead of re-applying
every incremental ever on each call (was: snapVersion stayed 0). Non-breaking —
snapshot key format unchanged.

Regression test TestReplicatorPushCursorCapturesEveryWrite simulates the
push->restore loop one write per round and asserts every boundary write reaches
the follower (fails under maxVersion+1, passes under the fix).

replicate.go

@@ -176,7 +176,13 @@ func (r *Replicator) Incremental(ctx context.Context) error {
 		return fmt.Errorf("upload %s: %w", key, err)
 	}
 
-	r.sinceVersion = maxVersion + 1
+	// Stream.SinceTs is EXCLUSIVE (entries with version <= SinceTs are
+	// ignored), so the NEXT backup must resume AT maxVersion, not maxVersion+1
+	// — otherwise the write at exactly maxVersion+1 is skipped by every
+	// incremental until the next full snapshot. Re-including maxVersion is a
+	// no-op (it is <= the exclusive bound), so there is no duplication. This
+	// matches the native cmd/replicate tool, which resumes at LastVersion=until.
+	r.sinceVersion = maxVersion
 	r.log.Infof("replicate: incremental uploaded %s (%d bytes, version %d)", key, size, maxVersion)
 	return nil
 }
@@ -217,7 +223,13 @@ func (r *Replicator) Snapshot(ctx context.Context) error {
 	}
 
 	r.lastSnapshot = time.Now()
-	r.sinceVersion = maxVersion + 1
+	// Stream.SinceTs is EXCLUSIVE (entries with version <= SinceTs are
+	// ignored), so the NEXT backup must resume AT maxVersion, not maxVersion+1
+	// — otherwise the write at exactly maxVersion+1 is skipped by every
+	// incremental until the next full snapshot. Re-including maxVersion is a
+	// no-op (it is <= the exclusive bound), so there is no duplication. This
+	// matches the native cmd/replicate tool, which resumes at LastVersion=until.
+	r.sinceVersion = maxVersion
 	r.log.Infof("replicate: snapshot uploaded %s (%d bytes, version %d)", key, size, maxVersion)
 	return nil
 }
@@ -251,9 +263,13 @@ func (r *Replicator) Restore(ctx context.Context) error {
 		if err := r.downloadAndLoad(ctx, latestSnap); err != nil {
 			return fmt.Errorf("restore snapshot: %w", err)
 		}
-		// Extract timestamp from snap key to filter incrementals.
-		// We still apply all incrementals since we don't know the exact version
-		// from the filename. The DB.Load handles duplicates safely.
+		// Seed snapVersion from the loaded DB's max version (db.Load advances
+		// it), so incrementals already contained in the snapshot are SKIPPED
+		// below rather than re-applied on every Restore. This bounds the
+		// restore-replay cost (previously snapVersion stayed 0 → every
+		// incremental ever was re-applied each cycle). Non-breaking: the
+		// snapshot key format is unchanged.
+		snapVersion = r.db.MaxVersion()
 	}
 
 	// Find and apply incrementals in order.
@@ -283,7 +299,10 @@ func (r *Replicator) Restore(ctx context.Context) error {
 		if err := r.downloadAndLoad(ctx, inc.key); err != nil {
 			return fmt.Errorf("restore incremental %s: %w", inc.key, err)
 		}
-		r.sinceVersion = inc.version + 1
+		// Resume the push cursor AT the last applied version (exclusive
+		// SinceTs), so a node promoted after a Restore does not skip the write
+		// at inc.version+1 on its first push.
+		r.sinceVersion = inc.version
 	}
 
 	r.log.Infof("replicate: restore complete (sinceVersion=%d)", r.sinceVersion)

replicate_test.go

@@ -222,6 +222,60 @@ func TestReplicatorIncrementalBackup(t *testing.T) {
 	require.NoError(t, db2.Close())
 }
 
+// TestReplicatorPushCursorCapturesEveryWrite is the regression test for the
+// incremental off-by-one. It simulates the Replicator's push→restore loop —
+// Backup(sinceVersion), then advance sinceVersion per the rule, then Load into
+// a follower — one write per round. Stream.SinceTs is EXCLUSIVE, so the write
+// landing at exactly the boundary (previous maxVersion+1) is dropped forever by
+// the maxVersion+1 cursor until a full snapshot. With the fix (sinceVersion =
+// maxVersion) EVERY write reaches the follower. Green here == fixed.
+func TestReplicatorPushCursorCapturesEveryWrite(t *testing.T) {
+	srcDir, err := os.MkdirTemp("", "zapdb-cursor-src")
+	require.NoError(t, err)
+	defer removeDir(srcDir)
+	dstDir, err := os.MkdirTemp("", "zapdb-cursor-dst")
+	require.NoError(t, err)
+	defer removeDir(dstDir)
+
+	src, err := Open(getTestOptions(srcDir))
+	require.NoError(t, err)
+	defer src.Close()
+	dst, err := Open(getTestOptions(dstDir))
+	require.NoError(t, err)
+	defer dst.Close()
+
+	const rounds = 6
+	var sinceVersion uint64 // the Replicator's cursor, seeded at 0
+	for i := 0; i < rounds; i++ {
+		k := []byte(fmt.Sprintf("k-%02d", i))
+		v := []byte(fmt.Sprintf("v-%02d", i))
+		require.NoError(t, src.Update(func(txn *Txn) error { return txn.Set(k, v) }))
+
+		var buf bytes.Buffer
+		maxV, err := src.Backup(&buf, sinceVersion)
+		require.NoError(t, err)
+		if buf.Len() > 0 {
+			require.NoError(t, dst.Load(&buf, 16))
+			// THE FIX: resume AT maxVersion (exclusive SinceTs), not maxVersion+1.
+			sinceVersion = maxV
+		}
+	}
+
+	// Every single-write round — including boundary writes — must be present.
+	require.NoError(t, dst.View(func(txn *Txn) error {
+		for i := 0; i < rounds; i++ {
+			k := []byte(fmt.Sprintf("k-%02d", i))
+			item, err := txn.Get(k)
+			require.NoError(t, err, "follower missing %s — a boundary write was dropped (off-by-one)", k)
+			require.NoError(t, item.Value(func(val []byte) error {
+				require.Equal(t, []byte(fmt.Sprintf("v-%02d", i)), val)
+				return nil
+			}))
+		}
+		return nil
+	}))
+}
+
 // TestReplicatorVersionFromKey tests version extraction from S3 keys.
 func TestReplicatorVersionFromKey(t *testing.T) {
 	tests := []struct {
@@ -295,13 +349,13 @@ func TestReplicatorEncryptedIncremental(t *testing.T) {
 }
 
 // TestReplicateE2E is the full end-to-end test:
-//   1. Create source DB, write 100 keys
-//   2. Full backup → age encrypt → buffer (simulating S3)
-//   3. Write 50 more keys to source
-//   4. Incremental backup → age encrypt → buffer
-//   5. Restore to fresh DB: decrypt full → load, decrypt inc → load
-//   6. Verify ALL 150 keys match
-//   7. Verify encrypted blobs cannot be loaded without decryption
+//  1. Create source DB, write 100 keys
+//  2. Full backup → age encrypt → buffer (simulating S3)
+//  3. Write 50 more keys to source
+//  4. Incremental backup → age encrypt → buffer
+//  5. Restore to fresh DB: decrypt full → load, decrypt inc → load
+//  6. Verify ALL 150 keys match
+//  7. Verify encrypted blobs cannot be loaded without decryption
 func TestReplicateE2E(t *testing.T) {
 	identity, err := age.GenerateX25519Identity()
 	require.NoError(t, err)