More effective views cleanup.

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -145,6 +145,11 @@ impl MemoryUsage for ViewInstanceState {
     }
 }
 
+pub struct ViewCleanupResult {
+    pub cleaned: usize,
+    pub backlog: bool,
+}
+
 impl ViewInstanceState {
     fn new(args: ViewInstanceArgs, last_used: Timestamp) -> Self {
         Self {
@@ -797,7 +802,11 @@ impl MutTxId {
         self.drop_st_view_param(view_id)?;
         self.drop_st_view_column(view_id)?;
         self.drop_st_view_sub(view_id)?;
-        for call in self.effective_view_instances_for_view(view_id).into_keys() {
+        let calls = self
+            .effective_view_instances_for_view(view_id)
+            .map(|(call, _)| call.clone())
+            .collect::<Vec<_>>();
+        for call in calls {
             self.view_instances.remove(call);
         }
         self.drop_view_from_committed_read_set(view_id);
@@ -2771,32 +2780,38 @@ impl MutTxId {
         self.view_instances.get_cloned(&self.committed_state_write_lock, call)
     }
 
-    fn effective_view_instances(&self) -> HashMap<ViewCallInfo, ViewInstanceState> {
-        let mut instances = self
+    fn effective_view_instances(&self) -> impl Iterator<Item = (&ViewCallInfo, &ViewInstanceState)> + '_ {
+        let committed = self
             .committed_state_write_lock
             .view_instances()
-            .map(|(call, state)| (call.clone(), state.clone()))
-            .collect::<HashMap<_, _>>();
+            .filter_map(|(call, state)| match self.view_instances.changes.get(call) {
+                Some(Some(tx_state)) => Some((call, tx_state)),
+                Some(None) => None,
+                None => Some((call, state)),
+            });
 
-        for (call, state) in &self.view_instances.changes {
-            match state {
-                Some(state) => {
-                    instances.insert(call.clone(), state.clone());
-                }
-                None => {
-                    instances.remove(call);
-                }
+        // The committed iterator above already applies tx-local overrides and
+        // removals for committed calls. This second half yields only tx-local
+        // inserts whose calls do not exist in committed state.
+        let tx_only = self.view_instances.changes.iter().filter_map(|(call, state)| {
+            if self.committed_state_write_lock.view_instance(call).is_some() {
+                None
+            } else {
+                state.as_ref().map(|state| (call, state))
             }
-        }
+        });
 
-        instances
+        committed.chain(tx_only)
     }
 
-    fn effective_view_instances_for_view(&self, view_id: ViewId) -> HashMap<ViewCallInfo, ViewInstanceState> {
+    fn effective_view_instances_for_view(
+        &self,
+        view_id: ViewId,
+    ) -> impl Iterator<Item = (&ViewCallInfo, &ViewInstanceState)> + '_ {
         self.effective_view_instances()
-            .into_iter()
-            .filter(|(call, _)| call.view_id == view_id)
-            .collect()
+            // FIXME: use a better data structure to store view instances in `CommittedState` and `MutTxId`,
+            // so that this can behave like an index scan rather than a full scan and filter.
+            .filter(move |(call, _)| call.view_id == view_id)
     }
 
     /// Is this view argument currently materialized?
@@ -2812,17 +2827,20 @@ impl MutTxId {
     /// Returns all materialized view instances for `view_id`.
     pub fn materialized_view_instances_for_view(&self, view_id: ViewId) -> Vec<ViewInstanceArgs> {
         self.effective_view_instances_for_view(view_id)
-            .into_values()
-            .map(|state| state.args)
+            .map(|(_, state)| state.args)
             .collect()
     }
 
     /// Returns active subscribers for a materialized view.
     #[cfg(any(test, feature = "test"))]
     pub fn active_subscribers_for_view(&self, view_id: ViewId) -> Vec<(Identity, u64)> {
         self.effective_view_instances_for_view(view_id)
-            .into_values()
-            .flat_map(|state| state.active_subscribers.into_iter())
+            .flat_map(|(_, state)| {
+                state
+                    .active_subscribers
+                    .iter()
+                    .map(|(identity, count)| (*identity, *count))
+            })
             .collect()
     }
 
@@ -2895,50 +2913,77 @@ impl MutTxId {
 
     /// Clean up materialized view arguments that have no subscribers and haven’t been used recently.
     ///
-    /// Returns a tuple `(cleaned, total_expired)`:
-    /// - `cleaned`: Number of expired materialized view arguments deleted in this run.
-    /// - `total_expired`: Total number of expired materialized view arguments found.
+    /// Expired views are cleared in batches. Each loop iteration selects up to
+    /// `batch_size` expired view calls and deletes their materialized rows.
+    ///
+    /// `max_duration` is checked between batches, so cleanup always finishes at
+    /// least one batch once it starts, even if that batch takes longer than
+    /// `max_duration`.
+    ///
+    /// Returns the number of calls cleaned and whether more expired work is
+    /// likely pending.
     pub fn clear_expired_views(
         &mut self,
         expiration_duration: Duration,
         max_duration: Duration,
-    ) -> Result<(usize, usize)> {
+        batch_size: usize,
+    ) -> Result<ViewCleanupResult> {
         let start = std::time::Instant::now();
         let expiration_threshold = Timestamp::now() - expiration_duration;
-        let mut cleaned_count = 0;
+        let is_expired = |state: &ViewInstanceState| !state.has_subscribers() && state.last_used < expiration_threshold;
+        let mut cleaned = 0;
+        let batch_size = batch_size.max(1);
+
+        loop {
+            let mut unexpired_visited_building_batch = 0;
+            let filter_and_count_view_instance = |(call, state): (&ViewCallInfo, &ViewInstanceState)| {
+                if is_expired(state) {
+                    Some(call.clone())
+                } else {
+                    unexpired_visited_building_batch += 1;
+                    None
+                }
+            };
+            // FIXME: Use a better datastructure for `view_instances` in `CommittedState` and `MutTxId`
+            // to avoid having to traverse live or recently-expired views to delete stale ones.
 
-        let expired_items = self
-            .effective_view_instances()
-            .into_iter()
-            .filter_map(|(call, state)| {
-                (!state.has_subscribers() && state.last_used < expiration_threshold).then_some(call)
-            })
-            .collect::<Vec<_>>();
+            let mut batch = self
+                .effective_view_instances()
+                .filter_map(filter_and_count_view_instance)
+                .take(batch_size + 1)
+                .collect::<Vec<_>>();
 
-        let total_expired = expired_items.len();
+            log::info!(
+                "[{}]: Traversed {} unexpired views to collect and delete a batch of {} expired views",
+                self.ctx.database_identity,
+                unexpired_visited_building_batch,
+                batch.len()
+            );
 
-        for call in expired_items {
-            if start.elapsed() >= max_duration {
-                break;
-            }
+            let backlog = batch.len() > batch_size;
+            batch.truncate(batch_size);
 
-            let StViewRow { table_id, .. } = self.lookup_st_view(call.view_id)?;
-            let table_id = table_id.expect("views have backing table");
-            let rows_to_delete = self
-                .iter_by_col_eq(table_id, VIEW_ARG_HASH_COL, &call.arg_hash)?
-                .map(|res| res.pointer())
-                .collect::<Vec<_>>();
+            for call in batch {
+                let StViewRow { table_id, .. } = self.lookup_st_view(call.view_id)?;
+                let table_id = table_id.expect("views have backing table");
+                let rows_to_delete = self
+                    .iter_by_col_eq(table_id, VIEW_ARG_HASH_COL, &call.arg_hash)?
+                    .map(|res| res.pointer())
+                    .collect::<Vec<_>>();
+
+                for row_ptr in rows_to_delete {
+                    self.delete(table_id, row_ptr)?;
+                }
 
-            for row_ptr in rows_to_delete {
-                self.delete(table_id, row_ptr)?;
+                self.drop_view_call_from_committed_read_set(call.clone());
+                self.view_instances.remove(call);
+                cleaned += 1;
             }
 
-            self.drop_view_call_from_committed_read_set(call.clone());
-            self.view_instances.remove(call);
-            cleaned_count += 1;
+            if start.elapsed() >= max_duration || !backlog {
+                return Ok(ViewCleanupResult { cleaned, backlog });
+            }
         }
-
-        Ok((cleaned_count, total_expired))
     }
 
     /// Clear all rows from all view tables without dropping them.

crates/engine/src/relational_db.rs

@@ -1126,52 +1126,77 @@ impl RelationalDB {
 /// Value is chosen arbitrarily; can be tuned later if needed.
 const VIEWS_EXPIRATION: std::time::Duration = std::time::Duration::from_secs(10 * 60);
 
+/// Normal interval between view cleanup runs.
+const VIEW_CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10 * 60);
+
+/// Lower bound for adaptive cleanup retries when expired views remain.
+const MIN_VIEW_CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);
+
 /// Duration to budget for each view cleanup job,
-/// so that it doesn't hold transaction lock for to long.
+/// so that it doesn't hold transaction lock for too long.
 //TODO: Make this value configurable
 const VIEW_CLEANUP_BUDGET: std::time::Duration = std::time::Duration::from_millis(10);
 
-/// Spawn a background task that periodically cleans up expired views
+/// Number of expired view calls to clean before checking the time budget.
+const VIEW_CLEANUP_BATCH_SIZE: usize = 1024;
+
+/// Spawn a background task that periodically cleans up expired views.
+///
+/// Each cleanup run has a fixed lock-hold budget. If a run leaves expired
+/// views behind, the next run is scheduled sooner, down to
+/// [`MIN_VIEW_CLEANUP_INTERVAL`]. Once cleanup catches up, the interval resets.
 pub fn spawn_view_cleanup_loop(
     db: Arc<RelationalDB>,
     handle: &spacetimedb_runtime::Handle,
 ) -> spacetimedb_runtime::AbortHandle {
-    fn run_view_cleanup(db: &RelationalDB) {
+    fn shorten_cleanup_interval(current: std::time::Duration) -> std::time::Duration {
+        (current / 2).max(MIN_VIEW_CLEANUP_INTERVAL)
+    }
+
+    fn run_view_cleanup(db: &RelationalDB) -> bool {
         match db.with_auto_commit(Workload::Internal, |tx| {
-            tx.clear_expired_views(VIEWS_EXPIRATION, VIEW_CLEANUP_BUDGET)
+            tx.clear_expired_views(VIEWS_EXPIRATION, VIEW_CLEANUP_BUDGET, VIEW_CLEANUP_BATCH_SIZE)
                 .map_err(DBError::from)
         }) {
-            Ok((cleared, total_expired)) => {
-                if cleared != total_expired {
+            Ok(result) => {
+                if result.backlog {
                     // TODO: metrics
                     log::info!(
-                        "[{}] DATABASE: cleared {} expired views ({} remaining)",
+                        "[{}] DATABASE: cleared {} expired views (more pending)",
                         db.database_identity(),
-                        cleared,
-                        total_expired - cleared
+                        result.cleaned,
                     );
                 }
+                result.backlog
             }
             Err(e) => {
                 log::error!(
                     "[{}] DATABASE: failed to clear expired views: {}",
                     db.database_identity(),
                     e
                 );
+                false
             }
         }
     }
 
     let handle_clone = handle.clone();
     handle
         .spawn(async move {
+            let mut interval = VIEW_CLEANUP_INTERVAL;
             loop {
                 // Offload actual cleanup to blocking thread pool, as `VIEW_CLEANUP_BUDGET` is defined
                 // in milliseconds, which may be too long for async tasks.
                 let db = db.clone();
-                handle_clone.spawn_blocking(move || run_view_cleanup(&db)).await;
+                let backlog = handle_clone.spawn_blocking(move || run_view_cleanup(&db)).await;
+
+                interval = if backlog {
+                    shorten_cleanup_interval(interval)
+                } else {
+                    VIEW_CLEANUP_INTERVAL
+                };
 
-                handle_clone.sleep(VIEWS_EXPIRATION).await;
+                handle_clone.sleep(interval).await;
             }
         })
         .abort_handle()
@@ -2731,6 +2756,7 @@ mod tests {
         tx.clear_expired_views(
             Instant::now().saturating_duration_since(before_sender2),
             VIEW_CLEANUP_BUDGET,
+            VIEW_CLEANUP_BATCH_SIZE,
         )?;
         stdb.commit_tx(tx)?;
 
@@ -2788,7 +2814,7 @@ mod tests {
         // 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)?;
+        let _result = tx.clear_expired_views(Duration::from_secs(1), VIEW_CLEANUP_BUDGET, VIEW_CLEANUP_BATCH_SIZE)?;
         stdb.commit_tx(tx)?;
 
         assert_eq!(
@@ -2832,7 +2858,7 @@ mod tests {
         // 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)?;
+        let _result = tx.clear_expired_views(Duration::from_secs(1), VIEW_CLEANUP_BUDGET, VIEW_CLEANUP_BATCH_SIZE)?;
         stdb.commit_tx(tx)?;
 
         assert!(