diff --git a/cmd/ledgerctl/store/bootstrap.go b/cmd/ledgerctl/store/bootstrap.go
index 75eddb9843..ef01793575 100644
--- a/cmd/ledgerctl/store/bootstrap.go
+++ b/cmd/ledgerctl/store/bootstrap.go
@@ -3,7 +3,6 @@ package store
 import (
 	"bufio"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -61,14 +60,20 @@ func runBootstrap(cmd *cobra.Command, _ []string) error {
 		yes, _      = cmd.Flags().GetBool("yes")
 	)
 
-	// Ensure data directory is fresh (no existing checkpoints).
-	_, hasCheckpoint, err := dal.ScanLatestCheckpointID(dataDir)
-	if err != nil {
-		return fmt.Errorf("scanning data directory: %w", err)
+	// Ensure the data directory is fresh: no RESTORED marker, no live/
+	// database, no checkpoints (normal startup prefers live/ over the
+	// restored checkpoint, silently booting the stale store under the
+	// marker's boundary). Marker first: ValidateFreshRestoreTarget's reclaim
+	// of a half-finalized checkpoint is only safe once the marker is known
+	// absent.
+	if marker, err := node.ReadRestoredMarker(dataDir); err != nil {
+		return fmt.Errorf("checking for RESTORED marker: %w", err)
+	} else if marker != nil {
+		return fmt.Errorf("data directory %s already contains a RESTORED marker; refusing to overwrite", dataDir)
 	}
 
-	if hasCheckpoint {
-		return fmt.Errorf("data directory %s already contains checkpoints; refusing to overwrite", dataDir)
+	if err := dal.ValidateFreshRestoreTarget(dataDir); err != nil {
+		return err
 	}
 
 	storageCfg, err := cmdutil.BackupStorageConfigFromFlags(cmd)
@@ -313,20 +318,19 @@ func runBootstrap(cmd *cobra.Command, _ []string) error {
 		return fmt.Errorf("hard linking staging to checkpoint: %w", err)
 	}
 
-	// Write RESTORED marker
-	marker := node.RestoredMarker{
+	// The marker is the restore's commit point — written only after the
+	// checkpoint is in place, atomically.
+	if err := node.WriteRestoredMarker(dataDir, node.RestoredMarker{
 		LastAppliedIndex:     lastAppliedIndex,
 		LastAppliedTimestamp: lastAppliedTimestamp,
-	}
-
-	markerData, err := json.Marshal(marker)
-	if err != nil {
-		return fmt.Errorf("marshaling restored marker: %w", err)
-	}
+	}); err != nil {
+		// Roll the placement back: a checkpoint without its marker would
+		// make the freshness guard refuse a re-run of this command.
+		if rmErr := os.RemoveAll(checkpointPath); rmErr != nil {
+			pterm.Warning.Printfln("Failed to remove checkpoint after marker write failure; delete %s manually before retrying: %v", checkpointPath, rmErr)
+		}
 
-	markerPath := filepath.Join(dataDir, "RESTORED")
-	if err := os.WriteFile(markerPath, markerData, 0o644); err != nil {
-		return fmt.Errorf("writing restored marker: %w", err)
+		return err
 	}
 
 	// Remove staging directory
diff --git a/docs/ops/backup-restore.md b/docs/ops/backup-restore.md
index dbd8528780..029d83d720 100644
--- a/docs/ops/backup-restore.md
+++ b/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. |
diff --git a/docs/ops/cli.md b/docs/ops/cli.md
index 718a522e47..323fa73b75 100644
--- a/docs/ops/cli.md
+++ b/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
 
diff --git a/docs/technical/architecture/subsystems/fsm/deterministic-fsm.md b/docs/technical/architecture/subsystems/fsm/deterministic-fsm.md
index 8157777777..be7d72b83b 100644
--- a/docs/technical/architecture/subsystems/fsm/deterministic-fsm.md
+++ b/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
 
diff --git a/internal/adapter/grpc/server_restore.go b/internal/adapter/grpc/server_restore.go
index 5cb66b827b..9c3c1d3ee6 100644
--- a/internal/adapter/grpc/server_restore.go
+++ b/internal/adapter/grpc/server_restore.go
@@ -2,7 +2,6 @@ package grpc
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -325,7 +324,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 {
@@ -353,22 +354,6 @@ func (s *RestoreServiceServerImpl) FinalizeRestore(_ context.Context, _ *restore
 	s.closeStagingStore()
 	s.mu.Unlock()
 
-	// Write RESTORED marker
-	marker := node.RestoredMarker{
-		LastAppliedIndex:     lastAppliedIndex,
-		LastAppliedTimestamp: lastAppliedTimestamp,
-	}
-
-	markerData, err := json.Marshal(marker)
-	if err != nil {
-		return nil, fmt.Errorf("marshaling restored marker: %w", err)
-	}
-
-	markerPath := filepath.Join(s.dataDir, "RESTORED")
-	if err := os.WriteFile(markerPath, markerData, 0o644); err != nil {
-		return nil, fmt.Errorf("writing restored marker: %w", err)
-	}
-
 	// Move staging to checkpoint 0
 	checkpointsDir := filepath.Join(s.dataDir, "checkpoints")
 	if err := os.MkdirAll(checkpointsDir, 0o755); err != nil {
@@ -384,6 +369,26 @@ func (s *RestoreServiceServerImpl) FinalizeRestore(_ context.Context, _ *restore
 		return nil, fmt.Errorf("hard linking staging to checkpoint: %w", err)
 	}
 
+	// The marker is the restore's commit point, so it goes in only after the
+	// checkpoint is in place: bootstrap plants the raft genesis snapshot at
+	// the marker's boundary, which only describes THIS checkpoint's store.
+	if err := node.WriteRestoredMarker(s.dataDir, node.RestoredMarker{
+		LastAppliedIndex:     lastAppliedIndex,
+		LastAppliedTimestamp: lastAppliedTimestamp,
+	}); err != nil {
+		// Roll the placement back: a checkpoint without its marker is a
+		// half-state the freshness guards would refuse to retry against
+		// (and the staging handle is already closed, so an in-process
+		// retry cannot re-finalize either). Undone, a restore-mode
+		// restart can run the download + finalize again.
+		if rmErr := os.RemoveAll(checkpointPath); rmErr != nil {
+			s.logger.WithFields(map[string]any{"error": rmErr}).
+				Errorf("Failed to remove checkpoint after marker write failure; delete %s manually before retrying", checkpointPath)
+		}
+
+		return nil, err
+	}
+
 	// Remove staging directory
 	if err := os.RemoveAll(stagingDir); err != nil {
 		s.logger.WithFields(map[string]any{"error": err}).Errorf("Failed to remove staging directory")
diff --git a/internal/bootstrap/module_restore.go b/internal/bootstrap/module_restore.go
index 0f5d24c625..3e311d86ad 100644
--- a/internal/bootstrap/module_restore.go
+++ b/internal/bootstrap/module_restore.go
@@ -15,6 +15,7 @@ import (
 
 	grpcadp "github.com/formancehq/ledger/v3/internal/adapter/grpc"
 	"github.com/formancehq/ledger/v3/internal/infra/monitoring/otlplogs"
+	"github.com/formancehq/ledger/v3/internal/infra/node"
 	"github.com/formancehq/ledger/v3/internal/storage/dal"
 )
 
@@ -52,18 +53,21 @@ func RestoreModule() fx.Option {
 			},
 		),
 		fx.Invoke(
-			// Validate that the data directory is fresh (no existing checkpoints)
+			// Validate that the data directory is fresh: no checkpoints, no
+			// live/ database (normal startup prefers it over the restored
+			// checkpoint, silently booting the stale store under the
+			// marker's boundary), no leftover RESTORED marker.
 			func(cfg Config) error {
-				_, hasCheckpoint, err := dal.ScanLatestCheckpointID(cfg.DataDir)
-				if err != nil {
-					return fmt.Errorf("scanning data directory: %w", err)
-				}
-
-				if hasCheckpoint {
-					return fmt.Errorf("restore mode requires a fresh data directory; checkpoints already exist in %s", cfg.DataDir)
+				// Marker first: ValidateFreshRestoreTarget's reclaim of a
+				// half-finalized checkpoint is only safe once the marker is
+				// known absent.
+				if marker, err := node.ReadRestoredMarker(cfg.DataDir); err != nil {
+					return fmt.Errorf("checking for RESTORED marker: %w", err)
+				} else if marker != nil {
+					return fmt.Errorf("restore mode requires a fresh data directory: a RESTORED marker already exists in %s", cfg.DataDir)
 				}
 
-				return nil
+				return dal.ValidateFreshRestoreTarget(cfg.DataDir)
 			},
 			// Register health service on ServiceServer
 			func(serviceServer *grpcadp.ServiceServer) {
diff --git a/internal/infra/attributes/prepare.go b/internal/infra/attributes/prepare.go
index 191f14bf2f..c4eb533bd8 100644
--- a/internal/infra/attributes/prepare.go
+++ b/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,60 @@ 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.
+	// Read the raw key: only genuine absence may fall back — a present but
+	// malformed value is a corrupt checkpoint and must fail closed, not be
+	// silently rewritten as the genesis fallback.
+	var genesisBoundary uint64
+
+	switch val, closer, err := s.Get([]byte{dal.ZoneGlobal, dal.SubGlobLastAppliedIndex}); {
+	case err == nil:
+		if len(val) != 8 {
+			_ = closer.Close()
+
+			return fmt.Errorf("invariant: checkpoint applied index has %d bytes (expected 8) — corrupt checkpoint", len(val))
+		}
+
+		genesisBoundary = binary.BigEndian.Uint64(val)
+		if err := closer.Close(); err != nil {
+			return fmt.Errorf("closing applied index read: %w", err)
+		}
+	case errors.Is(err, pebble.ErrNotFound):
+		// Genesis checkpoint: the key has never been written.
+	default:
+		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.
diff --git a/internal/infra/attributes/prepare_test.go b/internal/infra/attributes/prepare_test.go
index db04ffba34..3b8fa6a659 100644
--- a/internal/infra/attributes/prepare_test.go
+++ b/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,67 @@ 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")
+}
+
+// TestPrepareForBackupRejectsMalformedAppliedIndex asserts a present but
+// wrong-length applied index fails closed: only genuine absence may take the
+// genesis fallback — silently treating a corrupt value as 0 would rewrite it
+// as boundary 1 and misalign the restored raft genesis.
+func TestPrepareForBackupRejectsMalformedAppliedIndex(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, batch.SetBytes([]byte{dal.ZoneGlobal, dal.SubGlobLastAppliedIndex}, []byte{0x01, 0x02, 0x03}))
+	require.NoError(t, batch.Commit())
+
+	err = PrepareForBackup(s)
+	require.ErrorContains(t, err, "corrupt checkpoint")
+}
+
 // 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 +301,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 +382,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")
 	}()
 }
 
diff --git a/internal/infra/cache/cache.go b/internal/infra/cache/cache.go
index 7926f1c064..61069ce51f 100644
--- a/internal/infra/cache/cache.go
+++ b/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 {
diff --git a/internal/infra/node/node.go b/internal/infra/node/node.go
index d1d7c938f4..a49ed5f8ce 100644
--- a/internal/infra/node/node.go
+++ b/internal/infra/node/node.go
@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"math"
 	"slices"
 	"strings"
 	"sync"
@@ -284,10 +285,15 @@ func NewNode(
 		}
 
 		if marker != nil {
-			// Restore mode: bootstrap from restored data.
-			// The backup was compacted: all attribute indices are 0 and lastAppliedIndex is 0.
-			// We need to recover the FSM counters (nextLedgerID, nextSequenceID, etc.)
-			// from the Pebble data before creating the WAL snapshot.
+			// Restore mode: bootstrap from restored data. The restored store
+			// carries the genesis boundary PrepareForBackup established (the
+			// checkpoint's applied index, >= 1), so the WAL snapshot below
+			// lands there and the new log starts just above it: raft then has
+			// to route any fresh peer through the snapshot → checkpoint-sync
+			// path instead of "catching it up" by replaying the log onto an
+			// empty store, which would miss the entire restored FSM genesis.
+			// FSM counters (nextLedgerID, nextSequenceID, etc.) are recovered
+			// from Pebble before creating the WAL snapshot.
 			logger.WithFields(map[string]any{
 				"lastAppliedIndex":     marker.LastAppliedIndex,
 				"lastAppliedTimestamp": marker.LastAppliedTimestamp,
@@ -297,6 +303,22 @@ func NewNode(
 				return nil, fmt.Errorf("recovering FSM state from store: %w", err)
 			}
 
+			// A snapshot at 0 is no snapshot: the new log would claim
+			// completeness from index 1 and a fresh learner could be caught
+			// up by plain log replay onto an empty store, missing the whole
+			// restored genesis. PrepareForBackup guarantees a boundary >= 1;
+			// a 0 here means a marker written by another tool or by hand.
+			if marker.LastAppliedIndex == 0 {
+				return nil, errors.New("invariant: RESTORED marker carries lastAppliedIndex 0; the restored genesis must occupy index >= 1 so joiners are forced through checkpoint sync — re-run restore finalize / store bootstrap to regenerate the marker")
+			}
+
+			// Raft paths compute boundary+1 (WAL FirstIndex, FSM gap check),
+			// which would wrap. PrepareForBackup refuses this at finalize;
+			// here it means a hand-written or corrupt marker.
+			if marker.LastAppliedIndex == math.MaxUint64 {
+				return nil, errors.New("invariant: RESTORED marker carries lastAppliedIndex MaxUint64 — corrupt marker; re-run restore finalize / store bootstrap to regenerate it")
+			}
+
 			initialConfState = &raftpb.ConfState{
 				Voters: []uint64{cfg.NodeID},
 			}
diff --git a/internal/infra/node/restored_marker.go b/internal/infra/node/restored_marker.go
index fd697b89ce..5468ed7a62 100644
--- a/internal/infra/node/restored_marker.go
+++ b/internal/infra/node/restored_marker.go
@@ -12,6 +12,13 @@ const restoredMarkerFile = "RESTORED"
 
 // RestoredMarker contains the metadata written by FinalizeRestore.
 type RestoredMarker struct {
+	// LastAppliedIndex is the raft index the restored genesis occupies in the
+	// new log — the checkpoint's applied index, or 1 for a genesis checkpoint
+	// (PrepareForBackup guarantees >= 1). Bootstrap plants the WAL snapshot at
+	// this index so joiners must sync a checkpoint, and refuses a marker
+	// carrying 0. A boundary label only: after a full + incremental restore
+	// the restored state is NEWER than this index (exports are
+	// sequence-keyed and do not advance it).
 	LastAppliedIndex     uint64 `json:"lastAppliedIndex"`
 	LastAppliedTimestamp uint64 `json:"lastAppliedTimestamp"`
 }
@@ -38,6 +45,31 @@ func ReadRestoredMarker(dataDir string) (*RestoredMarker, error) {
 	return &marker, nil
 }
 
+// WriteRestoredMarker atomically writes the RESTORED marker (temp file +
+// rename) to the data directory. Callers must place the restored checkpoint
+// BEFORE writing the marker: the marker is the commit point of the restore —
+// a marker observed without its checkpoint would make bootstrap plant the
+// restore WAL snapshot over whatever store the next boot opens.
+func WriteRestoredMarker(dataDir string, marker RestoredMarker) error {
+	data, err := json.Marshal(marker)
+	if err != nil {
+		return fmt.Errorf("marshaling restored marker: %w", err)
+	}
+
+	tmpPath := filepath.Join(dataDir, restoredMarkerFile+".tmp")
+	if err := os.WriteFile(tmpPath, data, 0o644); err != nil {
+		return fmt.Errorf("writing restored marker temp file: %w", err)
+	}
+
+	if err := os.Rename(tmpPath, filepath.Join(dataDir, restoredMarkerFile)); err != nil {
+		_ = os.Remove(tmpPath)
+
+		return fmt.Errorf("renaming restored marker into place: %w", err)
+	}
+
+	return nil
+}
+
 // RemoveRestoredMarker removes the RESTORED marker file from the data directory.
 func RemoveRestoredMarker(dataDir string) error {
 	markerPath := filepath.Join(dataDir, restoredMarkerFile)
diff --git a/internal/infra/state/cache_snapshotter.go b/internal/infra/state/cache_snapshotter.go
index 0bd82beaa0..6903ff1345 100644
--- a/internal/infra/state/cache_snapshotter.go
+++ b/internal/infra/state/cache_snapshotter.go
@@ -20,6 +20,7 @@ import (
 	"github.com/formancehq/ledger/v3/internal/pkg/worker"
 	"github.com/formancehq/ledger/v3/internal/proto/commonpb"
 	"github.com/formancehq/ledger/v3/internal/proto/raftcmdpb"
+	"github.com/formancehq/ledger/v3/internal/query"
 	"github.com/formancehq/ledger/v3/internal/storage/dal"
 )
 
@@ -432,8 +433,9 @@ func (s *CacheSnapshotter) RestoreFromStore(store dal.RecoveryReader) error {
 	}
 
 	// Read cache-level metadata if present. Pre-rotation, this key does not
-	// exist; default to currentGeneration=0.
+	// exist; default to currentGeneration=0 and realign below.
 	currentGen := uint64(0)
+	metaAbsent := false
 
 	metaVal, closer, err := store.Get([]byte{dal.ZoneCache, dal.SubCacheMeta})
 	switch {
@@ -452,8 +454,13 @@ func (s *CacheSnapshotter) RestoreFromStore(store dal.RecoveryReader) error {
 
 		currentGen = meta.GetCurrentGeneration()
 	case errors.Is(err, pebble.ErrNotFound):
-		// Pre-rotation: the meta key does not exist yet; currentGeneration
-		// stays 0. Only genuine absence (ErrNotFound) is treated as "no meta".
+		// The meta key is written on rotation, so absence means either a
+		// young store still in generation 0, or a store whose applied index
+		// is real but whose cache zone carries no meta — a restored store
+		// (PrepareForBackup preserves the applied index as the genesis
+		// boundary but wipes the cache zone) or a pre-meta upgrade. The
+		// realignment below disambiguates via the applied index.
+		metaAbsent = true
 	default:
 		// Any other Get/closer failure is a real read error, not absence:
 		// silently defaulting to gen0 would restore the wrong generation
@@ -486,9 +493,24 @@ func (s *CacheSnapshotter) RestoreFromStore(store dal.RecoveryReader) error {
 
 	s.registry.Cache.SetCurrentGeneration(currentGen)
 
+	// Absent meta with a non-genesis applied index: realign the generation
+	// to the one the applied index falls into. Leaving it at 0 would make
+	// admission's CheckCache see Gen(appliedIndex+1) as 2+ generations ahead
+	// and reject every proposal as CacheUnreachable — and with admission
+	// refusing all proposals, no apply ever runs CheckRotationNeeded to
+	// catch the generation up.
+	if metaAbsent {
+		appliedIndex, err := query.ReadLastAppliedIndex(reader)
+		if err != nil {
+			return fmt.Errorf("reading applied index for generation realignment: %w", err)
+		}
+
+		s.registry.Cache.RealignGeneration(appliedIndex)
+	}
+
 	s.logger.WithFields(map[string]any{
 		"duration":          time.Since(restoreStart).String(),
-		"currentGeneration": currentGen,
+		"currentGeneration": s.registry.Cache.CurrentGeneration(),
 	}).Infof("Restored cache from Pebble")
 
 	// Restore bloom filters: load persisted blocks from Pebble, then replay
diff --git a/internal/infra/state/cache_snapshotter_test.go b/internal/infra/state/cache_snapshotter_test.go
index f12f92ea1b..d8c1304c42 100644
--- a/internal/infra/state/cache_snapshotter_test.go
+++ b/internal/infra/state/cache_snapshotter_test.go
@@ -296,6 +296,34 @@ func TestCacheSnapshotter_RestoreFromEmptyStore(t *testing.T) {
 	require.NoError(t, snapshotter.RestoreFromStore(dataStore))
 }
 
+// TestCacheSnapshotter_RestoreRealignsGenerationWhenMetaAbsent pins the
+// restored-store boot: PrepareForBackup preserves the applied index as the
+// genesis boundary but wipes the cache zone, so RestoreFromStore finds no
+// generation meta while the applied index sits far past generation 0. The
+// restore must realign the in-memory generation to that horizon — left at 0,
+// admission's CheckCache would see Gen(appliedIndex+1) as 2+ generations
+// ahead and classify proposals as CacheUnreachable until the first applied
+// entry (the election no-op) rotates the generation forward.
+func TestCacheSnapshotter_RestoreRealignsGenerationWhenMetaAbsent(t *testing.T) {
+	t.Parallel()
+
+	snapshotter, dataStore, registry := newTestCacheSnapshotter(t, nil)
+
+	batch := dataStore.OpenWriteSession()
+	require.NoError(t, SetAppliedIndex(batch, 5000))
+	require.NoError(t, batch.Commit())
+
+	require.NoError(t, snapshotter.RestoreFromStore(dataStore))
+
+	// The harness threshold is 1000 (newTestCacheSnapshotter): Gen(5000) = 4,
+	// whose canonical boundary is genEndIndex(3) = 4000.
+	require.Equal(t, uint64(4), registry.Cache.CurrentGeneration(),
+		"generation must realign to the preserved applied horizon")
+	require.Equal(t, uint64(4000), registry.Cache.BaseIndex.Gen0,
+		"gen0 base must sit at the canonical boundary of the realigned generation")
+	require.Equal(t, uint64(0), registry.Cache.BaseIndex.Gen1)
+}
+
 func TestCacheSnapshotter_PersistAndRestoreWithBloomFilters(t *testing.T) {
 	t.Parallel()
 
diff --git a/internal/proto/restorepb/restore.pb.go b/internal/proto/restorepb/restore.pb.go
index 8801b3d72c..92ebc93a9f 100644
--- a/internal/proto/restorepb/restore.pb.go
+++ b/internal/proto/restorepb/restore.pb.go
@@ -661,7 +661,7 @@ func (*PreviewRestoreRequest) Descriptor() ([]byte, []int) {
 
 type PreviewRestoreResponse struct {
 	state                protoimpl.MessageState `protogen:"open.v1"`
-	LastAppliedIndex     uint64                 `protobuf:"fixed64,1,opt,name=last_applied_index,json=lastAppliedIndex,proto3" json:"last_applied_index,omitempty"`             // Last Raft index applied
+	LastAppliedIndex     uint64                 `protobuf:"fixed64,1,opt,name=last_applied_index,json=lastAppliedIndex,proto3" json:"last_applied_index,omitempty"`             // Staged checkpoint's applied index, pre-preparation (may be 0 for a genesis checkpoint; finalize preserves it as the restore genesis boundary, 0 falling back to 1)
 	LastAppliedTimestamp uint64                 `protobuf:"fixed64,2,opt,name=last_applied_timestamp,json=lastAppliedTimestamp,proto3" json:"last_applied_timestamp,omitempty"` // Last applied timestamp (HLC microseconds)
 	LastSequence         uint64                 `protobuf:"fixed64,3,opt,name=last_sequence,json=lastSequence,proto3" json:"last_sequence,omitempty"`                           // Last log sequence number
 	LedgerCount          uint32                 `protobuf:"varint,4,opt,name=ledger_count,json=ledgerCount,proto3" json:"ledger_count,omitempty"`                               // Number of ledgers in the backup
diff --git a/internal/proto/restorepb/restore_grpc.pb.go b/internal/proto/restorepb/restore_grpc.pb.go
index ed990087db..d92597f559 100644
--- a/internal/proto/restorepb/restore_grpc.pb.go
+++ b/internal/proto/restorepb/restore_grpc.pb.go
@@ -52,7 +52,9 @@ type RestoreServiceClient interface {
 	ValidateRestore(ctx context.Context, in *ValidateRestoreRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ValidateRestoreEvent], error)
 	// PreviewRestore returns a summary of the staged backup data.
 	PreviewRestore(ctx context.Context, in *PreviewRestoreRequest, opts ...grpc.CallOption) (*PreviewRestoreResponse, error)
-	// FinalizeRestore compacts attributes, commits the staged backup as the live data.
+	// FinalizeRestore prepares the staged backup (applied index preserved as the
+	// restore genesis boundary, cluster-local state reset) and commits it as the
+	// live data.
 	FinalizeRestore(ctx context.Context, in *FinalizeRestoreRequest, opts ...grpc.CallOption) (*FinalizeRestoreResponse, error)
 }
 
@@ -158,7 +160,9 @@ type RestoreServiceServer interface {
 	ValidateRestore(*ValidateRestoreRequest, grpc.ServerStreamingServer[ValidateRestoreEvent]) error
 	// PreviewRestore returns a summary of the staged backup data.
 	PreviewRestore(context.Context, *PreviewRestoreRequest) (*PreviewRestoreResponse, error)
-	// FinalizeRestore compacts attributes, commits the staged backup as the live data.
+	// FinalizeRestore prepares the staged backup (applied index preserved as the
+	// restore genesis boundary, cluster-local state reset) and commits it as the
+	// live data.
 	FinalizeRestore(context.Context, *FinalizeRestoreRequest) (*FinalizeRestoreResponse, error)
 	mustEmbedUnimplementedRestoreServiceServer()
 }
diff --git a/internal/storage/dal/fresh_restore_target_test.go b/internal/storage/dal/fresh_restore_target_test.go
new file mode 100644
index 0000000000..1b9048723a
--- /dev/null
+++ b/internal/storage/dal/fresh_restore_target_test.go
@@ -0,0 +1,81 @@
+package dal_test
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/formancehq/ledger/v3/internal/storage/dal"
+)
+
+// TestValidateFreshRestoreTarget pins the restore freshness guard: a restore
+// stages a checkpoint whose RESTORED marker the next boot aligns the raft
+// genesis snapshot with, but normal startup prefers an existing live/
+// database over checkpoints — so any pre-existing store state in the target
+// directory would boot under the wrong boundary and must be refused.
+func TestValidateFreshRestoreTarget(t *testing.T) {
+	t.Parallel()
+
+	t.Run("fresh directory passes", func(t *testing.T) {
+		t.Parallel()
+		require.NoError(t, dal.ValidateFreshRestoreTarget(t.TempDir()))
+	})
+
+	t.Run("missing directory passes", func(t *testing.T) {
+		t.Parallel()
+		require.NoError(t, dal.ValidateFreshRestoreTarget(filepath.Join(t.TempDir(), "not-created-yet")))
+	})
+
+	t.Run("existing checkpoint is refused", func(t *testing.T) {
+		t.Parallel()
+
+		dir := t.TempDir()
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "checkpoints", "3"), 0o755))
+		require.ErrorContains(t, dal.ValidateFreshRestoreTarget(dir), "checkpoints already exist")
+	})
+
+	t.Run("orphaned checkpoint 0 is reclaimed", func(t *testing.T) {
+		// checkpoints/0 with no live database (and the caller having verified
+		// no RESTORED marker) is a finalize that died between checkpoint
+		// placement and the marker — the restore must be retryable, not
+		// permanently refused.
+		t.Parallel()
+
+		dir := t.TempDir()
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "checkpoints", "0"), 0o755))
+		require.NoError(t, dal.ValidateFreshRestoreTarget(dir))
+
+		_, err := os.Stat(filepath.Join(dir, "checkpoints"))
+		require.True(t, os.IsNotExist(err), "the orphaned checkpoint must be reclaimed")
+	})
+
+	t.Run("checkpoint 0 alongside higher checkpoints is refused", func(t *testing.T) {
+		t.Parallel()
+
+		dir := t.TempDir()
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "checkpoints", "0"), 0o755))
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "checkpoints", "5"), 0o755))
+		require.ErrorContains(t, dal.ValidateFreshRestoreTarget(dir), "checkpoints already exist")
+	})
+
+	t.Run("checkpoint 0 next to live is refused", func(t *testing.T) {
+		t.Parallel()
+
+		dir := t.TempDir()
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "checkpoints", "0"), 0o755))
+		require.NoError(t, os.MkdirAll(filepath.Join(dir, "live"), 0o755))
+		require.ErrorContains(t, dal.ValidateFreshRestoreTarget(dir), "live/ already exists")
+	})
+
+	for _, stale := range []string{"live", "live.staging", "live.discard"} {
+		t.Run("existing "+stale+" is refused", func(t *testing.T) {
+			t.Parallel()
+
+			dir := t.TempDir()
+			require.NoError(t, os.MkdirAll(filepath.Join(dir, stale), 0o755))
+			require.ErrorContains(t, dal.ValidateFreshRestoreTarget(dir), stale+"/ already exists")
+		})
+	}
+}
diff --git a/internal/storage/dal/store.go b/internal/storage/dal/store.go
index 99cb5f1b8b..b3d023617a 100644
--- a/internal/storage/dal/store.go
+++ b/internal/storage/dal/store.go
@@ -94,6 +94,49 @@ func ScanLatestCheckpointID(dataDir string) (latestID uint64, found bool, err er
 	return latestID, found, nil
 }
 
+// ValidateFreshRestoreTarget reports whether dataDir is safe to restore into,
+// reclaiming the one tolerable leftover. A restore stages a checkpoint and a
+// RESTORED marker; the next boot aligns the new raft log's WAL snapshot with
+// the store the marker describes. Normal startup prefers an existing live/
+// database over checkpoints, so restoring into a dir that already carries one
+// (or leftovers of an in-flight RestoreCheckpoint) would boot the STALE store
+// under the marker's boundary — a silent store/marker misalignment.
+//
+// The tolerated exception: exactly checkpoints/0 with no live database and —
+// the caller must have verified this first — no RESTORED marker. Only a
+// finalize that died between checkpoint placement and its marker (the
+// restore's commit point) produces that state; the orphan is removed so the
+// restore can run again instead of being permanently refused. Any other
+// checkpoint layout is refused.
+func ValidateFreshRestoreTarget(dataDir string) error {
+	for _, dir := range []string{liveDir, liveStagingDir, liveDiscardDir} {
+		if _, err := os.Stat(filepath.Join(dataDir, dir)); err == nil {
+			return fmt.Errorf("restore requires a fresh data directory: %s/ already exists in %s", dir, dataDir)
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return fmt.Errorf("checking %s directory: %w", dir, err)
+		}
+	}
+
+	latestID, hasCheckpoint, err := ScanLatestCheckpointID(dataDir)
+	if err != nil {
+		return fmt.Errorf("scanning data directory: %w", err)
+	}
+
+	if !hasCheckpoint {
+		return nil
+	}
+
+	if latestID != 0 {
+		return fmt.Errorf("restore requires a fresh data directory: checkpoints already exist in %s", dataDir)
+	}
+
+	if err := os.RemoveAll(filepath.Join(dataDir, checkpointsDir)); err != nil {
+		return fmt.Errorf("reclaiming half-finalized restore checkpoint: %w", err)
+	}
+
+	return nil
+}
+
 // Store is a Pebble implementation of dal.Store
 // It stores balances and account metadata.
 type Store struct {
diff --git a/misc/proto/restore.proto b/misc/proto/restore.proto
index 67d953c254..c1d69c32d9 100644
--- a/misc/proto/restore.proto
+++ b/misc/proto/restore.proto
@@ -36,7 +36,9 @@ service RestoreService {
   // PreviewRestore returns a summary of the staged backup data.
   rpc PreviewRestore(PreviewRestoreRequest) returns (PreviewRestoreResponse);
 
-  // FinalizeRestore compacts attributes, commits the staged backup as the live data.
+  // FinalizeRestore prepares the staged backup (applied index preserved as the
+  // restore genesis boundary, cluster-local state reset) and commits it as the
+  // live data.
   rpc FinalizeRestore(FinalizeRestoreRequest) returns (FinalizeRestoreResponse);
 }
 
@@ -121,7 +123,7 @@ message ValidateRestoreProgress {
 message PreviewRestoreRequest {}
 
 message PreviewRestoreResponse {
-  fixed64 last_applied_index = 1;      // Last Raft index applied
+  fixed64 last_applied_index = 1;      // Staged checkpoint's applied index, pre-preparation (may be 0 for a genesis checkpoint; finalize preserves it as the restore genesis boundary, 0 falling back to 1)
   fixed64 last_applied_timestamp = 2;  // Last applied timestamp (HLC microseconds)
   fixed64 last_sequence = 3;           // Last log sequence number
   uint32 ledger_count = 4;            // Number of ledgers in the backup
diff --git a/tests/e2e/cluster/restore_test.go b/tests/e2e/cluster/restore_test.go
index a21d7133c3..5559650c5e 100644
--- a/tests/e2e/cluster/restore_test.go
+++ b/tests/e2e/cluster/restore_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/formancehq/go-libs/v5/pkg/testing/testservice"
 	cmdserver "github.com/formancehq/ledger/v3/cmd/server"
 	"github.com/formancehq/ledger/v3/internal/infra/backup"
+	"github.com/formancehq/ledger/v3/internal/infra/node"
 	"github.com/formancehq/ledger/v3/internal/proto/clusterpb"
 	"github.com/formancehq/ledger/v3/internal/proto/commonpb"
 	"github.com/formancehq/ledger/v3/internal/proto/restorepb"
@@ -32,6 +33,7 @@ import (
 	"github.com/testcontainers/testcontainers-go/wait"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
 )
 
 const (
@@ -81,9 +83,9 @@ func newRestoreGRPCClient(grpcPort int) (restorepb.RestoreServiceClient, *grpc.C
 
 var _ = Describe("Restore", Ordered, func() {
 	const (
-		httpPort   = testutil.TestSingleHTTPPort
-		grpcPort   = testutil.TestSingleGRPCPort
-		raftPort   = grpcPort - 1000
+		httpPort    = testutil.TestSingleHTTPPort
+		grpcPort    = testutil.TestSingleGRPCPort
+		raftPort    = grpcPort - 1000
 		ledgerName  = "restore-ledger"
 		ledger2     = "restore-ledger-2"
 		chartLedger = "restore-chart-ledger"
@@ -532,6 +534,28 @@ var _ = Describe("Restore", Ordered, func() {
 			_, err = os.Stat(restoreDataDir + "/checkpoints/0")
 			Expect(err).To(Succeed(), "checkpoint 0 directory should exist")
 		})
+
+		It("should carry the checkpoint's applied index in the marker as the genesis boundary", func() {
+			// The restored bootstrap plants its WAL snapshot at the marker
+			// index, and the FSM gap check requires the first post-restore
+			// entry at exactly boundary+1 — so a marker diverging from the
+			// store, or a boundary of 0/1 here, would either fail Phase 3
+			// outright or silently re-open the learner log-replay hole.
+			data, err := os.ReadFile(restoreDataDir + "/RESTORED")
+			Expect(err).To(Succeed())
+
+			var marker node.RestoredMarker
+			Expect(json.Unmarshal(data, &marker)).To(Succeed())
+
+			manifest, err := readS3Manifest(ctx, s3Client)
+			Expect(err).To(Succeed())
+			Expect(manifest.Checkpoint).NotTo(BeNil())
+
+			Expect(marker.LastAppliedIndex).To(Equal(manifest.Checkpoint.LastAppliedIndex),
+				"marker must preserve the checkpoint's applied index")
+			Expect(marker.LastAppliedIndex).To(BeNumerically(">", 1),
+				"this suite's checkpoint is taken after real traffic, so the preserved boundary must exercise the non-fallback (N > 1) path")
+		})
 	})
 
 	// Phase 3: Restart a normal server on the restored data and verify all data.
@@ -556,6 +580,25 @@ var _ = Describe("Restore", Ordered, func() {
 				Output:    GinkgoWriter,
 			})
 			instruments = append(instruments, testserver.WithBootstrap())
+			// No background raft snapshot may exist when the learner-join spec
+			// below runs: it exercises the window where the leader's log is
+			// the only catch-up source, and a maintenance snapshot would
+			// legitimately force the MsgSnap path and mask a log-replay
+			// regression. The interval override wins over the default (pflag
+			// keeps the last occurrence); the margin blocks the snapshot
+			// trigger outright.
+			instruments = append(instruments,
+				testserver.WithMaintenanceInterval(time.Hour),
+				testserver.WithRaftCompactionMargin(1_000_000),
+			)
+			// A rotation threshold far below the preserved genesis boundary:
+			// admission's CheckCache compares Gen(boundary+1) against the
+			// in-memory generation, and the restored store carries no
+			// persisted generation meta (PrepareForBackup wipes the cache
+			// zone) — the boot-side realignment is what keeps proposals
+			// admissible; without it every write here trips the
+			// CacheUnreachable horizon.
+			instruments = append(instruments, testserver.WithCacheRotationThreshold(3))
 
 			server = testservice.New(cmdserver.NewRunCommand,
 				testservice.WithInstruments(instruments...),
@@ -787,5 +830,90 @@ var _ = Describe("Restore", Ordered, func() {
 			Expect(err).To(Succeed())
 			Expect(charlieResp.Volumes["USD"].Input).To(Equal("1000"))
 		})
+
+		It("should transfer the restored state to a learner joining before any raft snapshot", func() {
+			// A restored node's FSM genesis is the whole backup, but its raft
+			// log is brand new. Unless bootstrap plants a snapshot above the
+			// log start, the leader catches a fresh learner up by plain log
+			// replay from index 1 — the learner applies the few post-restore
+			// entries onto an EMPTY store and silently misses every restored
+			// row (found by the Antithesis model test as an aggregated volume
+			// imbalance on the joiner). The join must instead be forced
+			// through the snapshot → checkpoint-sync path.
+			const (
+				joinerGRPCPort = grpcPort + 7
+				joinerHTTPPort = httpPort + 7
+				joinerRaftPort = raftPort + 7
+			)
+
+			// Committed on the restored leader only: its replication to the
+			// learner proves the learner finished catching up on the
+			// post-restore log. Its visibility is NOT proof of a correct
+			// join — every proposal ships its cache preload, so entries
+			// re-materialize the rows they touch even on a hollow store.
+			// That masking is exactly why alice below is the real check.
+			_, err := client.Apply(ctx, servicepb.UnsignedApplyRequest("", actions.CreateTransactionAction(ledgerName, []*commonpb.Posting{
+				actions.NewPosting("world", "join-fence", big.NewInt(42), "USD"),
+			}, nil, nil)))
+			Expect(err).To(Succeed())
+
+			instruments := testserver.DefaultTestInstruments(testserver.TestNodeConfig{
+				NodeID:    2,
+				ClusterID: "test-cluster",
+				HTTPPort:  joinerHTTPPort,
+				RaftPort:  joinerRaftPort,
+				GRPCPort:  joinerGRPCPort,
+				WalDir:    GinkgoT().TempDir(),
+				DataDir:   GinkgoT().TempDir(),
+				Debug:     testutil.Debug,
+				Output:    GinkgoWriter,
+			})
+			instruments = append(instruments, testserver.WithJoin(fmt.Sprintf("127.0.0.1:%d", raftPort)))
+
+			joiner := testservice.New(cmdserver.NewRunCommand,
+				testservice.WithInstruments(instruments...),
+			)
+			Expect(joiner.Start(ctx)).To(Succeed())
+			DeferCleanup(func() {
+				stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+				defer cancel()
+				_ = joiner.Stop(stopCtx)
+			})
+
+			joinerClient, _, joinerConn, err := testutil.NewGRPCClient(joinerGRPCPort)
+			Expect(err).To(Succeed())
+			DeferCleanup(func() { _ = joinerConn.Close() })
+
+			// Stale consistency pins every read below to the learner's own
+			// store: linearizable reads on a syncing node transparently fall
+			// back to the leader (readCtrl's leader_fallback), which would
+			// let node 1 answer for the learner and pass this test without
+			// proving anything about the learner's state.
+			staleCtx := metadata.AppendToOutgoingContext(ctx, "x-consistency", "stale")
+
+			// This converges only once the learner itself applied the fence
+			// entry.
+			Eventually(func(g Gomega) {
+				resp, err := joinerClient.GetAccount(staleCtx, &servicepb.GetAccountRequest{Ledger: ledgerName, Address: "join-fence"})
+				g.Expect(err).To(Succeed())
+				g.Expect(resp.GetVolumes()["USD"].GetInput()).To(Equal("42"))
+			}).Within(60*time.Second).ProbeEvery(500*time.Millisecond).Should(Succeed(),
+				"learner never caught up on the post-restore raft log")
+
+			// alice was written only BEFORE the backup, so no post-restore
+			// entry (and no preload) can re-materialize her: she is on the
+			// learner iff the restored store itself was transferred.
+			aliceResp, err := joinerClient.GetAccount(staleCtx, &servicepb.GetAccountRequest{Ledger: ledgerName, Address: "alice"})
+			Expect(err).To(Succeed())
+			Expect(aliceResp.GetVolumes()).To(HaveKey("USD"),
+				"learner caught up by log replay alone: the restored state never reached it")
+			Expect(aliceResp.GetVolumes()["USD"].GetInput()).To(Equal("3000"))
+
+			treasuryResp, err := joinerClient.GetAccount(staleCtx, &servicepb.GetAccountRequest{Ledger: ledger2, Address: "treasury"})
+			Expect(err).To(Succeed())
+			Expect(treasuryResp.GetVolumes()).To(HaveKey("EUR"),
+				"ledger untouched since the restore must still reach the learner")
+			Expect(treasuryResp.GetVolumes()["EUR"].GetInput()).To(Equal("50000"))
+		})
 	})
 })