fmt

Author: Shubham8287

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

@@ -236,7 +236,6 @@ pub struct CallFromDatabaseQuery {
     msg_id: u64,
 }
 
-
 /// Call a reducer on behalf of another database, with deduplication.
 ///
 /// Endpoint: `POST /database/:name_or_identity/call-from-database/:reducer`
@@ -320,12 +319,10 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
                 ),
                 // 422 = reducer ran but returned Err; the IDC actor uses this to distinguish
                 // reducer failures from other  errors (which it retries).
-                ReducerOutcome::Failed(errmsg) => {
-                    (
-                        StatusCode::UNPROCESSABLE_ENTITY,
-                        axum::body::Body::from(errmsg.to_string()),
-                    )
-                }
+                ReducerOutcome::Failed(errmsg) => (
+                    StatusCode::UNPROCESSABLE_ENTITY,
+                    axum::body::Body::from(errmsg.to_string()),
+                ),
                 // This will be retried by IDC acttor
                 ReducerOutcome::BudgetExceeded => {
                     log::warn!(

crates/core/src/host/idc_actor.rs

@@ -18,9 +18,7 @@ use anyhow::anyhow;
 use bytes::Bytes;
 use spacetimedb_datastore::execution_context::{ReducerContext, Workload};
 use spacetimedb_datastore::locking_tx_datastore::MutTxId;
-use spacetimedb_datastore::system_tables::{
-    StInboundMsgResultStatus, StOutboundMsgRow, ST_OUTBOUND_MSG_ID,
-};
+use spacetimedb_datastore::system_tables::{StInboundMsgResultStatus, StOutboundMsgRow, ST_OUTBOUND_MSG_ID};
 use spacetimedb_datastore::traits::IsolationLevel;
 use spacetimedb_lib::{AlgebraicValue, Identity, ProductValue};
 use spacetimedb_primitives::{ColId, TableId};
@@ -249,11 +247,15 @@ fn reducer_workload(module: &ModuleInfo, params: &CallReducerParams) -> Workload
     })
 }
 
-fn duplicate_result_from_st_inbound_row(row: spacetimedb_datastore::system_tables::StInboundMsgRow) -> ReducerCallResult {
+fn duplicate_result_from_st_inbound_row(
+    row: spacetimedb_datastore::system_tables::StInboundMsgRow,
+) -> ReducerCallResult {
     let outcome = match row.result_status {
         StInboundMsgResultStatus::Success => ReducerOutcome::Committed,
         StInboundMsgResultStatus::ReducerError => ReducerOutcome::Failed(Box::new(
-            String::from_utf8_lossy(&row.result_payload).into_owned().into_boxed_str(),
+            String::from_utf8_lossy(&row.result_payload)
+                .into_owned()
+                .into_boxed_str(),
         )),
     };
 
@@ -344,11 +346,9 @@ where
     call_reducer(
         Some(tx),
         params,
-        Box::new(move |tx, _reducer_return_value| {
-            match action {
-                ReducerSuccessActionKind::DeleteOutboundMsg(msg_id) => {
-                    tx.delete_outbound_msg(msg_id).map_err(|e| anyhow!(e))
-                }
+        Box::new(move |tx, _reducer_return_value| match action {
+            ReducerSuccessActionKind::DeleteOutboundMsg(msg_id) => {
+                tx.delete_outbound_msg(msg_id).map_err(|e| anyhow!(e))
             }
         }),
     )
@@ -428,7 +428,6 @@ async fn finalize_message(
                 return;
             }
             Err(e) => {
-
                 delete_message(db, msg.msg_id);
                 log::error!(
                     "idc_actor: on_result reducer '{}' failed for msg_id={}: {e:?}",
@@ -438,7 +437,6 @@ async fn finalize_message(
             }
         }
     }
-
 }
 
 /// Load all messages from ST_OUTBOUND_MSG into the per-target queues, resolving delivery data
@@ -559,7 +557,9 @@ fn load_pending_into_targets(db: &RelationalDB, db_queues: &mut HashMap<Identity
     pending.sort_by_key(|m| m.msg_id);
 
     for msg in pending {
-        let state = db_queues.entry(msg.target_db_identity).or_insert_with(DatabaseQueue::new);
+        let state = db_queues
+            .entry(msg.target_db_identity)
+            .or_insert_with(DatabaseQueue::new);
         // Only add if not already in the queue (avoid duplicates after reload).
         let already_queued = state.queue.iter().any(|m| m.msg_id == msg.msg_id);
         if !already_queued {

crates/core/src/host/v8/mod.rs

@@ -1198,9 +1198,7 @@ async fn spawn_instance_worker(
                 |tx, params, on_success: crate::host::idc_actor::ReducerSuccessAction| {
                     instance_common.call_reducer_with_tx(tx, params, &mut inst, on_success)
                 };
-            let mut call_reducer = |tx, params| {
-                call_reducer_with_success(tx, params, Box::new(|_tx, _ret| Ok(())))
-            };
+            let mut call_reducer = |tx, params| call_reducer_with_success(tx, params, Box::new(|_tx, _ret| Ok(())));
             let mut should_exit = false;
 
             core_pinner.pin_if_changed();

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

@@ -514,7 +514,6 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         res
     }
 
-
     pub fn clear_all_clients(&self) -> anyhow::Result<()> {
         self.common.clear_all_clients()
     }
@@ -597,7 +596,8 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         F: FnOnce(&mut MutTxId, &Option<Bytes>) -> anyhow::Result<()>,
     {
         crate::callgrind_flag::invoke_allowing_callgrind(|| {
-            self.common.call_reducer_with_tx(tx, params, &mut self.instance, on_success)
+            self.common
+                .call_reducer_with_tx(tx, params, &mut self.instance, on_success)
         })
     }