diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs index 981bef18945..fe677648829 100644 --- a/crates/client-api/src/routes/database.rs +++ b/crates/client-api/src/routes/database.rs @@ -35,10 +35,11 @@ use spacetimedb::auth::identity::ConnectionAuthCtx; use spacetimedb::database_logger::DatabaseLogger; use spacetimedb::host::module_host::{ ClientConnectedError, DurabilityExited, DurableOffset, TransactionOffset, UpdateEnvironmentResult, + TRANSACTION_CONFIRMATION_TIMEOUT, }; use spacetimedb::host::{CallResult, UpdateDatabaseResult}; use spacetimedb::host::{FunctionArgs, MigratePlanResult}; -use spacetimedb::host::{ModuleHost, ReducerOutcome}; +use spacetimedb::host::{ModuleHost, ReducerCallResult, ReducerCallResultWithTxOffset, ReducerOutcome}; use spacetimedb::host::{ProcedureCallError, ReducerCallError}; use spacetimedb::identity::Identity; use spacetimedb::messages::control_db::{Database, HostType}; @@ -178,7 +179,7 @@ pub async fn call( let fut = async move |module: ModuleHost, caller_identity: Identity, connection_id: ConnectionId| { let result = match module - .call_reducer( + .call_reducer_with_offset( caller_identity, Some(connection_id), None, @@ -189,7 +190,9 @@ pub async fn call( ) .await { - Ok(rcr) => Ok(CallResult::Reducer(rcr)), + Ok(res) => Ok(CallResult::Reducer( + confirm_reducer_call(res, confirmed, module.durable_tx_offset()).await?, + )), Err(ReducerCallError::NoSuchReducer | ReducerCallError::ScheduleReducerNotFound) => { // Not a reducer — try procedure instead let ret = module @@ -238,17 +241,39 @@ pub async fn call( with_connection(module, caller_auth, caller_identity, fut).await } +/// If `confirmed`, waits for a committed reducer's transaction to be durable. +async fn confirm_reducer_call( + res: ReducerCallResultWithTxOffset, + confirmed: bool, + durable: Option, +) -> axum::response::Result { + if confirmed && matches!(res.result.outcome, ReducerOutcome::Committed) { + confirm_call_offset(Some(res.tx_offset), durable).await?; + } + Ok(res.result) +} + async fn confirm_call_offset( offset: Option, durable: Option, ) -> axum::response::Result<()> { if let (Some(offset), Some(mut durable)) = (offset, durable) { - let offset = offset - .await - .map_err(|_| log_and_500("transaction offset unavailable; outcome unknown"))?; - durable.wait_for(offset).await.map_err(|_| log_and_500( - "transaction completed, but durability could not be confirmed; outcome unknown, do not retry without deduplication", - ))?; + timeout(TRANSACTION_CONFIRMATION_TIMEOUT, async { + let offset = offset + .await + .map_err(|_| log_and_500("transaction offset unavailable; outcome unknown"))?; + durable.wait_for(offset).await.map_err(|_| log_and_500( + "transaction completed, but durability could not be confirmed; outcome unknown, do not retry without deduplication", + ))?; + Ok::<_, ErrorResponse>(()) + }) + .await + .map_err(|_| { + ErrorResponse::from(( + StatusCode::GATEWAY_TIMEOUT, + "durability confirmation timed out; outcome unknown, do not retry without deduplication", + )) + })??; } Ok(()) } @@ -1954,6 +1979,68 @@ mod tests { assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); } + #[tokio::test] + async fn reducer_call_reports_success_only_once_durable() { + use spacetimedb::energy::FunctionBudget; + use spacetimedb::subscription::module_subscription_manager::from_tx_offset; + use std::pin::pin; + use tokio::sync::watch; + + let call = |outcome, tx_offset| ReducerCallResultWithTxOffset { + result: ReducerCallResult { + outcome, + execution_budget_used: FunctionBudget::ZERO, + execution_duration: Duration::ZERO, + }, + tx_offset: from_tx_offset(tx_offset), + }; + let (durable, rx) = watch::channel(None); + + // A committed call doesn't return until its transaction is durable. + let mut res = pin!(confirm_reducer_call( + call(ReducerOutcome::Committed, 7), + true, + Some(rx.clone().into()) + )); + assert!(futures::poll!(&mut res).is_pending()); + durable.send_replace(Some(7)); + assert!(res.await.is_ok()); + + // Unconfirmed and failed calls return without waiting. + for (outcome, confirmed) in [ + (ReducerOutcome::Committed, false), + (ReducerOutcome::BudgetExceeded, true), + ] { + let res = pin!(confirm_reducer_call( + call(outcome, 8), + confirmed, + Some(rx.clone().into()) + )); + assert!(matches!(futures::poll!(res), std::task::Poll::Ready(Ok(_)))); + } + } + + #[tokio::test(start_paused = true)] + async fn call_confirmation_times_out_with_live_senders() { + use spacetimedb::subscription::module_subscription_manager::from_tx_offset; + use std::pin::pin; + + let (_sender, pending_offset) = oneshot::channel(); + let (_durable, rx) = tokio::sync::watch::channel(None); + for offset in [from_tx_offset(7), pending_offset] { + let mut confirmation = pin!(confirm_call_offset(Some(offset), Some(rx.clone().into()))); + assert!(futures::poll!(&mut confirmation).is_pending()); + tokio::time::advance(TRANSACTION_CONFIRMATION_TIMEOUT + Duration::from_millis(1)).await; + let std::task::Poll::Ready(Err(error)) = futures::poll!(confirmation) else { + panic!("confirmation must time out even while its senders are alive"); + }; + let response = Err::<(), _>(error).into_response(); + assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); + let body = axum::body::to_bytes(response.into_body(), 1024).await.unwrap(); + assert!(std::str::from_utf8(&body).unwrap().contains("outcome unknown")); + } + } + #[tokio::test] async fn publish_environment_error_identifies_only_typed_missing_required_keys() { use spacetimedb_lib::environment::EnvironmentSchemaError; diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 13022b7f68f..45fef04ed47 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -139,6 +139,7 @@ proptest.workspace = true pretty_assertions.workspace = true jsonwebtoken.workspace = true axum.workspace = true +tokio = { workspace = true, features = ["test-util"] } [lints] workspace = true diff --git a/crates/core/src/host/instance_env.rs b/crates/core/src/host/instance_env.rs index eca8e5b5369..ceb096ac192 100644 --- a/crates/core/src/host/instance_env.rs +++ b/crates/core/src/host/instance_env.rs @@ -2,7 +2,9 @@ use super::scheduler::{get_schedule_from_row, scheduled_row_hash, ScheduleError, use crate::database_logger::{BacktraceFrame, BacktraceProvider, LogLevel, ModuleBacktrace, Record}; use crate::db::relational_db::{MutTx, RelationalDB}; use crate::error::{DBError, DatastoreError, IndexError, NodesError}; -use crate::host::module_host::{DatabaseUpdate, EventStatus, ModuleEvent, ModuleFunctionCall}; +use crate::host::module_host::{ + DatabaseUpdate, EventStatus, ModuleEvent, ModuleFunctionCall, TRANSACTION_CONFIRMATION_TIMEOUT, +}; use crate::host::wasm_common::TimingSpan; use crate::replica_context::ReplicaContext; use crate::subscription::module_subscription_actor::{commit_and_broadcast_event, ModuleSubscriptions}; @@ -987,9 +989,12 @@ impl InstanceEnv { })(); async move { if let Some((offset, mut durable)) = confirmation? { - durable.wait_for(offset).await.context( - "transaction committed, but durability could not be confirmed; outcome unknown, do not retry without deduplication", - )?; + tokio::time::timeout(TRANSACTION_CONFIRMATION_TIMEOUT, durable.wait_for(offset)) + .await + .context("durability confirmation timed out; outcome unknown, do not retry without deduplication")? + .context( + "transaction committed, but durability could not be confirmed; outcome unknown, do not retry without deduplication", + )?; } Ok(()) } @@ -1577,7 +1582,7 @@ mod test { use crate::db::relational_db::Persistence; use std::pin::pin; - let runtime = tokio::runtime::Runtime::new()?; + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?; let (durable, rx) = tokio::sync::watch::channel(None); let persistence = Persistence::new( PendingDurability(rx), @@ -1601,6 +1606,7 @@ mod test { durable.send_replace(Some(offset - 1)); runtime.block_on(async { + tokio::time::pause(); let mut confirmation = pin!(confirmation); assert!(futures::poll!(&mut confirmation).is_pending()); @@ -1613,7 +1619,19 @@ mod test { assert!(futures::poll!(&mut confirmation).is_pending()); durable.send_replace(Some(offset)); - confirmation.await + confirmation.await?; + + // This read observes the second transaction, which is still unconfirmed. + env.start_mutable_tx()?; + env.commit_mutable_tx()?; + let mut confirmation = pin!(env.procedure_confirmation()); + assert!(futures::poll!(&mut confirmation).is_pending()); + tokio::time::advance(TRANSACTION_CONFIRMATION_TIMEOUT + Duration::from_millis(1)).await; + let std::task::Poll::Ready(Err(error)) = futures::poll!(confirmation) else { + panic!("confirmation must time out even while the durability sender is alive"); + }; + assert!(error.to_string().contains("outcome unknown")); + Ok(()) }) } diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index 150ca5acc26..f59fedeb6e5 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -78,6 +78,9 @@ use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; use tokio::sync::{oneshot, OwnedSemaphorePermit, Semaphore}; +/// Maximum duration of a transaction durability confirmation wait. +pub const TRANSACTION_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5); + #[derive(Debug, Default, Clone, From)] pub struct DatabaseUpdate { pub tables: SmallVec<[DatabaseTableUpdate; 1]>, @@ -2434,6 +2437,21 @@ impl ModuleHost { .map_err(Into::into) } + async fn call_reducer_with_params_and_offset( + &self, + reducer_name: &ReducerName, + params: CallReducerParams, + ) -> Result { + call_instance!( + self, + reducer_name, + params, + |p, inst| inst.call_reducer_with_offset(p), + |p, inst| { inst.call_reducer_with_offset(p).await }, + ) + .map_err(Into::into) + } + fn log_reducer_submit_error(&self, reducer_name: &str, err: &ReducerCallError) { let log_message = match err { ReducerCallError::NoSuchReducer => Some(no_such_function_log_message("reducer", reducer_name)), @@ -2505,6 +2523,31 @@ impl ModuleHost { .await } + /// Like [`Self::call_reducer`], but also returns the offset of the reducer's transaction, + /// so that the caller can wait for it to become durable. + pub async fn call_reducer_with_offset( + &self, + caller_identity: Identity, + caller_connection_id: Option, + client: Option>, + request_id: Option, + timer: Option, + reducer_name: &str, + args: FunctionArgs, + ) -> Result { + self.with_reducer_call( + caller_identity, + caller_connection_id, + client, + request_id, + timer, + reducer_name, + args, + async |call| self.call_reducer_with_params_and_offset(&call.name, call.params).await, + ) + .await + } + pub async fn enqueue_reducer( &self, caller_identity: Identity, diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs index 2b7bd84cf5f..3f5b4ddccd5 100644 --- a/crates/core/src/host/v8/mod.rs +++ b/crates/core/src/host/v8/mod.rs @@ -85,7 +85,9 @@ use crate::host::wasm_common::module_host_actor::{ ReducerExecuteResult, ReducerOp, ViewExecuteResult, ViewOp, WasmInstance, }; use crate::host::wasm_common::{RowIters, TimingSpanSet}; -use crate::host::{InitDatabaseResult, ModuleHost, ReducerCallError, ReducerCallResult, Scheduler}; +use crate::host::{ + InitDatabaseResult, ModuleHost, ReducerCallError, ReducerCallResult, ReducerCallResultWithTxOffset, Scheduler, +}; use crate::messages::control_db::HostType; use crate::module_host_context::ModuleCreationContext; use crate::replica_context::ReplicaContext; @@ -497,6 +499,11 @@ impl JsMainInstance { } pub async fn call_reducer(&self, params: CallReducerParams) -> ReducerCallResult { + self.request(CallReducerRequest { params }).await.result + } + + /// Like [`Self::call_reducer`], but also returns the offset of the reducer's transaction. + pub async fn call_reducer_with_offset(&self, params: CallReducerParams) -> ReducerCallResultWithTxOffset { self.request(CallReducerRequest { params }).await } @@ -649,7 +656,7 @@ js_main_request! { js_main_request! { CallReducerRequest { params: CallReducerParams, - } => "call_reducer", ReducerCallResult, CallReducer + } => "call_reducer", ReducerCallResultWithTxOffset, CallReducer } js_main_request! { @@ -837,7 +844,7 @@ enum JsMainWorkerRequest { }, /// See [`JsMainInstance::call_reducer`]. CallReducer { - reply_tx: JsReplyTx, + reply_tx: JsReplyTx, params: CallReducerParams, }, /// See [`JsMainInstance::enqueue_reducer`]. @@ -1443,8 +1450,7 @@ fn handle_main_worker_request( JsMainWorkerRequest::CallReducer { reply_tx, params } => { handle_worker_request("call_reducer", reply_tx, || { let mut call_reducer = |tx, params| instance_common.call_reducer_with_tx(tx, params, inst); - let (res, trapped) = call_reducer(None, params); - (res.result, trapped) + call_reducer(None, params) }) } JsMainWorkerRequest::CallReducerDetached { params, on_panic } => { diff --git a/crates/core/src/host/wasm_common/module_host_actor.rs b/crates/core/src/host/wasm_common/module_host_actor.rs index 2ba3c3635cc..fc8bf93bde4 100644 --- a/crates/core/src/host/wasm_common/module_host_actor.rs +++ b/crates/core/src/host/wasm_common/module_host_actor.rs @@ -518,6 +518,13 @@ impl WasmModuleInstance { res } + /// Like [`Self::call_reducer`], but also returns the offset of the reducer's transaction. + pub fn call_reducer_with_offset(&mut self, params: CallReducerParams) -> ReducerCallResultWithTxOffset { + let (res, trapped) = self.call_reducer_with_tx_offset(None, params); + self.trapped = trapped; + res + } + pub fn clear_all_clients(&self) -> anyhow::Result<()> { self.common.clear_all_clients() } diff --git a/docs/docs/00200-core-concepts/00200-functions/00400-procedures.md b/docs/docs/00200-core-concepts/00200-functions/00400-procedures.md index e94b15d77cc..c419b8e4e91 100644 --- a/docs/docs/00200-core-concepts/00200-functions/00400-procedures.md +++ b/docs/docs/00200-core-concepts/00200-functions/00400-procedures.md @@ -175,7 +175,7 @@ WebSocket procedures inherit the connection's setting. HTTP procedure calls accept the `confirmed` query parameter (default `true`); scheduled procedures and module HTTP handlers default to confirmed reads. Disabling confirmation skips the wait, so state may be lost after a crash. -If confirmation fails after commit, the procedure stops with an internal error. +If confirmation fails or times out after five seconds, the procedure stops with an internal error. The callback is not retried, and committed transactions are not rolled back. @@ -237,7 +237,7 @@ WebSocket procedures inherit the connection's setting. HTTP procedure calls accept the `confirmed` query parameter (default `true`); scheduled procedures and module HTTP handlers default to confirmed reads. Disabling confirmation skips the wait, so state may be lost after a crash. -If confirmation fails after commit, the procedure stops with an internal error. +If confirmation fails or times out after five seconds, the procedure stops with an internal error. The callback is not retried, and committed transactions are not rolled back. @@ -291,7 +291,7 @@ WebSocket procedures inherit the connection's setting. HTTP procedure calls accept the `confirmed` query parameter (default `true`); scheduled procedures and module HTTP handlers default to confirmed reads. Disabling confirmation skips the wait, so state may be lost after a crash. -If confirmation fails after commit, the procedure stops with an internal error. +If confirmation fails or times out after five seconds, the procedure stops with an internal error. The callback is not retried, and committed transactions are not rolled back. @@ -350,7 +350,7 @@ WebSocket procedures inherit the connection's setting. HTTP procedure calls accept the `confirmed` query parameter (default `true`); scheduled procedures and module HTTP handlers default to confirmed reads. Disabling confirmation skips the wait, so state may be lost after a crash. -If confirmation fails after commit, the procedure stops with an internal error. +If confirmation fails or times out after five seconds, the procedure stops with an internal error. The callback is not retried, and committed transactions are not rolled back. diff --git a/docs/docs/00300-resources/00200-reference/00200-http-api/00300-database.md b/docs/docs/00300-resources/00200-reference/00200-http-api/00300-database.md index 00f0640acd1..b24665dec1a 100644 --- a/docs/docs/00300-resources/00200-reference/00200-http-api/00300-database.md +++ b/docs/docs/00300-resources/00200-reference/00200-http-api/00300-database.md @@ -395,7 +395,7 @@ Invoke a reducer or procedure in a database. | Name | Value | | ---- | ----- | -| `confirmed` | Optional boolean for procedure calls, default `true`. Wait for durability before returning success and after each successful `withTx`, before procedure code continues. Set to `false` to allow results from state that has committed in memory but may be lost after a crash. This parameter has no effect on reducer calls. +| `confirmed` | Optional boolean, default `true`. Wait for durability before returning success. For procedure calls, also wait after each successful `withTx`, before procedure code continues. Set to `false` to allow results from state that has committed in memory but may be lost after a crash. #### Path parameters