Fix clean writer takeover after empty epochs

CHANGELOG.md

@@ -15,6 +15,8 @@
 
 ### Fixed
 
+- Filesystem writer takeover preserves unused LSNs after empty epochs and refuses unreconciled tails before changing the epoch ledger.
+
 - Incremental snapshot state roots now include the existing v1 domain separator and agree with canonical snapshot hashing. Previously emitted incorrect accumulator roots are not migrated.
 
 - The SPDX checker preserves Markdown frontmatter when it checks or repairs license headers.

crates/warp-core/src/causal_wal.rs

@@ -1586,7 +1586,7 @@ fn validate_writer_epoch_request(
                 if request.started_at_lsn <= final_lsn {
                     return Err(WalStoreError::WriterEpochLsnRegression);
                 }
-            } else if request.started_at_lsn <= previous_epoch.started_at_lsn {
+            } else if request.started_at_lsn < previous_epoch.started_at_lsn {
                 return Err(WalStoreError::WriterEpochLsnRegression);
             }
             if request.storage_fencing_token == previous_epoch.storage_fencing_token
@@ -5728,6 +5728,8 @@ impl FilesystemWalStore {
     /// reread. An active epoch left by a terminated process is closed under
     /// that lease before the successor is derived and admitted. A concurrently
     /// live writer retains the lease and prevents takeover.
+    /// Uncommitted or torn tails require writable recovery before takeover.
+    /// An empty predecessor leaves its starting LSN unused.
     pub fn acquire_fresh_writer_epoch(
         &mut self,
         minimum_started_at_lsn: Lsn,
@@ -5737,6 +5739,16 @@ impl FilesystemWalStore {
         }
         let writer_lock = acquire_writer_epoch_lock(&self.root)?;
         self.reload_writer_epoch_ledger()?;
+        let recovery = recover_filesystem_store(&self.root, RecoveryAccessMode::ReadOnly).map_err(
+            |error| match error {
+                WalRecoveryError::Store(error) => error,
+                WalRecoveryError::Validation(error) => WalStoreError::Validation(error),
+                WalRecoveryError::Index(error) => WalStoreError::RecoveryIndex(error),
+            },
+        )?;
+        if !matches!(recovery.tail_posture, RecoveryTailPosture::Clean) {
+            return Err(WalStoreError::SegmentHasUncommittedTail(self.segment_id));
+        }
 
         if self.active_epoch.is_some() {
             let previous_ledger = self.writer_epoch_ledger();
@@ -5761,11 +5773,12 @@ impl FilesystemWalStore {
             .and_then(|epoch| self.epoch_closures.get(&epoch.epoch_id))
             .copied()
             .unwrap_or_default();
-        let required_started_at_lsn = previous_closure
-            .final_lsn
-            .or_else(|| previous_epoch.map(|epoch| epoch.started_at_lsn))
-            .and_then(Lsn::checked_next)
-            .unwrap_or(minimum_started_at_lsn);
+        let required_started_at_lsn = match previous_closure.final_lsn {
+            Some(final_lsn) => final_lsn
+                .checked_next()
+                .ok_or(WalStoreError::WriterEpochChainGap)?,
+            None => previous_epoch.map_or(minimum_started_at_lsn, |epoch| epoch.started_at_lsn),
+        };
         let started_at_lsn = minimum_started_at_lsn.max(required_started_at_lsn);
         let ordinal = u64::try_from(self.closed_epochs.len())
             .map_err(|_| WalStoreError::WriterEpochChainGap)?
@@ -9927,6 +9940,9 @@ pub enum WalValidationError {
 /// WAL store errors.
 #[derive(Debug, Error, PartialEq, Eq)]
 pub enum WalStoreError {
+    /// Validated recovery indexes could not be reconstructed before takeover.
+    #[error(transparent)]
+    RecoveryIndex(#[from] WalRecoveryIndexError),
     /// A writer epoch is already active.
     #[error("WAL writer epoch already active")]
     WriterEpochAlreadyActive,

crates/warp-core/tests/causal_wal_hardening_tests.rs

@@ -964,7 +964,8 @@ fn filesystem_writer_lease_refuses_overlap_before_takeover() {
     drop(active);
     let successor = must_ok(contender.acquire_fresh_writer_epoch(Lsn::from_raw(0)));
     assert_eq!(successor.previous_epoch_id, Some(epoch_id()));
-    assert!(successor.started_at_lsn > Lsn::from_raw(0));
+    assert_eq!(successor.started_at_lsn, Lsn::from_raw(0));
+    assert_ne!(successor.epoch_id, epoch_id());
     drop(contender);
     must_ok(fs::remove_dir_all(root));
 }
@@ -2957,3 +2958,57 @@ fn segment_manifest_validation_gate_is_part_of_wal_release_readiness() {
         "manifest validation should be an explicit WAL release gate"
     );
 }
+
+#[test]
+fn filesystem_takeover_refuses_unreconciled_tail_before_reusing_its_lsn() {
+    let mut fixture = WalHardeningFixture::new("takeover-uncommitted");
+    fixture.append_uncommitted_submission_frame("uncommitted", Lsn::from_raw(0));
+    let root = fixture.root.clone();
+    drop(fixture);
+    let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)));
+    let error = must_err(
+        store.acquire_fresh_writer_epoch(Lsn::from_raw(0)),
+        "takeover must not reuse an occupied uncommitted coordinate",
+    );
+    assert!(matches!(error, WalStoreError::SegmentHasUncommittedTail(_)));
+    let report = must_ok(recover_filesystem_store(
+        &root,
+        RecoveryAccessMode::Writable,
+    ));
+    assert_eq!(report.tail_posture, RecoveryTailPosture::TruncatedAll);
+    let epoch = must_ok(store.acquire_fresh_writer_epoch(Lsn::from_raw(0)));
+    let builder = WalTransactionBuilder::new(
+        epoch.epoch_id,
+        WalSegmentId::from_raw(1),
+        transaction_id("reconciled"),
+        WalTransactionKind::SubmissionIntake,
+        WalAppendAuthority::SubmissionIntake,
+        Lsn::from_raw(0),
+        digest("hardening:previous-frame"),
+        digest("hardening:previous-commit"),
+        WalDurabilityMode::StrictFilesystem,
+        PayloadCodecId::from_hash(digest("hardening:codec")),
+        PayloadSchemaId::from_hash(digest("hardening:schema")),
+        1,
+        1,
+        digest("hardening:domain"),
+    );
+    must_ok(
+        store.append_transaction(must_ok(build_submission_acceptance_transaction(
+            builder,
+            submission_acceptance("reconciled"),
+            vec![frontier(
+                "reconciled",
+                AffectedFrontierKind::SubmissionQueue,
+            )],
+        ))),
+    );
+    drop(store);
+    let report = must_ok(recover_filesystem_store(
+        &root,
+        RecoveryAccessMode::ReadOnly,
+    ));
+    assert_eq!(report.tail_posture, RecoveryTailPosture::Clean);
+    assert_eq!(report.last_committed_lsn(), Some(Lsn::from_raw(1)));
+    must_ok(fs::remove_dir_all(root));
+}

docs/topics/WAL.md

@@ -371,6 +371,8 @@ Release-grade durability requires all of the following executable claims:
 - Retained evidence survives restart or produces a typed obstruction.
 - Required recovery artifacts are emitted by CI.
 
+Filesystem writer takeover acquires the process lease and checks the recovered tail before closing a predecessor epoch. An uncommitted or torn tail refuses takeover until writable recovery reconciles it. An empty predecessor leaves its starting LSN unused, so a distinct linked successor may start at that same coordinate; a committed predecessor requires the successor to start after its final LSN, with overflow refused.
+
 ## Evidence
 
 The runtime ACK and recovery witnesses live in