outbox macro change

Author: Shubham8287

crates/bindings-macro/src/table.rs

@@ -25,15 +25,26 @@ pub(crate) struct TableArgs {
     outbox: Option<OutboxArg>,
 }
 
-/// Parsed from `outbox(remote_reducer_fn)` optionally followed by `on_result(local_reducer_fn)`.
+/// Parsed from `outbox(remote_reducer_fn, on_result = local_reducer_fn)`.
+///
+/// For backwards compatibility, `on_result(local_reducer_fn)` is also accepted as a sibling
+/// table attribute and attached to the previously-declared outbox.
 struct OutboxArg {
     span: Span,
-    /// Path to the remote-side reducer function (used only for its name via `FnInfo::NAME`).
+    /// Path to the remote-side reducer function, used only for its final path segment.
     remote_reducer: Path,
     /// Path to the local `on_result` reducer, if any.
     on_result_reducer: Option<Path>,
 }
 
+fn path_tail_name(path: &Path) -> LitStr {
+    let segment = path
+        .segments
+        .last()
+        .expect("syn::Path should always contain at least one segment");
+    LitStr::new(&segment.ident.to_string(), segment.ident.span())
+}
+
 enum TableAccess {
     Public(Span),
     Private(Span),
@@ -268,18 +279,44 @@ impl ScheduledArg {
 }
 
 impl OutboxArg {
-    /// Parse `outbox(remote_reducer_path)`.
+    /// Parse `outbox(remote_reducer_path, on_result = local_reducer_path)`.
     ///
-    /// `on_result` is parsed separately via `parse_single_path_meta` and attached afterwards.
+    /// For backwards compatibility, `on_result` may also be parsed separately via
+    /// `parse_single_path_meta` and attached afterwards.
     fn parse_meta(meta: ParseNestedMeta) -> syn::Result<Self> {
         let span = meta.path.span();
         let mut remote_reducer: Option<Path> = None;
+        let mut on_result_reducer: Option<Path> = None;
 
         meta.parse_nested_meta(|meta| {
-            if meta.input.peek(syn::Token![=]) || meta.input.peek(syn::token::Paren) {
-                Err(meta.error("outbox takes a single function path, e.g. `outbox(my_remote_reducer)`"))
+            if meta.input.peek(Token![=]) {
+                if meta.path.is_ident("on_result") {
+                    check_duplicate_msg(
+                        &on_result_reducer,
+                        &meta,
+                        "can only specify one on_result reducer",
+                    )?;
+                    on_result_reducer = Some(meta.value()?.parse()?);
+                    Ok(())
+                } else {
+                    Err(meta.error(
+                        "outbox only supports `on_result = my_local_reducer` as a named argument",
+                    ))
+                }
+            } else if meta.input.peek(syn::token::Paren) {
+                Err(meta.error(
+                    "outbox expects a remote reducer path and optional `on_result = my_local_reducer`, e.g. `outbox(my_remote_reducer, on_result = my_local_reducer)`",
+                ))
+            } else if meta.path.is_ident("on_result") {
+                Err(meta.error(
+                    "outbox `on_result` must use `=`, e.g. `outbox(my_remote_reducer, on_result = my_local_reducer)`",
+                ))
             } else {
-                check_duplicate_msg(&remote_reducer, &meta, "can only specify one remote reducer for outbox")?;
+                check_duplicate_msg(
+                    &remote_reducer,
+                    &meta,
+                    "can only specify one remote reducer for outbox",
+                )?;
                 remote_reducer = Some(meta.path);
                 Ok(())
             }
@@ -291,7 +328,7 @@ impl OutboxArg {
         Ok(Self {
             span,
             remote_reducer,
-            on_result_reducer: None,
+            on_result_reducer,
         })
     }
 
@@ -1209,19 +1246,24 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
                 ));
             }
 
-            let remote_reducer = &ob.remote_reducer;
-            let on_result_reducer_name = match &ob.on_result_reducer {
-                Some(r) => quote!(Some(<#r as spacetimedb::rt::FnInfo>::NAME)),
+            let remote_reducer_name = path_tail_name(&ob.remote_reducer);
+            let on_result_reducer_name = match ob.on_result_reducer.as_ref().map(path_tail_name) {
+                Some(r) => quote!(Some(#r)),
                 None => quote!(None),
             };
             let desc = quote!(spacetimedb::table::OutboxDesc {
-                remote_reducer_name: <#remote_reducer as spacetimedb::rt::FnInfo>::NAME,
+                remote_reducer_name: #remote_reducer_name,
                 on_result_reducer_name: #on_result_reducer_name,
             });
 
             let primary_key_ty = primary_key_column.ty;
+            let callback_typecheck = ob
+                .on_result_reducer
+                .as_ref()
+                .map(|r| quote!(spacetimedb::rt::outbox_typecheck::<#original_struct_ident>(#r);));
             let typecheck = quote! {
                 spacetimedb::rt::assert_outbox_table_primary_key::<#primary_key_ty>();
+                #callback_typecheck
             };
 
             Ok((desc, typecheck))

crates/core/src/host/host_controller.rs

@@ -44,6 +44,7 @@ use spacetimedb_schema::def::{ModuleDef, RawModuleDefVersion};
 use spacetimedb_table::page_pool::PagePool;
 use std::future::Future;
 use std::ops::Deref;
+use std::sync::atomic::{AtomicU16, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{watch, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock as AsyncRwLock};
@@ -118,6 +119,8 @@ pub struct HostController {
     db_cores: JobCores,
     /// The pool of buffers used to build `BsatnRowList`s in subscriptions.
     pub bsatn_rlb_pool: BsatnRowListBuilderPool,
+    /// HTTP port used by local IDC delivery.
+    idc_http_port: Arc<AtomicU16>,
 }
 
 pub(crate) struct HostRuntimes {
@@ -233,6 +236,7 @@ impl HostController {
             page_pool: PagePool::new(default_config.page_pool_max_size),
             bsatn_rlb_pool: BsatnRowListBuilderPool::new(),
             db_cores,
+            idc_http_port: Arc::new(AtomicU16::new(80)),
         }
     }
 
@@ -241,6 +245,10 @@ impl HostController {
         self.program_storage = ps;
     }
 
+    pub fn set_idc_http_port(&self, port: u16) {
+        self.idc_http_port.store(port, Ordering::Relaxed);
+    }
+
     /// Get a [`ModuleHost`] managed by this controller, or launch it from
     /// persistent state.
     ///
@@ -1117,6 +1125,7 @@ impl Host {
             replica_ctx.relational_db().clone(),
             IdcActorConfig {
                 sender_identity: replica_ctx.database_identity,
+                http_port: host_controller.idc_http_port.load(Ordering::Relaxed),
             },
             module_host.downgrade(),
         );

crates/core/src/host/idc_actor.rs

@@ -4,7 +4,7 @@
 /// 1. Loads undelivered entries from `st_outbound_msg` on startup, resolving delivery data from outbox tables.
 /// 2. Accepts immediate notifications via an mpsc channel when new outbox rows are inserted.
 /// 3. Delivers each message in msg_id order via HTTP POST to
-///    `http://localhost:80/v1/database/{target_db}/call-from-database/{reducer}?sender_identity=<hex>&msg_id=<n>`
+///    `http://localhost:{http_port}/v1/database/{target_db}/call-from-database/{reducer}?sender_identity=<hex>&msg_id=<n>`
 /// 4. On transport errors (network, 5xx, 4xx except 422/402): retries infinitely with exponential
 ///    backoff, blocking only the affected target database (other targets continue unaffected).
 /// 5. On reducer errors (HTTP 422) or budget exceeded (HTTP 402): calls the configured
@@ -16,14 +16,13 @@ use crate::host::FunctionArgs;
 use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_datastore::system_tables::{StOutboundMsgRow, ST_OUTBOUND_MSG_ID};
 use spacetimedb_datastore::traits::IsolationLevel;
-use spacetimedb_lib::{AlgebraicValue, Identity};
+use spacetimedb_lib::{AlgebraicValue, Identity, ProductValue};
 use spacetimedb_primitives::{ColId, TableId};
 use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::sync::mpsc;
 
-const IDC_HTTP_PORT: u16 = 80;
 const INITIAL_BACKOFF: Duration = Duration::from_millis(100);
 const MAX_BACKOFF: Duration = Duration::from_secs(30);
 /// How long to wait before polling again when there is no work.
@@ -38,6 +37,7 @@ pub type IdcActorSender = mpsc::UnboundedSender<()>;
 /// The identity of this (sender) database, set when the IDC actor is started.
 pub struct IdcActorConfig {
     pub sender_identity: Identity,
+    pub http_port: u16,
 }
 
 /// A handle that, when dropped, stops the IDC actor background task.
@@ -86,6 +86,7 @@ struct PendingMessage {
     target_db_identity: Identity,
     target_reducer: String,
     args_bsatn: Vec<u8>,
+    request_row: ProductValue,
     /// From the outbox table's `TableSchema::on_result_reducer`.
     on_result_reducer: Option<String>,
 }
@@ -171,7 +172,7 @@ async fn run_idc_loop(
                         log::warn!(
                             "idc_actor: transport error delivering msg_id={} to {}: {reason}",
                             msg.msg_id,
-                            hex::encode(msg.target_db_identity.to_byte_array()),
+                            msg.target_db_identity.to_hex(),
                         );
                         state.record_transport_error();
                         // Do NOT pop the front — keep retrying this message for this target.
@@ -246,19 +247,23 @@ async fn finalize_message(
             return;
         };
 
-        // Encode (result_payload: String) as BSATN args.
-        // The on_result reducer is expected to accept a single String argument.
-        let args_bytes = match spacetimedb_sats::bsatn::to_vec(&result_payload) {
-            Ok(b) => b,
-            Err(e) => {
-                log::error!(
-                    "idc_actor: failed to encode on_result args for msg_id={}: {e}",
-                    msg.msg_id
-                );
-                delete_message(db, msg.msg_id);
-                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))
+        {
+            log::error!(
+                "idc_actor: failed to encode on_result args for msg_id={}: {e}",
+                msg.msg_id
+            );
+            delete_message(db, msg.msg_id);
+            return;
+        }
 
         let caller_identity = Identity::ZERO; // system call
         let result = host
@@ -366,12 +371,24 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
 
         let pv = outbox_row_ref.to_product_value();
 
-        // Col 1: target_db_identity (Identity stored as U256).
-        let target_db_identity = match pv.elements.get(1) {
-            Some(AlgebraicValue::U256(u)) => Identity::from_u256(**u),
+        // Col 1: target_db_identity stored as SATS `Identity`,
+        // i.e. the product wrapper `(__identity__: U256)`.
+        let target_db_identity: Identity = match pv.elements.get(1) {
+            Some(AlgebraicValue::Product(identity_pv)) if identity_pv.elements.len() == 1 => {
+                match &identity_pv.elements[0] {
+                    AlgebraicValue::U256(u) => Identity::from_u256(**u),
+                    other => {
+                        log::error!(
+                            "idc_actor: outbox row col 1 expected Identity inner U256, got {other:?} (msg_id={})",
+                            st_row.msg_id,
+                        );
+                        continue;
+                    }
+                }
+            }
             other => {
                 log::error!(
-                    "idc_actor: outbox row col 1 expected U256 (Identity), got {other:?} (msg_id={})",
+                    "idc_actor: outbox row col 1 expected Identity wrapper, got {other:?} (msg_id={})",
                     st_row.msg_id,
                 );
                 continue;
@@ -392,6 +409,7 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
             target_db_identity,
             target_reducer,
             args_bsatn,
+            request_row: pv,
             on_result_reducer,
         });
     }
@@ -412,17 +430,13 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
 }
 
 /// Attempt a single HTTP delivery of a message.
-async fn attempt_delivery(
-    client: &reqwest::Client,
-    config: &IdcActorConfig,
-    msg: &PendingMessage,
-) -> DeliveryOutcome {
-    let target_db_hex = hex::encode(msg.target_db_identity.to_byte_array());
-    let sender_hex = hex::encode(config.sender_identity.to_byte_array());
+async fn attempt_delivery(client: &reqwest::Client, config: &IdcActorConfig, msg: &PendingMessage) -> DeliveryOutcome {
+    let target_db_hex = msg.target_db_identity.to_hex();
+    let sender_hex = config.sender_identity.to_hex();
 
     let url = format!(
-        "http://localhost:{IDC_HTTP_PORT}/v1/database/{target_db_hex}/call-from-database/{}?sender_identity={sender_hex}&msg_id={}",
-        msg.target_reducer, msg.msg_id,
+        "http://localhost:{}/v1/database/{target_db_hex}/call-from-database/{}?sender_identity={sender_hex}&msg_id={}",
+        config.http_port, msg.target_reducer, msg.msg_id,
     );
 
     let result = client

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

@@ -1188,13 +1188,15 @@ async fn spawn_instance_worker(
                     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),
-                    );
+                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)
+                        });
                     worker_trapped.store(trapped, Ordering::Relaxed);
                     send_worker_reply("call_reducer", reply_tx, res, trapped);
                     should_exit = trapped;

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

@@ -472,7 +472,11 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         res
     }
 
-    pub fn call_reducer_delete_outbound_on_success(&mut self, params: CallReducerParams, msg_id: u64) -> ReducerCallResult {
+    pub fn call_reducer_delete_outbound_on_success(
+        &mut self,
+        params: CallReducerParams,
+        msg_id: u64,
+    ) -> ReducerCallResult {
         let (res, trapped) = crate::callgrind_flag::invoke_allowing_callgrind(|| {
             self.common
                 .call_reducer_with_tx_and_success_action(None, params, &mut self.instance, |tx| {

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -19,10 +19,10 @@ use crate::{
         is_built_in_meta_row, system_tables, table_id_is_reserved, StColumnRow, StConstraintData, StConstraintRow,
         StFields, StIndexRow, StSequenceFields, StSequenceRow, StTableFields, StTableRow, StViewRow, SystemTable,
         ST_CLIENT_ID, ST_CLIENT_IDX, ST_COLUMN_ID, ST_COLUMN_IDX, ST_COLUMN_NAME, ST_CONSTRAINT_ID, ST_CONSTRAINT_IDX,
-        ST_CONSTRAINT_NAME, ST_INBOUND_MSG_ID, ST_INDEX_ID, ST_INDEX_IDX, ST_INDEX_NAME, ST_MODULE_ID,
-        ST_MODULE_IDX, ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_IDX, ST_SCHEDULED_ID, ST_SCHEDULED_IDX,
-        ST_SEQUENCE_ID, ST_SEQUENCE_IDX, ST_SEQUENCE_NAME, ST_TABLE_ID, ST_TABLE_IDX, ST_VAR_ID, ST_VAR_IDX,
-        ST_VIEW_ARG_ID, ST_VIEW_ARG_IDX,
+        ST_CONSTRAINT_NAME, ST_INBOUND_MSG_ID, ST_INDEX_ID, ST_INDEX_IDX, ST_INDEX_NAME, ST_MODULE_ID, ST_MODULE_IDX,
+        ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_IDX, ST_SCHEDULED_ID, ST_SCHEDULED_IDX, ST_SEQUENCE_ID,
+        ST_SEQUENCE_IDX, ST_SEQUENCE_NAME, ST_TABLE_ID, ST_TABLE_IDX, ST_VAR_ID, ST_VAR_IDX, ST_VIEW_ARG_ID,
+        ST_VIEW_ARG_IDX,
     },
     traits::{EphemeralTables, TxData},
 };

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -2697,7 +2697,6 @@ impl MutTxId {
         })
     }
 
-
     pub fn insert_via_serialize_bsatn<'a, T: Serialize>(
         &'a mut self,
         table_id: TableId,

crates/datastore/src/system_tables.rs

@@ -710,7 +710,10 @@ fn system_module_def() -> ModuleDef {
 
     let st_outbound_msg_type = builder.add_type::<StOutboundMsgRow>();
     builder
-        .build_table(ST_OUTBOUND_MSG_NAME, *st_outbound_msg_type.as_ref().expect("should be ref"))
+        .build_table(
+            ST_OUTBOUND_MSG_NAME,
+            *st_outbound_msg_type.as_ref().expect("should be ref"),
+        )
         .with_type(TableType::System)
         .with_auto_inc_primary_key(StOutboundMsgFields::MsgId)
         .with_index_no_accessor_name(btree(StOutboundMsgFields::MsgId));