simplify

Shubham8287 · Shubham8287 · commit 3011ffd51abe · 2026-04-10T10:10:39.000+05:30
diff --git a/crates/core/src/host/idc_actor.rs b/crates/core/src/host/idc_actor.rs
@@ -11,10 +11,13 @@
 ///    `on_result_reducer` (read from the outbox table's schema) and deletes the st_outbound_msg row.
 /// 6. Enforces sequential delivery per target database: msg N+1 is only delivered after N is done.
 use crate::db::relational_db::RelationalDB;
-use crate::host::module_host::WeakModuleHost;
-use crate::host::FunctionArgs;
+use crate::energy::EnergyQuanta;
+use crate::host::module_host::{CallReducerParams, ModuleInfo, WeakModuleHost};
+use crate::host::{FunctionArgs, ReducerCallResult, ReducerOutcome};
+use anyhow::anyhow;
 use bytes::Bytes;
-use spacetimedb_datastore::execution_context::Workload;
+use spacetimedb_datastore::execution_context::{ReducerContext, Workload};
+use spacetimedb_datastore::locking_tx_datastore::MutTxId;
 use spacetimedb_datastore::system_tables::{st_inbound_msg_result_status, StOutboundMsgRow, ST_OUTBOUND_MSG_ID};
 use spacetimedb_datastore::traits::IsolationLevel;
 use spacetimedb_lib::{AlgebraicValue, Identity, ProductValue};
@@ -224,6 +227,119 @@ fn outcome_to_result(outcome: &DeliveryOutcome) -> (u8, Bytes) {
     }
 }
 
+pub(crate) type ReducerSuccessAction = Box<dyn FnOnce(&mut MutTxId, &Option<Bytes>) -> anyhow::Result<()> + Send>;
+
+fn reducer_workload(module: &ModuleInfo, params: &CallReducerParams) -> Workload {
+    let reducer_def = module.module_def.reducer_by_id(params.reducer_id);
+    Workload::Reducer(ReducerContext {
+        name: reducer_def.name.clone(),
+        caller_identity: params.caller_identity,
+        caller_connection_id: params.caller_connection_id,
+        timestamp: params.timestamp,
+        arg_bsatn: params.args.get_bsatn().clone(),
+    })
+}
+
+fn duplicate_result_from_row(row: spacetimedb_datastore::system_tables::StInboundMsgRow) -> ReducerCallResult {
+    let outcome = match row.result_status {
+        st_inbound_msg_result_status::SUCCESS => ReducerOutcome::Committed,
+        st_inbound_msg_result_status::REDUCER_ERROR => ReducerOutcome::Failed(Box::new(
+            String::from_utf8_lossy(&row.result_payload).into_owned().into_boxed_str(),
+        )),
+        status => {
+            log::warn!(
+                "IDC: unexpected inbound dedup result_status={} for sender {}",
+                status,
+                Identity::from(row.database_identity)
+            );
+            ReducerOutcome::Failed(Box::new("unexpected inbound dedup result status".into()))
+        }
+    };
+
+    ReducerCallResult {
+        outcome,
+        reducer_return_value: (row.result_status == st_inbound_msg_result_status::SUCCESS).then_some(row.result_payload),
+        energy_used: EnergyQuanta::ZERO,
+        execution_duration: Duration::ZERO,
+        tx_offset: None,
+    }
+}
+
+fn record_failed_inbound_result(db: &RelationalDB, sender_identity: Identity, sender_msg_id: u64, error: &str) {
+    let mut dedup_tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
+    if let Err(e) = dedup_tx.upsert_inbound_last_msg(
+        sender_identity,
+        sender_msg_id,
+        st_inbound_msg_result_status::REDUCER_ERROR,
+        error.to_string().into(),
+    ) {
+        log::error!("IDC: failed to record reducer error in dedup table for sender {sender_identity}: {e}");
+        let _ = db.rollback_mut_tx(dedup_tx);
+    } else if let Err(e) = db.commit_tx(dedup_tx) {
+        log::error!("IDC: failed to commit dedup error record for sender {sender_identity}: {e}");
+    }
+}
+
+pub(crate) fn call_reducer_from_database<F>(
+    module: &ModuleInfo,
+    db: &RelationalDB,
+    params: CallReducerParams,
+    sender_identity: Identity,
+    sender_msg_id: u64,
+    call_reducer: F,
+) -> (ReducerCallResult, bool)
+where
+    F: FnOnce(Option<MutTxId>, CallReducerParams, ReducerSuccessAction) -> (ReducerCallResult, bool),
+{
+    let tx = db.begin_mut_tx(IsolationLevel::Serializable, reducer_workload(module, &params));
+    if let Some(row) = tx.get_inbound_msg_row(sender_identity)
+        && sender_msg_id <= row.last_outbound_msg
+    {
+        let _ = db.rollback_mut_tx(tx);
+        return (duplicate_result_from_row(row), false);
+    }
+
+    let (result, trapped) = call_reducer(
+        Some(tx),
+        params,
+        Box::new(move |tx, reducer_return_value| {
+            tx.upsert_inbound_last_msg(
+                sender_identity,
+                sender_msg_id,
+                st_inbound_msg_result_status::SUCCESS,
+                reducer_return_value.clone().unwrap_or_default(),
+            )
+            .map_err(anyhow::Error::from)
+        }),
+    );
+
+    if let ReducerOutcome::Failed(err) = &result.outcome {
+        record_failed_inbound_result(db, sender_identity, sender_msg_id, err);
+    }
+
+    (result, trapped)
+}
+
+pub(crate) fn call_reducer_delete_outbound_on_success<F>(
+    module: &ModuleInfo,
+    db: &RelationalDB,
+    params: CallReducerParams,
+    msg_id: u64,
+    call_reducer: F,
+) -> (ReducerCallResult, bool)
+where
+    F: FnOnce(Option<MutTxId>, CallReducerParams, ReducerSuccessAction) -> (ReducerCallResult, bool),
+{
+    let tx = db.begin_mut_tx(IsolationLevel::Serializable, reducer_workload(module, &params));
+    call_reducer(
+        Some(tx),
+        params,
+        Box::new(move |tx, _reducer_return_value| {
+            tx.delete_outbound_msg(msg_id).map_err(|e| anyhow!(e))
+        }),
+    )
+}
+
 /// Finalize a delivered message: call the on_result reducer (if any), then delete from ST_OUTBOUND_MSG.
 ///
 /// On the happy path, `on_result_reducer` success and deletion of `st_outbound_msg`
diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs
@@ -624,16 +624,6 @@ pub struct CallReducerParams {
     pub timer: Option<Instant>,
     pub reducer_id: ReducerId,
     pub args: ArgsTuple,
-    /// If set, enables at-most-once delivery semantics for database-to-database calls.
-    ///
-    /// The tuple is `(sender_database_identity, sender_msg_id)`.
-    /// Before running the reducer, `st_inbound_msg` is consulted:
-    /// if `sender_msg_id` ≤ the stored last-delivered msg_id for the sender,
-    /// the call is a duplicate and returns the stored committed or failed result
-    /// without running the reducer again.
-    /// Otherwise the reducer runs, and the stored msg_id is updated atomically
-    /// within the same transaction.
-    pub dedup_sender: Option<(Identity, u64)>,
 }
 
 impl CallReducerParams {
@@ -654,7 +644,6 @@ impl CallReducerParams {
             timer: None,
             reducer_id,
             args,
-            dedup_sender: None,
         }
     }
 }
@@ -1578,7 +1567,6 @@ impl ModuleHost {
             timer,
             reducer_id,
             args,
-            dedup_sender: None,
         })
     }
 
@@ -1606,7 +1594,6 @@ impl ModuleHost {
             timer,
             reducer_id,
             args,
-            dedup_sender: None,
         };
 
         self.call(
@@ -1643,7 +1630,6 @@ impl ModuleHost {
             timer,
             reducer_id,
             args,
-            dedup_sender: None,
         };
 
         self.call(
@@ -1793,27 +1779,28 @@ impl ModuleHost {
                 return Err(ReducerCallError::NoSuchReducer);
             }
 
-            let args = args
-                .into_tuple_for_def(&self.info.module_def, reducer_def)
-                .map_err(InvalidReducerArguments)?;
-            let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
-            let call_reducer_params = CallReducerParams {
-                timestamp: Timestamp::now(),
+            let call_reducer_params = Self::call_reducer_params(
+                &self.info,
                 caller_identity,
                 caller_connection_id,
                 client,
                 request_id,
                 timer,
                 reducer_id,
+                reducer_def,
                 args,
-                dedup_sender: Some((sender_database_identity, sender_msg_id)),
-            };
+            )?;
 
             self.call(
                 &reducer_def.name,
-                call_reducer_params,
-                async |p, inst| Ok(inst.call_reducer(p)),
-                async |p, inst| inst.call_reducer(p).await,
+                (call_reducer_params, sender_database_identity, sender_msg_id),
+                async |(p, sender_database_identity, sender_msg_id), inst| {
+                    Ok(inst.call_reducer_from_database(p, sender_database_identity, sender_msg_id))
+                },
+                async |(p, sender_database_identity, sender_msg_id), inst| {
+                    inst.call_reducer_from_database(p, sender_database_identity, sender_msg_id)
+                        .await
+                },
             )
             .await?
         }
diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs
@@ -443,7 +443,7 @@ pub(super) async fn call_scheduled_function(
                 // print their message and backtrace when they occur, so we don't need to do
                 // anything with the error payload.
                 let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
-                    inst_common.call_reducer_with_tx(Some(tx), params, inst)
+                    inst_common.call_reducer_with_tx(Some(tx), params, inst, |_tx, _ret| Ok(()))
                 }));
                 let reschedule = delete_scheduled_function_row(module_info, db, id, None, None, inst_common, inst);
                 // Currently, we drop the return value from the function call. In the future,
diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs
@@ -586,6 +586,13 @@ enum JsWorkerRequest {
         reply_tx: JsReplyTx<ReducerCallResult>,
         params: CallReducerParams,
     },
+    /// See [`JsInstance::call_reducer_from_database`].
+    CallReducerFromDatabase {
+        reply_tx: JsReplyTx<ReducerCallResult>,
+        params: CallReducerParams,
+        sender_identity: Identity,
+        sender_msg_id: u64,
+    },
     /// See [`JsInstance::call_reducer_delete_outbound_on_success`].
     CallReducerDeleteOutboundOnSuccess {
         reply_tx: JsReplyTx<ReducerCallResult>,
@@ -633,7 +640,7 @@ enum JsWorkerRequest {
     },
 }
 
-static_assert_size!(CallReducerParams, 256);
+static_assert_size!(CallReducerParams, 192);
 
 fn send_worker_reply<T>(ctx: &str, reply_tx: JsReplyTx<T>, value: T, trapped: bool) {
     if reply_tx.send(JsWorkerReply { value, trapped }).is_err() {
@@ -908,6 +915,25 @@ impl JsInstanceLane {
         .map_err(|_| ReducerCallError::WorkerError(instance_lane_worker_error("call_reducer")))
     }
 
+    pub async fn call_reducer_from_database(
+        &self,
+        params: CallReducerParams,
+        sender_identity: Identity,
+        sender_msg_id: u64,
+    ) -> Result<ReducerCallResult, ReducerCallError> {
+        self.run_once("call_reducer", |inst: JsInstance| async move {
+            inst.send_request(|reply_tx| JsWorkerRequest::CallReducerFromDatabase {
+                reply_tx,
+                params,
+                sender_identity,
+                sender_msg_id,
+            })
+            .await
+        })
+        .await
+        .map_err(|_| ReducerCallError::WorkerError(instance_lane_worker_error("call_reducer")))
+    }
+
     pub async fn call_reducer_delete_outbound_on_success(
         &self,
         params: CallReducerParams,
@@ -1166,7 +1192,15 @@ async fn spawn_instance_worker(
         let mut requests_since_heap_check = 0u64;
         let mut last_heap_check_at = Instant::now();
         for request in request_rx.iter() {
-            let mut call_reducer = |tx, params| instance_common.call_reducer_with_tx(tx, params, &mut inst);
+            let info_for_idc = instance_common.info();
+            let db_for_idc = inst.replica_ctx().relational_db().clone();
+            let mut call_reducer_with_success =
+                |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 should_exit = false;
 
             core_pinner.pin_if_changed();
@@ -1188,15 +1222,36 @@ async fn spawn_instance_worker(
                     send_worker_reply("call_reducer", reply_tx, res, trapped);
                     should_exit = trapped;
                 }
+                JsWorkerRequest::CallReducerFromDatabase {
+                    reply_tx,
+                    params,
+                    sender_identity,
+                    sender_msg_id,
+                } => {
+                    let (res, trapped) = crate::host::idc_actor::call_reducer_from_database(
+                        info_for_idc.as_ref(),
+                        db_for_idc.as_ref(),
+                        params,
+                        sender_identity,
+                        sender_msg_id,
+                        |tx, params, on_success| call_reducer_with_success(tx, params, on_success),
+                    );
+                    worker_trapped.store(trapped, Ordering::Relaxed);
+                    send_worker_reply("call_reducer", reply_tx, res, trapped);
+                    should_exit = trapped;
+                }
                 JsWorkerRequest::CallReducerDeleteOutboundOnSuccess {
                     reply_tx,
                     params,
                     msg_id,
                 } => {
-                    let (res, trapped) =
-                        instance_common.call_reducer_with_tx_and_success_action(None, params, &mut inst, |tx| {
-                            tx.delete_outbound_msg(msg_id).map_err(anyhow::Error::from)
-                        });
+                    let (res, trapped) = crate::host::idc_actor::call_reducer_delete_outbound_on_success(
+                        info_for_idc.as_ref(),
+                        db_for_idc.as_ref(),
+                        params,
+                        msg_id,
+                        |tx, params, on_success| call_reducer_with_success(tx, params, on_success),
+                    );
                     worker_trapped.store(trapped, Ordering::Relaxed);
                     send_worker_reply("call_reducer", reply_tx, res, trapped);
                     should_exit = trapped;
diff --git a/crates/core/src/host/wasm_common/module_host_actor.rs b/crates/core/src/host/wasm_common/module_host_actor.rs
