Log and return ok on missing request_id

Author: joshua-spacetime

crates/core/src/db/relational_db.rs

@@ -897,11 +897,7 @@ impl RelationalDB {
 
     /// Send a tx to the durability worker, unless a durability barrier is active
     /// and the tx's offset exceeds the minimum active barrier. In that case, defer the tx.
-    fn request_durability_maybe_barrier(
-        &self,
-        reducer_context: Option<ReducerContext>,
-        tx_data: &Arc<TxData>,
-    ) {
+    fn request_durability_maybe_barrier(&self, reducer_context: Option<ReducerContext>, tx_data: &Arc<TxData>) {
         let Some(durability) = &self.durability else {
             return;
         };

crates/core/src/host/module_host.rs

@@ -890,7 +890,6 @@ pub struct ModuleHost {
     ///
     /// When this is true, most operations will fail with [`NoSuchModule`].
     closed: Arc<AtomicBool>,
-
 }
 
 impl fmt::Debug for ModuleHost {
@@ -1839,7 +1838,14 @@ impl ModuleHost {
             let _ = this
                 .call(
                     &reducer_name_owned,
-                    (params, prepare_id_clone, coordinator_identity, prepared_tx, decision_rx, commit_persist_rx),
+                    (
+                        params,
+                        prepare_id_clone,
+                        coordinator_identity,
+                        prepared_tx,
+                        decision_rx,
+                        commit_persist_rx,
+                    ),
                     async |(p, pid, cid, ptx, drx, cprx), inst| {
                         inst.call_reducer_prepare_and_hold(p, pid, cid, ptx, drx, cprx);
                         Ok::<(), ReducerCallError>(())
@@ -2056,7 +2062,13 @@ impl ModuleHost {
                 // The PREPARE PERSIST only stored the reducer inputs, not the row
                 // mutations. We re-run to get a fresh MutTxId with the mutations.
                 let new_prepare_id = match this
-                    .prepare_reducer(caller_identity, Some(caller_connection_id), &row.reducer_name, args, Some(coordinator_identity))
+                    .prepare_reducer(
+                        caller_identity,
+                        Some(caller_connection_id),
+                        &row.reducer_name,
+                        args,
+                        Some(coordinator_identity),
+                    )
                     .await
                 {
                     Ok((pid, result, _rv)) if !pid.is_empty() => {

crates/core/src/host/prepared_tx.rs

@@ -47,15 +47,19 @@ impl PreparedTransactions {
     /// The entry stays for Round 2 (commit_persist).
     pub fn send_decision(&self, id: &str, commit: bool) -> Result<(), String> {
         let guard = self.inner.lock().unwrap();
-        let info = guard.get(id).ok_or_else(|| format!("no such prepared transaction: {id}"))?;
+        let info = guard
+            .get(id)
+            .ok_or_else(|| format!("no such prepared transaction: {id}"))?;
         let _ = info.decision_sender.send(commit);
         Ok(())
     }
 
     /// Send a Round 2 COMMIT_PERSIST signal and remove the entry.
     pub fn send_commit_persist(&self, id: &str) -> Result<(), String> {
         let mut guard = self.inner.lock().unwrap();
-        let info = guard.remove(id).ok_or_else(|| format!("no such prepared transaction: {id}"))?;
+        let info = guard
+            .remove(id)
+            .ok_or_else(|| format!("no such prepared transaction: {id}"))?;
         let _ = info.commit_persist_sender.send(());
         Ok(())
     }

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

@@ -793,12 +793,7 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
 
             // Step 10: wait for COMMIT_PERSIST from coordinator (tokio oneshot).
             // Sender dropped without sending ⇒ coordinator restarted / aborted ⇒ treat as abort.
-            let persist_commit = match tokio::time::timeout(
-                Duration::from_secs(60),
-                commit_persist_rx,
-            )
-            .await
-            {
+            let persist_commit = match tokio::time::timeout(Duration::from_secs(60), commit_persist_rx).await {
                 Ok(Ok(())) => true,
                 Ok(Err(_)) => {
                     // Sender dropped: coordinator crashed / aborted.
@@ -851,19 +846,15 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
                 if let Err(e) = stdb.with_auto_commit::<_, _, anyhow::Error>(Workload::Internal, |del_tx| {
                     Ok(del_tx.delete_st_2pc_state(&prepare_id)?)
                 }) {
-                    log::error!(
-                        "call_reducer_prepare_and_hold: failed to delete st_2pc_state for {prepare_id}: {e}"
-                    );
+                    log::error!("call_reducer_prepare_and_hold: failed to delete st_2pc_state for {prepare_id}: {e}");
                 }
             } else {
                 // Round 2 abort: discard all deferred transactions.
                 stdb.abort_durability_barrier(barrier_offset);
                 // Trigger module restart via the on_panic hook rather than panicking,
                 // since we're in an async task outside the WASM executor thread.
                 if let Some(on_panic) = replica_ctx.on_panic.get() {
-                    log::error!(
-                        "2PC persistence aborted for {prepare_id}: triggering module restart"
-                    );
+                    log::error!("2PC persistence aborted for {prepare_id}: triggering module restart");
                     on_panic();
                 }
             }
@@ -1215,7 +1206,10 @@ impl InstanceCommon {
             // through), A's own tx must also be deferred. Set barrier at offset - 1
             // so A's tx (at offset) is > barrier and gets deferred.
             let barrier_offset = tx.next_tx_offset().saturating_sub(1);
-            self.info.subscriptions.relational_db().set_durability_barrier(barrier_offset);
+            self.info
+                .subscriptions
+                .relational_db()
+                .set_durability_barrier(barrier_offset);
             Some(barrier_offset)
         } else {
             None
@@ -1247,7 +1241,9 @@ impl InstanceCommon {
                         };
                         let url = format!(
                             "{}/v1/database/{}/2pc/abort/{}",
-                            base_url, db_identity.to_hex(), prepare_id,
+                            base_url,
+                            db_identity.to_hex(),
+                            prepare_id,
                         );
                         let mut req = client.post(&url);
                         if let Some(ref token) = auth_token {
@@ -1258,7 +1254,10 @@ impl InstanceCommon {
                                 log::info!("2PC abort: {prepare_id} on {db_identity}");
                             }
                             Ok(resp) => {
-                                log::error!("2PC abort: failed for {prepare_id} on {db_identity}: status {}", resp.status());
+                                log::error!(
+                                    "2PC abort: failed for {prepare_id} on {db_identity}: status {}",
+                                    resp.status()
+                                );
                             }
                             Err(e) => {
                                 log::error!("2PC abort: transport error for {prepare_id} on {db_identity}: {e}");
@@ -1289,7 +1288,9 @@ impl InstanceCommon {
                     };
                     let url = format!(
                         "{}/v1/database/{}/2pc/commit/{}",
-                        base_url, db_identity.to_hex(), prepare_id,
+                        base_url,
+                        db_identity.to_hex(),
+                        prepare_id,
                     );
                     let mut req = client.post(&url);
                     if let Some(ref token) = auth_token {
@@ -1300,7 +1301,10 @@ impl InstanceCommon {
                             log::info!("2PC commit (Round 1): {prepare_id} on {db_identity}");
                         }
                         Ok(resp) => {
-                            log::error!("2PC commit: failed for {prepare_id} on {db_identity}: status {}", resp.status());
+                            log::error!(
+                                "2PC commit: failed for {prepare_id} on {db_identity}: status {}",
+                                resp.status()
+                            );
                         }
                         Err(e) => {
                             log::error!("2PC commit: transport error for {prepare_id} on {db_identity}: {e}");
@@ -1367,7 +1371,9 @@ impl InstanceCommon {
                     };
                     let url = format!(
                         "{}/v1/database/{}/2pc/commit-persist/{}",
-                        base_url, db_identity.to_hex(), prepare_id,
+                        base_url,
+                        db_identity.to_hex(),
+                        prepare_id,
                     );
                     let mut req = client.post(&url);
                     if let Some(ref token) = auth_token {
@@ -1377,16 +1383,17 @@ impl InstanceCommon {
                         Ok(resp) if resp.status().is_success() => {
                             log::info!("2PC commit-persist: {prepare_id} on {db_identity}");
                             // Round 2 complete — delete coordinator log entry.
-                            if let Err(e) = stdb
-                                .with_auto_commit::<_, _, anyhow::Error>(Workload::Internal, |del_tx| {
-                                    Ok(del_tx.delete_st_2pc_coordinator_log(prepare_id)?)
-                                })
-                            {
+                            if let Err(e) = stdb.with_auto_commit::<_, _, anyhow::Error>(Workload::Internal, |del_tx| {
+                                Ok(del_tx.delete_st_2pc_coordinator_log(prepare_id)?)
+                            }) {
                                 log::warn!("delete_st_2pc_coordinator_log failed for {prepare_id}: {e}");
                             }
                         }
                         Ok(resp) => {
-                            log::error!("2PC commit-persist: failed for {prepare_id} on {db_identity}: status {}", resp.status());
+                            log::error!(
+                                "2PC commit-persist: failed for {prepare_id} on {db_identity}: status {}",
+                                resp.status()
+                            );
                         }
                         Err(e) => {
                             log::error!("2PC commit-persist: transport error for {prepare_id} on {db_identity}: {e}");

crates/core/src/startup.rs

@@ -413,10 +413,13 @@ impl Cores {
 
 #[cfg(target_os = "linux")]
 fn cores_to_cpuset(cores: &[CoreId]) -> Option<nix::sched::CpuSet> {
-    cores.iter().copied().try_fold(nix::sched::CpuSet::new(), |mut cpuset, core| {
-        cpuset.set(core.id).ok()?;
-        Some(cpuset)
-    })
+    cores
+        .iter()
+        .copied()
+        .try_fold(nix::sched::CpuSet::new(), |mut cpuset, core| {
+            cpuset.set(core.id).ok()?;
+            Some(cpuset)
+        })
 }
 
 #[cfg(target_os = "linux")]
@@ -591,10 +594,7 @@ mod tests {
         #[cfg(target_os = "linux")]
         {
             assert!(split.tokio.workers.is_none());
-            assert_eq!(
-                cpuset_cardinality(split.tokio.blocking.as_ref().unwrap()),
-                20
-            );
+            assert_eq!(cpuset_cardinality(split.tokio.blocking.as_ref().unwrap()), 20);
             assert!(split.rayon.dedicated.is_none());
             assert_eq!(split.rayon.shared.as_ref().unwrap().0, 20);
         }

crates/smoketests/tests/smoketests/cross_db_2pc.rs

@@ -125,13 +125,16 @@ fn test_cross_db_2pc_happy_path() {
     let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     // Publish participants first, then coordinator.
-    test.publish_module_named(&db_b_name, false).expect("failed to publish bank B");
+    test.publish_module_named(&db_b_name, false)
+        .expect("failed to publish bank B");
     let db_b_identity = test.database_identity.clone().expect("bank B identity not set");
 
-    test.publish_module_named(&db_c_name, false).expect("failed to publish bank C");
+    test.publish_module_named(&db_c_name, false)
+        .expect("failed to publish bank C");
     let db_c_identity = test.database_identity.clone().expect("bank C identity not set");
 
-    test.publish_module_named(&db_a_name, false).expect("failed to publish bank A");
+    test.publish_module_named(&db_a_name, false)
+        .expect("failed to publish bank A");
     let db_a_identity = test.database_identity.clone().expect("bank A identity not set");
 
     // Call transfer_funds; the return value is A's new alice balance.
@@ -167,13 +170,16 @@ fn test_cross_db_2pc_abort_insufficient_funds() {
 
     let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
-    test.publish_module_named(&db_b_name, false).expect("failed to publish bank B");
+    test.publish_module_named(&db_b_name, false)
+        .expect("failed to publish bank B");
     let db_b_identity = test.database_identity.clone().expect("bank B identity not set");
 
-    test.publish_module_named(&db_c_name, false).expect("failed to publish bank C");
+    test.publish_module_named(&db_c_name, false)
+        .expect("failed to publish bank C");
     let db_c_identity = test.database_identity.clone().expect("bank C identity not set");
 
-    test.publish_module_named(&db_a_name, false).expect("failed to publish bank A");
+    test.publish_module_named(&db_a_name, false)
+        .expect("failed to publish bank A");
     let db_a_identity = test.database_identity.clone().expect("bank A identity not set");
 
     // Transfer 110 from each — both only have 100, so B's debit panics → 2PC aborts all.
@@ -184,10 +190,25 @@ fn test_cross_db_2pc_abort_insufficient_funds() {
             &format!("[\"{db_b_identity}\", \"{db_c_identity}\", \"alice\", \"alice\", 110]"),
         )
         .expect("api_call failed");
-    assert!(!resp.is_success(), "Expected transfer_funds to fail due to insufficient funds");
+    assert!(
+        !resp.is_success(),
+        "Expected transfer_funds to fail due to insufficient funds"
+    );
 
     // All three accounts must still be at 100.
-    assert_eq!(call_balance(&test, &db_a_identity, "alice"), 100, "A alice should still be 100");
-    assert_eq!(call_balance(&test, &db_b_identity, "alice"), 100, "B alice should still be 100");
-    assert_eq!(call_balance(&test, &db_c_identity, "alice"), 100, "C alice should still be 100");
+    assert_eq!(
+        call_balance(&test, &db_a_identity, "alice"),
+        100,
+        "A alice should still be 100"
+    );
+    assert_eq!(
+        call_balance(&test, &db_b_identity, "alice"),
+        100,
+        "B alice should still be 100"
+    );
+    assert_eq!(
+        call_balance(&test, &db_c_identity, "alice"),
+        100,
+        "C alice should still be 100"
+    );
 }