Confirm HTTP reducer calls and bound transaction confirmation waits

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<S: ControlStateDelegate + NodeDelegate>(
 
     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<S: ControlStateDelegate + NodeDelegate>(
             )
             .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<S: ControlStateDelegate + NodeDelegate>(
     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<DurableOffset>,
+) -> axum::response::Result<ReducerCallResult> {
+    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<spacetimedb::subscription::module_subscription_manager::TransactionOffset>,
     durable: Option<spacetimedb::host::module_host::DurableOffset>,
 ) -> 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;

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

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(())
         })
     }
 

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<ReducerCallResultWithTxOffset, ReducerCallError> {
+        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<ConnectionId>,
+        client: Option<Arc<ClientConnectionSender>>,
+        request_id: Option<RequestId>,
+        timer: Option<Instant>,
+        reducer_name: &str,
+        args: FunctionArgs,
+    ) -> Result<ReducerCallResultWithTxOffset, ReducerCallError> {
+        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,

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<ReducerCallResult>,
+        reply_tx: JsReplyTx<ReducerCallResultWithTxOffset>,
         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 } => {

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -518,6 +518,13 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         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()
     }

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.
 
 </TabItem>
@@ -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.
 
 </TabItem>
@@ -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.
 
 </TabItem>
@@ -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.
 
 </TabItem>

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