Clear view tables on disconnect

Author: joshua-spacetime

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

@@ -101,7 +101,8 @@ pub async fn call<S: ControlStateDelegate + NodeDelegate>(
     };
 
     module
-        .call_identity_disconnected(caller_identity, connection_id)
+        // We don't clear views after reducer calls
+        .call_identity_disconnected(caller_identity, connection_id, false)
         .await
         .map_err(client_disconnected_error_to_response)?;
 
@@ -274,7 +275,8 @@ async fn procedure<S: ControlStateDelegate + NodeDelegate>(
     };
 
     module
-        .call_identity_disconnected(caller_identity, connection_id)
+        // We don't clear views after procedure calls
+        .call_identity_disconnected(caller_identity, connection_id, false)
         .await
         .map_err(client_disconnected_error_to_response)?;
 

crates/core/src/db/relational_db.rs

@@ -18,7 +18,7 @@ use spacetimedb_datastore::locking_tx_datastore::state_view::{
     IterByColEqMutTx, IterByColRangeMutTx, IterMutTx, IterTx, StateView,
 };
 use spacetimedb_datastore::locking_tx_datastore::{MutTxId, TxId};
-use spacetimedb_datastore::system_tables::{system_tables, StModuleRow};
+use spacetimedb_datastore::system_tables::{system_tables, StModuleRow, StViewRow, ST_VIEW_ID};
 use spacetimedb_datastore::system_tables::{StFields, StVarFields, StVarName, StVarRow, ST_MODULE_ID, ST_VAR_ID};
 use spacetimedb_datastore::traits::{
     InsertFlags, IsolationLevel, Metadata, MutTx as _, MutTxDatastore, Program, RowTypeForTable, Tx as _, TxDatastore,
@@ -1403,6 +1403,20 @@ impl RelationalDB {
         Ok(rows_deleted)
     }
 
+    /// Clear all rows from all view tables without dropping them.
+    pub fn clear_all_views(&self, tx: &mut MutTx) -> Result<(), DBError> {
+        for table_id in tx
+            .iter(ST_VIEW_ID)?
+            .map(StViewRow::try_from)
+            .collect::<Result<Vec<_>, _>>()?
+            .into_iter()
+            .filter_map(|row| row.table_id)
+        {
+            tx.clear_table(table_id)?;
+        }
+        Ok(())
+    }
+
     pub fn create_sequence(&self, tx: &mut MutTx, sequence_schema: SequenceSchema) -> Result<SequenceId, DBError> {
         Ok(self.inner.create_sequence_mut_tx(tx, sequence_schema)?)
     }

crates/core/src/host/host_controller.rs

@@ -847,9 +847,10 @@ impl Host {
         } = launched;
 
         // Disconnect dangling clients.
+        // No need to clear view tables here since we do it in `clear_all_clients`.
         for (identity, connection_id) in connected_clients {
             module_host
-                .call_identity_disconnected(identity, connection_id)
+                .call_identity_disconnected(identity, connection_id, false)
                 .await
                 .with_context(|| {
                     format!(

crates/core/src/host/module_host.rs

@@ -38,7 +38,7 @@ use spacetimedb_data_structures::error_stream::ErrorStream;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, IntMap};
 use spacetimedb_datastore::execution_context::{ExecutionContext, ReducerContext, Workload, WorkloadType};
 use spacetimedb_datastore::locking_tx_datastore::MutTxId;
-use spacetimedb_datastore::system_tables::{ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID};
+use spacetimedb_datastore::system_tables::{ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_CLIENT_ID};
 use spacetimedb_datastore::traits::{IsolationLevel, Program, TxData};
 use spacetimedb_durability::DurableOffset;
 use spacetimedb_execution::pipelined::{PipelinedProject, ViewProject};
@@ -903,7 +903,7 @@ impl ModuleHost {
                 // Call the `client_disconnected` reducer, if it exists.
                 // This is a no-op if the module doesn't define such a reducer.
                 this.subscriptions().remove_subscriber(client_id);
-                this.call_identity_disconnected_inner(client_id.identity, client_id.connection_id, inst)
+                this.call_identity_disconnected_inner(client_id.identity, client_id.connection_id, inst, true)
             })
             .await
         {
@@ -1024,6 +1024,7 @@ impl ModuleHost {
         caller_identity: Identity,
         caller_connection_id: ConnectionId,
         inst: &mut Instance,
+        clear_view_tables: bool,
     ) -> Result<(), ReducerCallError> {
         let reducer_lookup = self.info.module_def.lifecycle_reducer(Lifecycle::OnDisconnect);
         let reducer_name = reducer_lookup
@@ -1050,6 +1051,13 @@ impl ModuleHost {
         let fallback = || {
             let database_identity = me.info.database_identity;
             stdb.with_auto_commit(workload(), |mut_tx| {
+
+                if clear_view_tables {
+                    if let Err(err) = mut_tx.delete_view_data_for_client(caller_identity, caller_connection_id) {
+                        log::error!("`call_identity_disconnected`: failed to delete client view data: {err}");
+                    }
+                }
+
                 if !is_client_exist(mut_tx) {
                     // The client is already gone. Nothing to do.
                     log::debug!(
@@ -1076,7 +1084,13 @@ impl ModuleHost {
 
         if let Some((reducer_id, reducer_def)) = reducer_lookup {
             let stdb = me.module.replica_ctx().relational_db.clone();
-            let mut_tx = stdb.begin_mut_tx(IsolationLevel::Serializable, workload());
+            let mut mut_tx = stdb.begin_mut_tx(IsolationLevel::Serializable, workload());
+
+            if clear_view_tables {
+                if let Err(err) = mut_tx.delete_view_data_for_client(caller_identity, caller_connection_id) {
+                    log::error!("`call_identity_disconnected`: failed to delete client view data: {err}");
+                }
+            }
 
             if !is_client_exist(&mut_tx) {
                 // The client is already gone. Nothing to do.
@@ -1151,10 +1165,11 @@ impl ModuleHost {
         &self,
         caller_identity: Identity,
         caller_connection_id: ConnectionId,
+        clear_view_tables: bool,
     ) -> Result<(), ReducerCallError> {
         let me = self.clone();
         self.call("call_identity_disconnected", move |inst| {
-            me.call_identity_disconnected_inner(caller_identity, caller_connection_id, inst)
+            me.call_identity_disconnected_inner(caller_identity, caller_connection_id, inst, clear_view_tables)
         })
         .await?
     }
@@ -1166,8 +1181,10 @@ impl ModuleHost {
             let stdb = &me.module.replica_ctx().relational_db;
             let workload = Workload::Internal;
             stdb.with_auto_commit(workload, |mut_tx| {
+                stdb.clear_all_views(mut_tx)?;
                 stdb.clear_table(mut_tx, ST_CONNECTION_CREDENTIALS_ID)?;
                 stdb.clear_table(mut_tx, ST_CLIENT_ID)?;
+                stdb.clear_table(mut_tx, ST_VIEW_CLIENT_ID)?;
                 Ok::<(), DBError>(())
             })
         })

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -1252,9 +1252,9 @@ mod tests {
     use crate::system_tables::{
         system_tables, StColumnRow, StConnectionCredentialsFields, StConstraintData, StConstraintFields,
         StConstraintRow, StIndexAlgorithm, StIndexFields, StIndexRow, StRowLevelSecurityFields, StScheduledFields,
-        StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields, ST_CLIENT_NAME,
-        ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID,
-        ST_CONSTRAINT_NAME, ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE,
+        StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields, ST_CLIENT_ID,
+        ST_CLIENT_NAME, ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME,
+        ST_CONSTRAINT_ID, ST_CONSTRAINT_NAME, ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE,
         ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_NAME, ST_SCHEDULED_ID, ST_SCHEDULED_NAME, ST_SEQUENCE_ID,
         ST_SEQUENCE_NAME, ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_ARG_ID, ST_VIEW_ARG_NAME, ST_VIEW_CLIENT_ID,
         ST_VIEW_CLIENT_NAME, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_NAME, ST_VIEW_ID, ST_VIEW_NAME, ST_VIEW_PARAM_ID,

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -9,9 +9,9 @@ use super::{
     SharedMutexGuard, SharedWriteGuard,
 };
 use crate::system_tables::{
-    system_tables, ConnectionIdViaU128, StConnectionCredentialsFields, StConnectionCredentialsRow, StViewColumnFields,
-    StViewFields, StViewParamFields, StViewParamRow, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_COLUMN_ID, ST_VIEW_ID,
-    ST_VIEW_PARAM_ID,
+    system_tables, ConnectionIdViaU128, IdentityViaU256, StConnectionCredentialsFields, StConnectionCredentialsRow,
+    StViewClientFields, StViewClientRow, StViewColumnFields, StViewFields, StViewParamFields, StViewParamRow,
+    ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_CLIENT_ID, ST_VIEW_COLUMN_ID, ST_VIEW_ID, ST_VIEW_PARAM_ID,
 };
 use crate::traits::{InsertFlags, RowTypeForTable, TxData, UpdateFlags};
 use crate::{
@@ -1767,6 +1767,89 @@ impl<'a, I: Iterator<Item = RowRef<'a>>> Iterator for FilterDeleted<'a, I> {
 }
 
 impl MutTxId {
+    /// Delete view data and metadata for a client connection
+    pub fn delete_view_data_for_client(&mut self, sender: Identity, connection_id: ConnectionId) -> Result<()> {
+        for row in self.delete_st_view_client_rows(sender, connection_id)? {
+            if !self.is_identity_subscribed_to_view_args(row.view_id, row.arg_id, sender)? {
+                self.delete_view_rows_for_identity(row.view_id, row.arg_id, sender)?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Delete and return the rows in `st_view_client` for a client connection
+    fn delete_st_view_client_rows(
+        &mut self,
+        sender: Identity,
+        connection_id: ConnectionId,
+    ) -> Result<Vec<StViewClientRow>> {
+        let sender = IdentityViaU256(sender);
+        let conn_id = ConnectionIdViaU128(connection_id);
+        let cols = col_list![StViewClientFields::Identity, StViewClientFields::ConnectionId];
+        let value = AlgebraicValue::product([sender.into(), conn_id.into()]);
+        self.iter_by_col_eq(ST_VIEW_CLIENT_ID, cols, &value)?
+            .map(|row_ref| StViewClientRow::try_from(row_ref).map(|row| (row, row_ref.pointer())))
+            .collect::<Result<Vec<_>>>()?
+            .into_iter()
+            .map(|(row, ptr)| self.delete(ST_VIEW_CLIENT_ID, ptr).map(|_| row))
+            .collect()
+    }
+
+    /// Is anyone is subscribed to the view arguments identified by `arg_id`
+    fn is_identity_subscribed_to_view_args(&self, view_id: ViewId, arg_id: u64, sender: Identity) -> Result<bool> {
+        Ok(self
+            .iter_by_col_eq(
+                ST_VIEW_CLIENT_ID,
+                col_list![
+                    StViewClientFields::ViewId,
+                    StViewClientFields::ArgId,
+                    StViewClientFields::Identity
+                ],
+                &AlgebraicValue::product([view_id.into(), arg_id.into(), sender.into()]),
+            )?
+            .next()
+            .is_some())
+    }
+
+    /// Lookup a row in `st_view` by its primary key
+    fn st_view_row(&self, view_id: ViewId) -> Result<Option<StViewRow>> {
+        self.iter_by_col_eq(ST_VIEW_ID, col_list![StViewFields::ViewId], &view_id.into())?
+            .next()
+            .map(StViewRow::try_from)
+            .transpose()
+    }
+
+    /// Get the [`TableId`] for this view's backing table by probing `st_view`.
+    /// Note, all views with at least one subscriber are materialized.
+    fn get_table_id_for_view(&self, view_id: ViewId) -> Result<Option<(TableId, bool)>> {
+        Ok(self
+            .st_view_row(view_id)?
+            .and_then(|row| row.table_id.map(|id| (id, row.is_anonymous))))
+    }
+
+    /// Delete the rows of a view subscribed to by `sender`
+    fn delete_view_rows_for_identity(&mut self, view_id: ViewId, arg_id: u64, sender: Identity) -> Result<()> {
+        if let Some((table_id, is_anonymous)) = self.get_table_id_for_view(view_id)? {
+            let value = if is_anonymous {
+                let none_sender = AlgebraicValue::OptionNone();
+                AlgebraicValue::product([none_sender, arg_id.into()])
+            } else {
+                let sender = IdentityViaU256(sender);
+                let some_sender = AlgebraicValue::OptionSome(sender.into());
+                AlgebraicValue::product([some_sender, arg_id.into()])
+            };
+            for row_pointer in self
+                .iter_by_col_eq(table_id, col_list![0, 1], &value)?
+                .map(|row_ref| row_ref.pointer())
+                .collect::<Vec<_>>()
+                .into_iter()
+            {
+                self.delete(table_id, row_pointer)?;
+            }
+        }
+        Ok(())
+    }
+
     pub fn insert_st_client(
         &mut self,
         identity: Identity,

crates/datastore/src/system_tables.rs

@@ -937,6 +937,20 @@ pub struct StViewClientRow {
     pub connection_id: ConnectionIdViaU128,
 }
 
+impl TryFrom<RowRef<'_>> for StViewClientRow {
+    type Error = DatastoreError;
+
+    fn try_from(row: RowRef<'_>) -> Result<Self, Self::Error> {
+        read_via_bsatn(row)
+    }
+}
+
+impl From<StViewClientRow> for (Identity, ConnectionId) {
+    fn from(value: StViewClientRow) -> Self {
+        (value.identity.0, value.connection_id.0)
+    }
+}
+
 /// System table [ST_VIEW_ARG_NAME]
 ///
 /// | id | bytes   |
@@ -1238,6 +1252,12 @@ impl From<Identity> for IdentityViaU256 {
     }
 }
 
+impl From<IdentityViaU256> for AlgebraicValue {
+    fn from(val: IdentityViaU256) -> Self {
+        AlgebraicValue::U256(val.0.to_u256().into())
+    }
+}
+
 /// System table [ST_MODULE_NAME]
 /// This table holds exactly one row, describing the latest version of the
 /// SpacetimeDB module associated with the database:
@@ -1348,6 +1368,12 @@ impl TryFrom<RowRef<'_>> for StClientRow {
     }
 }
 
+impl From<StClientRow> for (Identity, ConnectionId) {
+    fn from(value: StClientRow) -> Self {
+        (value.identity.0, value.connection_id.0)
+    }
+}
+
 /// System table [ST_VAR_NAME]
 ///
 /// | name        | value     |