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

cmd/ledgerctl/store/bootstrap.go

@@ -3,7 +3,6 @@ package store
 import (
 	"bufio"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -61,14 +60,18 @@ 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 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.
+	if err := dal.ValidateFreshRestoreTarget(dataDir); err != nil {
+		return err
 	}
 
-	if hasCheckpoint {
-		return fmt.Errorf("data directory %s already contains checkpoints; refusing to overwrite", dataDir)
+	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)
 	}
 
 	storageCfg, err := cmdutil.BackupStorageConfigFromFlags(cmd)
@@ -313,20 +316,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

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

@@ -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")

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,15 +53,19 @@ 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 err := dal.ValidateFreshRestoreTarget(cfg.DataDir); err != nil {
+					return err
 				}
 
-				if hasCheckpoint {
-					return fmt.Errorf("restore mode requires a fresh data directory; checkpoints already exist in %s", cfg.DataDir)
+				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

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.