dead code

Author: Shubham8287

crates/core/src/host/instance_env.rs

@@ -1125,94 +1125,6 @@ impl InstanceEnv {
             result
         }
     }
-
-    /// Commit all prepared participants (called after coordinator's reducer succeeds).
-    pub fn commit_all_prepared(&mut self) -> impl Future<Output = ()> + use<> {
-        let participants = mem::take(&mut self.prepared_participants);
-        let client = self.replica_ctx.call_reducer_client.clone();
-        let router = self.replica_ctx.call_reducer_router.clone();
-        let auth_token = self.replica_ctx.call_reducer_auth_token.clone();
-
-        async move {
-            for (db_identity, prepare_id) in participants {
-                let base_url = match router.resolve_base_url(db_identity).await {
-                    Ok(url) => url,
-                    Err(e) => {
-                        log::error!("2PC commit: failed to resolve base URL for {db_identity}: {e}");
-                        continue;
-                    }
-                };
-                let url = format!(
-                    "{}/v1/database/{}/2pc/commit/{}",
-                    base_url,
-                    db_identity.to_hex(),
-                    prepare_id,
-                );
-                let mut req = client.post(&url);
-                if let Some(ref token) = auth_token {
-                    req = req.header(http::header::AUTHORIZATION, format!("Bearer {token}"));
-                }
-                match req.send().await {
-                    Ok(resp) if resp.status().is_success() => {
-                        log::info!("2PC commit: committed {prepare_id} on {db_identity}");
-                    }
-                    Ok(resp) => {
-                        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}");
-                    }
-                }
-            }
-        }
-    }
-
-    /// Abort all prepared participants (called when coordinator's reducer fails).
-    pub fn abort_all_prepared(&mut self) -> impl Future<Output = ()> + use<> {
-        let participants = mem::take(&mut self.prepared_participants);
-        let client = self.replica_ctx.call_reducer_client.clone();
-        let router = self.replica_ctx.call_reducer_router.clone();
-        let auth_token = self.replica_ctx.call_reducer_auth_token.clone();
-
-        async move {
-            for (db_identity, prepare_id) in participants {
-                let base_url = match router.resolve_base_url(db_identity).await {
-                    Ok(url) => url,
-                    Err(e) => {
-                        log::error!("2PC abort: failed to resolve base URL for {db_identity}: {e}");
-                        continue;
-                    }
-                };
-                let url = format!(
-                    "{}/v1/database/{}/2pc/abort/{}",
-                    base_url,
-                    db_identity.to_hex(),
-                    prepare_id,
-                );
-                let mut req = client.post(&url);
-                if let Some(ref token) = auth_token {
-                    req = req.header(http::header::AUTHORIZATION, format!("Bearer {token}"));
-                }
-                match req.send().await {
-                    Ok(resp) if resp.status().is_success() => {
-                        log::info!("2PC abort: aborted {prepare_id} on {db_identity}");
-                    }
-                    Ok(resp) => {
-                        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}");
-                    }
-                }
-            }
-        }
-    }
 }
 
 /// Default timeout for HTTP requests performed by [`InstanceEnv::http_request`].

crates/core/src/host/module_host.rs

@@ -1928,7 +1928,10 @@ impl ModuleHost {
                 let participant_identity = match Identity::from_hex(&row.participant_identity_hex) {
                     Ok(id) => id,
                     Err(e) => {
-                        log::error!("recover_2pc_coordinator: invalid participant identity hex {}: {e}", row.participant_identity_hex);
+                        log::error!(
+                            "recover_2pc_coordinator: invalid participant identity hex {}: {e}",
+                            row.participant_identity_hex
+                        );
                         continue;
                     }
                 };
@@ -1959,7 +1962,10 @@ impl ModuleHost {
                         }
                     }
                     Ok(resp) => {
-                        log::warn!("recover_2pc_coordinator: commit for {prepare_id} returned {}", resp.status());
+                        log::warn!(
+                            "recover_2pc_coordinator: commit for {prepare_id} returned {}",
+                            resp.status()
+                        );
                     }
                     Err(e) => {
                         log::warn!("recover_2pc_coordinator: transport error for {prepare_id}: {e}");
@@ -1994,14 +2000,18 @@ impl ModuleHost {
                 let coordinator_identity = match Identity::from_hex(&row.coordinator_identity_hex) {
                     Ok(id) => id,
                     Err(e) => {
-                        log::error!("recover_2pc_participant: invalid coordinator identity hex for {original_prepare_id}: {e}");
+                        log::error!(
+                            "recover_2pc_participant: invalid coordinator identity hex for {original_prepare_id}: {e}"
+                        );
                         continue;
                     }
                 };
                 let caller_identity = match Identity::from_hex(&row.caller_identity_hex) {
                     Ok(id) => id,
                     Err(e) => {
-                        log::error!("recover_2pc_participant: invalid caller identity hex for {original_prepare_id}: {e}");
+                        log::error!(
+                            "recover_2pc_participant: invalid caller identity hex for {original_prepare_id}: {e}"
+                        );
                         continue;
                     }
                 };
@@ -2051,7 +2061,8 @@ impl ModuleHost {
                             auth_token.clone(),
                             coordinator_identity,
                             &original_prepare_id,
-                        ).await;
+                        )
+                        .await;
                         match decision {
                             Some(commit) => {
                                 if commit {

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

@@ -432,7 +432,10 @@ async fn send_ack_commit_to_coordinator(
             log::info!("2PC ack-commit: notified coordinator for {prepare_id}");
         }
         Ok(resp) => {
-            log::warn!("2PC ack-commit: coordinator returned {} for {prepare_id}", resp.status());
+            log::warn!(
+                "2PC ack-commit: coordinator returned {} for {prepare_id}",
+                resp.status()
+            );
         }
         Err(e) => {
             log::warn!("2PC ack-commit: transport error for {prepare_id}: {e}");
@@ -709,8 +712,7 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         let _ = prepared_tx.send((res, return_value));
 
         // Step 4: wait for coordinator's decision (B never aborts on its own).
-        let commit =
-            Self::wait_for_2pc_decision(decision_rx, &prepare_id, coordinator_identity, &replica_ctx);
+        let commit = Self::wait_for_2pc_decision(decision_rx, &prepare_id, coordinator_identity, &replica_ctx);
 
         if commit {
             // Delete the marker in the same tx as the reducer changes (atomic commit).
@@ -1196,9 +1198,11 @@ impl InstanceCommon {
                                     // B acknowledged COMMIT — remove coordinator log entry
                                     // (best-effort; recovery will clean up on restart if missed).
                                     if committed {
-                                        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}");
                                         }
                                     }

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -22,11 +22,10 @@ use crate::{
     system_tables::{
         with_sys_table_buf, St2pcCoordinatorLogFields, St2pcCoordinatorLogRow, St2pcStateFields, St2pcStateRow,
         StClientFields, StClientRow, StColumnAccessorFields, StColumnAccessorRow, StColumnFields, StColumnRow,
-        StConstraintFields, StConstraintRow, StEventTableRow,
-        StFields as _, StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow, StRowLevelSecurityFields,
-        StRowLevelSecurityRow, StScheduledFields, StScheduledRow, StSequenceFields, StSequenceRow,
-        StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow, SystemTable,
-        ST_2PC_COORDINATOR_LOG_ID, ST_2PC_STATE_ID, ST_CLIENT_ID, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ID,
+        StConstraintFields, StConstraintRow, StEventTableRow, StFields as _, StIndexAccessorFields, StIndexAccessorRow,
+        StIndexFields, StIndexRow, StRowLevelSecurityFields, StRowLevelSecurityRow, StScheduledFields, StScheduledRow,
+        StSequenceFields, StSequenceRow, StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow,
+        SystemTable, ST_2PC_COORDINATOR_LOG_ID, ST_2PC_STATE_ID, ST_CLIENT_ID, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ID,
         ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INDEX_ACCESSOR_ID, ST_INDEX_ID, ST_ROW_LEVEL_SECURITY_ID,
         ST_SCHEDULED_ID, ST_SEQUENCE_ID, ST_TABLE_ACCESSOR_ID, ST_TABLE_ID,
     },
@@ -2771,9 +2770,7 @@ impl MutTxId {
         self.insert_via_serialize_bsatn(ST_2PC_COORDINATOR_LOG_ID, row)
             .map(|_| ())
             .inspect_err(|e| {
-                log::error!(
-                    "insert_st_2pc_coordinator_log: failed for prepare_id ({participant_prepare_id}): {e}"
-                );
+                log::error!("insert_st_2pc_coordinator_log: failed for prepare_id ({participant_prepare_id}): {e}");
             })
     }
 
@@ -2785,9 +2782,7 @@ impl MutTxId {
             St2pcCoordinatorLogFields::ParticipantPrepareId.col_id(),
             &AlgebraicValue::String(participant_prepare_id.into()),
         ) {
-            log::error!(
-                "delete_st_2pc_coordinator_log: no row for prepare_id ({participant_prepare_id}): {e}"
-            );
+            log::error!("delete_st_2pc_coordinator_log: no row for prepare_id ({participant_prepare_id}): {e}");
         }
         Ok(())
     }

crates/smoketests/tests/smoketests/cross_db_2pc_recovery.rs

@@ -166,10 +166,7 @@ fn setup_two_banks(test: &mut Smoketest, pid: u32, suffix: &str) -> (String, Str
 fn test_2pc_committed_data_survives_restart() {
     require_local_server!();
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, db_b_identity) = setup_two_banks(&mut test, pid, "dur");
 
@@ -178,8 +175,16 @@ fn test_2pc_committed_data_survives_restart() {
         .expect("transfer_funds failed");
 
     // Verify pre-restart state.
-    assert_eq!(alice_balance(&test, &db_a_identity), 150, "A should have 150 before restart");
-    assert_eq!(alice_balance(&test, &db_b_identity), 50, "B should have 50 before restart");
+    assert_eq!(
+        alice_balance(&test, &db_a_identity),
+        150,
+        "A should have 150 before restart"
+    );
+    assert_eq!(
+        alice_balance(&test, &db_b_identity),
+        50,
+        "B should have 50 before restart"
+    );
 
     // Restart the server — exercises recovery path even though there's nothing to recover.
     test.restart_server();
@@ -207,18 +212,23 @@ fn test_2pc_committed_data_survives_restart() {
 fn test_2pc_aborted_state_survives_restart() {
     require_local_server!();
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, db_b_identity) = setup_two_banks(&mut test, pid, "abort-dur");
 
     // Try to transfer 200 — B only has 100, so the remote debit panics → abort.
     let _ = test.call("transfer_funds", &[&db_b_identity, "alice", "alice", "200"]);
 
-    assert_eq!(alice_balance(&test, &db_a_identity), 100, "A should still be 100 after abort");
-    assert_eq!(alice_balance(&test, &db_b_identity), 100, "B should still be 100 after abort");
+    assert_eq!(
+        alice_balance(&test, &db_a_identity),
+        100,
+        "A should still be 100 after abort"
+    );
+    assert_eq!(
+        alice_balance(&test, &db_b_identity),
+        100,
+        "B should still be 100 after abort"
+    );
 
     test.restart_server();
 
@@ -243,10 +253,7 @@ fn test_2pc_aborted_state_survives_restart() {
 #[test]
 fn test_2pc_status_endpoint_unknown_returns_abort() {
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, _db_b_identity) = setup_two_banks(&mut test, pid, "status");
 
@@ -284,10 +291,7 @@ fn test_2pc_status_endpoint_unknown_returns_abort() {
 fn test_2pc_atomicity_under_crash() {
     require_local_server!();
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, db_b_identity) = setup_two_banks(&mut test, pid, "crash");
 
@@ -340,10 +344,7 @@ fn test_2pc_atomicity_under_crash() {
 fn test_2pc_coordinator_recovery() {
     require_local_server!();
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, db_b_identity) = setup_two_banks(&mut test, pid, "coord-rec");
 
@@ -378,10 +379,7 @@ fn test_2pc_coordinator_recovery() {
     let bal_a = alice_balance(&test, &db_a_identity);
     let bal_b = alice_balance(&test, &db_b_identity);
 
-    assert_eq!(
-        bal_a, 150,
-        "A should have committed (alice_a=150) before crash"
-    );
+    assert_eq!(bal_a, 150, "A should have committed (alice_a=150) before crash");
     assert_eq!(
         bal_b, 50,
         "B should have committed via coordinator recovery (alice_b=50), got {bal_b}"
@@ -407,10 +405,7 @@ fn test_2pc_coordinator_recovery() {
 fn test_2pc_participant_recovery_polls_and_aborts() {
     require_local_server!();
     let pid = std::process::id();
-    let mut test = Smoketest::builder()
-        .module_code(MODULE_CODE)
-        .autopublish(false)
-        .build();
+    let mut test = Smoketest::builder().module_code(MODULE_CODE).autopublish(false).build();
 
     let (db_a_identity, db_b_identity) = setup_two_banks(&mut test, pid, "part-rec");