fix macro

Author: Shubham8287

crates/bindings-macro/src/table.rs

@@ -1131,6 +1131,9 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
         }
     };
 
+    // Save a clone before the schedule closure captures `primary_key_column` by move.
+    let primary_key_column_for_outbox = primary_key_column.clone();
+
     let (schedule, schedule_typecheck) = args
         .scheduled
         .as_ref()
@@ -1189,6 +1192,44 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
         .unzip();
     let schedule = schedule.into_iter();
 
+    let (outbox, outbox_typecheck) = args
+        .outbox
+        .as_ref()
+        .map(|ob| {
+            let primary_key_column = primary_key_column_for_outbox.ok_or_else(|| {
+                syn::Error::new(
+                    ob.span,
+                    "outbox tables must have a `#[primary_key] #[auto_inc]` u64 column at position 0 (the row_id)",
+                )
+            })?;
+            if primary_key_column.index != 0 {
+                return Err(syn::Error::new(
+                    ob.span,
+                    "outbox tables must have the `#[primary_key] #[auto_inc]` column as the first column (col 0)",
+                ));
+            }
+
+            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)),
+                None => quote!(None),
+            };
+            let desc = quote!(spacetimedb::table::OutboxDesc {
+                remote_reducer_name: <#remote_reducer as spacetimedb::rt::FnInfo>::NAME,
+                on_result_reducer_name: #on_result_reducer_name,
+            });
+
+            let primary_key_ty = primary_key_column.ty;
+            let typecheck = quote! {
+                spacetimedb::rt::assert_outbox_table_primary_key::<#primary_key_ty>();
+            };
+
+            Ok((desc, typecheck))
+        })
+        .transpose()?
+        .unzip();
+    let outbox = outbox.into_iter();
+
     let unique_err = if !unique_columns.is_empty() {
         quote!(spacetimedb::UniqueConstraintViolation)
     } else {
@@ -1222,6 +1263,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
             #(const PRIMARY_KEY: Option<u16> = Some(#primary_col_id);)*
             const SEQUENCES: &'static [u16] = &[#(#sequence_col_ids),*];
             #(const SCHEDULE: Option<spacetimedb::table::ScheduleDesc<'static>> = Some(#schedule);)*
+            #(const OUTBOX: Option<spacetimedb::table::OutboxDesc<'static>> = Some(#outbox);)*
 
             #table_id_from_name_func
             #default_fn
@@ -1373,6 +1415,7 @@ pub(crate) fn table_impl(mut args: TableArgs, item: &syn::DeriveInput) -> syn::R
         const _: () = {
             #(let _ = <#field_types as spacetimedb::rt::TableColumn>::_ITEM;)*
             #schedule_typecheck
+            #outbox_typecheck
             #default_type_check
         };
 

crates/bindings/src/rt.rs

@@ -533,6 +533,9 @@ impl<T: SpacetimeType> TableColumn for T {}
 /// Assert that the primary_key column of a scheduled table is a u64.
 pub const fn assert_scheduled_table_primary_key<T: ScheduledTablePrimaryKey>() {}
 
+/// Assert that the primary_key column of an outbox table is a u64.
+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:
@@ -555,6 +558,13 @@ pub trait ScheduledTablePrimaryKey: sealed::Sealed {}
 impl sealed::Sealed for u64 {}
 impl ScheduledTablePrimaryKey for u64 {}
 
+#[diagnostic::on_unimplemented(
+    message = "outbox table primary key must be a `u64`",
+    label = "should be `u64`, not `{Self}`"
+)]
+pub trait OutboxTablePrimaryKey: sealed::Sealed {}
+impl OutboxTablePrimaryKey for u64 {}
+
 /// Used in the last type parameter of `Reducer` to indicate that the
 /// context argument *should* be passed to the reducer logic.
 pub struct ContextArg;

crates/core/src/host/idc_runtime.rs

@@ -77,7 +77,11 @@ impl IdcRuntime {
 #[derive(Clone)]
 struct PendingMessage {
     msg_id: u64,
+    /// Stored for future use (e.g. deleting the outbox row after delivery).
+    #[allow(dead_code)]
     outbox_table_id: TableId,
+    /// Stored for future use (e.g. deleting the outbox row after delivery).
+    #[allow(dead_code)]
     row_id: u64,
     target_db_identity: Identity,
     target_reducer: String,
@@ -325,9 +329,19 @@ fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity,
             }
         };
 
-        let table_name = schema.table_name.to_string();
-        let target_reducer = table_name.strip_prefix("__outbox_").unwrap_or(&table_name).to_string();
-        let on_result_reducer = schema.on_result_reducer.clone();
+        let outbox_schema = match schema.outbox.as_ref() {
+            Some(o) => o,
+            None => {
+                log::error!(
+                    "idc_runtime: table {:?} (msg_id={}) is not an outbox table",
+                    schema.table_name,
+                    st_row.msg_id,
+                );
+                continue;
+            }
+        };
+        let target_reducer = outbox_schema.remote_reducer.to_string();
+        let on_result_reducer = outbox_schema.on_result_reducer.as_ref().map(|id| id.to_string());
 
         // Look up the outbox row by its auto-inc PK (col 0) to get target identity and args.
         let outbox_row = db

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -44,7 +44,6 @@ use smallvec::SmallVec;
 use spacetimedb_data_structures::map::{HashMap, HashSet, IntMap};
 use spacetimedb_durability::TxOffset;
 use spacetimedb_execution::{dml::MutDatastore, Datastore, DeltaStore, Row};
-use spacetimedb_lib::bsatn::ToBsatn;
 use spacetimedb_lib::{db::raw_def::v9::RawSql, metrics::ExecutionMetrics, Timestamp};
 use spacetimedb_lib::{
     db::{auth::StAccess, raw_def::SEQUENCE_ALLOCATION_STEP},
@@ -2777,40 +2776,6 @@ impl MutTxId {
         })
     }
 
-    /// Returns the table IDs and table names of all tables that had rows inserted
-    /// in this transaction and whose name starts with `"__outbox_"`.
-    ///
-    /// Used by the IDC runtime to detect outbox inserts and enqueue ST_OUTBOUND_MSG entries.
-    pub fn outbox_insert_table_ids(&self) -> Vec<(TableId, String)> {
-        let mut result = Vec::new();
-        for table_id in self.tx_state.insert_tables.keys() {
-            if let Ok(schema) = self.schema_for_table(*table_id) {
-                let name = schema.table_name.to_string();
-                if name.starts_with("__outbox_") {
-                    result.push((*table_id, name));
-                }
-            }
-        }
-        result
-    }
-
-    /// Returns the raw BSATN bytes of all rows inserted into `table_id` in this transaction.
-    ///
-    /// Each `Vec<u8>` encodes a full outbox row. The first 32 bytes are the target
-    /// database identity (U256 little-endian), and the remaining bytes are the reducer args.
-    pub fn outbox_inserts_for_table(&mut self, table_id: TableId) -> Vec<Vec<u8>> {
-        let Some(inserted_table) = self.tx_state.insert_tables.get(&table_id) else {
-            return Vec::new();
-        };
-        // Collect blob-store-independent data: we hold a shared ref to insert_tables,
-        // so we can't also borrow committed_state mutably for the blob store.
-        // Use the blob store embedded in the TxState.
-        let blob_store = &self.tx_state.blob_store;
-        inserted_table
-            .scan_rows(blob_store)
-            .filter_map(|row_ref| row_ref.to_bsatn_vec().ok())
-            .collect()
-    }
 
     pub fn insert_via_serialize_bsatn<'a, T: Serialize>(
         &'a mut self,

crates/schema/src/schema.rs

@@ -201,9 +201,8 @@ pub struct TableSchema {
     /// Whether this is an event table.
     pub is_event: bool,
 
-    /// For outbox tables (`__outbox_<reducer>`): the name of the local reducer to call
-    /// with the result of the remote reducer invocation. `None` means no callback.
-    pub on_result_reducer: Option<String>,
+    /// Outbox configuration if this is an outbox table; `None` for non-outbox tables.
+    pub outbox: Option<OutboxSchema>,
 
     /// Cache for `row_type_for_table` in the data store.
     pub row_type: ProductType,
@@ -231,7 +230,7 @@ impl TableSchema {
         primary_key: Option<ColId>,
         is_event: bool,
         alias: Option<Identifier>,
-        on_result_reducer: Option<String>,
+        outbox: Option<OutboxSchema>,
     ) -> Self {
         Self {
             row_type: columns_to_row_type(&columns),
@@ -248,7 +247,7 @@ impl TableSchema {
             primary_key,
             is_event,
             alias,
-            on_result_reducer,
+            outbox,
         }
     }
 
@@ -1041,10 +1040,10 @@ impl Schema for TableSchema {
             .as_ref()
             .map(|schedule| ScheduleSchema::from_module_def(module_def, schedule, table_id, ScheduleId::SENTINEL));
 
-        let on_result_reducer = outbox
-            .as_ref()
-            .and_then(|o| o.on_result_reducer.as_ref())
-            .map(|r| r.to_string());
+        let outbox_schema = outbox.as_ref().map(|o| OutboxSchema {
+            remote_reducer: o.remote_reducer.clone(),
+            on_result_reducer: o.on_result_reducer.clone(),
+        });
 
         TableSchema::new(
             table_id,
@@ -1060,7 +1059,7 @@ impl Schema for TableSchema {
             *primary_key,
             *is_event,
             Some(accessor_name.clone()),
-            on_result_reducer,
+            outbox_schema,
         )
     }
 
@@ -1443,6 +1442,15 @@ impl Schema for ScheduleSchema {
     }
 }
 
+/// Marks a table as an outbox table for inter-database communication.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OutboxSchema {
+    /// The name of the reducer to invoke on the target database.
+    pub remote_reducer: Identifier,
+    /// The local reducer called with the delivery result, if any.
+    pub on_result_reducer: Option<Identifier>,
+}
+
 /// A struct representing the schema of a database index.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct IndexSchema {

crates/table/src/table.rs

@@ -108,7 +108,7 @@ pub struct Table {
     is_scheduler: bool,
     /// Indicates whether this is an outbox table or not.
     ///
-    /// Outbox tables follow the naming convention `__outbox_<reducer>`.
+    /// Set from `schema.outbox.is_some()`.
     /// This is an optimization to avoid checking the schema in e.g., `InstanceEnv::insert`.
     is_outbox: bool,
 }
@@ -549,7 +549,7 @@ impl Table {
         self.is_scheduler
     }
 
-    /// Returns whether this is an outbox table (`__outbox_<reducer>`).
+    /// Returns whether this is an outbox table (i.e., `schema.outbox.is_some()`).
     pub fn is_outbox(&self) -> bool {
         self.is_outbox
     }
@@ -2265,7 +2265,7 @@ impl Table {
         squashed_offset: SquashedOffset,
         pointer_map: Option<PointerMap>,
     ) -> Self {
-        let is_outbox = schema.table_name.starts_with("__outbox_");
+        let is_outbox = schema.outbox.is_some();
         Self {
             inner: TableInner {
                 row_layout,