Views cleanup for hotfix

crates/core/src/db/relational_db.rs

@@ -1084,53 +1084,78 @@ 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 rows 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>) -> tokio::task::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
             }
         }
     }
 
     tokio::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();
             let db_identity = db.database_identity();
-            tokio::task::spawn_blocking(move || run_view_cleanup(&db))
+            let backlog = tokio::task::spawn_blocking(move || run_view_cleanup(&db))
                 .await
                 .inspect_err(|e| {
                     log::error!("[{}] DATABASE: failed to run view cleanup task: {}", db_identity, e);
                 })
-                .ok();
+                .unwrap_or(false);
+
+            interval = if backlog {
+                shorten_cleanup_interval(interval)
+            } else {
+                VIEW_CLEANUP_INTERVAL
+            };
 
-            tokio::time::sleep(VIEWS_EXPIRATION).await;
+            tokio::time::sleep(interval).await;
         }
     })
     .abort_handle()
@@ -2698,6 +2723,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)?;
 
@@ -2721,6 +2747,49 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_view_cleanup_batches_expired_rows() -> ResultTest<()> {
+        let stdb = TestDB::durable()?;
+        let (view_id, table_id, _, _) = setup_view(&stdb)?;
+
+        let identity_from_u8 = |byte| Identity::from_byte_array([byte; 32]);
+        let senders = [identity_from_u8(1), identity_from_u8(2), identity_from_u8(3)];
+
+        for (idx, sender) in senders.iter().copied().enumerate() {
+            insert_view_row(&stdb, view_id, table_id, sender, idx as u8)?;
+
+            let mut tx = begin_mut_tx(&stdb);
+            tx.unsubscribe_view(view_id, ArgId::SENTINEL, sender)?;
+            stdb.commit_tx(tx)?;
+            update_last_called(&stdb, view_id, sender, Timestamp::UNIX_EPOCH)?;
+        }
+
+        let mut tx = begin_mut_tx(&stdb);
+        let result = tx.clear_expired_views(Duration::from_secs(1), Duration::ZERO, 1)?;
+        assert_eq!(result.cleaned, 1);
+        assert!(result.backlog);
+        stdb.commit_tx(tx)?;
+
+        let tx = begin_mut_tx(&stdb);
+        assert_eq!(tx.lookup_st_view_subs(view_id)?.len(), 2);
+        stdb.commit_tx(tx)?;
+
+        let mut tx = begin_mut_tx(&stdb);
+        let result = tx.clear_expired_views(Duration::from_secs(1), VIEW_CLEANUP_BUDGET, VIEW_CLEANUP_BATCH_SIZE)?;
+        assert_eq!(result.cleaned, 2);
+        assert!(!result.backlog);
+        stdb.commit_tx(tx)?;
+
+        let tx = begin_mut_tx(&stdb);
+        assert!(tx.lookup_st_view_subs(view_id)?.is_empty());
+        stdb.commit_tx(tx)?;
+        for sender in senders {
+            assert!(project_views(&stdb, table_id, sender).is_empty());
+        }
+
+        Ok(())
+    }
+
     /// Regression test for anonymous-view cleanup.
     ///
     /// If one subscriber row has expired but another subscriber for the same anonymous
@@ -2754,7 +2823,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!(
@@ -2799,7 +2868,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!(

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -82,6 +82,11 @@ pub struct ViewCallInfo {
     pub sender: Option<Identity>,
 }
 
+pub struct ViewCleanupResult {
+    pub cleaned: usize,
+    pub backlog: bool,
+}
+
 /// A data structure for tracking the database rows/keys that are read by views
 #[derive(Default)]
 pub struct ViewReadSets {
@@ -2492,80 +2497,94 @@ impl MutTxId {
     /// 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.
+    /// Expired views are cleared in batches. Each loop iteration selects up to
+    /// `batch_size` expired rows 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 a tuple `(cleaned, total_expired)`:
-    /// - `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).
+    /// Returns the number of rows 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 now = Timestamp::now();
         let expiration_threshold = now - expiration_duration;
-        let mut cleaned_count = 0;
+        let mut cleaned = 0;
+        let batch_size = batch_size.max(1);
 
-        // Collect all expired views from st_view_sub
-        let expired_items: Vec<(ViewId, Identity, RowPointer)> = self
-            .iter_by_col_eq(
+        loop {
+            let mut unexpired_visited_building_batch = 0;
+            let mut batch: Vec<(ViewId, Identity, RowPointer)> = Vec::with_capacity(batch_size + 1);
+
+            // Collect one batch of expired views from st_view_sub.
+            for row_ref in self.iter_by_col_eq(
                 ST_VIEW_SUB_ID,
                 StViewSubFields::HasSubscribers,
                 &AlgebraicValue::from(false),
-            )?
-            .filter_map(|row_ref| {
-                let row = StViewSubRow::try_from(row_ref).expect("Failed to deserialize st_view_sub row");
+            )? {
+                let row = StViewSubRow::try_from(row_ref)?;
 
                 if !row.has_subscribers && row.num_subscribers == 0 && row.last_called.0 < expiration_threshold {
-                    Some((row.view_id, row.identity.into(), row_ref.pointer()))
+                    batch.push((row.view_id, row.identity.into(), row_ref.pointer()));
+                    if batch.len() > batch_size {
+                        break;
+                    }
                 } else {
-                    None
+                    unexpired_visited_building_batch += 1;
                 }
-            })
-            .collect();
-
-        let total_expired = expired_items.len();
-
-        // 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 {
-                break;
             }
 
-            let StViewRow {
-                table_id, is_anonymous, ..
-            } = self.lookup_st_view(view_id)?;
-            let table_id = table_id.expect("views have backing table");
+            log::info!(
+                "[{}]: Traversed {} unexpired views to collect and delete a batch of {} expired views",
+                self.ctx.database_identity,
+                unexpired_visited_building_batch,
+                batch.len()
+            );
 
-            if is_anonymous {
-                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())?
-                    .map(|res| res.pointer())
-                    .collect::<Vec<_>>();
+            let backlog = batch.len() > batch_size;
+            batch.truncate(batch_size);
+
+            // 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 batch {
+                let StViewRow {
+                    table_id, is_anonymous, ..
+                } = self.lookup_st_view(view_id)?;
+                let table_id = table_id.expect("views have backing table");
+
+                if is_anonymous {
+                    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())?
+                        .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_with_sender_from_committed_read_set(view_id, sender);
                 }
 
-                self.drop_view_with_sender_from_committed_read_set(view_id, sender);
+                // Finally, delete the subscription row.
+                self.delete(ST_VIEW_SUB_ID, sub_row_ptr)?;
+                cleaned += 1;
             }
 
-            // Finally, delete the subscription row
-            self.delete(ST_VIEW_SUB_ID, sub_row_ptr)?;
-            cleaned_count += 1;
+            if start.elapsed() >= max_duration || !backlog {
+                return Ok(ViewCleanupResult { cleaned, backlog });
+            }
         }
-
-        Ok((cleaned_count, total_expired))
     }
 
     /// Decrement `num_subscribers` in `st_view_sub` to effectively unsubscribe a caller from a view.