Automigrate views using new schema for backing table (#5441)

# Description of Changes

#5300 changed the backing table schema of views. While this technically
isn't a breaking change because these tables are ephemeral, the schema
is not ephemeral, so it does require us to update the schema by dropping
and re-adding views once on initial startup.

# API and ABI breaking changes

None

# Expected complexity level and risk

1.5

# Testing

Auto-migration unit tests

Author: joshua-spacetime

Cargo.lock

@@ -37,7 +37,7 @@ use spacetimedb_datastore::execution_context::Workload;
 use spacetimedb_datastore::system_tables::ModuleKind;
 use spacetimedb_datastore::traits::Program;
 use spacetimedb_durability::{self as durability};
-use spacetimedb_lib::{AlgebraicValue, Identity, Timestamp};
+use spacetimedb_lib::{identity::AuthCtx, AlgebraicValue, Identity, Timestamp};
 use spacetimedb_paths::server::{ModuleLogsDir, ServerDataDir};
 use spacetimedb_runtime::AbortHandle;
 use spacetimedb_sats::hash::Hash;
@@ -941,6 +941,31 @@ impl<F: Fn() + Send + Sync + 'static> ModuleLauncher<F> {
     }
 }
 
+fn repair_stale_view_backing_tables_on_launch(launched: &LaunchedModule) -> anyhow::Result<()> {
+    let info = launched.module_host.info();
+    let stdb = info.relational_db().clone();
+    let Some(plan) = db::update::stale_view_backing_table_recreate_plan(stdb.as_ref(), &info.module_def)? else {
+        return Ok(());
+    };
+
+    info!(
+        "repairing stale view backing tables during module launch: {}",
+        info.database_identity
+    );
+    let system_logger = launched.replica_ctx.logger.system_logger();
+    system_logger.info("Repairing stale view backing tables");
+
+    let auth_ctx = AuthCtx::for_current(info.owner_identity);
+    stdb.with_auto_commit(Workload::Internal, |tx| -> anyhow::Result<()> {
+        match db::update::update_database(stdb.as_ref(), tx, auth_ctx, plan, system_logger)? {
+            db::update::UpdateResult::Success | db::update::UpdateResult::RequiresClientDisconnect => Ok(()),
+            db::update::UpdateResult::EvaluateSubscribedViews => {
+                bail!("startup view backing table repair unexpectedly requested view evaluation")
+            }
+        }
+    })
+}
+
 /// Update a module.
 ///
 /// If the `db` is not initialized yet (i.e. its program hash is `None`),
@@ -1208,6 +1233,7 @@ impl Host {
                 launched.module_host.durable_tx_offset(),
             ));
         } else {
+            repair_stale_view_backing_tables_on_launch(&launched)?;
             drop(program)
         }
 

crates/core/src/host/host_controller.rs

@@ -1,13 +1,16 @@
 use super::relational_db::RelationalDB;
 use crate::sql::rls::RowLevelExpr;
 use anyhow::Context;
-use spacetimedb_datastore::locking_tx_datastore::MutTxId;
+use spacetimedb_datastore::execution_context::Workload;
+use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
+use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
+use spacetimedb_datastore::system_tables::{StViewFields, StViewRow, ST_VIEW_ID};
 use spacetimedb_lib::db::auth::StTableType;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_lib::AlgebraicValue;
 use spacetimedb_primitives::{ColSet, TableId};
-use spacetimedb_schema::auto_migrate::{AutoMigratePlan, ManualMigratePlan, MigratePlan};
-use spacetimedb_schema::def::{ModuleDef, TableDef, ViewDef};
+use spacetimedb_schema::auto_migrate::{AutoMigratePlan, AutoMigrateStep, ManualMigratePlan, MigratePlan};
+use spacetimedb_schema::def::{ModuleDef, ModuleDefLookup, TableDef, ViewDef};
 use spacetimedb_schema::schema::{
     column_schemas_from_defs, ConstraintSchema, IndexSchema, Schema, SequenceSchema, TableSchema,
 };
@@ -26,6 +29,91 @@ pub enum UpdateResult {
     EvaluateSubscribedViews,
 }
 
+/// Build a repair-only migration plan for views whose stored backing table row
+/// layout is stale compared to the currently loaded module definition.
+pub fn stale_view_backing_table_recreate_plan<'def>(
+    stdb: &RelationalDB,
+    module_def: &'def ModuleDef,
+) -> anyhow::Result<Option<MigratePlan<'def>>> {
+    let steps = stale_view_backing_table_recreate_steps(stdb, module_def)?;
+    if steps.is_empty() {
+        return Ok(None);
+    }
+
+    Ok(Some(MigratePlan::Auto(AutoMigratePlan {
+        old: module_def,
+        new: module_def,
+        prechecks: Vec::new(),
+        steps,
+    })))
+}
+
+fn stale_view_backing_table_recreate_steps<'def>(
+    stdb: &RelationalDB,
+    module_def: &'def ModuleDef,
+) -> anyhow::Result<Vec<AutoMigrateStep<'def>>> {
+    stdb.with_read_only(Workload::Internal, |tx| -> anyhow::Result<_> {
+        let mut steps = Vec::new();
+
+        for view in module_def.views() {
+            if view_backing_table_needs_recreate(stdb, tx, module_def, view)? {
+                steps.extend([
+                    AutoMigrateStep::RemoveView(view.key()),
+                    AutoMigrateStep::AddView(view.key()),
+                ]);
+            }
+        }
+
+        if !steps.is_empty() {
+            ensure_disconnect_all_users(&mut steps);
+            steps.sort();
+        }
+
+        Ok(steps)
+    })
+}
+
+fn view_backing_table_needs_recreate(
+    stdb: &RelationalDB,
+    tx: &mut TxId,
+    module_def: &ModuleDef,
+    view: &ViewDef,
+) -> anyhow::Result<bool> {
+    let Some(table_id) = view_backing_table_id(tx, view)? else {
+        return Ok(false);
+    };
+
+    let actual = stdb.schema_for_table(tx, table_id)?;
+    let expected = TableSchema::from_view_def_for_datastore(module_def, view);
+
+    Ok(view_backing_row_layout_changed(&actual, &expected))
+}
+
+fn view_backing_table_id(tx: &mut TxId, view: &ViewDef) -> anyhow::Result<Option<TableId>> {
+    let view_name = AlgebraicValue::from(<Box<str>>::from(&*view.name));
+    let Some(row) = tx
+        .iter_by_col_eq(ST_VIEW_ID, StViewFields::ViewName, &view_name)?
+        .next()
+    else {
+        return Ok(None);
+    };
+
+    Ok(StViewRow::try_from(row)?.table_id)
+}
+
+fn view_backing_row_layout_changed(actual: &TableSchema, expected: &TableSchema) -> bool {
+    actual.row_type != expected.row_type
+}
+
+fn ensure_disconnect_all_users(steps: &mut Vec<AutoMigrateStep<'_>>) {
+    if !steps
+        .iter()
+        .any(|step| matches!(step, AutoMigrateStep::DisconnectAllUsers))
+    {
+        steps.push(AutoMigrateStep::DisconnectAllUsers);
+    }
+}
+
 /// Update the database according to the migration plan.
 ///
 /// The update is performed within the transactional context `tx`.
@@ -575,6 +663,86 @@ mod test {
             .expect("should be a valid module definition")
     }
 
+    fn view_module() -> ModuleDef {
+        let mut builder = RawModuleDefV10Builder::new();
+        let return_type_ref = builder.add_algebraic_type(
+            [],
+            "my_view_return_type",
+            AlgebraicType::product([("a", AlgebraicType::U64)]),
+            true,
+        );
+        builder.add_view(
+            "my_view",
+            0,
+            true,
+            true,
+            ProductType::unit(),
+            AlgebraicType::array(AlgebraicType::Ref(return_type_ref)),
+        );
+        builder.add_view_primary_key("my_view", ["a"]);
+        builder
+            .finish()
+            .try_into()
+            .expect("should be a valid module definition")
+    }
+
+    fn old_view_backing_schema(module_def: &ModuleDef) -> TableSchema {
+        let view = module_def.view("my_view").unwrap();
+        let mut schema = TableSchema::from_view_def_for_datastore(module_def, view);
+
+        schema.columns.remove(0);
+        for (pos, col) in schema.columns.iter_mut().enumerate() {
+            col.col_pos = pos.into();
+        }
+        schema.indexes.clear();
+        schema.constraints.clear();
+        schema.reset();
+
+        schema
+    }
+
+    fn create_backing_table(stdb: &TestDB, schema: TableSchema) -> anyhow::Result<()> {
+        let mut tx = begin_mut_tx(stdb);
+        stdb.create_table(&mut tx, schema)?;
+        stdb.commit_tx(tx)?;
+        Ok(())
+    }
+
+    #[test]
+    fn stale_view_backing_schema_generates_startup_repair_plan() -> anyhow::Result<()> {
+        let stdb = TestDB::durable()?;
+        let module_def = view_module();
+        create_backing_table(&stdb, old_view_backing_schema(&module_def))?;
+
+        let plan = stale_view_backing_table_recreate_plan(&stdb, &module_def)?.expect("expected repair plan");
+
+        let MigratePlan::Auto(plan) = &plan else {
+            panic!("expected auto migration");
+        };
+        let my_view = module_def.view("my_view").unwrap().key();
+        assert!(plan.steps.contains(&AutoMigrateStep::RemoveView(my_view)));
+        assert!(plan.steps.contains(&AutoMigrateStep::AddView(my_view)));
+        assert!(plan.steps.contains(&AutoMigrateStep::DisconnectAllUsers));
+        assert!(!plan.steps.contains(&AutoMigrateStep::UpdateView(my_view)));
+
+        Ok(())
+    }
+
+    #[test]
+    fn current_view_backing_schema_skips_startup_repair_plan() -> anyhow::Result<()> {
+        let stdb = TestDB::durable()?;
+        let module_def = view_module();
+        let view = module_def.view("my_view").unwrap();
+        let backing_schema = TableSchema::from_view_def_for_datastore(&module_def, view);
+        create_backing_table(&stdb, backing_schema)?;
+
+        let plan = stale_view_backing_table_recreate_plan(&stdb, &module_def)?;
+
+        assert!(plan.is_none(), "{plan:#?}");
+
+        Ok(())
+    }
+
     enum TakeSnapshot {
         None,
         BeforeAutomigration,

crates/engine/src/update.rs

@@ -131,14 +131,25 @@ impl MigrationPolicy {
         new_module_def: &'def ModuleDef,
     ) -> anyhow::Result<MigratePlan<'def>, MigrationPolicyError> {
         let plan = ponder_migrate(old_module_def, new_module_def).map_err(MigrationPolicyError::AutoMigrateFailure)?;
+        self.permits_migrate_plan(database_identity, old_module_hash, new_module_hash, &plan)?;
+        Ok(plan)
+    }
 
+    /// Validate an already-generated migration plan under this policy.
+    pub fn permits_migrate_plan(
+        &self,
+        database_identity: Identity,
+        old_module_hash: spacetimedb_lib::Hash,
+        new_module_hash: spacetimedb_lib::Hash,
+        plan: &MigratePlan<'_>,
+    ) -> anyhow::Result<(), MigrationPolicyError> {
         let token = MigrationToken {
             database_identity,
             old_module_hash,
             new_module_hash,
         };
-        self.permits_plan(&plan, &token)?;
-        Ok(plan)
+        self.permits_plan(plan, &token)?;
+        Ok(())
     }
 }
 

crates/schema/src/auto_migrate.rs

@@ -12,6 +12,7 @@ serde_json.workspace = true
 toml.workspace = true
 regex.workspace = true
 anyhow.workspace = true
+fs_extra.workspace = true
 which = "8.0.0"
 
 [dev-dependencies]

crates/smoketests/Cargo.toml

@@ -44,3 +44,55 @@ CARGO_NET_OFFLINE=true CARGO_TARGET_DIR="$TMP/target-old" \
 cp "$TMP/target-old/wasm32-unknown-unknown/release/upgrade_old_module.wasm" \
   crates/smoketests/fixtures/upgrade_old_module_v1.wasm
 ```
+
+## `stale-view-backing-table-v2.6.0`
+
+This is a standalone data-dir fixture created by `spacetimedb-standalone` `v2.6.0`.
+It contains database identity
+`c200f6ec405075e508c2ed6474019332d6a2a46c69614306cc4bd980e0b8b767`, published
+from `crates/smoketests/modules/views-sql`.
+
+The fixture exists to cover startup repair for view backing tables created before
+the `arg_hash` internal column. The persisted sender-scoped view backing tables
+begin with `sender`, and anonymous view backing tables do not have an internal
+argument column.
+
+The checked-in fixture intentionally keeps only the control database entry and
+commitlog segment required to replay that database state. The current server can
+recreate config, metadata, commitlog offset indexes (`*.stdb.ofs`), snapshots,
+lock files, and program byte storage from defaults and the commitlog during
+startup.
+
+To regenerate it:
+
+```bash
+TMP="$(mktemp -d)"
+git archive --format=tar v2.6.0 | tar -xf - -C "$TMP"
+
+cargo build --manifest-path "$TMP/Cargo.toml" \
+  -p spacetimedb-standalone --release \
+  --features spacetimedb-standalone/allow_loopback_http_for_tests
+
+"$TMP/target/release/spacetimedb-standalone" start \
+  --data-dir "$TMP/data" \
+  --jwt-key-dir "$TMP/keys" \
+  --listen-addr 127.0.0.1:0 \
+  --non-interactive
+
+# In another shell, publish to the printed server URL.
+CARGO_HOME="$TMP/cargo-home" CARGO_TARGET_DIR="$TMP/module-target" \
+  target/release/spacetimedb-cli --config-path "$TMP/client-config.toml" publish \
+  --server "http://127.0.0.1:<PORT>" \
+  --module-path crates/smoketests/modules/views-sql \
+  --yes stale-view-backing-table-v26
+
+# Stop the old server, update the identity constant if it changed, then copy the
+# minimal fixture.
+FIXTURE=crates/smoketests/fixtures/stale-view-backing-table-v2.6.0
+rsync -am \
+  --include='*/' \
+  --include='control-db/db' \
+  --include='replicas/1/clog/*.stdb.log' \
+  --exclude='*' \
+  "$TMP/data/" "$FIXTURE/"
+```

crates/smoketests/fixtures/README.md

@@ -0,0 +1,4 @@
+!replicas/
+!replicas/1/
+!replicas/1/clog/
+!replicas/1/clog/*.stdb.log