Fix anonymous view subscription cleanup

Author: joshua-spacetime

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))
     }
 
@@ -1388,6 +1351,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 {
@@ -1730,3 +1755,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(())
+    }
+}