system table naming

Author: Shubham8287

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

@@ -230,8 +230,8 @@ pub struct CallFromDatabaseParams {
 pub struct CallFromDatabaseQuery {
     /// Hex-encoded [`Identity`] of the sending database.
     sender_identity: String,
-    /// The inter-database message ID from the sender's st_msg_id.
-    /// Used for at-most-once delivery via `st_inbound_msg_id`.
+    /// The inter-database message ID from the sender's st_outbound_msg.
+    /// Used for at-most-once delivery via `st_inbound_msg`.
     msg_id: u64,
 }
 
@@ -241,9 +241,9 @@ pub struct CallFromDatabaseQuery {
 ///
 /// Required query params:
 /// - `sender_identity` — hex-encoded identity of the sending database.
-/// - `msg_id`          — the inter-database message ID from the sender's st_msg_id.
+/// - `msg_id`          — the inter-database message ID from the sender's st_outbound_msg.
 ///
-/// Before invoking the reducer, the receiver checks `st_inbound_msg_id`.
+/// Before invoking the reducer, the receiver checks `st_inbound_msg`.
 /// If the incoming `msg_id` is ≤ the last delivered msg_id for `sender_identity`,
 /// the call is a duplicate and 200 OK is returned immediately without running the reducer.
 /// Otherwise the reducer is invoked, the dedup index is updated atomically in the same

crates/core/src/host/host_controller.rs

@@ -889,7 +889,7 @@ struct Host {
     /// Handle to the task responsible for cleaning up old views.
     /// The task is aborted when [`Host`] is dropped.
     view_cleanup_task: AbortHandle,
-    /// IDC runtime: delivers outbound inter-database messages from `st_msg_id`.
+    /// IDC runtime: delivers outbound inter-database messages from `st_outbound_msg`.
     /// Stopped when [`Host`] is dropped.
     _idc_runtime: IdcRuntime,
 }

crates/core/src/host/idc_runtime.rs

@@ -1,20 +1,20 @@
 /// Inter-Database Communication (IDC) Runtime
 ///
 /// Background tokio task that:
-/// 1. Loads undelivered entries from `st_msg_id` on startup, resolving delivery data from outbox tables.
+/// 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>`
 /// 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
-///    `on_result_reducer` (read from the outbox table's schema) and deletes the st_msg_id row.
+///    `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 spacetimedb_datastore::execution_context::Workload;
-use spacetimedb_datastore::system_tables::{StMsgIdRow, ST_MSG_ID_ID};
+use spacetimedb_datastore::system_tables::{StOutboundMsgRow, ST_OUTBOUND_MSG_ID};
 use spacetimedb_datastore::traits::IsolationLevel;
 use spacetimedb_lib::{AlgebraicValue, Identity};
 use spacetimedb_primitives::{ColId, TableId};
@@ -207,19 +207,19 @@ 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_id_result_status;
+    use spacetimedb_datastore::system_tables::st_inbound_msg_result_status;
     match outcome {
-        DeliveryOutcome::Success => (st_inbound_msg_id_result_status::SUCCESS, String::new()),
-        DeliveryOutcome::ReducerError(msg) => (st_inbound_msg_id_result_status::REDUCER_ERROR, msg.clone()),
+        DeliveryOutcome::Success => (st_inbound_msg_result_status::SUCCESS, String::new()),
+        DeliveryOutcome::ReducerError(msg) => (st_inbound_msg_result_status::REDUCER_ERROR, msg.clone()),
         DeliveryOutcome::BudgetExceeded => (
-            st_inbound_msg_id_result_status::REDUCER_ERROR,
+            st_inbound_msg_result_status::REDUCER_ERROR,
             "budget exceeded".to_string(),
         ),
         DeliveryOutcome::TransportError(_) => unreachable!("transport errors never finalize"),
     }
 }
 
-/// Finalize a delivered message: call the on_result reducer (if any), then delete from ST_MSG_ID.
+/// Finalize a delivered message: call the on_result reducer (if any), then delete from ST_OUTBOUND_MSG.
 async fn finalize_message(
     db: &RelationalDB,
     module_host: &WeakModuleHost,
@@ -288,25 +288,28 @@ async fn finalize_message(
     delete_message(db, msg.msg_id);
 }
 
-/// Load all messages from ST_MSG_ID into the per-target queues, resolving delivery data
+/// Load all messages from ST_OUTBOUND_MSG into the per-target queues, resolving delivery data
 /// from the corresponding outbox table rows.
 ///
-/// A row's presence in ST_MSG_ID means it has not yet been processed.
+/// A row's presence in ST_OUTBOUND_MSG means it has not yet been processed.
 /// Messages already in a target's queue (by msg_id) are not re-added.
 fn load_pending_into_targets(db: &RelationalDB, targets: &mut HashMap<Identity, TargetState>) {
     let tx = db.begin_tx(Workload::Internal);
 
-    let st_msg_id_rows: Vec<StMsgIdRow> = db
-        .iter(&tx, ST_MSG_ID_ID)
-        .map(|iter| iter.filter_map(|row_ref| StMsgIdRow::try_from(row_ref).ok()).collect())
+    let st_outbound_msg_rows: Vec<StOutboundMsgRow> = db
+        .iter(&tx, ST_OUTBOUND_MSG_ID)
+        .map(|iter| {
+            iter.filter_map(|row_ref| StOutboundMsgRow::try_from(row_ref).ok())
+                .collect()
+        })
         .unwrap_or_else(|e| {
-            log::error!("idc_runtime: failed to read st_msg_id: {e}");
+            log::error!("idc_runtime: failed to read st_outbound_msg: {e}");
             Vec::new()
         });
 
-    let mut pending: Vec<PendingMessage> = Vec::with_capacity(st_msg_id_rows.len());
+    let mut pending: Vec<PendingMessage> = Vec::with_capacity(st_outbound_msg_rows.len());
 
-    for st_row in st_msg_id_rows {
+    for st_row in st_outbound_msg_rows {
         let outbox_table_id = TableId(st_row.outbox_table_id);
 
         // Read the outbox table schema for reducer name and on_result_reducer.
@@ -432,10 +435,10 @@ async fn attempt_delivery(
     }
 }
 
-/// Delete a message from ST_MSG_ID within a new transaction.
+/// Delete a message from ST_OUTBOUND_MSG within a new transaction.
 fn delete_message(db: &RelationalDB, msg_id: u64) {
     let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
-    if let Err(e) = tx.delete_msg_id(msg_id) {
+    if let Err(e) = tx.delete_outbound_msg(msg_id) {
         log::error!("idc_runtime: failed to delete msg_id={msg_id}: {e}");
         let _ = db.rollback_mut_tx(tx);
         return;

crates/core/src/host/instance_env.rs

@@ -432,16 +432,16 @@ impl InstanceEnv {
         Ok(())
     }
 
-    /// Enqueue an outbox row into ST_MSG_ID atomically within the current transaction,
+    /// Enqueue an outbox row into ST_OUTBOUND_MSG atomically within the current transaction,
     /// and notify the IDC runtime so it delivers without waiting for the next poll cycle.
     ///
     /// Outbox tables follow the naming convention `__outbox_<reducer>` and have:
-    ///   - Col 0: auto-inc primary key (u64) — stored as `row_id` in ST_MSG_ID.
+    ///   - Col 0: auto-inc primary key (u64) — stored as `row_id` in ST_OUTBOUND_MSG.
     ///   - Col 1: target database Identity (stored as U256).
     ///   - Remaining cols: args for the remote reducer.
     ///
     /// The `on_result_reducer` and delivery data are resolved at delivery time from the
-    /// outbox table's schema and row, so ST_MSG_ID only stores the minimal reference.
+    /// outbox table's schema and row, so ST_OUTBOUND_MSG only stores the minimal reference.
     fn enqueue_outbox_row(
         &self,
         _stdb: &RelationalDB,
@@ -454,7 +454,7 @@ impl InstanceEnv {
         let row_ref = tx.get(table_id, row_ptr).map_err(DBError::from)?.unwrap();
         let pv = row_ref.to_product_value();
 
-        // Col 0 is the auto-inc primary key — this is the row_id we store in ST_MSG_ID.
+        // Col 0 is the auto-inc primary key — this is the row_id we store in ST_OUTBOUND_MSG.
         let row_id = match pv.elements.first() {
             Some(AlgebraicValue::U64(id)) => *id,
             other => {
@@ -466,7 +466,7 @@ impl InstanceEnv {
             }
         };
 
-        tx.insert_st_msg_id(table_id.0, row_id).map_err(DBError::from)?;
+        tx.insert_st_outbound_msg(table_id.0, row_id).map_err(DBError::from)?;
 
         // Wake the IDC runtime immediately so it doesn't wait for the next poll cycle.
         let _ = self.idc_sender.send(());

crates/core/src/host/module_host.rs

@@ -627,7 +627,7 @@ pub struct CallReducerParams {
     /// 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_id` is consulted:
+    /// 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 [`ReducerCallError::Deduplicated`].
     /// Otherwise the reducer runs, and the stored msg_id is updated atomically
@@ -1670,7 +1670,7 @@ impl ModuleHost {
     /// Variant of [`Self::call_reducer`] for database-to-database calls.
     ///
     /// Behaves identically to `call_reducer`, except that it enforces at-most-once
-    /// delivery using the `st_inbound_msg_id` dedup index.
+    /// delivery using the `st_inbound_msg` dedup index.
     /// Before invoking the reducer, the receiver checks whether
     /// `sender_msg_id` ≤ the last delivered msg_id for `sender_database_identity`.
     /// If so, the call is a duplicate and [`ReducerOutcome::Deduplicated`] is returned

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

@@ -37,7 +37,7 @@ use spacetimedb_datastore::db_metrics::DB_METRICS;
 use spacetimedb_datastore::error::{DatastoreError, ViewError};
 use spacetimedb_datastore::execution_context::{self, ReducerContext, Workload};
 use spacetimedb_datastore::locking_tx_datastore::{FuncCallType, MutTxId, ViewCallInfo};
-use spacetimedb_datastore::system_tables::st_inbound_msg_id_result_status;
+use spacetimedb_datastore::system_tables::st_inbound_msg_result_status;
 use spacetimedb_datastore::traits::{IsolationLevel, Program};
 use spacetimedb_execution::pipelined::PipelinedProject;
 use spacetimedb_lib::buffer::DecodeError;
@@ -855,13 +855,13 @@ impl InstanceCommon {
         // 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_id_row(sender_identity);
-            if stored.as_ref().is_some_and(|r| sender_msg_id <= r.last_msg_id) {
+            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_id_result_status::SUCCESS => ReducerOutcome::Committed,
-                    s if s == st_inbound_msg_id_result_status::REDUCER_ERROR => {
+                    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()))
                     }
                     _ => ReducerOutcome::Deduplicated,
@@ -927,10 +927,10 @@ impl InstanceCommon {
                 // Atomically update the dedup index so this sender's tx_offset is recorded
                 // as delivered. This prevents re-processing if the message is retried.
                 let dedup_res = match dedup_sender {
-                    Some((sender_identity, sender_msg_id)) => tx.upsert_inbound_last_msg_id(
+                    Some((sender_identity, sender_msg_id)) => tx.upsert_inbound_last_msg(
                         sender_identity,
                         sender_msg_id,
-                        st_inbound_msg_id_result_status::SUCCESS,
+                        st_inbound_msg_result_status::SUCCESS,
                         String::new(),
                     ),
                     None => Ok(()),
@@ -996,16 +996,16 @@ impl InstanceCommon {
         let event = commit_and_broadcast_event(&info.subscriptions, client, event, out.tx).event;
 
         // For IDC (database-to-database) calls: if the reducer failed with a user error,
-        // record the failure in st_inbound_msg_id in a separate tx (since the reducer tx
+        // record the failure in st_inbound_msg in a separate tx (since the reducer tx
         // was rolled back). This allows the sending database to receive the error on dedup
         // rather than re-running the reducer.
         if let (Some((sender_identity, sender_msg_id)), EventStatus::FailedUser(err)) = (dedup_sender, &event.status) {
             let err_msg = err.clone();
             let mut dedup_tx = stdb.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
-            if let Err(e) = dedup_tx.upsert_inbound_last_msg_id(
+            if let Err(e) = dedup_tx.upsert_inbound_last_msg(
                 sender_identity,
                 sender_msg_id,
-                st_inbound_msg_id_result_status::REDUCER_ERROR,
+                st_inbound_msg_result_status::REDUCER_ERROR,
                 err_msg,
             ) {
                 log::error!("IDC: failed to record reducer error in dedup table for sender {sender_identity}: {e}");

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -19,7 +19,7 @@ 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_ID, ST_INDEX_ID, ST_INDEX_IDX, ST_INDEX_NAME, ST_MODULE_ID,
+        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,
@@ -30,8 +30,8 @@ use crate::{
     locking_tx_datastore::ViewCallInfo,
     system_tables::{
         ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_IDX, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX,
-        ST_EVENT_TABLE_ID, ST_EVENT_TABLE_IDX, ST_INBOUND_MSG_ID_IDX, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_IDX,
-        ST_MSG_ID_ID, ST_MSG_ID_IDX, ST_TABLE_ACCESSOR_ID, ST_TABLE_ACCESSOR_IDX, ST_VIEW_COLUMN_ID,
+        ST_EVENT_TABLE_ID, ST_EVENT_TABLE_IDX, ST_INBOUND_MSG_IDX, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_IDX,
+        ST_OUTBOUND_MSG_ID, ST_OUTBOUND_MSG_IDX, ST_TABLE_ACCESSOR_ID, ST_TABLE_ACCESSOR_IDX, ST_VIEW_COLUMN_ID,
         ST_VIEW_COLUMN_IDX, ST_VIEW_ID, ST_VIEW_IDX, ST_VIEW_PARAM_ID, ST_VIEW_PARAM_IDX, ST_VIEW_SUB_ID,
         ST_VIEW_SUB_IDX,
     },
@@ -478,8 +478,8 @@ impl CommittedState {
         self.create_table(ST_TABLE_ACCESSOR_ID, schemas[ST_TABLE_ACCESSOR_IDX].clone());
         self.create_table(ST_INDEX_ACCESSOR_ID, schemas[ST_INDEX_ACCESSOR_IDX].clone());
         self.create_table(ST_COLUMN_ACCESSOR_ID, schemas[ST_COLUMN_ACCESSOR_IDX].clone());
-        self.create_table(ST_INBOUND_MSG_ID_ID, schemas[ST_INBOUND_MSG_ID_IDX].clone());
-        self.create_table(ST_MSG_ID_ID, schemas[ST_MSG_ID_IDX].clone());
+        self.create_table(ST_INBOUND_MSG_ID, schemas[ST_INBOUND_MSG_IDX].clone());
+        self.create_table(ST_OUTBOUND_MSG_ID, schemas[ST_OUTBOUND_MSG_IDX].clone());
 
         // Insert the sequences into `st_sequences`
         let (st_sequences, blob_store, pool) =