Fix anonymous view subscription cleanup (#4646)

# Description of Changes

Fixes a bug during view cleanup where we would drop anonymous views that
had live subscribers.

The reason for this bug is that the `st_view_sub` does not have a single
entry for an anonymous view. Rather it has multiple entries - one per
identity subscribed. However the code was assuming only a single entry
per anonymous view, and so when view cleanup ran, we would only look at
the first entry in `st_view_sub` for an anonymous view, and if there
were no subscribers, we would drop the view's backing table and read
set. But there could still have been other clients subscribed to that
view, at which point they would stop receiving updates.

This change fixes the bug by making sure there aren't any rows in
`st_view_sub` with live subscribers before dropping the anonymous view's
backing table and read set.

The better fix would be to update `st_view_sub` so that there's only
ever at most one row per anonymous view. Unfortunately `st_view_sub`
mixes both anonymous and sender-scoped views, and the `sender` column is
of type `Identity`, not `Option<Identity>`. Instead of adding a new
system table to correct this, I chose to keep using `st_view_sub` and
delay this refactor until we add support for view parameters.

# API and ABI breaking changes

None

# Expected complexity level and risk

1.5

# Testing

Added regression tests that assert on the underlying system table rows.
A smoketest was **not** added because cleanup runs periodically and
currently there's no way to force a clean up.

crates/core/src/db/relational_db.rs

@@ -2349,12 +2349,12 @@ mod tests {
     use std::fs::OpenOptions;
     use std::path::PathBuf;
     use std::rc::Rc;
-    use std::time::Instant;
+    use std::time::{Duration, Instant};
 
     use super::tests_utils::begin_mut_tx;
     use super::*;
     use crate::db::relational_db::tests_utils::{
-        begin_tx, insert, make_snapshot, with_auto_commit, with_read_only, TestDB,
+        begin_tx, create_view_for_test, insert, make_snapshot, with_auto_commit, with_read_only, TestDB,
     };
     use anyhow::bail;
     use bytes::Bytes;
@@ -2518,6 +2518,15 @@ mod tests {
         Ok((view_id, table_id, module_def.clone(), view_def.clone()))
     }
 
+    fn setup_anonymous_view(stdb: &TestDB) -> ResultTest<(ViewId, TableId)> {
+        Ok(create_view_for_test(
+            stdb,
+            "my_anonymous_view",
+            &[("b", AlgebraicType::U8)],
+            true,
+        )?)
+    }
+
     fn insert_view_row(stdb: &TestDB, view_id: ViewId, table_id: TableId, sender: Identity, v: u8) -> ResultTest<()> {
         let row_pv = |v: u8| product![v];
 
@@ -2543,6 +2552,22 @@ mod tests {
             .collect()
     }
 
+    fn project_anonymous_views(stdb: &TestDB, table_id: TableId) -> Vec<ProductValue> {
+        let tx = begin_tx(stdb);
+
+        stdb.iter(&tx, table_id)
+            .unwrap()
+            .map(|row| row.to_product_value())
+            .collect()
+    }
+
+    fn update_last_called(stdb: &TestDB, view_id: ViewId, sender: Identity, last_called: Timestamp) -> ResultTest<()> {
+        let mut tx = begin_mut_tx(stdb);
+        tx.update_view_timestamp_at(view_id, ArgId::SENTINEL, sender, last_called)?;
+        stdb.commit_tx(tx)?;
+        Ok(())
+    }
+
     #[test]
     fn test_view_tables_are_ephemeral_in_commitlog() -> ResultTest<()> {
         let stdb = TestDB::durable_without_snapshot_repo()?;
@@ -2723,6 +2748,99 @@ mod tests {
 
         Ok(())
     }
+
+    /// Regression test for anonymous-view cleanup.
+    ///
+    /// If one subscriber row has expired but another subscriber for the same anonymous
+    /// view is still live, cleanup must delete only the stale bookkeeping row and keep
+    /// the shared materialized backing table intact.
+    #[test]
+    fn test_anonymous_view_cleanup_keeps_rows_for_live_subscribers() -> ResultTest<()> {
+        let stdb = TestDB::durable()?;
+        let (view_id, table_id) = setup_anonymous_view(&stdb)?;
+
+        let stale_sender = Identity::ONE;
+        let live_sender = Identity::ZERO;
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.subscribe_view(view_id, ArgId::SENTINEL, stale_sender)?;
+        tx.subscribe_view(view_id, ArgId::SENTINEL, live_sender)?;
+        stdb.materialize_anonymous_view(&mut tx, table_id, vec![product![42u8]])?;
+        stdb.commit_tx(tx)?;
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.unsubscribe_view(view_id, ArgId::SENTINEL, stale_sender)?;
+        stdb.commit_tx(tx)?;
+
+        // Make one row definitely expired without relying on wall-clock sleeps.
+        update_last_called(&stdb, view_id, stale_sender, Timestamp::UNIX_EPOCH)?;
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.update_view_timestamp(view_id, ArgId::SENTINEL, live_sender)?;
+        stdb.commit_tx(tx)?;
+
+        // Cleanup should remove only the stale subscriber row and keep the shared
+        // anonymous materialization because another subscriber is still live.
+        let mut tx = begin_mut_tx(&stdb);
+        let (_cleaned, _total_expired) = tx.clear_expired_views(Duration::from_secs(1), VIEW_CLEANUP_BUDGET)?;
+        stdb.commit_tx(tx)?;
+
+        assert_eq!(
+            project_anonymous_views(&stdb, table_id),
+            vec![product![42u8]],
+            "anonymous view rows should survive cleanup while another identity is still subscribed"
+        );
+
+        let tx = begin_mut_tx(&stdb);
+        let st_after = tx.lookup_st_view_subs(view_id)?;
+        assert_eq!(st_after.len(), 1);
+        assert_eq!(st_after[0].identity.0, live_sender);
+        assert!(st_after[0].has_subscribers);
+        assert_eq!(st_after[0].num_subscribers, 1);
+
+        Ok(())
+    }
+
+    /// Regression test for anonymous-view cleanup.
+    ///
+    /// Once the final subscriber row for an anonymous view has expired, cleanup must
+    /// remove both the stale bookkeeping row and the shared materialized backing table.
+    #[test]
+    fn test_anonymous_view_cleanup_clears_rows_when_unused() -> ResultTest<()> {
+        let stdb = TestDB::durable()?;
+        let (view_id, table_id) = setup_anonymous_view(&stdb)?;
+
+        let sender = Identity::ONE;
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.subscribe_view(view_id, ArgId::SENTINEL, sender)?;
+        stdb.materialize_anonymous_view(&mut tx, table_id, vec![product![42u8]])?;
+        stdb.commit_tx(tx)?;
+
+        let mut tx = begin_mut_tx(&stdb);
+        tx.unsubscribe_view(view_id, ArgId::SENTINEL, sender)?;
+        stdb.commit_tx(tx)?;
+
+        // Mark the unsubscribed row as expired so cleanup can process it immediately.
+        update_last_called(&stdb, view_id, sender, Timestamp::UNIX_EPOCH)?;
+
+        // With no remaining subscriber rows, cleanup should drop the shared
+        // anonymous materialization and remove the bookkeeping row.
+        let mut tx = begin_mut_tx(&stdb);
+        let (_cleaned, _total_expired) = tx.clear_expired_views(Duration::from_secs(1), VIEW_CLEANUP_BUDGET)?;
+        stdb.commit_tx(tx)?;
+
+        assert!(
+            project_anonymous_views(&stdb, table_id).is_empty(),
+            "anonymous view rows should be cleared once no entries remain"
+        );
+
+        let tx = begin_mut_tx(&stdb);
+        let st_after = tx.lookup_st_view_subs(view_id)?;
+        assert!(st_after.is_empty());
+
+        Ok(())
+    }
     #[test]
     fn test_table_name() -> ResultTest<()> {
         let stdb = TestDB::durable()?;

crates/core/src/host/module_host.rs

@@ -1884,7 +1884,12 @@ impl ModuleHost {
             let table_id = st_view_row.table_id.ok_or(ViewCallError::TableDoesNotExist(view_id))?;
             let is_anonymous = st_view_row.is_anonymous;
             let sender = if is_anonymous { None } else { Some(caller) };
-            if !tx.is_view_materialized(view_id, ArgId::SENTINEL, caller)? {
+            let is_materialized = if is_anonymous {
+                tx.is_anonymous_view_materialized(view_id)?
+            } else {
+                tx.is_view_materialized(view_id, ArgId::SENTINEL, caller)?
+            };
+            if !is_materialized {
                 let (res, trapped) =
                     Self::call_view(instance, tx, &view_name, view_id, table_id, Nullary, caller, sender)?;
                 tx = res.tx;

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -688,44 +688,7 @@ impl InstanceCommon {
         tx: MutTxId,
         inst: &mut I,
     ) -> Result<(ViewCallResult, bool), anyhow::Error> {
-        let views = self.info.module_def.views().collect::<Vec<_>>();
-        let owner_identity = self.info.owner_identity;
-
-        let mut view_calls = Vec::new();
-
-        for view in views {
-            let ViewDef {
-                name: view_name,
-                is_anonymous,
-                fn_ptr,
-                product_type_ref,
-                ..
-            } = view;
-
-            let st_view = tx
-                .view_from_name(view_name)?
-                .ok_or_else(|| anyhow::anyhow!("view {} not found in database", &view_name))?;
-
-            let view_id = st_view.view_id;
-            let table_id = st_view
-                .table_id
-                .ok_or_else(|| anyhow::anyhow!("view {} does not have a backing table in database", &view_name))?;
-
-            for sub in tx.lookup_st_view_subs(view_id)? {
-                view_calls.push(CallViewParams {
-                    view_name: view_name.clone(),
-                    view_id,
-                    table_id,
-                    fn_ptr: *fn_ptr,
-                    caller: owner_identity,
-                    sender: if *is_anonymous { None } else { Some(sub.identity.into()) },
-                    args: ArgsTuple::nullary(),
-                    row_type: *product_type_ref,
-                    timestamp: Timestamp::now(),
-                });
-            }
-        }
-
+        let view_calls = collect_subscribed_view_calls(&tx, &self.info.module_def, self.info.owner_identity)?;
         Ok(self.execute_view_calls(tx, view_calls, inst))
     }
 
@@ -1370,6 +1333,68 @@ impl InstanceCommon {
     }
 }
 
+fn collect_subscribed_view_calls(
+    tx: &MutTxId,
+    module_def: &ModuleDef,
+    owner_identity: Identity,
+) -> Result<Vec<CallViewParams>, anyhow::Error> {
+    let mut view_calls = Vec::new();
+
+    for view in module_def.views() {
+        let ViewDef {
+            name: view_name,
+            is_anonymous,
+            fn_ptr,
+            product_type_ref,
+            ..
+        } = view;
+
+        let st_view = tx
+            .view_from_name(view_name)?
+            .ok_or_else(|| anyhow::anyhow!("view {} not found in database", &view_name))?;
+
+        let view_id = st_view.view_id;
+        let table_id = st_view
+            .table_id
+            .ok_or_else(|| anyhow::anyhow!("view {} does not have a backing table in database", &view_name))?;
+        let subs = tx.lookup_st_view_subs(view_id)?;
+
+        if *is_anonymous {
+            if subs.is_empty() {
+                continue;
+            }
+            view_calls.push(CallViewParams {
+                view_name: view_name.clone(),
+                view_id,
+                table_id,
+                fn_ptr: *fn_ptr,
+                caller: owner_identity,
+                sender: None,
+                args: ArgsTuple::nullary(),
+                row_type: *product_type_ref,
+                timestamp: Timestamp::now(),
+            });
+            continue;
+        }
+
+        for sub in subs {
+            view_calls.push(CallViewParams {
+                view_name: view_name.clone(),
+                view_id,
+                table_id,
+                fn_ptr: *fn_ptr,
+                caller: owner_identity,
+                sender: Some(sub.identity.into()),
+                args: ArgsTuple::nullary(),
+                row_type: *product_type_ref,
+                timestamp: Timestamp::now(),
+            });
+        }
+    }
+
+    Ok(view_calls)
+}
+
 /// Pre-fetched VM metrics counters for all reducers and views in a module.
 /// Anonymous views have lazily fetched metrics counters.
 struct AllVmMetrics {
@@ -1712,3 +1737,91 @@ impl InstanceOp for ProcedureOp {
         FuncCallType::Procedure
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::collect_subscribed_view_calls;
+    use crate::db::relational_db::tests_utils::{begin_mut_tx, TestDB};
+    use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9Builder;
+    use spacetimedb_lib::{AlgebraicType, Identity, ProductType};
+    use spacetimedb_primitives::ArgId;
+    use spacetimedb_sats::raw_identifier::RawIdentifier;
+    use spacetimedb_schema::def::ModuleDef;
+
+    fn module_def_for_view(name: &str, is_anonymous: bool) -> ModuleDef {
+        let mut builder = RawModuleDefV9Builder::new();
+        let name = RawIdentifier::new(name);
+        let type_ref = builder.add_algebraic_type(
+            [],
+            name.clone(),
+            AlgebraicType::Product(ProductType::from_iter([("x", AlgebraicType::U8)])),
+            true,
+        );
+
+        builder.add_view(
+            name.clone(),
+            0,
+            true,
+            is_anonymous,
+            ProductType::unit(),
+            AlgebraicType::array(AlgebraicType::Ref(type_ref)),
+        );
+
+        builder.finish().try_into().expect("test module def should be valid")
+    }
+
+    /// Regression test for evaluating anonymous views.
+    ///
+    /// Anonymous views have one shared materialization,
+    /// so we should only re-evaluate once even if there are multiple subscribers.
+    #[test]
+    fn test_dedup_anonymous_view_calls() -> anyhow::Result<()> {
+        let stdb = TestDB::in_memory()?;
+        let module_def = module_def_for_view("anonymous_view", true);
+        let view_def = module_def.view("anonymous_view").expect("view should exist");
+
+        let mut tx = begin_mut_tx(&stdb);
+        let (view_id, _table_id) = stdb.create_view(&mut tx, &module_def, view_def)?;
+        tx.subscribe_view(view_id, ArgId::SENTINEL, Identity::ZERO)?;
+        tx.subscribe_view(view_id, ArgId::SENTINEL, Identity::ONE)?;
+
+        // Two subscriber rows exist, but anonymous views should still be reevaluated once
+        // because they share a single materialization.
+        let calls = collect_subscribed_view_calls(&tx, &module_def, Identity::ZERO)?;
+
+        assert_eq!(
+            calls.len(),
+            1,
+            "anonymous views should only be reevaluated once even with multiple subscriber rows"
+        );
+        assert_eq!(calls[0].view_id, view_id);
+        assert_eq!(calls[0].sender, None);
+        Ok(())
+    }
+
+    /// Regression test for evaluating sender-scoped views.
+    ///
+    /// These views have separate materializations per sender,
+    /// so reevaluation must emit one call per subscribed sender.
+    #[test]
+    fn test_distinct_sender_scoped_view_calls() -> anyhow::Result<()> {
+        let stdb = TestDB::in_memory()?;
+        let module_def = module_def_for_view("sender_view", false);
+        let view_def = module_def.view("sender_view").expect("view should exist");
+
+        let mut tx = begin_mut_tx(&stdb);
+        let (view_id, _table_id) = stdb.create_view(&mut tx, &module_def, view_def)?;
+        tx.subscribe_view(view_id, ArgId::SENTINEL, Identity::ZERO)?;
+        tx.subscribe_view(view_id, ArgId::SENTINEL, Identity::ONE)?;
+
+        // Sender-backed views keep one materialization per sender, so reevaluation must
+        // preserve both callers.
+        let calls = collect_subscribed_view_calls(&tx, &module_def, Identity::ZERO)?;
+        let senders: Vec<_> = calls.iter().filter_map(|call| call.sender).collect();
+
+        assert_eq!(calls.len(), 2, "sender views should still reevaluate once per sender");
+        assert!(senders.contains(&Identity::ZERO));
+        assert!(senders.contains(&Identity::ONE));
+        Ok(())
+    }
+}

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -2283,18 +2283,36 @@ impl MutTxId {
         Ok(self.iter_by_col_eq(ST_VIEW_SUB_ID, cols, &value)?.next().is_some())
     }
 
+    /// Does any `st_view_sub` row exist for this anonymous view?
+    pub fn is_anonymous_view_materialized(&self, view_id: ViewId) -> Result<bool> {
+        let cols = StViewSubFields::ViewId;
+        let value = view_id.into();
+        Ok(self.iter_by_col_eq(ST_VIEW_SUB_ID, cols, &value)?.next().is_some())
+    }
+
     /// Updates the `last_called` timestamp in `st_view_sub`.
     /// Inserts a row into `st_view_sub` with no subscribers if the row does not exist.
     ///
     /// This is invoked when calling a view, but not subscribing to it.
     /// Such is the case for the sql http api.
     pub fn update_view_timestamp(&mut self, view_id: ViewId, arg_id: ArgId, sender: Identity) -> Result<()> {
+        self.update_view_timestamp_at(view_id, arg_id, sender, Timestamp::now())
+    }
+
+    /// Updates the `last_called` timestamp in `st_view_sub` to an explicit value.
+    pub fn update_view_timestamp_at(
+        &mut self,
+        view_id: ViewId,
+        arg_id: ArgId,
+        sender: Identity,
+        last_called: Timestamp,
+    ) -> Result<()> {
         use StViewSubFields::*;
 
         let identity = IdentityViaU256(sender);
         let cols = col_list![ViewId, ArgId, Identity];
         let value = AlgebraicValue::product([view_id.into(), arg_id.into(), identity.into()]);
-        let last_called = Timestamp::now().into();
+        let last_called = last_called.into();
 
         // Update `last_called` of `st_view_sub` row
         if let Some((row, ptr)) = self
@@ -2374,18 +2392,18 @@ impl MutTxId {
     /// - `has_subscribers == false`, `num_subscribers == 0`.
     /// - `last_called` is older than `expiration_duration`.
     ///
-    /// For each such expired view:
-    /// 1. It clears the backing table,
-    /// 2. Removes the view from the committed read set, and
-    /// 3. Deletes the subscription row.
+    /// For each such expired row:
+    /// 1. It deletes the expired `st_view_sub` row.
+    /// 2. If that row was the last remaining materialization entry for the view,
+    ///    it clears the backing table and removes the view from the committed read set.
     ///
     /// The cleanup is bounded by a total `max_duration`. The function stops when either:
     /// - all expired views have been processed, or
     /// - the `max_duration` budget is reached.
     ///
     /// Returns a tuple `(cleaned, total_expired)`:
-    /// - `cleaned`: Number of views actually cleaned (deleted) in this run.
-    /// - `total_expired`: Total number of expired views found (even if not all were cleaned due to time budget).
+    /// - `cleaned`: Number of expired `st_view_sub` rows deleted in this run.
+    /// - `total_expired`: Total number of expired rows found (even if not all were cleaned due to time budget).
     pub fn clear_expired_views(
         &mut self,
         expiration_duration: Duration,
@@ -2416,7 +2434,8 @@ impl MutTxId {
 
         let total_expired = expired_items.len();
 
-        // For each expired view subscription, clear the backing table and delete the subscription
+        // For each expired subscription row, clear the backing table only if that row
+        // was the last remaining entry for the shared materialization.
         for (view_id, sender, sub_row_ptr) in expired_items {
             // Check if we've exceeded our time budget
             if start.elapsed() >= max_duration {
@@ -2429,8 +2448,10 @@ impl MutTxId {
             let table_id = table_id.expect("views have backing table");
 
             if is_anonymous {
-                self.clear_table(table_id)?;
-                self.drop_view_from_committed_read_set(view_id);
+                if !self.has_other_st_view_sub_entries(view_id, sub_row_ptr)? {
+                    self.clear_table(table_id)?;
+                    self.drop_view_from_committed_read_set(view_id);
+                }
             } else {
                 let rows_to_delete = self
                     .iter_by_col_eq(table_id, 0, &sender.into())?
@@ -2548,6 +2569,16 @@ impl MutTxId {
             .collect::<Result<Vec<_>>>()
     }
 
+    /// Does this `view_id` have other entries in `st_view_sub` besides `current_ptr`?
+    /// Can be true for anonymous views with multiple subscribers.
+    fn has_other_st_view_sub_entries(&self, view_id: ViewId, current_ptr: RowPointer) -> Result<bool> {
+        let cols = StViewSubFields::ViewId;
+        let value = view_id.into();
+        Ok(self
+            .iter_by_col_eq(ST_VIEW_SUB_ID, cols, &value)?
+            .any(|row_ref| row_ref.pointer() != current_ptr))
+    }
+
     /// 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())?