lint

Author: Shubham8287

crates/client-api/src/routes/database.rs

@@ -291,10 +291,9 @@ pub async fn prepare<S: ControlStateDelegate + NodeDelegate>(
             )
                 .into_response();
             if !prepare_id.is_empty() {
-                response.headers_mut().insert(
-                    "X-Prepare-Id",
-                    http::HeaderValue::from_str(&prepare_id).unwrap(),
-                );
+                response
+                    .headers_mut()
+                    .insert("X-Prepare-Id", http::HeaderValue::from_str(&prepare_id).unwrap());
             }
             Ok(response)
         }

crates/core/src/db/relational_db.rs

@@ -76,7 +76,6 @@ type RowCountFn = Arc<dyn Fn(TableId, &str) -> i64 + Send + Sync>;
 /// The type of transactions committed by [RelationalDB].
 pub type Txdata = commitlog::payload::Txdata<ProductValue>;
 
-
 /// We've added a module version field to the system tables, but we don't yet
 /// have the infrastructure to support multiple versions.
 /// All modules are currently locked to this version, but this will be
@@ -460,9 +459,7 @@ impl RelationalDB {
     /// If non-empty, the caller must resume these transactions: retransmit PREPARED to
     /// the coordinator and await a COMMIT or ABORT decision before allowing normal operation.
     pub fn pending_2pc_prepares(&self) -> Result<Vec<String>, DBError> {
-        self.with_auto_commit(Workload::Internal, |tx| {
-            tx.scan_st_2pc_state().map_err(DBError::from)
-        })
+        self.with_auto_commit(Workload::Internal, |tx| tx.scan_st_2pc_state().map_err(DBError::from))
     }
 
     /// Read the set of clients currently connected to the database.

crates/core/src/host/instance_env.rs

@@ -1034,7 +1034,10 @@ impl InstanceEnv {
             let result = async {
                 let response = req.send().await.map_err(|e| NodesError::HttpError(e.to_string()))?;
                 let status = response.status().as_u16();
-                let body = response.bytes().await.map_err(|e| NodesError::HttpError(e.to_string()))?;
+                let body = response
+                    .bytes()
+                    .await
+                    .map_err(|e| NodesError::HttpError(e.to_string()))?;
                 Ok((status, body))
             }
             .await;
@@ -1102,7 +1105,10 @@ impl InstanceEnv {
                     .get("X-Prepare-Id")
                     .and_then(|v| v.to_str().ok())
                     .map(|s| s.to_owned());
-                let body = response.bytes().await.map_err(|e| NodesError::HttpError(e.to_string()))?;
+                let body = response
+                    .bytes()
+                    .await
+                    .map_err(|e| NodesError::HttpError(e.to_string()))?;
                 Ok((status, body, prepare_id))
             }
             .await;
@@ -1121,9 +1127,7 @@ impl InstanceEnv {
     }
 
     /// Commit all prepared participants (called after coordinator's reducer succeeds).
-    pub fn commit_all_prepared(
-        &mut self,
-    ) -> impl Future<Output = ()> + use<> {
+    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();
@@ -1167,9 +1171,7 @@ impl InstanceEnv {
     }
 
     /// Abort all prepared participants (called when coordinator's reducer fails).
-    pub fn abort_all_prepared(
-        &mut self,
-    ) -> impl Future<Output = ()> + use<> {
+    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();

crates/core/src/host/module_host.rs

@@ -1791,8 +1791,7 @@ impl ModuleHost {
         let prepare_id = format!("prepare-{}", PREPARE_COUNTER.fetch_add(1, Ordering::Relaxed));
 
         // Channel for signalling PREPARED result back to this task.
-        let (prepared_tx, prepared_rx) =
-            tokio::sync::oneshot::channel::<(ReducerCallResult, Option<Bytes>)>();
+        let (prepared_tx, prepared_rx) = tokio::sync::oneshot::channel::<(ReducerCallResult, Option<Bytes>)>();
         // Channel for sending the COMMIT/ABORT decision to the executor thread.
         let (decision_tx, decision_rx) = std::sync::mpsc::channel::<bool>();
 
@@ -1855,7 +1854,8 @@ impl ModuleHost {
 
     /// Finalize a prepared transaction as COMMIT.
     pub fn commit_prepared(&self, prepare_id: &str) -> Result<(), String> {
-        let info = self.prepared_txs
+        let info = self
+            .prepared_txs
             .remove(prepare_id)
             .ok_or_else(|| format!("no such prepared transaction: {prepare_id}"))?;
         // Unblock the executor thread to commit.
@@ -1865,7 +1865,8 @@ impl ModuleHost {
 
     /// Abort a prepared transaction.
     pub fn abort_prepared(&self, prepare_id: &str) -> Result<(), String> {
-        let info = self.prepared_txs
+        let info = self
+            .prepared_txs
             .remove(prepare_id)
             .ok_or_else(|| format!("no such prepared transaction: {prepare_id}"))?;
         // Unblock the executor thread to abort.

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

@@ -654,7 +654,9 @@ 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: abort: failed to delete st_2pc_state for {prepare_id}: {e}");
+                log::error!(
+                    "call_reducer_prepare_and_hold: abort: failed to delete st_2pc_state for {prepare_id}: {e}"
+                );
             }
         }
     }
@@ -953,11 +955,9 @@ impl InstanceCommon {
             std::thread::scope(|s| {
                 s.spawn(|| {
                     handle.block_on(async {
-                        if committed {
-                            if let Some(mut durable_offset) = stdb.durable_tx_offset() {
-                                let current: u64 = durable_offset.last_seen().unwrap_or(0);
-                                let _ = durable_offset.wait_for(current + 1).await;
-                            }
+                        if committed && let Some(mut durable_offset) = stdb.durable_tx_offset() {
+                            let current: u64 = durable_offset.last_seen().unwrap_or(0);
+                            let _ = durable_offset.wait_for(current + 1).await;
                         }
 
                         let client = replica_ctx.call_reducer_client.clone();

crates/core/src/host/wasmtime/wasm_instance_env.rs

@@ -2076,12 +2076,10 @@ impl WasmInstanceEnv {
             match result {
                 Ok((status, body, prepare_id)) => {
                     // If we got a prepare_id, register this participant.
-                    if let Some(pid) = prepare_id {
-                        if status < 300 {
-                            env.instance_env
-                                .prepared_participants
-                                .push((database_identity, pid));
-                        }
+                    if let Some(pid) = prepare_id
+                        && status < 300
+                    {
+                        env.instance_env.prepared_participants.push((database_identity, pid));
                     }
                     let bytes_source = WasmInstanceEnv::create_bytes_source(env, body)?;
                     bytes_source.0.write_to(mem, out)?;

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -1430,13 +1430,11 @@ impl CommittedState {
         row: &ProductValue,
     ) -> Result<TxData> {
         let (table, blob_store, pool) = self.get_table_and_blob_store_or_create(table_id, schema);
-        table
-            .insert(pool, blob_store, row)
-            .map_err(|e| match e {
-                InsertError::Duplicate(e) => DatastoreError::from(TableError::Duplicate(e)),
-                InsertError::Bflatn(e) => DatastoreError::from(TableError::Bflatn(e)),
-                InsertError::IndexError(e) => DatastoreError::from(IndexError::UniqueConstraintViolation(e)),
-            })?;
+        table.insert(pool, blob_store, row).map_err(|e| match e {
+            InsertError::Duplicate(e) => DatastoreError::from(TableError::Duplicate(e)),
+            InsertError::Bflatn(e) => DatastoreError::from(TableError::Bflatn(e)),
+            InsertError::IndexError(e) => DatastoreError::from(IndexError::UniqueConstraintViolation(e)),
+        })?;
 
         let row_arc: Arc<[ProductValue]> = Arc::from([row.clone()]);
         let mut tx_data = TxData::default();

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -20,13 +20,13 @@ use crate::{
 use crate::{
     error::{IndexError, SequenceError, TableError},
     system_tables::{
-        with_sys_table_buf, 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_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, St2pcStateFields, St2pcStateRow, ST_2PC_STATE_ID,
+        with_sys_table_buf, 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_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,
     },
 };
 use crate::{execution_context::ExecutionContext, system_tables::StViewColumnRow};
@@ -2774,9 +2774,7 @@ impl MutTxId {
         // Collect deletes: row pointers live in the committed state; read them
         // without deleting.
         for (table_id, delete_table) in &self.tx_state.delete_tables {
-            if let Ok((table, blob_store, _)) =
-                self.committed_state_write_lock.get_table_and_blob_store(*table_id)
-            {
+            if let Ok((table, blob_store, _)) = self.committed_state_write_lock.get_table_and_blob_store(*table_id) {
                 let rows: std::sync::Arc<[ProductValue]> = delete_table
                     .iter()
                     .map(|row_ptr| {

crates/smoketests/tests/smoketests/cross_db_2pc.rs

@@ -91,18 +91,12 @@ fn test_cross_db_2pc_happy_path() {
     // Publish bank B (the participant that will be debited).
     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");
+    let db_b_identity = test.database_identity.clone().expect("bank B identity not set");
 
     // Publish bank A (the coordinator that will be credited).
     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");
+    let _db_a_identity = test.database_identity.clone().expect("bank A identity not set");
 
     // Transfer 50 from B's alice to A's alice.
     // The coordinator is bank A. It credits locally, then calls debit on B via 2PC.
@@ -155,10 +149,7 @@ fn test_cross_db_2pc_abort_insufficient_funds() {
     // 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");
+    let db_b_identity = test.database_identity.clone().expect("bank B identity not set");
 
     // Publish bank A.
     test.publish_module_named(&db_a_name, false)
@@ -167,7 +158,10 @@ fn test_cross_db_2pc_abort_insufficient_funds() {
     // Try to transfer 200 -- B only has 100, so the remote debit will fail.
     let result = test.call("transfer_funds", &[&db_b_identity, "alice", "alice", "200"]);
     // The call should fail because the remote debit panicked.
-    assert!(result.is_err(), "Expected transfer_funds to fail due to insufficient funds");
+    assert!(
+        result.is_err(),
+        "Expected transfer_funds to fail due to insufficient funds"
+    );
 
     // Verify bank A: alice should still have 100 (the local credit was rolled back).
     let result_a = test

tools/tpcc-runner/src/client.rs

@@ -86,14 +86,10 @@ impl ModuleClient {
         increment_pending(pending);
         let pending_for_callback = Arc::clone(pending);
         let errors = Arc::clone(errors);
-        if let Err(err) = self
-            .conn
-            .reducers
-            .load_remote_warehouses_then(rows, move |_, res| {
-                handle_reducer_result("load_remote_warehouses", res, &errors);
-                decrement_pending(&pending_for_callback);
-            })
-        {
+        if let Err(err) = self.conn.reducers.load_remote_warehouses_then(rows, move |_, res| {
+            handle_reducer_result("load_remote_warehouses", res, &errors);
+            decrement_pending(&pending_for_callback);
+        }) {
             decrement_pending(pending);
             return Err(anyhow!("load_remote_warehouses send error: {err}"));
         }