Simplify PersistenceBarrier to two states: Inactive and Active

Remove the Armed state. No race is possible because the barrier is
activated on the same thread that just released the write lock, and
the PREPARE is sent to the durability worker directly (not through
send_or_buffer_durability) before the barrier activates.

Author: cloutiertyler

crates/core/2PC-IMPLEMENTATION-PLAN.md

@@ -153,17 +153,19 @@ On replay, when encountering a PREPARE:
 
 ## Persistence barrier
 
-The barrier in `relational_db.rs` has three states: `Inactive`, `Armed`, `Active`.
+The barrier in `relational_db.rs` has two states: `Inactive` and `Active`.
 
 - **Inactive**: normal operation, durability requests go through.
-- **Armed**: set BEFORE committing the transaction (while write lock is held). The NEXT durability request (the PREPARE) goes through to the worker and transitions the barrier to Active.
-- **Active**: all subsequent durability requests are buffered.
+- **Active**: all durability requests are buffered.
 
-This ensures no race between the write lock release and the barrier activation. Since the barrier is Armed while the write lock is held, no other transaction can commit and send a durability request before the barrier transitions to Active.
+No race is possible because the barrier is activated on the same thread that holds the write lock. The sequence on both coordinator and participant is:
+
+1. Commit in-memory (releases write lock)
+2. Send PREPARE to durability worker (direct call, bypasses barrier)
+3. Activate barrier
+
+Steps 1-3 happen sequentially on one thread. No other transaction can commit between 1 and 3 because steps 2 and 3 are immediate (no async, no lock release between them). By the time another transaction acquires the write lock and commits, the barrier is already active and its durability request is buffered.
 
-Used by both coordinator and participant:
-- Arm before committing the 2PC transaction
-- The commit's durability request (the PREPARE) transitions Armed -> Active
 - On COMMIT: deactivate, flush buffered requests
 - On ABORT: deactivate, discard buffered requests
 

crates/core/src/db/relational_db.rs

@@ -96,24 +96,12 @@ pub struct PersistenceBarrier {
     inner: std::sync::Mutex<PersistenceBarrierInner>,
 }
 
-#[derive(Default, PartialEq, Eq, Debug, Clone, Copy)]
-enum BarrierState {
-    /// No 2PC in progress. Durability requests go through normally.
-    #[default]
-    Inactive,
-    /// A 2PC is about to commit. The NEXT durability request is the PREPARE
-    /// and should go through to the worker. After that request, the barrier
-    /// transitions to Active automatically.
-    Armed,
-    /// A 2PC PREPARE has been sent to durability. All subsequent durability
-    /// requests are buffered until the barrier is deactivated (COMMIT or ABORT).
-    Active,
-}
-
 #[derive(Default)]
 struct PersistenceBarrierInner {
-    state: BarrierState,
-    /// Buffered durability requests that arrived while the barrier was Active.
+    /// Whether the barrier is active. When active, all durability requests
+    /// are buffered instead of being sent to the worker.
+    active: bool,
+    /// Buffered durability requests that arrived while the barrier was active.
     buffered: Vec<BufferedDurabilityRequest>,
 }
 
@@ -122,62 +110,42 @@ impl PersistenceBarrier {
         Self::default()
     }
 
-    /// Arm the barrier. The next durability request will go through (it's the
-    /// PREPARE), and then the barrier transitions to Active, buffering all
-    /// subsequent requests.
+    /// Activate the barrier. All subsequent durability requests will be buffered.
     ///
-    /// This must be called BEFORE the transaction commits, while the write lock
-    /// is still held. This ensures no other transaction can send a durability
-    /// request between the PREPARE and the barrier activation.
-    pub fn arm(&self) {
+    /// Called after committing in-memory and sending PREPARE to the durability
+    /// worker. No race is possible because this runs on the same thread that
+    /// just released the write lock, before any other transaction can commit.
+    pub fn activate(&self) {
         let mut inner = self.inner.lock().unwrap();
-        assert_eq!(
-            inner.state,
-            BarrierState::Inactive,
-            "persistence barrier must be Inactive to arm, but is {:?}",
-            inner.state,
-        );
-        inner.state = BarrierState::Armed;
+        assert!(!inner.active, "persistence barrier already active");
+        inner.active = true;
         inner.buffered.clear();
     }
 
-    /// Called by `send_or_buffer_durability` for every durability request.
-    ///
-    /// Returns `Some(reducer_context)` if the request should be sent to the
-    /// durability worker (barrier is Inactive, or barrier is Armed and this is
-    /// the PREPARE). Returns `None` if the request was buffered (barrier is Active).
-    pub fn filter_durability_request(
+    /// If the barrier is active, buffer the durability request and return None.
+    /// If inactive, return the arguments back (caller should send normally).
+    pub fn try_buffer(
         &self,
         reducer_context: Option<ReducerContext>,
         tx_data: &Arc<TxData>,
     ) -> Option<Option<ReducerContext>> {
         let mut inner = self.inner.lock().unwrap();
-        match inner.state {
-            BarrierState::Inactive => {
-                // No barrier. Let it through.
-                Some(reducer_context)
-            }
-            BarrierState::Armed => {
-                // This is the PREPARE request. Let it through, then go Active.
-                inner.state = BarrierState::Active;
-                Some(reducer_context)
-            }
-            BarrierState::Active => {
-                // Buffer this request.
-                inner.buffered.push(BufferedDurabilityRequest {
-                    reducer_context,
-                    tx_data: tx_data.clone(),
-                });
-                None
-            }
+        if inner.active {
+            inner.buffered.push(BufferedDurabilityRequest {
+                reducer_context,
+                tx_data: tx_data.clone(),
+            });
+            None
+        } else {
+            Some(reducer_context)
         }
     }
 
     /// Deactivate the barrier and return the buffered requests.
     /// Called on COMMIT (to flush them) or ABORT (to discard them).
     pub fn deactivate(&self) -> Vec<BufferedDurabilityRequest> {
         let mut inner = self.inner.lock().unwrap();
-        inner.state = BarrierState::Inactive;
+        inner.active = false;
         std::mem::take(&mut inner.buffered)
     }
 }
@@ -952,32 +920,26 @@ impl RelationalDB {
 
     /// Send a durability request, or buffer it if the persistence barrier is active.
     fn send_or_buffer_durability(&self, reducer_context: Option<ReducerContext>, tx_data: &Arc<TxData>) {
-        match self.persistence_barrier.filter_durability_request(reducer_context, tx_data) {
+        match self.persistence_barrier.try_buffer(reducer_context, tx_data) {
+            None => {
+                // Buffered behind the persistence barrier.
+            }
             Some(reducer_context) => {
-                // Either barrier is Inactive (normal path) or Armed (this is the PREPARE).
-                // Send to durability worker.
+                // Barrier not active. Send to durability worker.
                 if let Some(durability) = &self.durability {
                     durability.request_durability(reducer_context, tx_data);
                 }
             }
-            None => {
-                // Buffered behind the persistence barrier (Active state).
-            }
         }
     }
 
-    /// Arm the persistence barrier for a 2PC PREPARE.
-    ///
-    /// Call this BEFORE committing the transaction (while the write lock is
-    /// still held). The next durability request (the PREPARE) will go through
-    /// to the worker normally. After that, all subsequent durability requests
-    /// are buffered until `finalize_prepare_commit()` or `finalize_prepare_abort()`.
+    /// Activate the persistence barrier for a 2PC PREPARE.
     ///
-    /// This ensures no speculative transaction can reach the durability worker
-    /// between the PREPARE and the COMMIT/ABORT decision, even though the
-    /// write lock is released by `commit_tx_downgrade`.
-    pub fn arm_persistence_barrier(&self) {
-        self.persistence_barrier.arm();
+    /// Call this AFTER committing in-memory and sending PREPARE to the
+    /// durability worker. All subsequent durability requests will be buffered
+    /// until `finalize_prepare_commit()` or `finalize_prepare_abort()`.
+    pub fn activate_persistence_barrier(&self) {
+        self.persistence_barrier.activate();
     }
 
     /// Finalize a 2PC transaction as COMMIT.

crates/core/src/host/module_host.rs

@@ -1787,9 +1787,7 @@ impl ModuleHost {
             // been sent to the durability worker (via the normal commit path).
             // The barrier prevents any subsequent transactions from being persisted
             // until we finalize with COMMIT or ABORT.
-            //
-            // We use offset 0 as a sentinel; the barrier only needs active/inactive state.
-            self.relational_db().persistence_barrier().activate(0);
+            self.relational_db().activate_persistence_barrier();
 
             let info = super::prepared_tx::PreparedTxInfo {
                 tx_offset: 0, // TODO: thread TxOffset from commit path

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -993,7 +993,7 @@ impl InstanceCommon {
                 // (via commit_and_broadcast_event -> commit_tx_downgrade -> send_or_buffer_durability).
                 // No subsequent transactions should be persisted until we confirm all
                 // participants are prepared and we decide COMMIT.
-                stdb.persistence_barrier().activate(0);
+                stdb.activate_persistence_barrier();
             }
 
             let replica_ctx = inst.replica_ctx().clone();