From 8029f8485f042670efd9cd219e95e38ad8beeeb1 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 17:01:22 +0530 Subject: [PATCH 1/8] exponential backoff for view cleanup sleep interval --- crates/engine/src/relational_db.rs | 31 +++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 5ba71da15cb..503be23cd36 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -61,6 +61,7 @@ use spacetimedb_table::indexes::RowPointer; use spacetimedb_table::page_pool::PagePool; use spacetimedb_table::table::{RowRef, TableScanIter}; use spacetimedb_table::table_index::IndexKey; +use sqlparser::keywords::MIN; use std::borrow::Cow; use std::io; use std::ops::RangeBounds; @@ -1126,17 +1127,27 @@ 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 +/// 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 run_view_cleanup(db: &RelationalDB) -> bool { match db.with_auto_commit(Workload::Internal, |tx| { tx.clear_expired_views(VIEWS_EXPIRATION, VIEW_CLEANUP_BUDGET) .map_err(DBError::from) @@ -1151,6 +1162,7 @@ pub fn spawn_view_cleanup_loop( total_expired - cleared ); } + cleared < total_expired } Err(e) => { log::error!( @@ -1158,6 +1170,7 @@ pub fn spawn_view_cleanup_loop( db.database_identity(), e ); + false } } } @@ -1165,13 +1178,21 @@ 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; + + // Calculate next cleanup interval based on backlog + interval = if backlog { + (interval / 2).max(MIN_VIEW_CLEANUP_INTERVAL) + } else { + VIEW_CLEANUP_INTERVAL + }; - handle_clone.sleep(VIEWS_EXPIRATION).await; + handle_clone.sleep(interval).await; } }) .abort_handle() From d7c952d8263e38f11f2254ffdc3e55ae57c900c4 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 17:01:38 +0530 Subject: [PATCH 2/8] straming and batching --- .../src/locking_tx_datastore/mut_tx.rs | 136 +++++++++++------- crates/engine/src/relational_db.rs | 29 ++-- 2 files changed, 99 insertions(+), 66 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 3debc259afe..f2abbfbc3b2 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,33 @@ 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); - } + 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() + .filter(move |(call, _)| call.view_id == view_id) } /// Is this view argument currently materialized? @@ -2812,8 +2822,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 +2830,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 +2908,65 @@ 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. + /// Each cleanup batch retains at most `batch_size` expired view calls while + /// deleting materialized rows. A `batch_size` of zero is treated as one. + /// + /// `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; + + // Process expired calls in bounded batches so cleanup does not retain + // every expired call while deleting materialized rows. + loop { + let mut batch = self + .effective_view_instances() + .filter_map(|(call, state)| is_expired(state).then_some(call.clone())) + .take(batch_size + 1) + .collect::>(); + let backlog = batch.len() > batch_size; + batch.truncate(batch_size); + + if batch.is_empty() { + return Ok(ViewCleanupResult { + cleaned, + backlog: false, + }); + } - 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::>(); + 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::>(); - let total_expired = expired_items.len(); + for row_ptr in rows_to_delete { + self.delete(table_id, row_ptr)?; + } - for call in expired_items { - if start.elapsed() >= max_duration { - break; + self.drop_view_call_from_committed_read_set(call.clone()); + self.view_instances.remove(call); + cleaned += 1; } - 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)?; + if start.elapsed() >= max_duration || !backlog { + return Ok(ViewCleanupResult { cleaned, backlog }); } - - self.drop_view_call_from_committed_read_set(call.clone()); - self.view_instances.remove(call); - cleaned_count += 1; } - - 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 503be23cd36..a94b675bf96 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -61,7 +61,6 @@ use spacetimedb_table::indexes::RowPointer; use spacetimedb_table::page_pool::PagePool; use spacetimedb_table::table::{RowRef, TableScanIter}; use spacetimedb_table::table_index::IndexKey; -use sqlparser::keywords::MIN; use std::borrow::Cow; use std::io; use std::ops::RangeBounds; @@ -1138,6 +1137,9 @@ const MIN_VIEW_CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from //TODO: Make this value configurable const VIEW_CLEANUP_BUDGET: std::time::Duration = std::time::Duration::from_millis(10); +/// 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 @@ -1147,22 +1149,25 @@ pub fn spawn_view_cleanup_loop( db: Arc, handle: &spacetimedb_runtime::Handle, ) -> spacetimedb_runtime::AbortHandle { + 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, ); } - cleared < total_expired + result.backlog } Err(e) => { log::error!( @@ -1185,9 +1190,8 @@ pub fn spawn_view_cleanup_loop( let db = db.clone(); let backlog = handle_clone.spawn_blocking(move || run_view_cleanup(&db)).await; - // Calculate next cleanup interval based on backlog interval = if backlog { - (interval / 2).max(MIN_VIEW_CLEANUP_INTERVAL) + shorten_cleanup_interval(interval) } else { VIEW_CLEANUP_INTERVAL }; @@ -2752,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)?; @@ -2809,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!( @@ -2853,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!( From cc8b5e50dc4c2e8a2686c0d2e1e7df8c3487093c Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 17:15:42 +0530 Subject: [PATCH 3/8] comments --- crates/datastore/src/locking_tx_datastore/mut_tx.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index f2abbfbc3b2..b9b1b94d5b6 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2790,6 +2790,9 @@ impl MutTxId { None => Some((call, state)), }); + // 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 @@ -2908,8 +2911,8 @@ impl MutTxId { /// Clean up materialized view arguments that have no subscribers and haven’t been used recently. /// - /// Each cleanup batch retains at most `batch_size` expired view calls while - /// deleting materialized rows. A `batch_size` of zero is treated as one. + /// 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 @@ -2928,8 +2931,6 @@ impl MutTxId { let is_expired = |state: &ViewInstanceState| !state.has_subscribers() && state.last_used < expiration_threshold; let mut cleaned = 0; - // Process expired calls in bounded batches so cleanup does not retain - // every expired call while deleting materialized rows. loop { let mut batch = self .effective_view_instances() From c05fcf5deab7ab6c7c99adef68d366760d5a53a6 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 17:39:54 +0530 Subject: [PATCH 4/8] remove redundant code --- crates/datastore/src/locking_tx_datastore/mut_tx.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index b9b1b94d5b6..9c7c4d7b488 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2930,6 +2930,7 @@ impl MutTxId { let expiration_threshold = Timestamp::now() - expiration_duration; 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 batch = self @@ -2940,13 +2941,6 @@ impl MutTxId { let backlog = batch.len() > batch_size; batch.truncate(batch_size); - if batch.is_empty() { - return Ok(ViewCleanupResult { - cleaned, - backlog: false, - }); - } - 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"); From 11e7ae1133dae4869f0ed5b0118f099e5794e596 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 21:25:25 +0530 Subject: [PATCH 5/8] Update crates/datastore/src/locking_tx_datastore/mut_tx.rs Co-authored-by: Phoebe Goldman Signed-off-by: Shubham Mishra --- .../src/locking_tx_datastore/mut_tx.rs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 9c7c4d7b488..427dd592366 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2933,11 +2933,26 @@ impl MutTxId { 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 + } + }; let mut batch = self .effective_view_instances() - .filter_map(|(call, state)| is_expired(state).then_some(call.clone())) + .filter_map(filter_and_count_view_instance) .take(batch_size + 1) .collect::>(); + log::info!( + "[{}]: Traversed {} unexpired views to collect and delete a batch of {} expired views", + self.ctx.database_identity, + unexpired_visited_building_batch, + batch.len() + ); let backlog = batch.len() > batch_size; batch.truncate(batch_size); From 05615f10b6d688a73006281c1d8f03afbcec2aa6 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 21:44:27 +0530 Subject: [PATCH 6/8] Update crates/datastore/src/locking_tx_datastore/mut_tx.rs Co-authored-by: Phoebe Goldman Signed-off-by: Shubham Mishra --- crates/datastore/src/locking_tx_datastore/mut_tx.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 427dd592366..85175c5d761 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2948,11 +2948,13 @@ impl MutTxId { .take(batch_size + 1) .collect::>(); log::info!( - "[{}]: Traversed {} unexpired views to collect and delete a batch of {} expired views", - self.ctx.database_identity, - unexpired_visited_building_batch, - batch.len() - ); + // 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 mut batch = self + .effective_view_instances() + .filter_map(|(call, state)| is_expired(state).then_some(call.clone())) + .take(batch_size + 1) +.collect::>(); let backlog = batch.len() > batch_size; batch.truncate(batch_size); From 12ccfc2ab21f15a8d3b8d9fac51162d1f3ff93ce Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 21:45:10 +0530 Subject: [PATCH 7/8] comment Co-authored-by: Phoebe Goldman Signed-off-by: Shubham Mishra --- crates/datastore/src/locking_tx_datastore/mut_tx.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 85175c5d761..9b17c4d324a 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2809,6 +2809,8 @@ impl MutTxId { view_id: ViewId, ) -> impl Iterator + '_ { self.effective_view_instances() + // 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) } From 9be7ddbb0a1b4a0f1a3a967ecf79e3780dcb3c78 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 9 Jul 2026 22:00:28 +0530 Subject: [PATCH 8/8] syntax fixes --- .../src/locking_tx_datastore/mut_tx.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index 9b17c4d324a..d230c40c7fe 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -2944,19 +2944,22 @@ impl MutTxId { 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 mut batch = self .effective_view_instances() .filter_map(filter_and_count_view_instance) .take(batch_size + 1) .collect::>(); + log::info!( - // 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 mut batch = self - .effective_view_instances() - .filter_map(|(call, state)| is_expired(state).then_some(call.clone())) - .take(batch_size + 1) -.collect::>(); + "[{}]: Traversed {} unexpired views to collect and delete a batch of {} expired views", + self.ctx.database_identity, + unexpired_visited_building_batch, + batch.len() + ); + let backlog = batch.len() > batch_size; batch.truncate(batch_size);