diff --git a/crates/core/src/db/relational_db.rs b/crates/core/src/db/relational_db.rs index d149c437b59..cc47b0fba34 100644 --- a/crates/core/src/db/relational_db.rs +++ b/crates/core/src/db/relational_db.rs @@ -1084,28 +1084,45 @@ 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) -> 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!( @@ -1113,24 +1130,32 @@ pub fn spawn_view_cleanup_loop(db: Arc) -> tokio::task::AbortHandl 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!( diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 3383e9a7fb3..0e12eca392f 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -82,6 +82,11 @@ pub struct ViewCallInfo { pub sender: Option, } +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 { 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::>(); + 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::>(); + + 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.