Drop views, metadadta, and disconnect clients when auto-migrating

Author: joshua-spacetime

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -30,7 +30,7 @@ use core::{convert::Infallible, ops::RangeBounds};
 use spacetimedb_data_structures::map::{HashSet, IntMap, IntSet};
 use spacetimedb_durability::TxOffset;
 use spacetimedb_lib::{db::auth::StTableType, Identity};
-use spacetimedb_primitives::{ColId, ColList, ColSet, IndexId, TableId};
+use spacetimedb_primitives::{ColId, ColList, ColSet, IndexId, TableId, ViewId};
 use spacetimedb_sats::{algebraic_value::de::ValueDeserializer, memory_usage::MemoryUsage, Deserialize};
 use spacetimedb_sats::{AlgebraicValue, ProductValue};
 use spacetimedb_schema::{
@@ -624,6 +624,10 @@ impl CommittedState {
         tx_data.has_rows_or_connect_disconnect(ctx.reducer_context())
     }
 
+    pub(super) fn drop_view_from_read_sets(&mut self, view_id: ViewId) {
+        self.read_sets.remove_view(view_id)
+    }
+
     pub(super) fn merge(&mut self, tx_state: TxState, read_sets: ViewReadSets, ctx: &ExecutionContext) -> TxData {
         let mut tx_data = TxData::default();
         let mut truncates = IntSet::default();

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -109,6 +109,14 @@ impl ViewReadSets {
         self.tables.entry(table_id).or_default().insert_scan(call);
     }
 
+    /// Removes keys for `view_id` from the read set
+    pub fn remove_view(&mut self, view_id: ViewId) {
+        self.tables.retain(|_, readset| {
+            readset.remove_view(view_id);
+            !readset.is_empty()
+        });
+    }
+
     /// Merge or union read sets together
     pub fn merge(&mut self, readset: Self) {
         for (table_id, rs) in readset.tables {
@@ -134,6 +142,16 @@ impl TableReadSet {
         self.table_scans.iter()
     }
 
+    /// Is this read set empty?
+    fn is_empty(&self) -> bool {
+        self.table_scans.is_empty()
+    }
+
+    /// Removes keys for `view_id` from the read set
+    fn remove_view(&mut self, view_id: ViewId) {
+        self.table_scans.retain(|info| info.view_id != view_id);
+    }
+
     /// Merge or union two read sets for this table
     fn merge(&mut self, readset: TableReadSet) {
         self.table_scans.extend(readset.table_scans);
@@ -199,6 +217,12 @@ impl MutTxId {
             .collect::<HashSet<_>>()
             .into_iter()
     }
+
+    /// Removes keys for `view_id` from the committed read set.
+    /// Used for dropping views in an auto-migration.
+    pub fn drop_view_from_committed_read_set(&mut self, view_id: ViewId) {
+        self.committed_state_write_lock.drop_view_from_read_sets(view_id)
+    }
 }
 
 impl Datastore for MutTxId {
@@ -369,6 +393,8 @@ impl MutTxId {
         self.drop_st_view(view_id)?;
         self.drop_st_view_param(view_id)?;
         self.drop_st_view_column(view_id)?;
+        self.drop_st_view_sub(view_id)?;
+        self.drop_view_from_committed_read_set(view_id);
 
         // Drop the view's backing table if materialized
         if let StViewRow {
@@ -637,6 +663,11 @@ impl MutTxId {
         self.delete_col_eq(ST_VIEW_COLUMN_ID, StViewColumnFields::ViewId.col_id(), &view_id.into())
     }
 
+    /// Drops the rows in `st_view_sub` for this `view_id`
+    fn drop_st_view_sub(&mut self, view_id: ViewId) -> Result<()> {
+        self.delete_col_eq(ST_VIEW_SUB_ID, StViewSubFields::ViewId.col_id(), &view_id.into())
+    }
+
     pub fn drop_table(&mut self, table_id: TableId) -> Result<()> {
         self.clear_table(table_id)?;
 

crates/schema/src/auto_migrate.rs

@@ -546,7 +546,7 @@ fn auto_migrate_view<'def>(plan: &mut AutoMigratePlan<'def>, old: &'def ViewDef,
     // 2. If we change the order of the columns or parameters
     // 3. If we change the types of the columns or parameters
     // 4. If we change the context parameter
-    let Any(incompatible_return_type) = diff(plan.old, plan.new, |def| {
+    let Any(_incompatible_return_type) = diff(plan.old, plan.new, |def| {
         def.lookup_expect::<ViewDef>(key).return_columns.iter()
     })
     .map(|col_diff| {
@@ -575,7 +575,7 @@ fn auto_migrate_view<'def>(plan: &mut AutoMigratePlan<'def>, old: &'def ViewDef,
     })
     .collect();
 
-    let Any(incompatible_param_types) = diff(plan.old, plan.new, |def| {
+    let Any(_incompatible_param_types) = diff(plan.old, plan.new, |def| {
         def.lookup_expect::<ViewDef>(key).param_columns.iter()
     })
     .map(|col_diff| {
@@ -604,15 +604,24 @@ fn auto_migrate_view<'def>(plan: &mut AutoMigratePlan<'def>, old: &'def ViewDef,
     })
     .collect();
 
-    if old.is_anonymous != new.is_anonymous || incompatible_return_type || incompatible_param_types {
-        plan.steps.push(AutoMigrateStep::AddView(new.key()));
-        plan.steps.push(AutoMigrateStep::RemoveView(old.key()));
-
-        if !plan.disconnects_all_users() {
-            plan.steps.push(AutoMigrateStep::DisconnectAllUsers);
-        }
-    } else {
-        plan.steps.push(AutoMigrateStep::UpdateView(old.key()));
+    // TODO: Uncomment and re-enable view auto-migrations without disconnecting clients
+    //
+    // if old.is_anonymous != new.is_anonymous || incompatible_return_type || incompatible_param_types {
+    //     plan.steps.push(AutoMigrateStep::AddView(new.key()));
+    //     plan.steps.push(AutoMigrateStep::RemoveView(old.key()));
+
+    //     if !plan.disconnects_all_users() {
+    //         plan.steps.push(AutoMigrateStep::DisconnectAllUsers);
+    //     }
+    // } else {
+    //     plan.steps.push(AutoMigrateStep::UpdateView(old.key()));
+    // }
+
+    plan.steps.push(AutoMigrateStep::AddView(new.key()));
+    plan.steps.push(AutoMigrateStep::RemoveView(old.key()));
+
+    if !plan.disconnects_all_users() {
+        plan.steps.push(AutoMigrateStep::DisconnectAllUsers);
     }
 
     Ok(())
@@ -1904,18 +1913,31 @@ mod tests {
             let plan = ponder_auto_migrate(&old_def, &new_def).expect("auto migration should succeed");
             let steps = &plan.steps[..];
 
-            assert!(!plan.disconnects_all_users(), "{name}, plan: {plan:#?}");
+            // TODO: Assert that we don't disconnect users once we have automatic view update in auto-migrations
+            //
+            // assert!(!plan.disconnects_all_users(), "{name}, plan: {plan:#?}");
+
+            // assert!(
+            //     steps.contains(&AutoMigrateStep::UpdateView(&my_view)),
+            //     "{name}, steps: {steps:?}"
+            // );
+            // assert!(
+            //     !steps.contains(&AutoMigrateStep::AddView(&my_view)),
+            //     "{name}, steps: {steps:?}"
+            // );
+            // assert!(
+            //     !steps.contains(&AutoMigrateStep::RemoveView(&my_view)),
+            //     "{name}, steps: {steps:?}"
+            // );
+
+            assert!(plan.disconnects_all_users(), "{name}, plan: {plan:#?}");
 
             assert!(
-                steps.contains(&AutoMigrateStep::UpdateView(&my_view)),
-                "{name}, steps: {steps:?}"
-            );
-            assert!(
-                !steps.contains(&AutoMigrateStep::AddView(&my_view)),
+                steps.contains(&AutoMigrateStep::AddView(&my_view)),
                 "{name}, steps: {steps:?}"
             );
             assert!(
-                !steps.contains(&AutoMigrateStep::RemoveView(&my_view)),
+                steps.contains(&AutoMigrateStep::RemoveView(&my_view)),
                 "{name}, steps: {steps:?}"
             );
         }