libsql-server: reconcile indeterminate fence commits and restart at each boundary

Add crash-restart tests for the namespace fence: each server lifetime runs on
its own runtime and is ended without any shutdown code while a fence command
is parked at a hook point, so the next start takes the real dirty-recovery
path on the same directory. They cover every persistence boundary of
AcquireSourceWriteFence and ReleaseSourceWriteFence (including a marker that
lags the metastore commit), a restart in SOURCE_DRAINING with a writer active
at the crash, indeterminate commits through the drain path (applied and not
applied), and lost acquisition responses resolved by replay and inspection.

The tests exposed that a source restarted while draining could never finish
its drain: dirty recovery rebuilds the replication log under a new log id,
and completing the drain refused a boundary on a log other than the one the
fence was acquired against, leaving the namespace in SOURCE_DRAINING for
good. The frozen boundary now names the log that is live when the drain is
proven, the record's identity keeps the acquisition log id, and the server
warns when the two differ. Write admission was durably closed throughout, so
the data at the boundary is unchanged.

The BeforeMetastoreCommit test hook can now report a commit as indeterminate
without running it. The contract document describes the restart and log
rebuild semantics.

Co-authored-by: Tomasz Szymczyszyn <tomasz.szymczyszyn@shopify.com>

Author: shopify-river

libsql-server/src/namespace/fence/controller.rs

@@ -458,11 +458,17 @@ impl Transition {
             }
         }
 
-        if let HookOutcome::Fail(e) = controller.hook(HookPoint::BeforeMetastoreCommit).await {
-            return Err(e.into());
-        }
+        let result = match controller.hook(HookPoint::BeforeMetastoreCommit).await {
+            HookOutcome::Continue => run.await,
+            HookOutcome::Fail(e) => return Err(e.into()),
+            // A commit that failed without applying, but whose outcome the controller cannot
+            // know (test hook).
+            HookOutcome::Indeterminate => {
+                Err(indeterminate(key, "the commit was not acknowledged (test hook)").into())
+            }
+        };
 
-        let result = match run.await {
+        let result = match result {
             Ok(commit) => match controller.hook(HookPoint::AfterMetastoreCommit).await {
                 HookOutcome::Continue => Ok(commit),
                 HookOutcome::Indeterminate | HookOutcome::Fail(_) => Err(indeterminate(

libsql-server/src/namespace/fence/drain.rs

@@ -121,6 +121,16 @@ pub async fn acquire_source_write_fence(
     };
 
     // Step 7.
+    let acquired_on = commit.record.as_ref().and_then(|r| r.identity.log_id);
+    if acquired_on.is_some_and(|log_id| log_id != boundary.log_id) {
+        tracing::warn!(
+            namespace = %controller.namespace(),
+            acquired_on = ?acquired_on,
+            boundary_log_id = %boundary.log_id,
+            "the replication log was rebuilt since the write fence was acquired (the source \
+             restarted while draining); the frozen boundary names the rebuilt log"
+        );
+    }
     ctx.now_ms = now_ms();
     transition
         .complete_drain(

libsql-server/src/namespace/fence/hooks.rs

@@ -56,7 +56,9 @@ pub enum HookAction {
     /// Fail at this point with `error`, as if the step had failed before it took effect.
     Fail(FenceError),
     /// At `AfterMetastoreCommit`: report the commit as indeterminate even though it happened,
-    /// which is what a lost commit acknowledgement looks like to the controller.
+    /// which is what a lost commit acknowledgement looks like to the controller. At
+    /// `BeforeMetastoreCommit`: report it as indeterminate without running it, which is a
+    /// commit that failed without applying but whose outcome the controller cannot know.
     Indeterminate,
 }
 

libsql-server/src/namespace/fence/mod.rs

@@ -28,6 +28,9 @@ pub mod state;
 pub mod store;
 pub mod transition;
 
+#[cfg(test)]
+mod tests;
+
 #[allow(clippy::all)]
 pub(crate) mod proto {
     include!("../../generated/namespace_fence.rs");