Preserve IDC reducer return values

Author: Shubham8287

crates/bindings/src/rt.rs

@@ -538,11 +538,12 @@ pub const fn assert_outbox_table_primary_key<T: OutboxTablePrimaryKey>() {}
 
 /// Verify at compile time that a function has the correct signature for an outbox `on_result` reducer.
 ///
-/// The reducer must accept `(OutboxRow, Result<(), String>)` as its user-supplied arguments:
-/// `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<(), String>)`
-pub const fn outbox_typecheck<'de, OutboxRow>(_x: impl Reducer<'de, (OutboxRow, Result<(), String>)>)
+/// The reducer must accept `(OutboxRow, Result<T, String>)` as its user-supplied arguments:
+/// `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<T, String>)`
+pub const fn outbox_typecheck<'de, OutboxRow, T>(_x: impl Reducer<'de, (OutboxRow, Result<T, String>)>)
 where
     OutboxRow: spacetimedb_lib::SpacetimeType + Serialize + Deserialize<'de>,
+    T: spacetimedb_lib::SpacetimeType + Serialize + Deserialize<'de>,
 {
     core::mem::forget(_x);
 }

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

@@ -310,15 +310,26 @@ pub async fn call_from_database<S: ControlStateDelegate + NodeDelegate>(
     match result {
         Ok(rcr) => {
             let (status, body) = match rcr.outcome {
-                ReducerOutcome::Committed => (StatusCode::OK, "".into()),
+                ReducerOutcome::Committed => (
+                    StatusCode::OK,
+                    axum::body::Body::from(rcr.reducer_return_value.unwrap_or_default()),
+                ),
                 // 422 = reducer ran but returned Err; the IDC actor uses this to distinguish
                 // reducer failures from transport errors (which it retries).
-                ReducerOutcome::Failed(errmsg) => (StatusCode::UNPROCESSABLE_ENTITY, *errmsg),
+                ReducerOutcome::Failed(errmsg) => {
+                    (
+                        StatusCode::UNPROCESSABLE_ENTITY,
+                        axum::body::Body::from(errmsg.to_string()),
+                    )
+                }
                 ReducerOutcome::BudgetExceeded => {
                     log::warn!(
                         "Node's energy budget exceeded for identity: {owner_identity} while executing {reducer}"
                     );
-                    (StatusCode::PAYMENT_REQUIRED, "Module energy budget exhausted.".into())
+                    (
+                        StatusCode::PAYMENT_REQUIRED,
+                        axum::body::Body::from("Module energy budget exhausted."),
+                    )
                 }
             };
             Ok((

crates/core/src/host/host_controller.rs

@@ -23,6 +23,7 @@ use crate::util::jobs::{AllocatedJobCore, JobCores};
 use crate::worker_metrics::WORKER_METRICS;
 use anyhow::{anyhow, bail, Context};
 use async_trait::async_trait;
+use bytes::Bytes;
 use durability::{Durability, EmptyHistory};
 use log::{info, trace, warn};
 use parking_lot::Mutex;
@@ -139,6 +140,7 @@ impl HostRuntimes {
 #[derive(Debug)]
 pub struct ReducerCallResult {
     pub outcome: ReducerOutcome,
+    pub reducer_return_value: Option<Bytes>,
     pub energy_used: EnergyQuanta,
     pub execution_duration: Duration,
     pub tx_offset: Option<TransactionOffset>,

crates/core/src/host/idc_actor.rs

@@ -13,8 +13,9 @@
 use crate::db::relational_db::RelationalDB;
 use crate::host::module_host::WeakModuleHost;
 use crate::host::FunctionArgs;
+use bytes::Bytes;
 use spacetimedb_datastore::execution_context::Workload;
-use spacetimedb_datastore::system_tables::{StOutboundMsgRow, ST_OUTBOUND_MSG_ID};
+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};
 use spacetimedb_primitives::{ColId, TableId};
@@ -130,7 +131,7 @@ impl TargetState {
 /// Outcome of a delivery attempt.
 enum DeliveryOutcome {
     /// Reducer succeeded (HTTP 200).
-    Success,
+    Success(Bytes),
     /// Reducer ran but returned Err (HTTP 422).
     ReducerError(String),
     /// Budget exceeded (HTTP 402).
@@ -211,14 +212,13 @@ async fn run_idc_loop(
 }
 
 /// Decode the delivery outcome into `(result_status, result_payload)` for recording.
-fn outcome_to_result(outcome: &DeliveryOutcome) -> (u8, String) {
-    use spacetimedb_datastore::system_tables::st_inbound_msg_result_status;
+fn outcome_to_result(outcome: &DeliveryOutcome) -> (u8, Bytes) {
     match outcome {
-        DeliveryOutcome::Success => (st_inbound_msg_result_status::SUCCESS, String::new()),
-        DeliveryOutcome::ReducerError(msg) => (st_inbound_msg_result_status::REDUCER_ERROR, msg.clone()),
+        DeliveryOutcome::Success(payload) => (st_inbound_msg_result_status::SUCCESS, payload.clone()),
+        DeliveryOutcome::ReducerError(msg) => (st_inbound_msg_result_status::REDUCER_ERROR, Bytes::from(msg.clone())),
         DeliveryOutcome::BudgetExceeded => (
             st_inbound_msg_result_status::REDUCER_ERROR,
-            "budget exceeded".to_string(),
+            Bytes::from("budget exceeded".to_string()),
         ),
         DeliveryOutcome::TransportError(_) => unreachable!("transport errors never finalize"),
     }
@@ -232,8 +232,8 @@ async fn finalize_message(
     db: &RelationalDB,
     module_host: &WeakModuleHost,
     msg: &PendingMessage,
-    _result_status: u8,
-    result_payload: String,
+    result_status: u8,
+    result_payload: Bytes,
 ) {
     // Call the on_result reducer if configured.
     if let Some(on_result_reducer) = &msg.on_result_reducer {
@@ -247,23 +247,37 @@ async fn finalize_message(
             return;
         };
 
-        // Encode `(request_row: OutboxRow, result: Result<(), String>)` as BSATN args.
-        let result = if result_payload.is_empty() {
-            Ok::<(), String>(())
-        } else {
-            Err::<(), String>(result_payload)
-        };
         let mut args_bytes = Vec::new();
-        if let Err(e) = spacetimedb_sats::bsatn::to_writer(&mut args_bytes, &msg.request_row)
-            .and_then(|_| spacetimedb_sats::bsatn::to_writer(&mut args_bytes, &result))
-        {
+        if let Err(e) = spacetimedb_sats::bsatn::to_writer(&mut args_bytes, &msg.request_row) {
             log::error!(
                 "idc_actor: failed to encode on_result args for msg_id={}: {e}",
                 msg.msg_id
             );
             delete_message(db, msg.msg_id);
             return;
         }
+        match result_status {
+            st_inbound_msg_result_status::SUCCESS => {
+                args_bytes.push(0);
+                args_bytes.extend_from_slice(&result_payload);
+            }
+            st_inbound_msg_result_status::REDUCER_ERROR => {
+                let err = String::from_utf8_lossy(&result_payload).into_owned();
+                if let Err(e) = spacetimedb_sats::bsatn::to_writer(&mut args_bytes, &Err::<(), String>(err)) {
+                    log::error!(
+                        "idc_actor: failed to encode on_result error args for msg_id={}: {e}",
+                        msg.msg_id
+                    );
+                    delete_message(db, msg.msg_id);
+                    return;
+                }
+            }
+            status => {
+                log::error!("idc_actor: unexpected result status {status} for msg_id={}", msg.msg_id);
+                delete_message(db, msg.msg_id);
+                return;
+            }
+        }
 
         let caller_identity = Identity::ZERO; // system call
         let result = host
@@ -452,7 +466,7 @@ async fn attempt_delivery(client: &reqwest::Client, config: &IdcActorConfig, msg
             let status = resp.status();
             if status.is_success() {
                 // HTTP 200: reducer committed successfully.
-                DeliveryOutcome::Success
+                DeliveryOutcome::Success(resp.bytes().await.unwrap_or_default())
             } else if status.as_u16() == 422 {
                 // HTTP 422: reducer ran but returned Err(...).
                 let body = resp.text().await.unwrap_or_default();

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

@@ -2,7 +2,7 @@ use super::instrumentation::CallTimes;
 use super::*;
 use crate::client::ClientActorId;
 use crate::database_logger;
-use crate::energy::{EnergyMonitor, EnergyQuanta, FunctionBudget, FunctionFingerprint};
+use crate::energy::{EnergyMonitor, FunctionBudget, FunctionFingerprint};
 use crate::error::DBError;
 use crate::host::host_controller::CallProcedureReturn;
 use crate::host::instance_env::{InstanceEnv, TxSlot};
@@ -880,41 +880,6 @@ impl InstanceCommon {
         let workload = Workload::Reducer(ReducerContext::from(op.clone()));
         let tx = tx.unwrap_or_else(|| stdb.begin_mut_tx(IsolationLevel::Serializable, workload));
 
-        // Check the dedup index atomically within this transaction.
-        // If the incoming msg_id is ≤ the last delivered msg_id for this sender,
-        // the message is a duplicate; discard it and return the stored result.
-        if let Some((sender_identity, sender_msg_id)) = dedup_sender {
-            let stored = tx.get_inbound_msg_row(sender_identity);
-            if stored.as_ref().is_some_and(|r| sender_msg_id <= r.last_outbound_msg) {
-                let _ = stdb.rollback_mut_tx(tx);
-                let stored = stored.unwrap();
-                let outcome = match stored.result_status {
-                    s if s == st_inbound_msg_result_status::SUCCESS => ReducerOutcome::Committed,
-                    s if s == st_inbound_msg_result_status::REDUCER_ERROR => {
-                        ReducerOutcome::Failed(Box::new(stored.result_payload.into()))
-                    }
-                    _ => {
-                        log::warn!(
-                            "IDC: unexpected inbound dedup result_status={} for sender {}, msg_id={}",
-                            stored.result_status,
-                            sender_identity,
-                            sender_msg_id
-                        );
-                        ReducerOutcome::Committed
-                    }
-                };
-                return (
-                    ReducerCallResult {
-                        outcome,
-                        energy_used: EnergyQuanta::ZERO,
-                        execution_duration: Duration::ZERO,
-                        tx_offset: None,
-                    },
-                    false,
-                );
-            }
-        }
-
         let mut tx_slot = inst.tx_slot();
 
         let vm_metrics = self.vm_metrics.get_for_reducer_id(reducer_id);
@@ -969,7 +934,7 @@ impl InstanceCommon {
                         sender_identity,
                         sender_msg_id,
                         st_inbound_msg_result_status::SUCCESS,
-                        String::new(),
+                        return_value.clone().unwrap_or_default(),
                     ),
                     None => Ok(()),
                 };
@@ -1049,7 +1014,7 @@ impl InstanceCommon {
                 sender_identity,
                 sender_msg_id,
                 st_inbound_msg_result_status::REDUCER_ERROR,
-                err_msg,
+                err_msg.into(),
             ) {
                 log::error!("IDC: failed to record reducer error in dedup table for sender {sender_identity}: {e}");
                 let _ = stdb.rollback_mut_tx(dedup_tx);
@@ -1060,6 +1025,7 @@ impl InstanceCommon {
 
         let res = ReducerCallResult {
             outcome: ReducerOutcome::from(&event.status),
+            reducer_return_value: event.reducer_return_value.clone(),
             energy_used: energy_quanta_used,
             execution_duration: total_duration,
             tx_offset: Some(tx_offset),

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -35,6 +35,7 @@ use crate::{
     locking_tx_datastore::state_view::ScanOrIndex,
     traits::{InsertFlags, RowTypeForTable, TxData, UpdateFlags},
 };
+use bytes::Bytes;
 use core::{cell::RefCell, iter, mem, ops::RangeBounds};
 use itertools::Either;
 use smallvec::SmallVec;
@@ -2636,7 +2637,7 @@ impl MutTxId {
         sender_identity: Identity,
         last_outbound_msg: u64,
         result_status: u8,
-        result_payload: String,
+        result_payload: Bytes,
     ) -> Result<()> {
         // Delete the existing row if present.
         self.delete_col_eq(

crates/datastore/src/system_tables.rs

@@ -11,6 +11,7 @@
 //! - Use [`st_fields_enum`] to define its column enum.
 //! - Register its schema in [`system_module_def`], making sure to call `validate_system_table` at the end of the function.
 
+use bytes::Bytes;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, HashMap};
 use spacetimedb_lib::db::auth::{StAccess, StTableType};
 use spacetimedb_lib::db::raw_def::v9::{btree, RawSql};
@@ -1936,8 +1937,10 @@ pub struct StInboundMsgRow {
     pub last_outbound_msg: u64,
     /// See [st_inbound_msg_result_status] for values.
     pub result_status: u8,
-    /// Error message if result_status is REDUCER_ERROR, empty otherwise.
-    pub result_payload: String,
+    /// Reducer return payload encoded as raw bytes.
+    /// For `SUCCESS`, this stores the committed reducer return value.
+    /// For `REDUCER_ERROR`, this stores the error message bytes.
+    pub result_payload: Bytes,
 }
 
 impl TryFrom<RowRef<'_>> for StInboundMsgRow {

crates/lib/src/db/raw_def/v10.rs

@@ -280,7 +280,7 @@ pub struct RawColumnDefaultValueV10 {
 ///
 /// The `remote_reducer` is the name of the reducer to call on the target database.
 /// If `on_result_reducer` is set, that local reducer is called when delivery completes,
-/// with signature `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<(), String>)`.
+/// with signature `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<T, String>)`.
 #[derive(Debug, Clone, SpacetimeType)]
 #[sats(crate = crate)]
 #[cfg_attr(feature = "test", derive(PartialEq, Eq, PartialOrd, Ord))]

crates/schema/src/def.rs

@@ -723,7 +723,7 @@ pub struct OutboxDef {
 
     /// The local reducer called with the delivery result, if any.
     ///
-    /// Signature: `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<(), String>)`.
+    /// Signature: `fn on_result(ctx: &ReducerContext, request: OutboxRow, result: Result<T, String>)`.
     pub on_result_reducer: Option<Identifier>,
 }