[2/5] Async IDC: outbox schema. (#6104)

# Description of Changes

Based on #6084

PR can be reviewed commit wise, It does 3 things:
 ### Commit 1: Add IDC outbox in ModuleDef

This introduces `OutboxDef` in `ModuleDef` / `RawModuleDefV10`, wires it
through schema construction, and validates
that outbox definitions reference valid tables, columns, reducers,
target identity columns, argument columns, and
  result handlers.

  ### Commit 2: Record outbound IDC messages.

When an outbox table row is written, the datastore advances the durable
outbound stream state in
`st_outbound_stream` and records the queued message in
`st_outbound_msg`.

  ### Commit 3: Support IDC outbox schema migration

Outbox metadata changes such as `remote_reducer`, `arg_columns`,
`on_result_reducer`, and `signature_hash` are
handled through `ChangeOutbox`, while unsafe presence/identity changes
such as adding, removing outbox rejects.

Table migrations such as adding columns change table id, we preserve
existing IDC outbound stream state so
users can evolve schemas used by dependent reducers without losing
stream continuity.

# API and ABI breaking changes
NA

# Rollback safety impact
n/a

# Expected complexity level and risk
2.5

# Testing
unit tests.

Author: Shubham8287

crates/bindings-csharp/Runtime/Internal/Autogen/RawModuleDefV10Section.g.cs

@@ -472,8 +472,13 @@ impl InstanceEnv {
                 },
             )?;
 
-        if insert_flags.is_scheduler_table {
-            self.schedule_row(stdb, tx, table_id, row_ptr)?;
+        if insert_flags.has_post_write_work() {
+            if insert_flags.is_scheduler_table {
+                self.schedule_row(stdb, tx, table_id, row_ptr)?;
+            }
+            if insert_flags.is_outbox_table {
+                tx.record_outbox_insert(table_id, row_ptr).map_err(DBError::from)?;
+            }
         }
 
         // Note, we update the metric for bytes written after the insert.
@@ -554,8 +559,13 @@ impl InstanceEnv {
                 },
             )?;
 
-        if update_flags.is_scheduler_table {
-            self.schedule_row(stdb, tx, table_id, row_ptr)?;
+        if update_flags.has_post_write_work() {
+            if update_flags.is_scheduler_table {
+                self.schedule_row(stdb, tx, table_id, row_ptr)?;
+            }
+            if update_flags.is_outbox_table {
+                tx.record_outbox_insert(table_id, row_ptr).map_err(DBError::from)?;
+            }
         }
         tx.metrics.bytes_written += buffer.len();
         tx.metrics.rows_updated += 1;

crates/bindings-csharp/Runtime/Internal/Autogen/RawOutboxDefV10.g.cs

@@ -239,11 +239,12 @@ pub(crate) mod tests {
     use spacetimedb_lib::bsatn::ToBsatn;
     use spacetimedb_lib::db::auth::{StAccess, StTableType};
     use spacetimedb_lib::error::{ResultTest, TestError};
-    use spacetimedb_lib::{AlgebraicValue, Identity};
+    use spacetimedb_lib::{AlgebraicValue, Identity, ScheduleAt, TimeDuration};
     use spacetimedb_primitives::{col_list, ColId, TableId};
     use spacetimedb_sats::{product, AlgebraicType, ArrayValue, ProductType};
     use spacetimedb_schema::identifier::Identifier;
-    use spacetimedb_schema::schema::{ColumnSchema, TableSchema};
+    use spacetimedb_schema::reducer_name::ReducerName;
+    use spacetimedb_schema::schema::{ColumnSchema, OutboxSchema, ScheduleSchema, TableSchema};
     use spacetimedb_schema::table_name::TableName;
 
     #[test]
@@ -400,6 +401,7 @@ pub(crate) mod tests {
                 None,
                 false,
                 None,
+                None,
             ),
         )?;
         let schema = db.schema_for_table_mut(tx, table_id)?;
@@ -425,6 +427,69 @@ pub(crate) mod tests {
         Ok((stdb, TestRows { data: rows }))
     }
 
+    fn create_outbox_table() -> ResultTest<(TestDB, TableId)> {
+        let stdb = TestDB::durable()?;
+        let table_id = with_auto_commit(&stdb, |tx| {
+            let mut schema = TableSchema::new(
+                TableId::SENTINEL,
+                TableName::for_test("outbound"),
+                None,
+                vec![
+                    ColumnSchema::for_test(0, "msg_id", AlgebraicType::U64),
+                    ColumnSchema::for_test(1, "target", AlgebraicType::identity()),
+                ],
+                vec![],
+                vec![],
+                vec![],
+                StTableType::User,
+                StAccess::Public,
+                None,
+                Some(ColId(0)),
+                false,
+                None,
+                None,
+            );
+            schema.outbox = Some(OutboxSchema {
+                remote_reducer: ReducerName::for_test("receive"),
+                target_column: ColId(1),
+                arg_columns: vec![],
+                on_result_reducer: None,
+                signature_hash: "test-signature".into(),
+            });
+            stdb.create_table(tx, schema)
+        })?;
+        Ok((stdb, table_id))
+    }
+
+    fn create_scheduled_table() -> ResultTest<(TestDB, TableId)> {
+        let stdb = TestDB::durable()?;
+        let table_id = with_auto_commit(&stdb, |tx| {
+            stdb.create_table(
+                tx,
+                TableSchema::new(
+                    TableId::SENTINEL,
+                    TableName::for_test("scheduled"),
+                    None,
+                    vec![
+                        ColumnSchema::for_test(0, "id", AlgebraicType::U64),
+                        ColumnSchema::for_test(1, "at", ScheduleAt::get_type()),
+                    ],
+                    vec![],
+                    vec![],
+                    vec![],
+                    StTableType::User,
+                    StAccess::Public,
+                    Some(ScheduleSchema::for_test("schedule", "reducer", 1)),
+                    Some(ColId(0)),
+                    false,
+                    None,
+                    None,
+                ),
+            )
+        })?;
+        Ok((stdb, table_id))
+    }
+
     fn create_identity_table(table_name: &str) -> ResultTest<(TestDB, TestRows)> {
         let stdb = TestDB::durable()?;
         let head = ProductType::from([("identity", AlgebraicType::identity())]);
@@ -1451,6 +1516,56 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_sql_dml_rejects_outbox_tables() -> ResultTest<()> {
+        let (db, table_id) = create_outbox_table()?;
+        insert_rows(&db, table_id, [product![1u64, Identity::ZERO]]).unwrap();
+        let target = "0".repeat(63) + "1";
+
+        for sql in [
+            format!("INSERT INTO outbound (msg_id, target) VALUES (2, 0x{target})"),
+            format!("UPDATE outbound SET target = 0x{target} WHERE msg_id = 1"),
+            "DELETE FROM outbound WHERE msg_id = 1".to_string(),
+        ] {
+            let err = run_for_testing(&db, &sql).unwrap_err().to_string();
+            assert!(
+                err.contains("SQL writes to outbox tables are not supported"),
+                "expected `{err}` to reject SQL writes to outbox tables"
+            );
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_sql_dml_rejects_scheduled_tables() -> ResultTest<()> {
+        let (db, table_id) = create_scheduled_table()?;
+        insert_rows(
+            &db,
+            table_id,
+            [product![
+                1u64,
+                spacetimedb_sats::algebraic_value::ser::value_serialize(&ScheduleAt::Interval(
+                    TimeDuration::from_micros(42)
+                ))
+            ]],
+        )
+        .unwrap();
+
+        for sql in [
+            "UPDATE scheduled SET id = 2 WHERE id = 1".to_string(),
+            "DELETE FROM scheduled WHERE id = 1".to_string(),
+        ] {
+            let err = run_for_testing(&db, &sql).unwrap_err().to_string();
+            assert!(
+                err.contains("SQL writes to scheduled tables are not supported"),
+                "expected `{err}` to reject SQL writes to scheduled tables"
+            );
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_multi_column() -> ResultTest<()> {
         let (db, _input) = create_data(1)?;

crates/core/src/host/instance_env.rs

@@ -33,9 +33,9 @@ use crate::{
         ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_IDX, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX,
         ST_ENV_ID, ST_ENV_IDX, ST_EVENT_TABLE_ID, ST_EVENT_TABLE_IDX, ST_INBOUND_MSG_ID, ST_INBOUND_MSG_IDX,
         ST_INBOUND_STREAM_ID, ST_INBOUND_STREAM_IDX, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_IDX, ST_OUTBOUND_MSG_ID,
-        ST_OUTBOUND_MSG_IDX, ST_OUTBOUND_STREAM_ID, ST_OUTBOUND_STREAM_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,
+        ST_OUTBOUND_MSG_IDX, ST_OUTBOUND_STREAM_ID, ST_OUTBOUND_STREAM_IDX, ST_OUTBOX_ID, ST_OUTBOX_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,
     },
 };
 use anyhow::anyhow;
@@ -399,6 +399,7 @@ impl CommittedState {
         self.create_table(ST_OUTBOUND_MSG_ID, schemas[ST_OUTBOUND_MSG_IDX].clone());
         self.create_table(ST_INBOUND_STREAM_ID, schemas[ST_INBOUND_STREAM_IDX].clone());
         self.create_table(ST_INBOUND_MSG_ID, schemas[ST_INBOUND_MSG_IDX].clone());
+        self.create_table(ST_OUTBOX_ID, schemas[ST_OUTBOX_IDX].clone());
 
         // Insert the sequences into `st_sequences`
         let (st_sequences, blob_store, pool) =
@@ -875,6 +876,11 @@ impl CommittedState {
                 let table = self.tables.get_mut(&table_id)?;
                 table.with_mut_schema(|s| s.primary_key = old_pk.and_then(|cl| cl.as_singleton()));
             }
+            // A table's outbox metadata changed. Change back to the old one.
+            TableAlterOutbox(table_id, old_outbox) => {
+                let table = self.tables.get_mut(&table_id)?;
+                table.set_outbox(old_outbox);
+            }
             // A table's row type was changed. Change back to the old one.
             // The row representation of old rows hasn't changed,
             // so it's safe to not rewrite the rows and merely change the type back.