diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 3debc259afe..d230c40c7fe 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/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::>(); + 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 { - let mut instances = self + fn effective_view_instances(&self) -> impl Iterator + '_ { + let committed = self .committed_state_write_lock .view_instances() - .map(|(call, state)| (call.clone(), state.clone())) - .collect::>(); + .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 { + fn effective_view_instances_for_view( + &self, + view_id: ViewId, + ) -> impl Iterator + '_ { 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,8 +2827,7 @@ impl MutTxId { /// Returns all materialized view instances for `view_id`. pub fn materialized_view_instances_for_view(&self, view_id: ViewId) -> Vec { self.effective_view_instances_for_view(view_id) - .into_values() - .map(|state| state.args) + .map(|(_, state)| state.args) .collect() } @@ -2821,8 +2835,12 @@ impl MutTxId { #[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 { 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::>(); + let mut batch = self + .effective_view_instances() + .filter_map(filter_and_count_view_instance) + .take(batch_size + 1) + .collect::>(); - 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::>(); + 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::>(); + + 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. diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 5ba71da15cb..a94b675bf96 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -1126,31 +1126,48 @@ 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, 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!( @@ -1158,6 +1175,7 @@ pub fn spawn_view_cleanup_loop( db.database_identity(), e ); + false } } } @@ -1165,13 +1183,20 @@ pub fn spawn_view_cleanup_loop( 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!(