fix(restore): force joiners of a restored cluster through checkpoint sync

docs/ops/backup-restore.md

@@ -122,7 +122,7 @@ The temporary checkpoint is removed from the leader's filesystem after the backu
 
 Backup preparation is performed on the **restore side** (during `FinalizeRestore` or `store bootstrap`), not during backup. It resets cluster-local and checkpoint-era zones and leaves the attribute zone **byte-for-byte intact** — there is no attribute compaction, because each canonical key holds exactly one Pebble entry (no per-index history to fold):
 
-1. **Reset lastAppliedIndex**: The Raft applied index is reset to 0, so the restored cluster starts fresh.
+1. **Preserve lastAppliedIndex as the genesis boundary**: The checkpoint's applied index is kept (a genesis checkpoint at index 0 gets the fallback boundary 1; MaxUint64 is refused). The restored bootstrap plants its WAL snapshot at this index, so the new log starts just above it and any fresh peer is routed through the snapshot → checkpoint-sync path (plain log replay from index 1 would land on an empty store and miss the restored state). The boundary labels the new log's start — it is NOT the restored state's provenance: incremental exports are sequence-keyed and never advance it, so after a full + incremental restore the state is newer than the boundary.
 2. **Remove persisted config**: Node and cluster IDs are stripped for portability.
 3. **Wipe the cluster-transient zone**: In-flight-only tracking (e.g. running backup jobs) has no meaning on the restored cluster.
 4. **Drop persisted bloom blocks**: Stale bloom blocks are cleared so the booting node rebuilds the bloom from a full attribute scan using its own config.
@@ -142,7 +142,7 @@ The backup is a complete Pebble database that contains:
 | Per-Ledger | `0x03` | Per-ledger data |
 | Cold | `0x04` | Transaction logs (`{0x04, 0x01}`), audit entries (`{0x04, 0x02}`) |
 | Idempotency | `0x05` | Idempotency keys |
-| Global | `0x06` | Last applied index (reset to 0), last applied timestamp, signing keys, signing config, chapters, sink configs, sink cursors, sink statuses |
+| Global | `0x06` | Last applied index (preserved on restore as the genesis boundary), last applied timestamp, signing keys, signing config, chapters, sink configs, sink cursors, sink statuses |
 
 > **Note**: If chapters have been archived before the backup, the archived logs and audit entries are no longer in the backup (they have been purged to cold storage). Attributes remain.
 
@@ -381,11 +381,12 @@ ledgerctl restore finalize --yes
 
 Calls `RestoreService.FinalizeRestore` (unary). This commits the staged backup as live data:
 
-1. Opens the staging directory read-only to extract `lastAppliedIndex` and `lastAppliedTimestamp`.
-2. Writes the **RESTORED marker** JSON file to `{dataDir}/RESTORED`.
-3. Creates `{dataDir}/checkpoints/` directory.
-4. **Atomically** hard-links the staging directory to `{dataDir}/checkpoints/0` (using a temp directory + `os.Rename` for crash safety).
-5. Removes the staging directory.
+1. Prepares the staged store ([Backup Preparation on Restore](#backup-preparation-on-restore): applied index preserved as the genesis boundary, persisted config/bloom/peers/cache reset).
+2. Reads back `lastAppliedIndex` and `lastAppliedTimestamp` from the prepared store.
+3. Writes the **RESTORED marker** JSON file to `{dataDir}/RESTORED`.
+4. Creates `{dataDir}/checkpoints/` directory.
+5. **Atomically** hard-links the staging directory to `{dataDir}/checkpoints/0` (using a temp directory + `os.Rename` for crash safety).
+6. Removes the staging directory.
 
 On next startup, `ScanLatestCheckpointID()` scans the `checkpoints/` directory for the highest numbered subdirectory to find the active checkpoint.
 
@@ -464,7 +465,7 @@ ledgerctl store bootstrap --driver s3 --s3-bucket my-bucket --s3-region us-east-
 4. **Preview**: Opens the staging as a read-only Pebble database, reads metadata (last applied index, timestamp, ledger list), and displays a summary table.
 5. **Validate** (optional): If `--validate` is set, runs the full integrity checker (`check.Checker`) -- the same checker used by `store check` and `restore validate`.
 6. **Confirm**: Unless `--yes` is set, prompts for user confirmation.
-7. **Prepare**: Prepares attributes for backup (Global-zone resets: applied index → 0, persisted config stripped, persisted bloom blocks dropped); the attribute zone is left intact.
+7. **Prepare**: Prepares attributes for backup (Global-zone resets: applied index preserved as the genesis boundary — the restored genesis' WAL-snapshot index — persisted config stripped, persisted bloom blocks dropped); the attribute zone is left intact.
 8. **Finalize**: Hard-links staging to `{data-dir}/checkpoints/0`, writes the `RESTORED` marker JSON.
 9. **Cleanup**: Removes the staging directory.
 
@@ -510,7 +511,7 @@ On startup, the node detects the `RESTORED` marker in `NewNode()`:
      - `nextSequenceID` from the last log sequence
      - `lastAuditHash` and `nextAuditSequenceID` from the last audit entry
    - Creates an FSM snapshot (`fsm.CreateSnapshot()`)
-   - Creates a WAL snapshot at `marker.LastAppliedIndex` with `ConfState{Voters: [nodeID]}` (single-node bootstrap)
+   - Creates a WAL snapshot at `marker.LastAppliedIndex` with `ConfState{Voters: [nodeID]}` (single-node bootstrap); a marker carrying 0 is refused — the snapshot must sit above the log start so joiners are forced through checkpoint sync
    - Removes the RESTORED marker
    - Continues with normal Raft startup
 
@@ -524,12 +525,12 @@ The `RESTORED` file is a JSON file written to the data directory during `Finaliz
 
 ```json
 {
-  "lastAppliedIndex": 12345,
+  "lastAppliedIndex": 8500,
   "lastAppliedTimestamp": 1700000000000000
 }
 ```
 
-- `lastAppliedIndex`: The Raft index of the last applied entry in the backup (reset to 0 during backup preparation, so this is always 0 in practice).
+- `lastAppliedIndex`: The Raft index the restored genesis occupies in the new log — the checkpoint's applied index preserved by backup preparation (1 for a genesis checkpoint, never 0). The restored bootstrap creates its WAL snapshot at this index. A boundary label only: after a full + incremental restore the restored state is newer than this index.
 - `lastAppliedTimestamp`: The HLC timestamp (microseconds) of the last applied entry.
 
 **File**: `internal/infra/node/restored_marker.go`
@@ -542,7 +543,7 @@ The `RESTORED` file is a JSON file written to the data directory during `Finaliz
 |-----------|-----------|
 | **Consistent snapshot** | Backup checkpoint is created as a direct Pebble checkpoint. Boundaries are always up-to-date in Pebble (written on every commit), so the checkpoint is consistent without Raft consensus or FSM gating. |
 | **Incremental efficiency** | SST files are immutable — same name means same content. Only new/changed files are uploaded; stale files are deleted. |
-| **Self-contained on restore** | During restore finalize, the applied index, persisted config and bloom blocks are reset (Global-zone); the attribute zone is preserved byte-for-byte. No dependency on the original cluster's Raft indices. |
+| **Self-contained on restore** | During restore finalize, the applied index is preserved as the genesis boundary and persisted config and bloom blocks are reset (Global-zone); the attribute zone is preserved byte-for-byte. The new log's index space starts at the boundary; nothing depends on the original cluster's later Raft indices. |
 | **Data integrity (content)** | `ValidateRestore` runs the full integrity checker: log sequence continuity, volume balance verification, metadata consistency. |
 | **Fresh directory required** | Restore mode refuses to start if existing checkpoints are found in `checkpoints/`, preventing accidental overwrites. |
 | **Atomic finalize** | Checkpoint placement uses `HardLink()` (temp directory + atomic `os.Rename`) for crash safety. |

docs/ops/cli.md

@@ -2147,7 +2147,7 @@ ledgerctl store bootstrap --driver s3 --s3-bucket <bucket> --data-dir /path/to/d
 4. Opens the staging as a read-only Pebble database and displays a preview (ledger count, timestamps)
 5. If `--validate` is set, runs the full integrity checker (same as `store check`); aborts before finalizing if any integrity error is found
 6. Prompts for confirmation (unless `--yes`)
-7. Prepares attributes for backup (Global-zone resets): resets applied index to 0, strips persisted config, drops persisted bloom blocks; the attribute zone is left intact
+7. Prepares attributes for backup (Global-zone resets): preserves the applied index as the genesis boundary (the WAL-snapshot index the restored genesis occupies, so joiners must sync a checkpoint; fallback 1 for a genesis checkpoint), strips persisted config, drops persisted bloom blocks; the attribute zone is left intact
 8. Hard-links staging to `checkpoints/0`, writes `RESTORED` marker
 9. Cleans up the staging directory
 

docs/technical/architecture/subsystems/fsm/deterministic-fsm.md

@@ -460,7 +460,7 @@ FSM flushes at apply time (batch, no sync):
 
 ### 12.3 Backup preparation (backup only)
 
-There is exactly one entry per canonical key, so there is no attribute compaction. Backup preparation (`PrepareForBackup`) leaves the attribute zone byte-for-byte intact and only resets cluster-local and checkpoint-era state: applied index → 0, persisted config deleted, cluster-transient zone wiped, persisted bloom blocks and Raft peers dropped, and the cache zone cleared (checkpoint-era cache rows predate the replayed delta and would otherwise serve stale CacheHits on the restored node).
+There is exactly one entry per canonical key, so there is no attribute compaction. Backup preparation (`PrepareForBackup`) leaves the attribute zone byte-for-byte intact and only resets cluster-local and checkpoint-era state: applied index preserved as the genesis boundary (the restored genesis’ WAL-snapshot index — the new log starts just above it, forcing joiners through checkpoint sync; a genesis checkpoint falls back to 1), persisted config deleted, cluster-transient zone wiped, persisted bloom blocks and Raft peers dropped, and the cache zone cleared (checkpoint-era cache rows predate the replayed delta and would otherwise serve stale CacheHits on the restored node).
 
 ### 12.4 Pebble key layout
 

internal/adapter/grpc/server_restore.go

@@ -325,7 +325,9 @@ func (s *RestoreServiceServerImpl) FinalizeRestore(_ context.Context, _ *restore
 
 	stagingDir := s.stagingDir()
 
-	// Reset Global-zone keys so the staged backup is restartable on a fresh cluster.
+	// Prepare Global-zone keys (applied index pinned to the restore-genesis
+	// index, cluster-local state reset) so the staged backup is restartable
+	// on a fresh cluster.
 	s.logger.Infof("Preparing backup for restore compatibility")
 
 	if err := attributes.PrepareForBackup(store); err != nil {

internal/infra/attributes/prepare.go

@@ -1,7 +1,10 @@
 package attributes
 
 import (
+	"encoding/binary"
+	"errors"
 	"fmt"
+	"math"
 
 	"github.com/cockroachdb/pebble/v2"
 
@@ -18,8 +21,9 @@ import (
 // versions to fold. The attribute zone is left byte-for-byte intact.
 //
 // The six resets are:
-//  1. lastAppliedIndex -> 0, so the restored cluster starts fresh without
-//     raft-index conflicts.
+//  1. lastAppliedIndex is preserved as the restored genesis boundary — the
+//     raft index the restored FSM genesis occupies in the new log (see
+//     below); a genesis checkpoint (index 0) gets the fallback boundary 1.
 //  2. persisted config (nodeId, clusterId) deleted, so the backup is portable
 //     to any cluster.
 //  3. ZoneClusterTransient wiped — in-flight-only tracking (backup jobs) has
@@ -36,13 +40,42 @@ import (
 // the checkpoint was taken. The backup flow achieves this by running the flush
 // and checkpoint atomically on the Raft loop.
 func PrepareForBackup(s *dal.Store) error {
+	// The checkpoint's applied index becomes the restored genesis boundary:
+	// the RESTORED bootstrap plants its WAL snapshot at this index, so the
+	// new log starts at boundary+1 and raft must route any fresh peer
+	// through the snapshot → checkpoint-sync path. At 0 the snapshot would
+	// be empty and the log would claim completeness from index 1 — a learner
+	// joining before the first post-restore raft snapshot would then be
+	// "caught up" by plain log replay onto an empty store, missing every
+	// restored row — so a genesis checkpoint gets the fallback boundary 1.
+	//
+	// The boundary is a label for the new log's start, NOT the state's
+	// source-cluster provenance: incremental exports are sequence-keyed and
+	// never advance this key, so after a full + incremental restore the
+	// state is newer than the boundary.
+	genesisBoundary, err := dal.ReadUint64(s, []byte{dal.ZoneGlobal, dal.SubGlobLastAppliedIndex}, 0)
+	if err != nil {
+		return fmt.Errorf("reading checkpoint applied index: %w", err)
+	}
+
+	// Several raft paths compute boundary+1 (first entry, FSM gap check).
+	if genesisBoundary == math.MaxUint64 {
+		return errors.New("invariant: checkpoint applied index is MaxUint64 — corrupt checkpoint")
+	}
+
+	if genesisBoundary == 0 {
+		genesisBoundary = 1
+	}
+
 	batch := s.OpenWriteSession()
 
-	// Reset lastAppliedIndex to 0 so the restored cluster starts fresh.
-	if err := batch.SetBytes([]byte{dal.ZoneGlobal, dal.SubGlobLastAppliedIndex}, make([]byte, 8)); err != nil {
+	appliedIndex := make([]byte, 8)
+	binary.BigEndian.PutUint64(appliedIndex, genesisBoundary)
+
+	if err := batch.SetBytes([]byte{dal.ZoneGlobal, dal.SubGlobLastAppliedIndex}, appliedIndex); err != nil {
 		_ = batch.Cancel()
 
-		return fmt.Errorf("resetting applied index: %w", err)
+		return fmt.Errorf("writing genesis boundary: %w", err)
 	}
 
 	// Remove persisted config (nodeId, clusterId) so the backup is portable to any cluster.

internal/infra/attributes/prepare_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"io"
+	"math"
 	"os"
 	"path/filepath"
 	"testing"
@@ -157,8 +158,9 @@ func TestPrepareForBackupPreservesAttributesByteForByte(t *testing.T) {
 	require.Equal(t, before, after, "attribute zone must be byte-for-byte identical after PrepareForBackup")
 }
 
-// TestPrepareForBackupResetsGlobalZone asserts the three Global-zone resets:
-// applied index -> 0, persisted config deleted, persisted bloom blocks dropped.
+// TestPrepareForBackupResetsGlobalZone asserts the Global-zone preparation:
+// applied index preserved as the genesis boundary, persisted config deleted,
+// persisted bloom blocks dropped.
 func TestPrepareForBackupResetsGlobalZone(t *testing.T) {
 	t.Parallel()
 
@@ -186,7 +188,8 @@ func TestPrepareForBackupResetsGlobalZone(t *testing.T) {
 
 	idx, err := readLastAppliedIndex(s)
 	require.NoError(t, err)
-	require.Equal(t, uint64(0), idx, "applied index must be reset to 0")
+	require.Equal(t, uint64(200), idx,
+		"the checkpoint's applied index must be preserved as the genesis boundary — the WAL-snapshot index the restored FSM genesis occupies")
 
 	_, _, err = s.Get([]byte{dal.ZoneGlobal, dal.SubGlobPersistedConfig})
 	require.ErrorIs(t, err, pebble.ErrNotFound, "persisted config must be deleted")
@@ -199,6 +202,46 @@ func TestPrepareForBackupResetsGlobalZone(t *testing.T) {
 	require.ErrorIs(t, err, pebble.ErrNotFound, "persisted Raft peers must be dropped (EN-1413)")
 }
 
+// TestPrepareForBackupGenesisCheckpointFallback asserts a checkpoint taken at
+// applied index 0 (genesis full backup, possibly followed by incremental
+// exports) gets the fallback boundary 1: the restored bootstrap must never
+// plant a WAL snapshot at 0 — that is the empty-snapshot shape that lets a
+// learner catch up by log replay onto an empty store.
+func TestPrepareForBackupGenesisCheckpointFallback(t *testing.T) {
+	t.Parallel()
+
+	tmpDir := t.TempDir()
+	logger := logging.FromContext(logging.TestingContext())
+	s, err := dal.OpenDirect(tmpDir, logger)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = s.Close() })
+
+	require.NoError(t, PrepareForBackup(s))
+
+	idx, err := readLastAppliedIndex(s)
+	require.NoError(t, err)
+	require.Equal(t, uint64(1), idx, "a genesis checkpoint must get the fallback boundary 1")
+}
+
+// TestPrepareForBackupRejectsMaxUint64 asserts a checkpoint carrying a
+// MaxUint64 applied index is refused: raft paths compute boundary+1.
+func TestPrepareForBackupRejectsMaxUint64(t *testing.T) {
+	t.Parallel()
+
+	tmpDir := t.TempDir()
+	logger := logging.FromContext(logging.TestingContext())
+	s, err := dal.OpenDirect(tmpDir, logger)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = s.Close() })
+
+	batch := s.OpenWriteSession()
+	require.NoError(t, setAppliedIndex(batch, math.MaxUint64))
+	require.NoError(t, batch.Commit())
+
+	err = PrepareForBackup(s)
+	require.ErrorContains(t, err, "MaxUint64")
+}
+
 // TestPrepareForBackupClearsCacheZone asserts the cache zone is dropped in
 // full: per-entry rows in both generation slots, per-generation rotation
 // metadata, and the global snapshot meta. Checkpoint-era cache rows predate
@@ -237,7 +280,8 @@ func TestPrepareForBackupClearsCacheZone(t *testing.T) {
 
 // TestPrepareForBackupRestorableOnFreshCluster runs the full backup->restore
 // pipeline (write -> prepare -> tar -> extract -> reopen) and asserts the
-// attribute values survive and the applied index is 0 on the fresh cluster.
+// attribute values survive and the seeded applied index reaches the fresh
+// cluster as its genesis boundary.
 func TestPrepareForBackupRestorableOnFreshCluster(t *testing.T) {
 	t.Parallel()
 
@@ -317,7 +361,8 @@ func TestPrepareForBackupRestorableOnFreshCluster(t *testing.T) {
 
 		idx, err := readLastAppliedIndex(s)
 		require.NoError(t, err)
-		require.Equal(t, uint64(0), idx, "applied index must be 0 on the fresh cluster")
+		require.Equal(t, uint64(5), idx,
+			"the checkpoint's applied index must reach the fresh cluster as its genesis boundary")
 	}()
 }
 

internal/infra/cache/cache.go

@@ -413,6 +413,28 @@ func (c *Cache) ResetWithThreshold(threshold, raftIndex uint64) {
 	}
 }
 
+// RealignGeneration aligns currentGeneration and BaseIndex to the generation
+// raftIndex falls into under the current threshold, without touching cache
+// data or the epoch. For use when no persisted generation meta exists but the
+// store's applied index is not at genesis — a restored store (PrepareForBackup
+// wipes the cache zone while preserving the applied index as the genesis
+// boundary), or a pre-meta store after an upgrade. In both, "meta absent"
+// does NOT imply generation 0: leaving 0 makes admission's CheckCache observe
+// Gen(appliedIndex+1) as 2+ generations ahead and reject every proposal as
+// CacheUnreachable — and no apply ever runs to rotate the generation forward.
+func (c *Cache) RealignGeneration(raftIndex uint64) {
+	threshold := c.generationThreshold.Load()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// Forward only: g == 0 needs no realignment (and genEndIndex(g-1) would
+	// underflow), and moving a generation backward is never correct.
+	if g := Gen(raftIndex, threshold); g > c.currentGeneration.Load() {
+		c.rotateLocked(genEndIndex(g-1, threshold), g)
+	}
+}
+
 // clearLocked clears all cache data without incrementing the epoch.
 func (c *Cache) clearLocked() {
 	for _, ac := range c.caches {