diff --git a/Cargo.lock b/Cargo.lock index 112ef9d1b74..c7b53a3d4fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8587,6 +8587,7 @@ dependencies = [ "regex", "reqwest 0.12.24", "serde_json", + "socket2 0.5.10", "spacetimedb-core", "spacetimedb-guard", "tempfile", diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs index 893c0ec9c15..f3894d1c10b 100644 --- a/crates/client-api/src/routes/database.rs +++ b/crates/client-api/src/routes/database.rs @@ -1,4 +1,5 @@ use std::borrow::Cow; +use std::future::Future; use std::num::NonZeroU8; use std::str::FromStr; use std::time::Duration; @@ -43,7 +44,7 @@ use spacetimedb_client_api_messages::name::{ }; use spacetimedb_lib::db::raw_def::v10::RawModuleDefV10; use spacetimedb_lib::db::raw_def::v9::RawModuleDefV9; -use spacetimedb_lib::http as st_http; +use spacetimedb_lib::{http as st_http, ConnectionId}; use spacetimedb_lib::{sats, AlgebraicValue, Hash, ProductValue, Timestamp}; use spacetimedb_schema::auto_migrate::{ MigrationPolicy as SchemaMigrationPolicy, MigrationToken, PrettyPrintStyle as AutoMigratePrettyPrintStyle, @@ -152,78 +153,67 @@ pub async fn call( let args = FunctionArgs::Json(body); - // HTTP callers always need a connection ID to provide to connect/disconnect, - // so generate one. - let connection_id = generate_random_connection_id(); + let caller_auth: ConnectionAuthCtx = auth.into(); let (module, Database { owner_identity, .. }) = find_module_and_database(&worker_ctx, name_or_identity).await?; - // Call the database's `client_connected` reducer, if any. - // If it fails or rejects the connection, bail. - module - .call_identity_connected(auth.into(), connection_id) - .await - .map_err(client_connected_error_to_response)?; - - let result = match module - .call_reducer( - caller_identity, - Some(connection_id), - None, - None, - None, - &reducer, - args.clone(), - ) - .await - { - Ok(rcr) => Ok(CallResult::Reducer(rcr)), - Err(ReducerCallError::NoSuchReducer | ReducerCallError::ScheduleReducerNotFound) => { - // Not a reducer — try procedure instead - match module - .call_procedure(caller_identity, Some(connection_id), None, &reducer, args) - .await - .result - { - Ok(res) => Ok(CallResult::Procedure(res)), - Err(e) => Err(map_procedure_error(e, &reducer)), + let fut = async move |module: ModuleHost, caller_identity: Identity, connection_id: ConnectionId| { + let result = match module + .call_reducer( + caller_identity, + Some(connection_id), + None, + None, + None, + &reducer, + args.clone(), + ) + .await + { + Ok(rcr) => Ok(CallResult::Reducer(rcr)), + Err(ReducerCallError::NoSuchReducer | ReducerCallError::ScheduleReducerNotFound) => { + // Not a reducer — try procedure instead + match module + .call_procedure(caller_identity, Some(connection_id), None, &reducer, args) + .await + .result + { + Ok(res) => Ok(CallResult::Procedure(res)), + Err(e) => Err(map_procedure_error(e, &reducer)), + } } + Err(e) => Err(map_reducer_error(e, &reducer)), + }; + + match result { + Ok(CallResult::Reducer(result)) => { + let (status, body) = reducer_outcome_response(&owner_identity, &reducer, result.outcome); + Ok(( + status, + TypedHeader(SpacetimeEnergyUsed(result.execution_budget_used)), + TypedHeader(SpacetimeExecutionDurationMicros(result.execution_duration)), + body, + ) + .into_response()) + } + Ok(CallResult::Procedure(result)) => { + // Procedures don't assign a special meaning to error returns, unlike reducers, + // as there's no transaction for them to automatically abort. + // Instead, we just pass on their return value with the OK status so long as we successfully invoked the procedure. + let (status, body) = procedure_outcome_response(result.return_val); + Ok(( + status, + TypedHeader(SpacetimeExecutionDurationMicros(result.execution_duration)), + body, + ) + .into_response()) + } + Err(e) => Err((e.0, e.1).into()), } - Err(e) => Err(map_reducer_error(e, &reducer)), }; - module - .call_identity_disconnected(caller_identity, connection_id) - .await - .map_err(client_disconnected_error_to_response)?; - - match result { - Ok(CallResult::Reducer(result)) => { - let (status, body) = reducer_outcome_response(&owner_identity, &reducer, result.outcome); - Ok(( - status, - TypedHeader(SpacetimeEnergyUsed(result.execution_budget_used)), - TypedHeader(SpacetimeExecutionDurationMicros(result.execution_duration)), - body, - ) - .into_response()) - } - Ok(CallResult::Procedure(result)) => { - // Procedures don't assign a special meaning to error returns, unlike reducers, - // as there's no transaction for them to automatically abort. - // Instead, we just pass on their return value with the OK status so long as we successfully invoked the procedure. - let (status, body) = procedure_outcome_response(result.return_val); - Ok(( - status, - TypedHeader(SpacetimeExecutionDurationMicros(result.execution_duration)), - body, - ) - .into_response()) - } - Err(e) => Err((e.0, e.1).into()), - } + with_connection(module, caller_auth, caller_identity, fut).await } - #[derive(Deserialize)] pub struct HttpRouteRootParams { name_or_identity: NameOrIdentity, @@ -709,6 +699,46 @@ pub struct SqlQueryParams { pub confirmed: Option, } +/// Runs `fut` inside an HTTP client connection lifecycle. +/// +/// The lifecycle is detached from the request task, so once `client_connected` +/// succeeds, `client_disconnected` still runs if the handler is dropped or +/// `fut` returns an error. +async fn with_connection( + module: ModuleHost, + caller_auth: ConnectionAuthCtx, + caller_identity: Identity, + fut: F, +) -> axum::response::Result +where + F: FnOnce(ModuleHost, Identity, ConnectionId) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, + R: Send + 'static, +{ + tokio::spawn(async move { + let connection_id = generate_random_connection_id(); + + // Run the module's client_connected reducer, if any. + // If it rejects the connection, bail before executing `fut` + module + .call_identity_connected(caller_auth, connection_id) + .await + .map_err(client_connected_error_to_response)?; + + let result = fut(module.clone(), caller_identity, connection_id).await; + + // Always disconnect, even if authorization or execution failed. + module + .call_identity_disconnected(caller_identity, connection_id) + .await + .map_err(client_disconnected_error_to_response)?; + + result + }) + .await + .map_err(log_and_500)? +} + pub async fn sql_direct( worker_ctx: S, SqlParams { name_or_identity }: SqlParams, @@ -718,21 +748,12 @@ pub async fn sql_direct( sql: String, ) -> axum::response::Result>> where - S: NodeDelegate + ControlStateDelegate + Authorization, + S: NodeDelegate + ControlStateDelegate + Authorization + 'static, { - let connection_id = generate_random_connection_id(); - let (host, database) = find_leader_and_database(&worker_ctx, name_or_identity).await?; - // Run the module's client_connected reducer, if any. - // If it rejects the connection, bail before executing SQL. let module = host.module().await.map_err(log_and_500)?; - module - .call_identity_connected(caller_auth, connection_id) - .await - .map_err(client_connected_error_to_response)?; - - let result = async { + let fut = async move |_module: ModuleHost, caller_identity: Identity, _connection_id: ConnectionId| { let sql_auth = worker_ctx .authorize_sql(caller_identity, database.database_identity) .await?; @@ -744,16 +765,9 @@ where sql, ) .await - } - .await; - - // Always disconnect, even if authorization or execution failed. - module - .call_identity_disconnected(caller_identity, connection_id) - .await - .map_err(client_disconnected_error_to_response)?; + }; - result + with_connection(module, caller_auth, caller_identity, fut).await } pub async fn sql( @@ -764,7 +778,7 @@ pub async fn sql( body: String, ) -> axum::response::Result where - S: NodeDelegate + ControlStateDelegate + Authorization, + S: NodeDelegate + ControlStateDelegate + Authorization + 'static, { let caller_identity = auth.claims.identity; let caller_auth: ConnectionAuthCtx = auth.into(); diff --git a/crates/commitlog/src/payload/txdata.rs b/crates/commitlog/src/payload/txdata.rs index 36a09a8848b..ab9dc570d27 100644 --- a/crates/commitlog/src/payload/txdata.rs +++ b/crates/commitlog/src/payload/txdata.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{io, sync::Arc}; use bitflags::bitflags; use spacetimedb_sats::buffer::{BufReader, BufWriter, DecodeError}; @@ -528,6 +528,8 @@ pub enum DecoderError { Visitor(V), #[error(transparent)] Traverse(#[from] error::Traversal), + #[error(transparent)] + Io(#[from] io::Error), } /// A free standing implementation of [`crate::Decoder::skip_record`] diff --git a/crates/core/src/db/relational_db.rs b/crates/core/src/db/relational_db.rs index d149c437b59..f760d131a07 100644 --- a/crates/core/src/db/relational_db.rs +++ b/crates/core/src/db/relational_db.rs @@ -7,7 +7,7 @@ use crate::worker_metrics::WORKER_METRICS; use anyhow::{anyhow, Context}; use enum_map::EnumMap; use spacetimedb_commitlog::repo::OnNewSegmentFn; -use spacetimedb_commitlog::{self as commitlog, Commitlog, SizeOnDisk}; +use spacetimedb_commitlog::{self as commitlog, SizeOnDisk}; use spacetimedb_data_structures::map::HashSet; use spacetimedb_datastore::db_metrics::DB_METRICS; use spacetimedb_datastore::error::{DatastoreError, TableError, ViewError}; @@ -34,6 +34,7 @@ use spacetimedb_datastore::{ }, traits::TxData, }; +use spacetimedb_durability::local::LocalHistory; use spacetimedb_durability::{self as durability, History}; use spacetimedb_lib::bsatn::ToBsatn; use spacetimedb_lib::db::auth::StAccess; @@ -1084,28 +1085,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 +1131,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() @@ -1751,7 +1777,7 @@ pub async fn local_durability_with_options( /// Currently, this is simply a read-only copy of the commitlog. pub async fn local_history(replica_dir: &ReplicaDir) -> io::Result + use<>> { let commitlog_dir = replica_dir.commit_log(); - asyncify(move || Commitlog::open(commitlog_dir, <_>::default(), None)).await + asyncify(move || LocalHistory::open(commitlog_dir)).await } /// Watches snapshot creation events and compresses all commitlog segments older @@ -2698,6 +2724,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 +2748,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 +2824,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 +2869,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. diff --git a/crates/datastore/src/locking_tx_datastore/replay.rs b/crates/datastore/src/locking_tx_datastore/replay.rs index 95778b6fcff..8c87536f3d4 100644 --- a/crates/datastore/src/locking_tx_datastore/replay.rs +++ b/crates/datastore/src/locking_tx_datastore/replay.rs @@ -29,6 +29,7 @@ use spacetimedb_schema::table_name::TableName; use spacetimedb_table::indexes::RowPointer; use spacetimedb_table::table::{InsertError, RowRef}; use std::cell::RefCell; +use std::io; use std::sync::Arc; use thiserror::Error; @@ -104,6 +105,8 @@ pub enum ReplayError { Db(#[from] DatastoreError), #[error(transparent)] Any(#[from] anyhow::Error), + #[error(transparent)] + Io(#[from] io::Error), } /// A [`spacetimedb_commitlog::Decoder`] suitable for replaying a transaction diff --git a/crates/durability/src/imp/local.rs b/crates/durability/src/imp/local.rs index e3eca56e5d9..4411523da6a 100644 --- a/crates/durability/src/imp/local.rs +++ b/crates/durability/src/imp/local.rs @@ -1,5 +1,6 @@ use std::{ io, + marker::PhantomData, num::NonZeroUsize, sync::{ atomic::{AtomicU64, Ordering::Relaxed}, @@ -12,13 +13,13 @@ use itertools::Itertools as _; use log::{info, trace, warn}; use scopeguard::ScopeGuard; use spacetimedb_commitlog::{ - error, + commitlog, error, payload::Txdata, - repo::{Fs, Repo, RepoWithoutLockFile}, - Commit, Commitlog, CompressionStats, Decoder, Encode, Transaction, + repo::{self, Fs, Repo, RepoWithoutLockFile}, + Commit, Commitlog, CompressionStats, Decoder, Encode, Transaction, DEFAULT_LOG_FORMAT_VERSION, }; use spacetimedb_fs_utils::lockfile::advisory::{LockError, LockedFile}; -use spacetimedb_paths::server::ReplicaDir; +use spacetimedb_paths::server::{CommitLogDir, ReplicaDir}; use spacetimedb_runtime::{Handle, JoinHandle}; use thiserror::Error; use tokio::sync::watch; @@ -92,6 +93,10 @@ where { /// The [`Commitlog`] this [`Durability`] and [`History`] impl wraps. clog: Arc, R>>, + /// Copy of the underlying [`Repo`]. + /// + /// Used to obtain a [LocalHistory] via [Self::as_history]. + repo: R, /// The durable transaction offset, as reported by the background /// [`FlushAndSyncTask`]. durable_offset: watch::Receiver>, @@ -133,12 +138,9 @@ impl Local { // yet for backwards-compatibility, we keep using the `db.lock` file. let lock = LockedFile::lock(replica_dir.0.join("db.lock"))?; - let clog = Arc::new(Commitlog::open( - replica_dir.commit_log(), - opts.commitlog, - on_new_segment, - )?); - Self::open_inner(clog, rt, opts, Some(lock)) + let repo = repo::Fs::new(replica_dir.commit_log(), on_new_segment)?; + let clog = Commitlog::open_with_repo(repo.clone(), opts.commitlog).map(Arc::new)?; + Self::open_inner(clog, repo, rt, opts, Some(lock)) } } @@ -150,8 +152,8 @@ where /// Create a [`Local`] instance backed by the provided commitlog repo. pub fn open_with_repo(repo: R, rt: Handle, opts: Options) -> Result { info!("open local durability"); - let clog = Arc::new(Commitlog::open_with_repo(repo, opts.commitlog)?); - Self::open_inner(clog, rt, opts, None) + let clog = Arc::new(Commitlog::open_with_repo(repo.clone(), opts.commitlog)?); + Self::open_inner(clog, repo, rt, opts, None) } } @@ -162,6 +164,7 @@ where { fn open_inner( clog: Arc, R>>, + repo: R, rt: Handle, opts: Options, lock: Option, @@ -184,6 +187,7 @@ where Ok(Self { clog, + repo, durable_offset: durable_rx, queue, queue_depth, @@ -193,7 +197,16 @@ where /// Obtain a read-only copy of the durable state that implements [History]. pub fn as_history(&self) -> impl History> + use { - self.clog.clone() + let tx_range = { + let min = self.clog.min_committed_offset().unwrap_or_default(); + let max = self.clog.max_committed_offset(); + (min, max) + }; + LocalHistory { + repo: self.repo.clone(), + tx_range, + _payload: PhantomData, + } } } @@ -383,39 +396,65 @@ where } } -impl History for Commitlog, R> -where - T: Encode + 'static, - R: Repo + Send + Sync + 'static, -{ +/// [History] impl that operates on a local [Repo] `R`. +pub struct LocalHistory { + repo: R, + tx_range: (TxOffset, Option), + _payload: PhantomData, +} + +impl LocalHistory { + /// Open the commitlog at `dir` as a read-only [History]. + pub fn open(dir: CommitLogDir) -> io::Result { + let repo = repo::Fs::new(dir, None)?; + let meta = commitlog::committed_meta(&repo)?; + let tx_range = match meta { + // Log is empty + None => (0, None), + Some(committed) => { + let max = committed.metadata().tx_range.end.saturating_sub(1); + let min = repo.existing_offsets()?.first().copied().unwrap_or_default(); + + (min, Some(max)) + } + }; + + Ok(Self { + repo, + tx_range, + _payload: PhantomData, + }) + } +} + +impl History for LocalHistory { type TxData = Txdata; fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From, + D::Error: From + From, { - self.fold_transactions_from(offset, decoder) + commitlog::fold_transactions_from(&self.repo, DEFAULT_LOG_FORMAT_VERSION, offset, decoder) } fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a, { - self.transactions_from(offset, decoder) + commitlog::transactions_from(&self.repo, DEFAULT_LOG_FORMAT_VERSION, offset, decoder) + .into_iter() + .flatten() } fn tx_range_hint(&self) -> (TxOffset, Option) { - let min = self.min_committed_offset().unwrap_or_default(); - let max = self.max_committed_offset(); - - (min, max) + self.tx_range } } diff --git a/crates/durability/src/lib.rs b/crates/durability/src/lib.rs index 8bdcbb3b922..3ef4a28de1c 100644 --- a/crates/durability/src/lib.rs +++ b/crates/durability/src/lib.rs @@ -1,4 +1,4 @@ -use std::{iter, marker::PhantomData, sync::Arc}; +use std::{io, iter, marker::PhantomData, sync::Arc}; use futures::future::BoxFuture; use thiserror::Error; @@ -190,17 +190,17 @@ pub trait History { fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From; + D::Error: From + From; /// Obtain an iterator over the history of transactions, starting from `offset`. fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a; /// Get the maximum transaction offset contained in this history. @@ -224,19 +224,19 @@ impl History for Arc { fn fold_transactions_from(&self, offset: TxOffset, decoder: D) -> Result<(), D::Error> where D: Decoder, - D::Error: From, + D::Error: From + From, { (**self).fold_transactions_from(offset, decoder) } fn transactions_from<'a, D>( - &self, + &'a self, offset: TxOffset, decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, - D::Error: From, + D::Error: From + From, Self::TxData: 'a, { (**self).transactions_from(offset, decoder) @@ -273,7 +273,7 @@ impl History for EmptyHistory { &self, _offset: TxOffset, _decoder: &'a D, - ) -> impl Iterator, D::Error>> + ) -> impl Iterator, D::Error>> + 'a where D: Decoder, D::Error: From, diff --git a/crates/pg/src/pg_server.rs b/crates/pg/src/pg_server.rs index ebca2b61b95..15170a35866 100644 --- a/crates/pg/src/pg_server.rs +++ b/crates/pg/src/pg_server.rs @@ -150,7 +150,7 @@ struct PgSpacetimeDB { impl PgSpacetimeDB where - T: ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone, + T: ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone + 'static, { async fn exe_sql(&self, query: String) -> PgWireResult> { let params = self.cached.lock().await.clone().unwrap(); @@ -314,7 +314,7 @@ impl SimpleQueryHandler for PgSpacetimeDB where - T: Sync + Send + ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone, + T: Sync + Send + ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone + 'static, { async fn do_query(&self, _client: &mut C, query: &str) -> PgWireResult> where @@ -347,7 +347,7 @@ impl PgSpacetimeDBFactory { impl PgWireServerHandlers for PgSpacetimeDBFactory where - T: Sync + Send + ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone, + T: Sync + Send + ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone + 'static, { fn simple_query_handler(&self) -> Arc { self.handler.clone() diff --git a/crates/smoketests/Cargo.toml b/crates/smoketests/Cargo.toml index e100b3c5b2b..5de97156840 100644 --- a/crates/smoketests/Cargo.toml +++ b/crates/smoketests/Cargo.toml @@ -19,6 +19,7 @@ spacetimedb-core.workspace = true cargo_metadata.workspace = true assert_cmd = "2" predicates = "3" +socket2.workspace = true tokio.workspace = true tokio-postgres.workspace = true reqwest = { workspace = true, features = ["blocking"] } diff --git a/crates/smoketests/modules/Cargo.lock b/crates/smoketests/modules/Cargo.lock index 5f63514cacc..e937e85a17c 100644 --- a/crates/smoketests/modules/Cargo.lock +++ b/crates/smoketests/modules/Cargo.lock @@ -609,6 +609,14 @@ dependencies = [ "spacetimedb", ] +[[package]] +name = "smoketest-module-client-connection-http-cancel" +version = "0.1.0" +dependencies = [ + "log", + "spacetimedb", +] + [[package]] name = "smoketest-module-client-connection-reject" version = "0.1.0" diff --git a/crates/smoketests/modules/Cargo.toml b/crates/smoketests/modules/Cargo.toml index c23de0030f6..c5a3faffebb 100644 --- a/crates/smoketests/modules/Cargo.toml +++ b/crates/smoketests/modules/Cargo.toml @@ -82,6 +82,7 @@ members = [ "delete-database", "client-connection-reject", "client-connection-disconnect-panic", + "client-connection-http-cancel", "sql-connect-hook", # Log filtering tests diff --git a/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml b/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml new file mode 100644 index 00000000000..46ecd40b509 --- /dev/null +++ b/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "smoketest-module-client-connection-http-cancel" +version = "0.1.0" +edition = "2021" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +spacetimedb.workspace = true +log.workspace = true diff --git a/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs new file mode 100644 index 00000000000..1830b7f862d --- /dev/null +++ b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs @@ -0,0 +1,25 @@ +use spacetimedb::{log, ReducerContext}; + +#[spacetimedb::table(accessor = marker, public)] +pub struct Marker { + id: u32, +} + +#[spacetimedb::reducer(client_connected)] +pub fn connected(ctx: &ReducerContext) { + log::info!("http_cancel_repro: connected {}", ctx.sender()); +} + +#[spacetimedb::reducer(client_disconnected)] +pub fn disconnected(ctx: &ReducerContext) { + log::info!("http_cancel_repro: disconnected {}", ctx.sender()); +} + +#[spacetimedb::reducer] +pub fn slow(_ctx: &ReducerContext) { + log::info!("http_cancel_repro: slow reducer started"); + for i in 0..300_000_000u64 { + core::hint::black_box(i); + } + log::info!("http_cancel_repro: slow reducer finished"); +} diff --git a/crates/smoketests/src/lib.rs b/crates/smoketests/src/lib.rs index 0ebfb70e935..3374920d905 100644 --- a/crates/smoketests/src/lib.rs +++ b/crates/smoketests/src/lib.rs @@ -823,6 +823,10 @@ impl Smoketest { pub fn spacetime_cmd(&self, args: &[&str]) -> Output { let start = Instant::now(); let cli_path = ensure_binaries_built(); + + let cmd_name = args.first().unwrap_or(&"unknown"); + eprintln!("[TIMING] spacetime {cmd_name}: starting at {start:?}"); + let output = Command::new(&cli_path) .arg("--config-path") .arg(&self.config_path) @@ -831,8 +835,8 @@ impl Smoketest { .output() .expect("Failed to execute spacetime command"); - let cmd_name = args.first().unwrap_or(&"unknown"); - eprintln!("[TIMING] spacetime {}: {:?}", cmd_name, start.elapsed()); + eprintln!("[TIMING] spacetime {}: completed in {:?}", cmd_name, start.elapsed()); + output } diff --git a/crates/smoketests/tests/smoketests/client_connection_errors.rs b/crates/smoketests/tests/smoketests/client_connection_errors.rs index 7033c56024d..7911aec39f7 100644 --- a/crates/smoketests/tests/smoketests/client_connection_errors.rs +++ b/crates/smoketests/tests/smoketests/client_connection_errors.rs @@ -1,4 +1,9 @@ -use spacetimedb_smoketests::Smoketest; +use socket2::SockRef; +use spacetimedb_smoketests::{require_local_server, Smoketest}; +use std::io::Write; +use std::net::TcpStream; +use std::thread; +use std::time::{Duration, Instant}; /// Test that client_connected returning an error rejects the connection #[test] @@ -57,3 +62,121 @@ fn test_client_disconnected_error_still_deletes_st_client() { "Expected at most 1 st_client row (the SQL connection itself), got {row_count}: {sql_out}", ); } + +#[test] +fn test_http_reducer_call_cancel_still_deletes_st_client() { + require_local_server!(); + + let test = Smoketest::builder() + .precompiled_module("client-connection-http-cancel") + .build(); + + let stream = send_http_reducer_call_and_hold_connection(&test, "slow"); + wait_for_log( + &test, + "http_cancel_repro: slow reducer started", + Duration::from_secs(30), + ); + abort_tcp_stream(stream); + + wait_for_log( + &test, + "http_cancel_repro: slow reducer finished", + Duration::from_secs(30), + ); + + let (st_client_count, st_connection_credentials_count) = wait_for_client_rows_to_clear(&test); + assert!( + st_client_count <= 1, + "Expected at most 1 st_client row (the SQL connection itself) after dropping an HTTP call post-connect, got {st_client_count}", + ); + assert!( + st_connection_credentials_count <= 1, + "Expected at most 1 st_connection_credentials row (the SQL connection itself) after dropping an HTTP call post-connect, got {st_connection_credentials_count}", + ); +} + +fn send_http_reducer_call_and_hold_connection(test: &Smoketest, reducer: &str) -> TcpStream { + let host = test.server_host(); + let request = http_reducer_call_request(test, reducer); + + let mut stream = TcpStream::connect(host).expect("Failed to connect to local server"); + stream.set_nodelay(true).expect("Failed to set TCP_NODELAY"); + stream + .write_all(request.as_bytes()) + .expect("Failed to write HTTP request"); + stream.flush().expect("Failed to flush HTTP request"); + stream +} + +fn abort_tcp_stream(stream: TcpStream) { + SockRef::from(&stream) + .set_linger(Some(Duration::ZERO)) + .expect("Failed to set SO_LINGER=0"); + drop(stream); +} + +fn http_reducer_call_request(test: &Smoketest, reducer: &str) -> String { + let identity = test.database_identity.as_ref().expect("No database published"); + let host = test.server_host(); + let token = test.read_token().expect("Failed to read auth token"); + let path = format!("/v1/database/{identity}/call/{reducer}"); + + // Keep the HTTP/1.1 connection alive so dropping the TCP stream is an + // unexpected client disconnect while the reducer call is still active. + format!( + "POST {path} HTTP/1.1\r\n\ + Host: {host}\r\n\ + Authorization: Bearer {token}\r\n\ + Content-Type: application/json\r\n\ + Content-Length: 2\r\n\ + Connection: keep-alive\r\n\ + \r\n\ + []" + ) +} + +fn wait_for_log(test: &Smoketest, needle: &str, timeout: Duration) -> Vec { + let start = Instant::now(); + loop { + let logs = test.logs(200).unwrap(); + if logs.iter().any(|l| l.contains(needle)) { + return logs; + } + assert!( + start.elapsed() < timeout, + "Timed out waiting for log containing {needle:?}: {:?}", + logs + ); + thread::sleep(Duration::from_millis(1)); + } +} + +fn wait_for_client_rows_to_clear(test: &Smoketest) -> (usize, usize) { + let start = Instant::now(); + let mut last = (usize::MAX, usize::MAX); + + while start.elapsed() < Duration::from_secs(5) { + last = ( + sql_count(test, "st_client"), + sql_count(test, "st_connection_credentials"), + ); + if last.0 <= 1 && last.1 <= 1 { + return last; + } + thread::sleep(Duration::from_millis(50)); + } + + last +} + +fn sql_count(test: &Smoketest, table: &str) -> usize { + let output = test + .sql(&format!("SELECT COUNT(*) AS count FROM {table}")) + .unwrap_or_else(|err| panic!("Failed to count rows in {table}: {err:#}")); + + output + .lines() + .find_map(|line| line.trim().parse::().ok()) + .unwrap_or_else(|| panic!("Failed to parse COUNT(*) output for {table}: {output}")) +}