feat: Allow toggling event flag during auto-migration for empty tables

## Summary

Currently, toggling `#[spacetimedb::table(event)]` on an existing table
fails with `AutoMigrateError::ChangeTableEventFlag` and requires a
manual migration. This PR allows the flip in either direction as a
live auto-migration step when the table has zero committed rows.
Non-empty tables fail with an actionable error guiding the user to
clear the table first. Clients are disconnected on the flip because
the change is observable to subscribers (event tables have no
committed state).

## Changes

- **`tx_state.rs`**: New `PendingSchemaChange::TableAlterEventFlag`
  variant storing the old `is_event` value for rollback.
- **`committed_state.rs`**: Rollback branch restores the old value on
  the live schema.
- **`mut_tx.rs`**: New `alter_table_event_flag` — dual-write to the
  tx + commit table schemas and to `st_event_table`. Idempotent
  no-op early returns before pushing a pending change. New
  `delete_st_event_table_row` helper using the existing
  `delete_col_eq` utility (hits the unique btree index on col 0).
- **`replay.rs`**: New hook on `ST_EVENT_TABLE_ID` insert/delete
  mirrors `reschema_table_for_st_table_update` — flips `is_event`
  on the referenced user-table's cached schema during replay. This
  is load-bearing for cold replay across the flip point.
- **`relational_db.rs`**: Thin `alter_table_event_flag` wrapper.
- **`auto_migrate.rs`**: `event_ok` error branch replaced with
  `AutoMigrateStep::ChangeEventFlag` + `ensure_disconnect_all_users`.
  Removed dead `AutoMigrateError::ChangeTableEventFlag` variant.
- **`formatter.rs` / `termcolor_formatter.rs`**: New
  `format_change_event_flag` mirroring `format_change_access`.
- **`update.rs`**: New `ChangeEventFlag` handler with an O(1)
  row-count precheck before any mutation.

## Safety

- **Transaction safety**: Precheck (row count) and all three writes
  (st_event_table, tx schema, commit schema) run in the same `MutTx`.
- **Rollback**: `TableAlterEventFlag` stores the old flag value so
  failed txs revert `is_event` on the live schema. Idempotent flips
  do not push a pending change.
- **Replay correctness**: Without the replay hook, cold replay from
  a pre-migration snapshot would miss the schema flip and
  post-migration inserts would silently land in committed state.
  The hook mirrors the existing `st_table`/`st_column` pattern.
- **Client contract**: Flipping `event` changes observability — v1
  subscribers stop seeing updates; v2 subscribers see a different
  message variant. `ensure_disconnect_all_users` forces reconnection.

## Example error output

```
Cannot change `event` flag on table `my_table`: table contains data.
Clear the table's rows (e.g. via a reducer) before toggling the
`event` annotation.
```

## Test plan

- [x] `cargo test -p spacetimedb-datastore --features test` — 87 pass
  (including 4 new `alter_table_event_flag` tests)
- [x] `cargo test -p spacetimedb-schema` — 103 pass (including 3 new
  `change_event_flag` plan tests)
- [x] `cargo test -p spacetimedb-core` — 192 pass (including 2 new
  empty/non-empty integration tests)
- [x] `cargo clippy -p spacetimedb-datastore -p spacetimedb-schema -p spacetimedb-core --tests` clean
- [x] Pre-existing event-table tests still pass (10 tests)

Author: Ludv1gL

crates/core/src/db/relational_db.rs

@@ -989,6 +989,10 @@ impl RelationalDB {
         Ok(self.inner.alter_table_access_mut_tx(tx, name, access)?)
     }
 
+    pub(crate) fn alter_table_event_flag(&self, tx: &mut MutTx, name: &str, is_event: bool) -> Result<(), DBError> {
+        Ok(self.inner.alter_table_event_flag_mut_tx(tx, name, is_event)?)
+    }
+
     pub(crate) fn alter_table_primary_key(
         &self,
         tx: &mut MutTx,

crates/core/src/db/update.rs

@@ -282,6 +282,27 @@ fn auto_migrate_database(
                 let table_def = plan.new.stored_in_table_def(&table_name.clone().into()).unwrap();
                 stdb.alter_table_access(tx, table_name, table_def.table_access.into())?;
             }
+            spacetimedb_schema::auto_migrate::AutoMigrateStep::ChangeEventFlag(table_name) => {
+                let table_def: &TableDef = plan.new.expect_lookup(table_name);
+                let table_id = stdb
+                    .table_id_from_name_mut(tx, table_name)?
+                    .expect("ChangeEventFlag references a table that should exist");
+
+                // Pre-validate: flipping is only safe when the table has no committed rows.
+                if stdb.table_row_count_mut(tx, table_id).unwrap_or(0) > 0 {
+                    anyhow::bail!(
+                        "Cannot change `event` flag on table `{table_name}`: table contains data. \
+                         Clear the table's rows (e.g. via a reducer) before toggling the `event` annotation."
+                    );
+                }
+
+                log!(
+                    logger,
+                    "Changing `event` flag on table `{table_name}` to `{}`",
+                    table_def.is_event
+                );
+                stdb.alter_table_event_flag(tx, table_name, table_def.is_event)?;
+            }
             spacetimedb_schema::auto_migrate::AutoMigrateStep::ChangePrimaryKey(table_name) => {
                 let table_def = plan.new.stored_in_table_def(&table_name.clone().into()).unwrap();
                 log!(logger, "Changing primary key for table `{table_name}`");
@@ -339,6 +360,7 @@ mod test {
         db::relational_db::tests_utils::{begin_mut_tx, insert, TestDB},
         host::module_host::create_table_from_def,
     };
+    use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
     use spacetimedb_datastore::locking_tx_datastore::PendingSchemaChange;
     use spacetimedb_lib::db::raw_def::v9::{btree, RawModuleDefV9Builder, TableAccess};
     use spacetimedb_sats::{product, AlgebraicType, AlgebraicType::U64};
@@ -580,4 +602,112 @@ mod test {
         );
         Ok(())
     }
+
+    /// Build a minimal v10 module with a single user table `events` whose
+    /// `is_event` flag matches `is_event`.
+    fn single_event_table_module_v10(is_event: bool) -> ModuleDef {
+        use spacetimedb_lib::db::raw_def::v10::RawModuleDefV10Builder;
+
+        let mut builder = RawModuleDefV10Builder::new();
+        let t = builder.build_table_with_new_type("events", [("id", U64)], true);
+        let t = if is_event { t.with_event(true) } else { t };
+        t.finish();
+        builder
+            .finish()
+            .try_into()
+            .expect("should be a valid v10 module definition")
+    }
+
+    #[test]
+    fn change_event_flag_empty_table_succeeds() -> anyhow::Result<()> {
+        let auth_ctx = AuthCtx::for_testing();
+        let stdb = TestDB::durable()?;
+
+        let old = single_event_table_module_v10(false);
+        let new = single_event_table_module_v10(true);
+
+        // Create the non-event table.
+        let mut tx = begin_mut_tx(&stdb);
+        for def in old.tables() {
+            create_table_from_def(&stdb, &mut tx, &old, def)?;
+        }
+        let table_id = stdb
+            .table_id_from_name_mut(&tx, "events")?
+            .expect("table should exist");
+        assert!(
+            !tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            "fresh non-event table should have `is_event=false`"
+        );
+        stdb.commit_tx(tx)?;
+
+        // Apply the auto-migration flipping `is_event=true` on an empty table.
+        let mut tx = begin_mut_tx(&stdb);
+        let plan = ponder_migrate(&old, &new)?;
+        let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?;
+
+        assert!(
+            matches!(res, UpdateResult::RequiresClientDisconnect),
+            "flipping the `event` flag should disconnect clients"
+        );
+        assert!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            "live schema should reflect the new `is_event=true`"
+        );
+        assert!(
+            tx.pending_schema_changes()
+                .iter()
+                .any(|c| matches!(c, PendingSchemaChange::TableAlterEventFlag(_, false))),
+            "flipping `is_event` should record a TableAlterEventFlag pending change: {:?}",
+            tx.pending_schema_changes()
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn change_event_flag_nonempty_table_fails() -> anyhow::Result<()> {
+        let auth_ctx = AuthCtx::for_testing();
+        let stdb = TestDB::durable()?;
+
+        let old = single_event_table_module_v10(false);
+        let new = single_event_table_module_v10(true);
+
+        // Create the table, insert a row, then attempt to flip `is_event`.
+        let mut tx = begin_mut_tx(&stdb);
+        for def in old.tables() {
+            create_table_from_def(&stdb, &mut tx, &old, def)?;
+        }
+        let table_id = stdb
+            .table_id_from_name_mut(&tx, "events")?
+            .expect("table should exist");
+        insert(&stdb, &mut tx, table_id, &product![42u64])?;
+        stdb.commit_tx(tx)?;
+
+        let mut tx = begin_mut_tx(&stdb);
+        let plan = ponder_migrate(&old, &new)?;
+        let result = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger);
+        let err = result
+            .err()
+            .expect("flipping `is_event` on a non-empty table should fail");
+        assert!(
+            err.to_string().contains("contains data"),
+            "error should mention that the table contains data, got: {err}"
+        );
+        // The schema should be untouched and there should be no pending changes to roll back.
+        assert!(
+            !tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            "failed migration must leave the `is_event` flag unchanged"
+        );
+        assert!(
+            tx.pending_schema_changes().is_empty(),
+            "failed migration should leave no pending schema changes: {:?}",
+            tx.pending_schema_changes()
+        );
+        Ok(())
+    }
 }

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -921,6 +921,11 @@ impl CommittedState {
                 let table = self.tables.get_mut(&table_id)?;
                 table.with_mut_schema(|s| s.table_access = access);
             }
+            // A table's `is_event` flag was changed. Change back to the old one.
+            TableAlterEventFlag(table_id, old_is_event) => {
+                let table = self.tables.get_mut(&table_id)?;
+                table.with_mut_schema(|s| s.is_event = old_is_event);
+            }
             // A table's primary key was changed. Change back to the old one.
             TableAlterPrimaryKey(table_id, old_pk) => {
                 let table = self.tables.get_mut(&table_id)?;

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -320,6 +320,14 @@ impl Locking {
         tx.alter_table_access(table_id, access)
     }
 
+    pub fn alter_table_event_flag_mut_tx(&self, tx: &mut MutTxId, name: &str, is_event: bool) -> Result<()> {
+        let table_id = self
+            .table_id_from_name_mut_tx(tx, name)?
+            .ok_or_else(|| TableError::NotFound(name.into()))?;
+
+        tx.alter_table_event_flag(table_id, is_event)
+    }
+
     pub fn alter_table_primary_key_mut_tx(
         &self,
         tx: &mut MutTxId,
@@ -1011,7 +1019,7 @@ pub(crate) mod tests {
     use crate::locking_tx_datastore::tx_state::PendingSchemaChange;
     use crate::system_tables::{
         system_tables, StColumnRow, StConnectionCredentialsFields, StConstraintData, StConstraintFields,
-        StConstraintRow, StEventTableFields, StIndexAlgorithm, StIndexFields, StIndexRow, StRowLevelSecurityFields,
+        StConstraintRow, StEventTableFields, StFields as _, StIndexAlgorithm, StIndexFields, StIndexRow, StRowLevelSecurityFields,
         StScheduledFields, StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields,
         ST_CLIENT_ID, ST_CLIENT_NAME, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_NAME, ST_COLUMN_ID, ST_COLUMN_NAME,
         ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID, ST_CONSTRAINT_NAME,
@@ -3181,6 +3189,166 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    /// Returns whether `st_event_table` contains a row referencing `table_id`.
+    fn st_event_table_has_row(datastore: &Locking, tx: &MutTxId, table_id: TableId) -> bool {
+        datastore
+            .iter_by_col_eq_mut_tx(
+                tx,
+                ST_EVENT_TABLE_ID,
+                ColList::from(StEventTableFields::TableId.col_id()),
+                &table_id.into(),
+            )
+            .expect("st_event_table lookup should succeed")
+            .next()
+            .is_some()
+    }
+
+    #[test]
+    fn test_alter_table_event_flag_non_event_to_event() -> ResultTest<()> {
+        // Create a non-event table.
+        let (datastore, tx, table_id) = setup_table()?;
+        commit(&datastore, tx)?;
+
+        // Flip `is_event` from `false` to `true`.
+        let mut tx = begin_mut_tx(&datastore);
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            false
+        );
+        assert!(
+            !st_event_table_has_row(&datastore, &tx, table_id),
+            "fresh non-event table must not have a row in `st_event_table`"
+        );
+
+        tx.alter_table_event_flag(table_id, true)?;
+        assert_matches!(
+            tx.pending_schema_changes(),
+            &[PendingSchemaChange::TableAlterEventFlag(t, false)] if t == table_id
+        );
+        commit(&datastore, tx)?;
+
+        // After commit, the schema should reflect the flipped flag
+        // and `st_event_table` should contain the row.
+        let tx = begin_mut_tx(&datastore);
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            true
+        );
+        assert!(
+            st_event_table_has_row(&datastore, &tx, table_id),
+            "after flipping to event, `st_event_table` should have the row"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_alter_table_event_flag_event_to_non_event() -> ResultTest<()> {
+        // Create an event table.
+        let (datastore, tx, table_id) = setup_event_table()?;
+        commit(&datastore, tx)?;
+
+        // Sanity check: `st_event_table` should have the row.
+        let mut tx = begin_mut_tx(&datastore);
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            true
+        );
+        assert!(
+            st_event_table_has_row(&datastore, &tx, table_id),
+            "event table should have a row in `st_event_table`"
+        );
+
+        // Flip `is_event` from `true` to `false`.
+        tx.alter_table_event_flag(table_id, false)?;
+        assert_matches!(
+            tx.pending_schema_changes(),
+            &[PendingSchemaChange::TableAlterEventFlag(t, true)] if t == table_id
+        );
+        commit(&datastore, tx)?;
+
+        // After commit, the schema should reflect the flipped flag
+        // and `st_event_table` should NOT contain the row.
+        let tx = begin_mut_tx(&datastore);
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            false
+        );
+        assert!(
+            !st_event_table_has_row(&datastore, &tx, table_id),
+            "after flipping to non-event, `st_event_table` should not have the row"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_alter_table_event_flag_rollback_reverts_live_state_and_st_event_table() -> ResultTest<()> {
+        // Create a non-event table.
+        let (datastore, tx, table_id) = setup_table()?;
+        commit(&datastore, tx)?;
+
+        // Start a new tx, flip, check pending change, then rollback.
+        let mut tx = begin_mut_tx(&datastore);
+        assert!(!st_event_table_has_row(&datastore, &tx, table_id));
+
+        tx.alter_table_event_flag(table_id, true)?;
+        assert_matches!(
+            tx.pending_schema_changes(),
+            &[PendingSchemaChange::TableAlterEventFlag(t, false)] if t == table_id
+        );
+        // The in-tx view must reflect the flip.
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            true
+        );
+        assert!(
+            st_event_table_has_row(&datastore, &tx, table_id),
+            "after flipping within the tx, `st_event_table` should have the row"
+        );
+        let _ = datastore.rollback_mut_tx(tx);
+
+        // After rollback, the schema and `st_event_table` should be back to pre-state.
+        let tx = begin_mut_tx(&datastore);
+        assert!(
+            tx.pending_schema_changes().is_empty(),
+            "rollback should clear pending schema changes"
+        );
+        assert_eq!(
+            tx.get_schema(table_id)
+                .map(|s| s.is_event)
+                .expect("schema should exist"),
+            false,
+            "rollback should revert `is_event` to its pre-tx value"
+        );
+        assert!(
+            !st_event_table_has_row(&datastore, &tx, table_id),
+            "rollback should revert the `st_event_table` row"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_alter_table_event_flag_idempotent_no_pending_change() -> ResultTest<()> {
+        // Create a non-event table.
+        let (datastore, tx, table_id) = setup_table()?;
+        commit(&datastore, tx)?;
+
+        // Flipping to the same value must be a no-op with no pending change.
+        let mut tx = begin_mut_tx(&datastore);
+        tx.alter_table_event_flag(table_id, false)?;
+        assert_matches!(tx.pending_schema_changes(), &[]);
+        Ok(())
+    }
+
     #[test]
     fn test_alter_table_row_type_rejects_some_bad_changes() -> ResultTest<()> {
         let datastore = get_datastore()?;