diff --git a/Cargo.lock b/Cargo.lock index 24f385f39c6..cf8bd1ba07c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9162,6 +9162,7 @@ dependencies = [ "lazy_static", "log", "rand 0.9.2", + "scopeguard", "serde", "serde_json", "serial_test", @@ -9170,9 +9171,12 @@ dependencies = [ "spacetimedb-client-api-messages", "spacetimedb-core", "spacetimedb-data-structures", + "spacetimedb-datastore", "spacetimedb-guard", "spacetimedb-lib", "spacetimedb-paths", + "spacetimedb-primitives", + "spacetimedb-sats", "spacetimedb-schema", "spacetimedb-standalone", "tempfile", diff --git a/crates/core/src/host/host_controller.rs b/crates/core/src/host/host_controller.rs index 55e52350a41..43432bafc13 100644 --- a/crates/core/src/host/host_controller.rs +++ b/crates/core/src/host/host_controller.rs @@ -1,3 +1,4 @@ +use super::idc_actor::{self, IdcActor, IdcActorStarter, IdcConfig}; use super::module_host::{ DurableOffset, EventStatus, InitDatabaseResult, ModuleHost, ModuleInfo, NoSuchModule, UpdateEnvironmentResult, }; @@ -31,6 +32,7 @@ use anyhow::{bail, Context}; use async_trait::async_trait; use durability::{Durability, EmptyHistory}; use log::{info, trace, warn}; +use once_cell::sync::OnceCell; use parking_lot::Mutex; use scopeguard::{defer, guard}; use spacetimedb_client_api_messages::name::EnvironmentVersionConflict; @@ -219,6 +221,10 @@ pub struct HostController { db_cores: JobCores, /// The pool of buffers used to build `BsatnRowList`s in subscriptions. pub bsatn_rlb_pool: BsatnRowListBuilderPool, + /// Transport settings configured before starting database actors. + idc_config: OnceCell, + /// HTTP connection pool shared by all IDC actors managed by this controller. + idc_client: reqwest::Client, } pub(crate) struct HostRuntimes { @@ -384,6 +390,8 @@ impl HostController { page_pool: PagePool::new(default_config.page_pool_max_size), bsatn_rlb_pool: BsatnRowListBuilderPool::new(), db_cores, + idc_config: OnceCell::new(), + idc_client: idc_actor::http_client(), } } @@ -398,6 +406,13 @@ impl HostController { self.program_storage = ps; } + /// Configure IDC transport before starting database actors. + pub fn set_idc_config(&self, config: IdcConfig) -> anyhow::Result<()> { + self.idc_config + .set(config) + .map_err(|_| anyhow::anyhow!("IDC transport already configured")) + } + /// Get a [`ModuleHost`] managed by this controller, or launch it from /// persistent state. /// @@ -569,9 +584,10 @@ impl HostController { .init_database_with_environment(program, environment) .await; let info = launched.module_host.info.clone(); - // Validation never starts scheduled work. Release its receiver before - // waiting for scheduler closure, including when initialization failed. + // Validation never starts background work. Release its receivers before + // waiting for closure, including when initialization failed. drop(launched.scheduler_starter); + drop(launched.idc_starter); launched.module_host.exit().await; let InitDatabaseResult { reducer, .. } = result?; if let Some(call_result) = reducer { @@ -1014,6 +1030,7 @@ async fn make_module_host( runtimes: Arc, replica_ctx: Arc, scheduler: Scheduler, + idc_sender: crate::host::idc_actor::IdcActorSender, program: Program, energy_monitor: Arc, unregister: impl Fn() + Send + Sync + 'static, @@ -1029,6 +1046,7 @@ async fn make_module_host( let mcc = ModuleCreationContext { replica_ctx, scheduler, + idc_sender, program_hash: program.hash, energy_monitor, }; @@ -1066,6 +1084,7 @@ struct LaunchedModule { module_host: ModuleHost, scheduler: Scheduler, scheduler_starter: SchedulerStarter, + idc_starter: IdcActorStarter, } struct HostInit { @@ -1110,10 +1129,12 @@ impl ModuleLauncher { .await .map(Arc::new)?; let (scheduler, scheduler_starter) = Scheduler::open(replica_ctx.relational_db().clone()); + let (idc_starter, idc_sender) = idc_actor::channel(); let (program, module_host) = make_module_host( self.runtimes.clone(), replica_ctx.clone(), scheduler.clone(), + idc_sender.clone(), self.program, self.energy_monitor, self.on_panic, @@ -1130,6 +1151,7 @@ impl ModuleLauncher { module_host, scheduler, scheduler_starter, + idc_starter, }, )) } @@ -1213,6 +1235,13 @@ struct Host { /// Handle to the task responsible for cleaning up old views. /// The task is aborted when [`Host`] is dropped. view_cleanup_task: AbortHandle, + /// Transport settings reused when starting a replacement IDC actor after a module update. + idc_config: IdcConfig, + /// Shared client retained across IDC actor replacements. + idc_client: reqwest::Client, + /// IDC actor for the current module. [`ModuleHost::exit`] drains it; + /// dropping this handle aborts the task if shutdown cannot finish. + idc_actor: IdcActor, } impl Host { @@ -1442,6 +1471,7 @@ impl Host { module_host, scheduler, scheduler_starter, + idc_starter, } = launched; // Disconnect dangling clients. @@ -1465,6 +1495,18 @@ impl Host { module_host.clear_all_clients().await?; scheduler_starter.start(&module_host)?; + let idc_config = host_controller + .idc_config + .get() + .ok_or_else(|| anyhow::anyhow!("IDC transport is not configured"))? + .clone(); + let idc_actor = idc_starter.start( + replica_ctx.relational_db().clone(), + replica_ctx.database_identity, + host_controller.idc_client.clone(), + idc_config.clone(), + module_host.downgrade(), + )?; let disk_metrics_recorder_task: spacetimedb_runtime::AbortHandle = tokio::spawn(metric_reporter(replica_ctx.clone())).abort_handle().into(); let view_cleanup_task = spawn_view_cleanup_loop(replica_ctx.relational_db().clone(), &runtime); @@ -1480,6 +1522,9 @@ impl Host { disk_metrics_recorder_task, tx_metrics_recorder_task, view_cleanup_task, + idc_config, + idc_client: host_controller.idc_client.clone(), + idc_actor, }, bootstrap_completion, }) @@ -1557,11 +1602,13 @@ impl Host { ) -> anyhow::Result { let replica_ctx = &self.replica_ctx; let (scheduler, scheduler_starter) = Scheduler::open(self.replica_ctx.relational_db().clone()); + let (idc_starter, idc_sender) = idc_actor::channel(); let (program, module) = make_module_host( - runtimes, + runtimes.clone(), replica_ctx.clone(), scheduler.clone(), + idc_sender, program, energy_monitor, on_panic, @@ -1588,8 +1635,15 @@ impl Host { UpdateDatabaseResult::NoUpdateNeeded | UpdateDatabaseResult::UpdatePerformed { .. } => { self.scheduler = scheduler; scheduler_starter.start(&module)?; - let old_module = self.module.send_replace(module); + let old_module = self.module.send_replace(module.clone()); old_module.exit().await; + self.idc_actor = idc_starter.start( + replica_ctx.relational_db().clone(), + replica_ctx.database_identity, + self.idc_client.clone(), + self.idc_config.clone(), + module.downgrade(), + )?; } // In this case, we need to disconnect all clients connected to the old module @@ -1612,10 +1666,17 @@ impl Host { self.scheduler = scheduler; scheduler_starter.start(&module)?; - // exit the old module, drop the `old_watcher` afterwards, - // which will signal websocket clients that the module is gone. + // Drain the old module and actor before starting replacement deliveries. + // Dropping old_watcher afterwards disconnects its remaining watchers. let old_module = old_watcher.borrow().clone(); old_module.exit().await; + self.idc_actor = idc_starter.start( + replica_ctx.relational_db().clone(), + replica_ctx.database_identity, + self.idc_client.clone(), + self.idc_config.clone(), + module.downgrade(), + )?; } _ => {} } diff --git a/crates/core/src/host/idc/mod.rs b/crates/core/src/host/idc/mod.rs new file mode 100644 index 00000000000..74b6f9ed304 --- /dev/null +++ b/crates/core/src/host/idc/mod.rs @@ -0,0 +1,3 @@ +mod streams; + +pub(super) use streams::{DeliveryStreams, StreamId}; diff --git a/crates/core/src/host/idc/streams.rs b/crates/core/src/host/idc/streams.rs new file mode 100644 index 00000000000..6ecbae38faa --- /dev/null +++ b/crates/core/src/host/idc/streams.rs @@ -0,0 +1,204 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::fmt::Display; +use std::time::{Duration, Instant}; + +// Retry delays start at 100 ms and double up to 30 seconds. +const INITIAL_BACKOFF: Duration = Duration::from_millis(100); +const MAX_BACKOFF: Duration = Duration::from_secs(30); + +#[derive(Debug, Eq, PartialEq, Hash, Clone, Copy)] +pub(in crate::host) struct StreamId(pub u64); + +impl Display for StreamId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Allows one message per stream to run through delivery, callback, and cleanup. +/// TODO(idc): Support pipelining. +struct OrderedStream { + /// The sequence being processed, or `None` between attempts. + active_seq: Option, + /// Earliest time a failed attempt may retry. + blocked_until: Option, + /// Delay after the next failed attempt. Resets when cleanup succeeds. + backoff: Duration, +} + +impl OrderedStream { + fn new() -> Self { + Self { + active_seq: None, + blocked_until: None, + backoff: INITIAL_BACKOFF, + } + } + + fn is_ready(&self) -> bool { + self.active_seq.is_none() && self.blocked_until.is_none_or(|until| Instant::now() >= until) + } +} + +/// Keeps stream state and the ready queue in sync. Each stream is queued at most once. +/// It maintains a FIFO ready queue poll streams. +#[derive(Default)] +pub(in crate::host) struct DeliveryStreams { + streams: HashMap, + ready: VecDeque, + queued: HashSet, +} + +impl DeliveryStreams { + /// Wake a stream after an insert or retry deadline without interrupting an active attempt or backoff. + pub fn wake(&mut self, stream_id: StreamId) { + let state = self.streams.entry(stream_id).or_insert_with(OrderedStream::new); + if state.is_ready() && self.queued.insert(stream_id) { + self.ready.push_back(stream_id); + } + } + + /// Take a stream for a database lookup. Call `start` if it has work, or `remove_idle` otherwise. + pub fn pop_ready(&mut self) -> Option { + let stream_id = self.ready.pop_front()?; + self.queued.remove(&stream_id); + Some(stream_id) + } + + /// Keep this stream busy until delivery, callback, and cleanup finish. + pub fn start(&mut self, stream_id: StreamId, seq: u64) { + assert!( + !self.queued.contains(&stream_id), + "stream must be taken from the ready queue first" + ); + let state = self + .streams + .get_mut(&stream_id) + .expect("ready stream has delivery state"); + assert!( + state.is_ready(), + "ordered stream must be ready before starting a message" + ); + state.active_seq = Some(seq); + } + + /// Queue the next message after cleanup commits and reset the retry delay. + pub fn finish(&mut self, stream_id: StreamId, seq: u64) { + let state = self + .streams + .get_mut(&stream_id) + .expect("active stream has delivery state"); + assert_eq!(state.active_seq, Some(seq), "completion must match the active sequence"); + state.active_seq = None; + state.blocked_until = None; + state.backoff = INITIAL_BACKOFF; + self.wake(stream_id); + } + + /// End the failed attempt and return its retry delay. New inserts do not shorten this delay. + pub fn fail(&mut self, stream_id: StreamId, seq: u64) -> Duration { + let state = self + .streams + .get_mut(&stream_id) + .expect("active stream has delivery state"); + assert_eq!(state.active_seq, Some(seq), "failure must match the active sequence"); + // Clear the active sequence so the same message can run again after backoff. + state.active_seq = None; + let delay = state.backoff; + state.blocked_until = Some(Instant::now() + delay); + state.backoff = (delay * 2).min(MAX_BACKOFF); + delay + } + + /// Forget an empty stream after its database lookup. A later insert recreates it. + pub fn remove_idle(&mut self, stream_id: StreamId) { + assert!( + !self.queued.contains(&stream_id), + "stream must be taken from the ready queue first" + ); + let state = self.streams.get(&stream_id).expect("idle stream has delivery state"); + assert!(state.is_ready(), "cannot remove an active or blocked stream"); + self.streams.remove(&stream_id); + } + + /// Find the next stream to retry and how long it needs to wait. + pub fn next_retry_delay(&self) -> Option<(StreamId, Duration)> { + self.streams + .iter() + .filter(|(_, state)| state.active_seq.is_none()) + .filter_map(|(&id, state)| state.blocked_until.map(|deadline| (id, deadline))) + .min_by_key(|(_, deadline)| *deadline) + .map(|(id, deadline)| (id, deadline.saturating_duration_since(Instant::now()))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn notifications_and_completion_queue_each_stream_once() { + let mut streams = DeliveryStreams::default(); + let first = StreamId(1); + let second = StreamId(2); + streams.wake(first); + streams.wake(first); + streams.wake(second); + assert_eq!(streams.pop_ready(), Some(first)); + streams.start(first, 1); + streams.wake(first); + assert_eq!(streams.pop_ready(), Some(second)); + streams.start(second, 1); + assert_eq!(streams.pop_ready(), None); + + streams.finish(first, 1); + streams.wake(first); + assert_eq!(streams.pop_ready(), Some(first)); + assert_eq!(streams.pop_ready(), None); + streams.remove_idle(first); + assert!(!streams.streams.contains_key(&first)); + streams.wake(first); + assert_eq!(streams.pop_ready(), Some(first)); + assert_eq!(streams.streams[&second].active_seq, Some(1)); + } + + #[test] + fn retries_preserve_backoff_and_skip_active_streams() { + let mut streams = DeliveryStreams::default(); + let first = StreamId(1); + let second = StreamId(2); + for id in [first, second] { + streams.wake(id); + assert_eq!(streams.pop_ready(), Some(id)); + streams.start(id, 1); + assert_eq!(streams.fail(id, 1), INITIAL_BACKOFF); + // Use a future deadline so this check does not depend on test execution speed. + streams.streams.get_mut(&id).unwrap().blocked_until = Some(Instant::now() + Duration::from_secs(3600)); + streams.wake(id); + assert_eq!(streams.pop_ready(), None); + } + + let now = Instant::now(); + streams.streams.get_mut(&first).unwrap().blocked_until = Some(now); + streams.streams.get_mut(&second).unwrap().blocked_until = Some(now - INITIAL_BACKOFF); + assert_eq!(streams.next_retry_delay(), Some((second, Duration::ZERO))); + streams.wake(second); + assert_eq!(streams.pop_ready(), Some(second)); + streams.start(second, 1); + assert_eq!(streams.next_retry_delay(), Some((first, Duration::ZERO))); + streams.wake(first); + assert_eq!(streams.pop_ready(), Some(first)); + streams.start(first, 1); + assert_eq!(streams.next_retry_delay(), None); + assert_eq!(streams.fail(first, 1), INITIAL_BACKOFF * 2); + + streams.streams.get_mut(&first).unwrap().blocked_until = Some(now); + streams.wake(first); + assert_eq!(streams.pop_ready(), Some(first)); + streams.start(first, 1); + streams.finish(first, 1); + assert_eq!(streams.pop_ready(), Some(first)); + streams.start(first, 2); + assert_eq!(streams.fail(first, 2), INITIAL_BACKOFF); + } +} diff --git a/crates/core/src/host/idc_actor.rs b/crates/core/src/host/idc_actor.rs new file mode 100644 index 00000000000..af99bcbc0c7 --- /dev/null +++ b/crates/core/src/host/idc_actor.rs @@ -0,0 +1,1227 @@ +use crate::auth::idc::IdcTokenSigner; +use crate::db::relational_db::{RelationalDB, Tx}; +use crate::host::idc::{DeliveryStreams, StreamId}; +use crate::host::module_host::WeakModuleHost; +use crate::host::module_host::{CallReducerParams, ModuleInfo}; +use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance}; +use crate::host::{FunctionArgs, ReducerOutcome}; +use anyhow::Context; +use bytes::Bytes; +use futures::stream::FuturesUnordered; +use futures::StreamExt; +use spacetimedb_client_api_messages::idc::IdcResponse; +use spacetimedb_datastore::execution_context::{ExecutionContext, ReducerContext, Workload}; +use spacetimedb_datastore::locking_tx_datastore::MutTxId; +use spacetimedb_datastore::system_tables::{StInboundMsgResultStatus, StOutboundMsgRow, StOutboundStreamRow}; +use spacetimedb_datastore::traits::IsolationLevel; +use spacetimedb_lib::{Hash, Identity, ProductValue, Timestamp}; +use spacetimedb_primitives::{ColId, TableId}; +use spacetimedb_schema::reducer_name::ReducerName; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +const NOTIFICATION_BATCH_SIZE: usize = 16; + +#[derive(Clone)] +pub struct IdcActorSender { + tx: mpsc::UnboundedSender, + shutdown: mpsc::UnboundedSender<()>, +} + +impl IdcActorSender { + pub fn send(&self, msg: StOutboundMsgRow) -> Result<(), mpsc::error::SendError> { + self.tx.send(msg) + } + + /// Stop dispatching new work and drain active attempts. + pub fn close(&self) { + let _ = self.shutdown.send(()); + } + + /// Wait until the actor has stopped, including its active attempts. + pub async fn closed(&self) { + self.shutdown.closed().await; + } +} + +/// HTTP and signing settings shared by the host's IDC actors. +#[derive(Clone)] +pub struct IdcConfig { + /// Local HTTP port used to send requests. + pub http_port: u16, + /// Signs the sender's database identity. These tokens are not exposed to module code. + pub token_signer: Arc, +} + +pub struct IdcActor { + task: tokio::task::JoinHandle<()>, +} + +impl Drop for IdcActor { + fn drop(&mut self) { + self.task.abort(); + } +} + +pub struct IdcActorStarter { + rx: mpsc::UnboundedReceiver, + shutdown_rx: mpsc::UnboundedReceiver<()>, +} + +impl IdcActorStarter { + pub fn start( + self, + db: Arc, + database_identity: Identity, + client: reqwest::Client, + config: IdcConfig, + module_host: WeakModuleHost, + ) -> anyhow::Result { + let mut streams = DeliveryStreams::default(); + restore_pending_streams(&db, &mut streams)?; + let task = tokio::spawn(async move { + if let Err(error) = run_idc_loop( + db, + database_identity, + client, + config, + module_host, + self.rx, + streams, + self.shutdown_rx, + ) + .await + { + log::error!( + "idc_actor: sender_identity={}: stopped: {error:#}", + database_identity.to_hex(), + ); + } + }); + Ok(IdcActor { task }) + } +} + +/// Create the channels used to notify and stop an actor, which is started separately. +pub fn channel() -> (IdcActorStarter, IdcActorSender) { + let (tx, rx) = mpsc::unbounded_channel(); + let (shutdown, shutdown_rx) = mpsc::unbounded_channel(); + (IdcActorStarter { rx, shutdown_rx }, IdcActorSender { tx, shutdown }) +} + +/// An outbox message and its metadata from the `st_outbound_*` tables. +#[derive(Clone)] +struct PendingMessage { + stream_id: StreamId, + /// Primary key of the user outbox row. + message_id: u64, + sequence: u64, + outbox_table_id: TableId, + target_identity: Identity, + /// All sequences through this value are complete. The receiver can discard their results. + ack_prefix: u64, + target_reducer: ReducerName, + signature_hash: Hash, + /// BSATN argument product in receiver parameter order, without field names or a field count. + reducer_args: Bytes, + /// Full row used for delivery and passed to `on_result`. + outbox_row: ProductValue, + on_result_reducer: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct MessageResult { + status: StInboundMsgResultStatus, + /// Raw HTTP response body. Errors are UTF-8 text, not BSATN strings. + /// Success bodies are empty for new calls. Retries may return saved BSATN return bytes. + /// These bytes are stored but are not decoded for `on_result`. + payload: Bytes, +} + +/// A message whose receiver result has been stored locally. +#[derive(Clone)] +pub(super) struct DeliveredMessage { + message: PendingMessage, + result: MessageResult, +} + +enum LoadedMessage { + /// Needs delivery to the receiver. + Pending(PendingMessage), + /// Has a stored result and is ready for callback and cleanup. + Delivered(DeliveredMessage), +} + +impl LoadedMessage { + fn message(&self) -> &PendingMessage { + match self { + Self::Pending(message) => message, + Self::Delivered(delivered) => &delivered.message, + } + } +} + +/// Process one message at a time per stream. Other streams can run at the same time. +/// +/// For each message: +/// 1. Send it using its stored stream ID and sequence. If the reply is lost, retry +/// with those same IDs. The receiver returns the saved result instead of +/// executing the reducer again. +/// 2. Save the reply locally before calling `on_result`. If a retry or restart +/// finds a saved reply, skip the HTTP request and use that reply. +/// 3. Run `on_result`, if configured and the outbox row still matches. Commit its +/// changes, delete the message, and advance `ack_prefix` in one transaction. +/// If the callback fails, roll back its changes and clean up separately. +/// 4. Move to the next message after cleanup commits. If no result was received, +/// wait and retry the same sequence. A reducer or callback failure does not +/// require sending the message again. +/// +/// Load pending streams at startup. After that, new inserts, completed attempts, +/// and retry timers wake the loop. Read each stream's next message from the database. +/// Send acknowledgments with later messages. Idle streams do not send a separate ack. +/// If loading a message fails, stop the actor instead of retrying the read. +#[allow(clippy::too_many_arguments)] +async fn run_idc_loop( + db: Arc, + database_identity: Identity, + client: reqwest::Client, + config: IdcConfig, + module_host: WeakModuleHost, + mut notify_rx: mpsc::UnboundedReceiver, + mut streams: DeliveryStreams, + mut shutdown: mpsc::UnboundedReceiver<()>, +) -> anyhow::Result<()> { + let mut in_flight = FuturesUnordered::new(); + let mut notifications = Vec::with_capacity(NOTIFICATION_BATCH_SIZE); + let mut notifications_open = true; + + let result = async { + loop { + // Honor explicit shutdown before normal work; dropping senders must not stop pending retries. + if shutdown.try_recv().is_ok() { + break Ok(()); + } + + // Start ready messages. Each future includes delivery, callback, and cleanup. + while let Some(msg) = take_ready_message(&db, &mut streams)? { + let stream_id = msg.message().stream_id; + let seq = msg.message().sequence; + let (db, client, config, module_host) = (&db, &client, &config, &module_host); + in_flight.push(async move { + let result = handle_message(db, client, database_identity, config, module_host, msg).await; + (stream_id, seq, result) + }); + } + + let next_retry = streams.next_retry_delay(); + if !notifications_open && in_flight.is_empty() && next_retry.is_none() { + log::debug!( + "idc_actor: sender_identity={}: stopping; notification channel closed and no deliveries or retries remain", + database_identity.to_hex(), + ); + break Ok(()); + } + + tokio::select! { + Some(()) = shutdown.recv() => break Ok(()), + + // Start the next message after cleanup succeeds. Retry failed attempts after backoff. + Some((stream_id, seq, result)) = in_flight.next(), if !in_flight.is_empty() => { + match result { + Ok(()) => { + streams.finish(stream_id, seq); + } + Err(error) => { + let delay = streams.fail(stream_id, seq); + log::warn!( + "idc_actor: sender_identity={} stream_id={stream_id} sequence={seq}: retrying after {delay:?}: {error:#}", + database_identity.to_hex(), + ); + } + } + }, + + // Wake the stream with the earliest retry deadline. + Some(stream_id) = async { + let (stream_id, delay) = next_retry?; + // An overdue retry must stay ready even when other branches keep winning. + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + Some(stream_id) + }, if next_retry.is_some() => { + streams.wake(stream_id); + }, + + // New inserts wake ready streams, but do not clear backoff. + // Bound each batch so a steady flow of inserts cannot hold up completions or retries. + count = notify_rx.recv_many(&mut notifications, NOTIFICATION_BATCH_SIZE), + if notifications_open => + { + // Zero means the channel is closed and drained. + notifications_open = count != 0; + for msg in notifications.drain(..) { + streams.wake(StreamId(msg.stream_id)); + } + }, + }; + } + } + .await; + + // Submitted callbacks continue on the module executor even if their futures + // are dropped. Drain them before the caller can shut down the database. + // Failed attempts remain pending; shutdown must not schedule another retry. + while let Some((stream_id, seq, result)) = in_flight.next().await { + if let Err(error) = result { + log::warn!("idc_actor: stream_id={stream_id} sequence={seq}: left pending during shutdown: {error:#}"); + } + } + // Closing the shutdown channel reports completion only after active work drains. + drop(shutdown); + result +} + +/// Keep the stream active through delivery, callback, and cleanup. +/// Errors leave the message pending for retry. A saved result skips the HTTP request. +async fn handle_message( + db: &RelationalDB, + client: &reqwest::Client, + database_identity: Identity, + config: &IdcConfig, + module_host: &WeakModuleHost, + msg: LoadedMessage, +) -> anyhow::Result<()> { + let delivered = match msg { + LoadedMessage::Pending(msg) => { + deliver_message_and_record_result(db, client, database_identity, config, msg).await? + } + LoadedMessage::Delivered(msg) => Some(msg), + }; + let Some(delivered) = delivered else { + // A late reply arrived after the message was already cleaned up. + return Ok(()); + }; + + let host = module_host + .upgrade() + .context("IDC module host is no longer available")?; + host.run_idc_callback_and_cleanup(delivered).await +} + +/// Take the next ready message and mark its stream as active until processing finishes. +fn take_ready_message(db: &RelationalDB, streams: &mut DeliveryStreams) -> anyhow::Result> { + while let Some(stream_id) = streams.pop_ready() { + // With 1 inflight msg per stream, ack_prefix + 1 selects the next message or retry. + match read_next_message(db, stream_id, None) + .with_context(|| format!("failed to read next message for stream {stream_id}"))? + { + Some(msg) => { + streams.start(stream_id, msg.message().sequence); + return Ok(Some(msg)); + } + None => { + // There is no work left. A later insert adds the stream back. + streams.remove_idle(stream_id); + } + } + } + Ok(None) +} + +/// Restore pending streams at startup without loading message bodies. +fn restore_pending_streams(db: &RelationalDB, streams: &mut DeliveryStreams) -> anyhow::Result<()> { + for stream_id in read_pending_streams(db)? { + streams.wake(stream_id); + } + Ok(()) +} + +fn read_pending_streams(db: &RelationalDB) -> anyhow::Result> { + let tx = db.begin_tx(Workload::Internal); + let mut pending = Vec::new(); + for stream in db.st_outbound_streams(&tx)? { + let stream = stream?; + // next_seq - 1 is the last assigned sequence, so anything beyond ack_prefix is still pending. + if stream.ack_prefix < stream.next_seq - 1 { + pending.push(StreamId(stream.stream_id)); + } + } + Ok(pending) +} + +/// Read the message after `after_seq`, or resume from the acknowledgment prefix when absent. +fn read_next_message( + db: &RelationalDB, + stream_id: StreamId, + after_seq: Option, +) -> anyhow::Result> { + let tx = db.begin_tx(Workload::Internal); + // An insert may have rolled back or cleanup may have finished before we read its notification. + let Some(stream) = db.st_outbound_stream_by_id(&tx, stream_id.0)? else { + return Ok(None); + }; + let after_seq = after_seq.unwrap_or(stream.ack_prefix); + anyhow::ensure!( + after_seq >= stream.ack_prefix, + "IDC stream {stream_id} cursor {after_seq} is behind acknowledgment {}", + stream.ack_prefix + ); + let seq = after_seq + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("IDC stream {stream_id} sequence overflow"))?; + if seq == stream.next_seq { + // No new row + return Ok(None); + } + anyhow::ensure!( + seq < stream.next_seq, + "IDC stream {stream_id} requested sequence {seq} exceeds its next sequence {}", + stream.next_seq + ); + // A missing pending row is not an acknowledgment. Never skip a gap in the stream. + let row = db + .outbound_msg_by_stream_seq(&tx, stream_id.0, seq)? + .ok_or_else(|| anyhow::anyhow!("IDC stream {stream_id} is missing pending sequence {seq}"))?; + let row = StOutboundMsgRow::try_from(row)?; + load_pending_message(db, tx, &stream, row).map(Some) +} + +fn load_pending_message( + db: &RelationalDB, + tx: Tx, + stream: &StOutboundStreamRow, + message_row: StOutboundMsgRow, +) -> anyhow::Result { + let schema = db.schema_for_table(&tx, stream.outbox_table_id)?; + let outbox = schema + .outbox + .as_ref() + .ok_or_else(|| anyhow::anyhow!("table is no longer an outbox"))?; + let outbox_row = db + .outbox_row(&tx, stream.outbox_table_id, message_row.msg_id)? + .ok_or_else(|| anyhow::anyhow!("pending IDC message has no outbox row {}", message_row.msg_id))?; + let reducer_args = encode_reducer_args(&outbox_row, &outbox.arg_columns).into(); + let (_, metrics, reducer) = db.release_tx(tx); + db.report_read_tx_metrics(reducer, metrics); + + let message = PendingMessage { + stream_id: StreamId(message_row.stream_id), + message_id: message_row.msg_id, + sequence: message_row.seq, + outbox_table_id: stream.outbox_table_id, + target_identity: stream.target_identity.0, + ack_prefix: stream.ack_prefix, + target_reducer: outbox.remote_reducer.clone(), + signature_hash: Hash::from_hex(&outbox.signature_hash)?, + reducer_args, + outbox_row, + on_result_reducer: outbox.on_result_reducer.clone(), + }; + + match (message_row.result_status, message_row.result_payload) { + (None, None) => Ok(LoadedMessage::Pending(message)), + (Some(status), Some(payload)) => Ok(LoadedMessage::Delivered(DeliveredMessage { + message, + result: MessageResult { status, payload }, + })), + _ => anyhow::bail!("pending IDC message has an incomplete stored result"), + } +} + +/// Encode the mapped columns as a BSATN reducer argument product. +/// +/// `arg_columns` lists columns in receiver parameter order. BSATN products store fields +/// back to back, without names or a field count. Write only the mapped columns. +fn encode_reducer_args(row: &ProductValue, arg_columns: &[ColId]) -> Vec { + let mut out = Vec::new(); + for col in arg_columns { + let elem = row + .elements + .get(col.idx()) + .expect("validated outbox arg column should exist in row"); + spacetimedb_sats::bsatn::to_writer(&mut out, elem).expect("writing outbox row args to BSATN should never fail"); + } + out +} + +/// Send the message and save its result before running the callback. +/// Saving the result does not advance `ack_prefix`. That happens during callback and cleanup. +/// A saving result is more important in pipelined version, where responses can arrive in any order. +/// but we need to hold them to run callbacks serially. +/// Returns `None` for a late reply to a message we already cleaned up and acknowledged. +async fn deliver_message_and_record_result( + db: &RelationalDB, + client: &reqwest::Client, + database_identity: Identity, + config: &IdcConfig, + msg: PendingMessage, +) -> anyhow::Result> { + let response = attempt_delivery(client, database_identity, config, &msg).await; + match response.and_then(|response| record_delivery_response(db, &msg, response)) { + Ok(result) => Ok(result.map(|result| DeliveredMessage { message: msg, result })), + Err(error) => { + log::warn!( + "idc_actor: sender_identity={} stream_id={} sequence={} target_identity={} outbox_table_id={:?}: delivery failed: {error:#}", + database_identity.to_hex(), + msg.stream_id, + msg.sequence, + msg.target_identity.to_hex(), + msg.outbox_table_id, + ); + record_transport_error(db, &msg, format!("{error:#}")); + Err(error) + } + } +} + +/// Create the HTTP client shared by a host controller's IDC actors. +pub fn http_client() -> reqwest::Client { + reqwest::Client::builder() + // The local proxy handles routing. A redirect is not an IDC result. + .redirect(reqwest::redirect::Policy::none()) + .http2_prior_knowledge() + .http2_keep_alive_interval(Some(Duration::from_millis(200))) + .http2_keep_alive_timeout(Duration::from_secs(5)) + .http2_keep_alive_while_idle(true) + .timeout(Duration::from_secs(5)) + .build() + .expect("failed to build IDC reqwest client") +} + +/// Send a request through localhost and let the host's proxy route it to the receiver. +/// +/// The request body is the BSATN argument product in [`PendingMessage::reducer_args`]. +/// Response bodies use the format described by [`MessageResult::payload`]. +/// +/// Read the full body before interpreting the HTTP status with [`IdcResponse`]. Errors leave +/// the message pending, even when someone needs to fix the underlying problem. +async fn attempt_delivery( + client: &reqwest::Client, + database_identity: Identity, + config: &IdcConfig, + msg: &PendingMessage, +) -> anyhow::Result { + let token = config + .token_signer + .sign(database_identity, msg.target_identity) + .context("IDC signing failed")?; + let target_db_hex = msg.target_identity.to_hex(); + let url = format!( + "http://localhost:{}/v1/database/{target_db_hex}/call-from-database/{}?stream_id={}&seq={}&ack_prefix={}", + config.http_port, msg.target_reducer, msg.stream_id, msg.sequence, msg.ack_prefix, + ); + let resp = client + .post(&url) + .query(&[("signature_hash", msg.signature_hash.to_hex().as_str())]) + .bearer_auth(token) + .header("Content-Type", "application/octet-stream") + .body(msg.reducer_args.clone()) + .send() + .await?; + + let status = resp.status(); + let payload = resp + .bytes() + .await + .with_context(|| format!("HTTP {status}: failed to read response"))?; + Ok(IdcResponse::from_http_parts(status, payload)) +} + +/// Save the receiver's result. Ignore `AlreadyAcked` only if local cleanup already finished. +fn record_delivery_response( + db: &RelationalDB, + msg: &PendingMessage, + response: IdcResponse, +) -> anyhow::Result> { + let (status, payload) = match response { + IdcResponse::Success(payload) => (StInboundMsgResultStatus::Ok, payload), + IdcResponse::TerminalError(payload) => (StInboundMsgResultStatus::Err, payload), + IdcResponse::BudgetExceeded(payload) => (StInboundMsgResultStatus::BudgetExceeded, payload), + IdcResponse::DeliveryError(error) => return Err(error.into()), + IdcResponse::AlreadyAcked => { + // We should have finished cleanup before acknowledging this message. + // Check that here. This reply cannot replace a missing result or tell + // us whether the reducer succeeded. Do not advance ack_prefix or call + // on_result again. + let tx = db.begin_tx(Workload::Internal); + let retired = db + .st_outbound_stream_by_id(&tx, msg.stream_id.0)? + .is_some_and(|stream| msg.sequence <= stream.ack_prefix); + anyhow::ensure!( + retired, + "receiver has retired the outcome for pending stream {}, sequence {}", + msg.stream_id, + msg.sequence, + ); + return Ok(None); + } + }; + let result = MessageResult { status, payload }; + record_delivery_result(db, msg, &result)?; + Ok(Some(result)) +} + +/// Save the receiver's result and clear the last transport error. Keep the message +/// until its callback and acknowledgment are handled. +fn record_delivery_result(db: &RelationalDB, msg: &PendingMessage, result: &MessageResult) -> anyhow::Result<()> { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + tx.record_outbound_idc_result(msg.stream_id.0, msg.sequence, result.status, result.payload.clone())?; + db.commit_tx(tx)?; + Ok(()) +} + +/// Save the latest transport error and increment the retry count. Keep the message +/// pending even if saving the error fails. +fn record_transport_error(db: &RelationalDB, msg: &PendingMessage, reason: String) { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + if let Err(e) = tx.record_outbound_idc_transport_error(msg.stream_id.0, msg.sequence, reason) { + log::error!( + "idc_actor: sender_identity={} stream_id={} sequence={}: failed to record transport error: {e:#}", + db.database_identity().to_hex(), + msg.stream_id, + msg.sequence, + ); + let _ = db.rollback_mut_tx(tx); + } else if let Err(e) = db.commit_tx(tx) { + log::error!( + "idc_actor: sender_identity={} stream_id={} sequence={}: failed to commit transport error: {e:#}", + db.database_identity().to_hex(), + msg.stream_id, + msg.sequence, + ); + } +} + +/// Runs on the module executor. Check the row and run the callback in the same +/// transaction so the callback cannot run for a deleted or replaced row. +/// Successful callback changes commit atomically with cleanup: deleting the matching +/// outbox row and pending message, and advancing `ack_prefix`. +/// If the callback fails, roll back its changes and attempt cleanup in a new transaction. +pub(super) fn run_callback_and_cleanup( + info: &ModuleInfo, + msg: DeliveredMessage, + common: &mut InstanceCommon, + inst: &mut impl WasmInstance, +) -> (anyhow::Result<()>, bool) { + let db = info.relational_db(); + let sender_identity = info.database_identity.to_hex(); + let stream_id = msg.message.stream_id; + let sequence = msg.message.sequence; + let mut trapped = false; + let result = (|| { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + if !message_exists(db, &tx, &msg.message)? { + return Ok(()); + } + + let params = if current_row_matches(db, &tx, &msg.message)? { + msg.message + .on_result_reducer + .as_deref() + .map(|name| callback_params(info, &msg, name)) + .transpose() + } else { + Ok(None) + }; + let params = match params { + Ok(params) => params, + Err(error) => { + log::error!( + "idc_actor: sender_identity={sender_identity} stream_id={stream_id} sequence={sequence}: cannot prepare on_result: {error:#}" + ); + None + } + }; + let Some(params) = params else { + // No callback to run. Clean up the message in the open transaction. + let (result, did_trap) = cleanup_message_with_views(tx, info, &msg, common, inst); + trapped = did_trap; + return result; + }; + + let name = info.module_def.reducer_by_id(params.reducer_id).name.clone(); + tx.ctx = ExecutionContext::with_workload( + tx.ctx.database_identity(), + Workload::Reducer(ReducerContext { + name: name.clone(), + caller_identity: params.caller_identity, + caller_connection_id: params.caller_connection_id, + timestamp: params.timestamp, + arg_bsatn: params.args.get_bsatn().clone(), + }), + ); + + // Successful callbacks commit their changes and cleanup together. + let callback_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + common.call_reducer_with_success_action(Some(tx), params, inst, |tx, _| cleanup_message(db, tx, &msg)) + })); + match callback_result { + Ok((result, did_trap)) => { + trapped = did_trap; + if matches!(result.result.outcome, ReducerOutcome::Committed) { + return Ok(()); + } + log::error!( + "idc_actor: sender_identity={sender_identity} stream_id={stream_id} sequence={sequence}: on_result {name} failed: {:?}", + result.result.outcome, + ); + } + Err(_) => { + trapped = true; + log::error!( + "idc_actor: sender_identity={sender_identity} stream_id={stream_id} sequence={sequence}: on_result {name} panicked" + ); + } + } + + // The callback transaction rolled back. Start a new transaction for cleanup. + // If cleanup also fails, the message remains pending and a later attempt + // can run the failed callback again. + let cleanup_tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let (result, cleanup_trapped) = cleanup_message_with_views(cleanup_tx, info, &msg, common, inst); + trapped |= cleanup_trapped; + result + })(); + (result, trapped) +} + +/// Commit cleanup and view updates without running the callback. +fn cleanup_message_with_views( + mut tx: MutTxId, + info: &ModuleInfo, + msg: &DeliveredMessage, + common: &mut InstanceCommon, + inst: &mut impl WasmInstance, +) -> (anyhow::Result<()>, bool) { + if let Err(error) = cleanup_message(info.relational_db(), &mut tx, msg) { + return (Err(error), false); + } + // User-row cleanup must refresh views and notify subscriptions, just like + // scheduled-row cleanup. Do not bypass the module's commit path. + common.refresh_views_then_commit_and_broadcast(tx, inst) +} + +fn callback_params(info: &ModuleInfo, msg: &DeliveredMessage, name: &str) -> anyhow::Result { + let mut bytes = spacetimedb_sats::bsatn::to_vec(&msg.message.outbox_row)?; + let result: Result<(), String> = match msg.result.status { + StInboundMsgResultStatus::Ok => Ok(()), + StInboundMsgResultStatus::Err | StInboundMsgResultStatus::BudgetExceeded => { + Err(String::from_utf8_lossy(&msg.result.payload).into_owned()) + } + }; + spacetimedb_sats::bsatn::to_writer(&mut bytes, &result)?; + let (id, def, owning) = info + .module_def + .reducer_by_name_with_module(name) + .ok_or_else(|| anyhow::anyhow!("on_result reducer {name} not found"))?; + let args = FunctionArgs::Bsatn(bytes.into()).into_tuple_for_def(owning, def)?; + Ok(CallReducerParams::from_system( + Timestamp::now(), + Identity::ZERO, + id, + args, + )) +} + +fn message_exists(db: &RelationalDB, tx: &MutTxId, msg: &PendingMessage) -> anyhow::Result { + Ok(db + .outbound_msg_by_stream_seq(tx, msg.stream_id.0, msg.sequence)? + .is_some()) +} + +fn current_row_matches(db: &RelationalDB, tx: &MutTxId, msg: &PendingMessage) -> anyhow::Result { + Ok(db + .outbox_row(tx, msg.outbox_table_id, msg.message_id)? + .is_some_and(|row| row == msg.outbox_row)) +} + +/// Repeated cleanup is a no-op. Only delete the row used for this delivery. +/// Leave replacement rows alone. +fn cleanup_message(db: &RelationalDB, tx: &mut MutTxId, delivered: &DeliveredMessage) -> anyhow::Result<()> { + let msg = &delivered.message; + if !message_exists(db, tx, msg)? { + return Ok(()); + } + if current_row_matches(db, tx, msg)? { + db.delete_by_rel(tx, msg.outbox_table_id, [msg.outbox_row.clone()]); + } + tx.ack_outbound_idc_msg(msg.stream_id.0, msg.sequence)?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::extract::{Path, Query}; + use axum::http::StatusCode; + use axum::routing::post; + use axum::Router; + use spacetimedb_datastore::system_tables::{ + StOutboundMsgRow, StOutboundStreamRow, ST_OUTBOUND_MSG_ID, ST_OUTBOUND_STREAM_ID, + }; + use spacetimedb_lib::db::auth::{StAccess, StTableType}; + use spacetimedb_primitives::ColId; + use spacetimedb_sats::bsatn::to_vec; + use spacetimedb_sats::{product, AlgebraicType}; + use spacetimedb_schema::def::BTreeAlgorithm; + use spacetimedb_schema::reducer_name::ReducerName; + use spacetimedb_schema::schema::{ColumnSchema, ConstraintSchema, IndexSchema, OutboxSchema, TableSchema}; + use spacetimedb_schema::table_name::TableName; + use spacetimedb_table::page_pool::PagePool; + use std::collections::HashMap; + use std::sync::atomic::Ordering; + use tokio::net::TcpListener; + + fn open_test_db() -> Arc { + let (db, _) = RelationalDB::open( + Identity::ZERO, + Identity::ZERO, + spacetimedb_durability::EmptyHistory::new(), + None, + None, + PagePool::new_for_test(), + ) + .expect("failed to open test database"); + Arc::new(db) + } + + fn outbox_schema() -> TableSchema { + TableSchema::new( + TableId::SENTINEL, + TableName::for_test("outbox"), + None, + vec![ + ColumnSchema::for_test(0, "target", AlgebraicType::identity()), + ColumnSchema::for_test(1, "msg_id", AlgebraicType::U64), + ColumnSchema::for_test(2, "value", AlgebraicType::U32), + ], + vec![IndexSchema::for_test("outbox_msg_id", BTreeAlgorithm::from(1))], + vec![ConstraintSchema::unique_for_test("outbox_msg_id_unique", 1)], + vec![], + StTableType::User, + StAccess::Public, + None, + Some(ColId(1)), + false, + None, + Some(OutboxSchema { + remote_reducer: ReducerName::for_test("receive_value"), + target_column: ColId(0), + arg_columns: vec![ColId(2)], + on_result_reducer: None, + signature_hash: spacetimedb_schema::def::reducer_signature_hash( + &spacetimedb_sats::ProductType::from([("value", AlgebraicType::U32)]), + &AlgebraicType::unit(), + &AlgebraicType::String, + &spacetimedb_sats::Typespace::default(), + ) + .to_hex() + .to_string(), + }), + ) + } + + fn insert_outbox_message(db: &RelationalDB, target: Identity, value: u32) -> anyhow::Result { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let table_id = db.create_table(&mut tx, outbox_schema())?; + db.commit_tx(tx)?; + insert_outbox_row(db, table_id, target, 7, value)?; + Ok(table_id) + } + + fn insert_outbox_row( + db: &RelationalDB, + table_id: TableId, + target: Identity, + msg_id: u64, + value: u32, + ) -> anyhow::Result<()> { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let row = to_vec(&product![target, msg_id, value])?; + let (row_ptr, insert_flags) = { + let (_, row_ref, insert_flags) = db.insert(&mut tx, table_id, &row)?; + (row_ref.pointer(), insert_flags) + }; + assert!(insert_flags.is_outbox_table); + tx.record_outbox_insert(table_id, row_ptr)?; + db.commit_tx(tx)?; + Ok(()) + } + + fn next_message(db: &RelationalDB) -> anyhow::Result { + let streams = read_pending_streams(db)?; + assert_eq!(streams.len(), 1); + let LoadedMessage::Pending(message) = read_next_message(db, streams[0], None)?.unwrap() else { + anyhow::bail!("expected a message without a result"); + }; + Ok(message) + } + + // These tests cover delivery and queue state without a module instance. + // They do not exercise callbacks, view refresh, or subscription updates. + fn cleanup_without_module(db: &RelationalDB, msg: DeliveredMessage) -> anyhow::Result<()> { + anyhow::ensure!( + msg.message.on_result_reducer.is_none(), + "test cleanup cannot run callbacks" + ); + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + cleanup_message(db, &mut tx, &msg)?; + db.commit_tx(tx)?; + Ok(()) + } + + fn store_success(db: &RelationalDB, message: PendingMessage) -> anyhow::Result { + let result = MessageResult { + status: StInboundMsgResultStatus::Ok, + payload: Bytes::new(), + }; + record_delivery_result(db, &message, &result)?; + Ok(DeliveredMessage { message, result }) + } + + #[test] + fn idc_actor_reads_after_cursor_before_acknowledgment() -> anyhow::Result<()> { + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + insert_outbox_row(&db, table, Identity::ONE, 8, 43)?; + let first = store_success(&db, next_message(&db)?)?; + let stream_id = first.message.stream_id; + + // Recovery still finds the first message's saved result until cleanup commits. + assert!(matches!( + read_next_message(&db, stream_id, None)?, + Some(LoadedMessage::Delivered(_)) + )); + // A caller can load the next message while the first awaits cleanup. + let second = read_next_message(&db, stream_id, Some(first.message.sequence))?.unwrap(); + assert!(matches!(second, LoadedMessage::Pending(_))); + assert_eq!(second.message().sequence, 2); + assert_eq!(second.message().ack_prefix, 0); + assert!(read_next_message(&db, stream_id, Some(2))?.is_none()); + assert!(read_next_message(&db, stream_id, Some(u64::MAX)).is_err()); + + cleanup_without_module(&db, first)?; + for cursor in [None, Some(1)] { + assert_eq!( + read_next_message(&db, stream_id, cursor)?.unwrap().message().sequence, + 2 + ); + } + let error = read_next_message(&db, stream_id, Some(0)) + .err() + .expect("a cursor behind acknowledgment must fail"); + assert!(error.to_string().contains("cursor 0 is behind acknowledgment 1")); + Ok(()) + } + + #[test] + fn idc_actor_only_delivers_committed_inserts() -> anyhow::Result<()> { + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + let first = store_success(&db, next_message(&db)?)?; + let (mut env, _runtime) = super::super::instance_env::test::instance_env(db.clone())?; + // Delivery must wait until the insert commits. + let (mut starter, sender) = channel(); + env.idc_sender = sender; + + let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let (tx, result) = env.tx.clone().set(tx, || { + env.insert(table, &mut to_vec(&product![Identity::ONE, 8u64, 43u32]).unwrap()) + }); + result?; + let _ = db.rollback_mut_tx(tx); + let msg = starter.rx.try_recv()?; + let stream_id = StreamId(msg.stream_id); + assert_eq!(read_next_message(&db, stream_id, None)?.unwrap().message().sequence, 1); + + // After cleaning up the first message, the rolled-back insert leaves no work. + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + cleanup_message(&db, &mut tx, &first)?; + db.commit_tx(tx)?; + assert!(read_next_message(&db, stream_id, None)?.is_none()); + + let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let (tx, result) = env.tx.clone().set(tx, || { + env.insert(table, &mut to_vec(&product![Identity::ONE, 8u64, 43u32]).unwrap()) + }); + result?; + db.commit_tx(tx)?; + let msg = starter.rx.try_recv()?; + assert_eq!(StreamId(msg.stream_id), stream_id); + let next = read_next_message(&db, stream_id, None)?.unwrap(); + assert_eq!(next.message().message_id, 8); + assert_eq!(next.message().sequence, 2); + let next = store_success(&db, next_message(&db)?)?; + // Cleanup can finish before an old notification is processed. + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + cleanup_message(&db, &mut tx, &next)?; + cleanup_message(&db, &mut tx, &next)?; + db.commit_tx(tx)?; + assert!(read_next_message(&db, stream_id, None)?.is_none()); + assert!(read_pending_streams(&db)?.is_empty()); + + // Rolling back the first insert of another stream leaves no stream to wake. + let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let (tx, result) = env.tx.clone().set(tx, || { + env.insert(table, &mut to_vec(&product![Identity::ZERO, 9u64, 44u32]).unwrap()) + }); + result?; + let _ = db.rollback_mut_tx(tx); + let msg = starter.rx.try_recv()?; + assert!(read_next_message(&db, StreamId(msg.stream_id), None)?.is_none()); + Ok(()) + } + + #[tokio::test] + async fn idc_actor_cleanup_preserves_replacements_and_rejects_gaps() -> anyhow::Result<()> { + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + let msg = store_success(&db, next_message(&db)?)?; + let replacement = product![Identity::ONE, 7u64, 99u32]; + + // Internal writes can replace a row. Cleanup of the old message must leave it alone. + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + db.delete_by_rel(&mut tx, table, [msg.message.outbox_row.clone()]); + db.insert(&mut tx, table, &to_vec(&replacement)?)?; + assert!(!current_row_matches(&db, &tx, &msg.message)?); + cleanup_message(&db, &mut tx, &msg)?; + cleanup_message(&db, &mut tx, &msg)?; + assert!(!message_exists(&db, &tx, &msg.message)?); + assert_eq!(db.iter_mut(&tx, table)?.next().unwrap().to_product_value(), replacement); + db.commit_tx(tx)?; + + // Neither a missing user row nor a missing pending row permits skipping a sequence. + for delete_pending in [false, true] { + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + insert_outbox_row(&db, table, Identity::ONE, 8, 43)?; + let first = next_message(&db)?; + let stream_id = first.stream_id; + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + if delete_pending { + let row = db + .outbound_msg_by_stream_seq(&tx, first.stream_id.0, first.sequence)? + .unwrap(); + let ptr = row.pointer(); + db.delete(&mut tx, ST_OUTBOUND_MSG_ID, [ptr]); + } else { + db.delete_by_rel(&mut tx, table, [first.outbox_row]); + } + db.commit_tx(tx)?; + let mut streams = DeliveryStreams::default(); + streams.wake(stream_id); + assert!(take_ready_message(&db, &mut streams).is_err()); + assert_eq!(streams.next_retry_delay(), None); + let tx = db.begin_tx(Workload::Internal); + assert_eq!(db.st_outbound_stream_by_id(&tx, stream_id.0)?.unwrap().ack_prefix, 0); + assert!(db.outbound_msg_by_stream_seq(&tx, stream_id.0, 2)?.is_some()); + let _ = db.release_tx(tx); + } + Ok(()) + } + + #[tokio::test] + async fn idc_actor_stores_complete_results_and_retries_transport_failures() -> anyhow::Result<()> { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let keys = crate::auth::JwtKeys::generate()?; + let response_status = Arc::new(std::sync::atomic::AtomicU16::new(200)); + let app = Router::new().route( + "/v1/database/:target_db/call-from-database/:reducer", + post({ + let response_status = response_status.clone(); + let public_key = keys.public; + move |Path((target, reducer)): Path<(String, String)>, + Query(query): Query>, + headers: axum::http::HeaderMap, + body: Bytes| { + let status = response_status.load(Ordering::Relaxed); + let public_key = public_key.clone(); + async move { + assert_eq!(target, Identity::ONE.to_hex().to_string()); + assert_eq!(reducer, "receive_value"); + let token = headers[axum::http::header::AUTHORIZATION].to_str().unwrap(); + assert_eq!( + crate::auth::idc::verify( + &public_key, + token.strip_prefix("Bearer ").unwrap(), + Identity::ONE + ) + .unwrap(), + Identity::ZERO + ); + assert_eq!(query["seq"], "1"); + assert_eq!(query["ack_prefix"], "0"); + assert_eq!(query["signature_hash"], outbox_schema().outbox.unwrap().signature_hash); + assert_eq!(body.as_ref(), to_vec(&42u32).unwrap()); + (StatusCode::from_u16(status).unwrap(), "receiver detail") + } + } + }), + ); + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let http_port = listener.local_addr()?.port(); + let server = tokio::spawn(async move { axum::serve(listener, app).await }); + let mut config = IdcConfig { + http_port, + token_signer: Arc::new(keys.private), + }; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(Duration::from_secs(2)) + .build()?; + + // Save and reload each result without changing its status or payload. + // Budget errors are stored results too. + for (http_status, stored_status) in [ + (200, StInboundMsgResultStatus::Ok), + (422, StInboundMsgResultStatus::Err), + (402, StInboundMsgResultStatus::BudgetExceeded), + ] { + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + insert_outbox_row(&db, table, Identity::ONE, 8, 43)?; + let msg = next_message(&db)?; + let seq = msg.sequence; + response_status.store(http_status, Ordering::Relaxed); + let delivered = deliver_message_and_record_result(&db, &client, Identity::ZERO, &config, msg) + .await? + .expect("terminal response must produce a delivered message"); + assert_eq!(delivered.result.status, stored_status); + assert_eq!(delivered.result.payload.as_ref(), b"receiver detail"); + + // A stored result does not advance the sequence until cleanup commits. + let stored = read_next_message(&db, delivered.message.stream_id, None)?.unwrap(); + let LoadedMessage::Delivered(stored) = stored else { + panic!("a stored result must load as a delivered message"); + }; + assert_eq!(stored.message.sequence, seq); + assert_eq!(stored.result, delivered.result); + { + let tx = db.begin_tx(Workload::Internal); + assert_eq!(db.iter(&tx, table)?.count(), 2); + assert_eq!( + db.st_outbound_stream_by_id(&tx, stored.message.stream_id.0)? + .unwrap() + .ack_prefix, + 0 + ); + } + // Cleanup commits the saved result without another delivery attempt. + cleanup_without_module(&db, stored)?; + assert_eq!(next_message(&db)?.sequence, 2); + } + + // Keep these messages pending and save the error. Do not acknowledge them + // or run on_result. + for status in [208, 401, 403, 404, 425, 500] { + let db = open_test_db(); + insert_outbox_message(&db, Identity::ONE, 42)?; + let msg = next_message(&db)?; + response_status.store(status, Ordering::Relaxed); + let error = deliver_message_and_record_result(&db, &client, Identity::ZERO, &config, msg) + .await + .err() + .expect("response must leave the message pending"); + let tx = db.begin_tx(Workload::Internal); + let row = StOutboundMsgRow::try_from(db.iter(&tx, ST_OUTBOUND_MSG_ID)?.next().unwrap())?; + assert!(row.result_status.is_none(), "HTTP {status}"); + assert_eq!(row.retry_count, 1); + let diagnostic = row.last_transport_error.unwrap(); + assert_eq!(diagnostic, format!("{error:#}")); + if status == 208 { + assert!(diagnostic.contains("receiver has retired the outcome"), "{diagnostic}"); + } else { + assert!(diagnostic.contains(&format!("HTTP {status}")), "{diagnostic}"); + assert!(diagnostic.contains("receiver detail"), "{diagnostic}"); + } + let stream = StOutboundStreamRow::try_from(db.iter(&tx, ST_OUTBOUND_STREAM_ID)?.next().unwrap())?; + assert_eq!(stream.ack_prefix, 0); + let _ = db.release_tx(tx); + } + // A late 208 must not repeat cleanup or the callback, or touch the next message. + let db = open_test_db(); + let table = insert_outbox_message(&db, Identity::ONE, 42)?; + insert_outbox_row(&db, table, Identity::ONE, 8, 43)?; + let stale = next_message(&db)?; + cleanup_without_module(&db, store_success(&db, stale.clone())?)?; + response_status.store(208, Ordering::Relaxed); + assert!( + deliver_message_and_record_result(&db, &client, Identity::ZERO, &config, stale) + .await? + .is_none() + ); + assert_eq!(next_message(&db)?.sequence, 2); + { + let tx = db.begin_tx(Workload::Internal); + assert_eq!(db.iter(&tx, table)?.count(), 1); + let row = StOutboundMsgRow::try_from(db.iter(&tx, ST_OUTBOUND_MSG_ID)?.next().unwrap())?; + assert_eq!(row.retry_count, 0); + assert!(row.result_status.is_none()); + } + server.abort(); + let _ = server.await; + // A terminal status alone is not enough. We need the whole response body. + for status in [200, 422, 402] { + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let http_port = listener.local_addr()?.port(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await?; + let mut request = [0; 4096]; + let received = socket.read(&mut request).await?; + anyhow::ensure!(received != 0, "client closed before sending a request"); + // Send the status, then close the connection before the body is complete. + socket + .write_all( + format!("HTTP/1.1 {status} Test\r\nContent-Length: 100\r\nConnection: close\r\n\r\npartial") + .as_bytes(), + ) + .await?; + socket.shutdown().await?; + anyhow::Ok(()) + }); + let db = open_test_db(); + insert_outbox_message(&db, Identity::ONE, 42)?; + let msg = next_message(&db)?; + config.http_port = http_port; + assert!( + deliver_message_and_record_result(&db, &client, Identity::ZERO, &config, msg) + .await + .is_err() + ); + let tx = db.begin_tx(Workload::Internal); + let row = StOutboundMsgRow::try_from(db.iter(&tx, ST_OUTBOUND_MSG_ID)?.next().unwrap())?; + assert!(row.result_status.is_none()); + assert!(row.last_transport_error.unwrap().contains("failed to read response")); + let stream = StOutboundStreamRow::try_from(db.iter(&tx, ST_OUTBOUND_STREAM_ID)?.next().unwrap())?; + assert_eq!(stream.ack_prefix, 0); + let _ = db.release_tx(tx); + server.await??; + } + + // A stored result requires both fields. Never resend or complete a partial result. + for (result_status, result_payload) in [(Some(StInboundMsgResultStatus::Ok), None), (None, Some(Bytes::new()))] + { + let db = open_test_db(); + insert_outbox_message(&db, Identity::ONE, 42)?; + let message = next_message(&db)?; + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let row = db + .outbound_msg_by_stream_seq(&tx, message.stream_id.0, message.sequence)? + .unwrap(); + let mut row = StOutboundMsgRow::try_from(row)?; + db.delete_by_rel(&mut tx, ST_OUTBOUND_MSG_ID, [row.clone().into()]); + row.result_status = result_status; + row.result_payload = result_payload; + db.insert(&mut tx, ST_OUTBOUND_MSG_ID, &to_vec(&row)?)?; + db.commit_tx(tx)?; + let error = read_next_message(&db, message.stream_id, None) + .err() + .expect("partial result must fail"); + assert!(error.to_string().contains("incomplete stored result")); + } + Ok(()) + } +} diff --git a/crates/core/src/host/instance_env.rs b/crates/core/src/host/instance_env.rs index ceb096ac192..93c746f3f5a 100644 --- a/crates/core/src/host/instance_env.rs +++ b/crates/core/src/host/instance_env.rs @@ -1,3 +1,4 @@ +use super::idc_actor::IdcActorSender; use super::scheduler::{get_schedule_from_row, scheduled_row_hash, ScheduleError, Scheduler}; use crate::database_logger::{BacktraceFrame, BacktraceProvider, LogLevel, ModuleBacktrace, Record}; use crate::db::relational_db::{MutTx, RelationalDB}; @@ -44,6 +45,7 @@ use std::vec::IntoIter; pub struct InstanceEnv { pub replica_ctx: Arc, pub scheduler: Scheduler, + pub idc_sender: IdcActorSender, pub tx: TxSlot, /// The timestamp the current function began running. pub start_time: Timestamp, @@ -233,10 +235,11 @@ impl ChunkedWriter { // Generic 'instance environment' delegated to from various host types. impl InstanceEnv { - pub fn new(replica_ctx: Arc, scheduler: Scheduler) -> Self { + pub fn new(replica_ctx: Arc, scheduler: Scheduler, idc_sender: IdcActorSender) -> Self { Self { replica_ctx, scheduler, + idc_sender, tx: TxSlot::default(), start_time: Timestamp::now(), start_instant: Instant::now(), @@ -484,7 +487,9 @@ impl InstanceEnv { self.schedule_row(stdb, tx, table_id, row_ptr)?; } if insert_flags.is_outbox_table { - tx.record_outbox_insert(table_id, row_ptr).map_err(DBError::from)?; + let msg = tx.record_outbox_insert(table_id, row_ptr).map_err(DBError::from)?; + // The actor rechecks committed state, so a rolled-back insert is harmless. + let _ = self.idc_sender.send(msg); } } @@ -566,13 +571,8 @@ impl InstanceEnv { }, )?; - if update_flags.has_post_write_work() { - if update_flags.is_scheduler_table { - self.schedule_row(stdb, tx, table_id, row_ptr)?; - } - if update_flags.is_outbox_table { - tx.record_outbox_insert(table_id, row_ptr).map_err(DBError::from)?; - } + if update_flags.is_scheduler_table { + self.schedule_row(stdb, tx, table_id, row_ptr)?; } tx.metrics.bytes_written += buffer.len(); tx.metrics.rows_updated += 1; @@ -1494,7 +1494,7 @@ impl Display for GetTxError { impl std::error::Error for GetTxError {} #[cfg(test)] -mod test { +pub(super) mod test { use super::*; use std::{ops::Bound, sync::Arc}; @@ -1555,10 +1555,14 @@ mod test { } /// An `InstanceEnv` used for testing the database syscalls. - fn instance_env(db: Arc) -> Result<(InstanceEnv, tokio::runtime::Runtime)> { + pub(in crate::host) fn instance_env(db: Arc) -> Result<(InstanceEnv, tokio::runtime::Runtime)> { let (scheduler, _) = Scheduler::open(db.clone()); + let (_, idc_sender) = crate::host::idc_actor::channel(); let (replica_context, runtime) = replica_ctx(db, crate::config::ModuleHttpConfig::default())?; - Ok((InstanceEnv::new(Arc::new(replica_context), scheduler), runtime)) + Ok(( + InstanceEnv::new(Arc::new(replica_context), scheduler, idc_sender), + runtime, + )) } struct PendingDurability(tokio::sync::watch::Receiver>); @@ -1652,8 +1656,9 @@ mod test { fn module_http_config_disables_requests() -> Result<()> { let db = relational_db()?; let (scheduler, _) = Scheduler::open(db.clone()); + let (_, idc_sender) = crate::host::idc_actor::channel(); let (replica_context, _runtime) = replica_ctx(db, crate::config::ModuleHttpConfig { enabled: false })?; - let mut env = InstanceEnv::new(Arc::new(replica_context), scheduler); + let mut env = InstanceEnv::new(Arc::new(replica_context), scheduler, idc_sender); let request = st_http::Request { method: st_http::Method::Get, headers: std::iter::empty().collect(), diff --git a/crates/core/src/host/mod.rs b/crates/core/src/host/mod.rs index c3eb7936d47..635a15df4b8 100644 --- a/crates/core/src/host/mod.rs +++ b/crates/core/src/host/mod.rs @@ -12,6 +12,8 @@ use spacetimedb_schema::def::ModuleDef; mod disk_storage; mod host_controller; +mod idc; +pub mod idc_actor; mod module_common; #[allow(clippy::too_many_arguments)] pub mod module_host; @@ -29,6 +31,7 @@ pub use host_controller::{ HostController, HostRuntimeConfig, InitialEnvironmentSource, MigratePlanResult, ModuleHostWithBootstrap, ProcedureCallResult, ProgramStorage, ReducerCallResult, ReducerCallResultWithTxOffset, ReducerOutcome, }; +pub use idc_actor::IdcConfig; pub use module_host::{ IdcDeliveryFailure, IdcReducerCallOutcome, InitDatabaseResult, ModuleHost, NoSuchModule, ProcedureCallError, ReducerCallError, UpdateDatabaseResult, diff --git a/crates/core/src/host/module_common.rs b/crates/core/src/host/module_common.rs index 49f1126b54b..b1e1fda6e32 100644 --- a/crates/core/src/host/module_common.rs +++ b/crates/core/src/host/module_common.rs @@ -4,6 +4,7 @@ use crate::{ energy::EnergyMonitor, host::{ + idc_actor::IdcActorSender, module_host::ModuleInfo, wasm_common::{module_host_actor::DescribeError, DESCRIBE_MODULE_DUNDER}, Scheduler, @@ -37,7 +38,13 @@ pub fn build_common_module_from_raw( replica_ctx.subscriptions.clone(), ); - Ok(ModuleCommon::new(replica_ctx, mcc.scheduler, info, mcc.energy_monitor)) + Ok(ModuleCommon::new( + replica_ctx, + mcc.scheduler, + mcc.idc_sender, + info, + mcc.energy_monitor, + )) } /// Non-runtime-specific parts of a module. @@ -45,6 +52,7 @@ pub fn build_common_module_from_raw( pub(crate) struct ModuleCommon { replica_context: Arc, scheduler: Scheduler, + idc_sender: IdcActorSender, info: Arc, energy_monitor: Arc, } @@ -54,12 +62,14 @@ impl ModuleCommon { fn new( replica_context: Arc, scheduler: Scheduler, + idc_sender: IdcActorSender, info: Arc, energy_monitor: Arc, ) -> Self { Self { replica_context, scheduler, + idc_sender, info, energy_monitor, } @@ -84,6 +94,10 @@ impl ModuleCommon { pub fn scheduler(&self) -> &Scheduler { &self.scheduler } + + pub fn idc_sender(&self) -> IdcActorSender { + self.idc_sender.clone() + } } /// Runs the describer of modules in `run` and does some logging around it. diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index e39e0de5e26..527fffcf219 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -2905,6 +2905,19 @@ impl ModuleHost { Ok(()) } + pub(super) async fn run_idc_callback_and_cleanup( + &self, + msg: super::idc_actor::DeliveredMessage, + ) -> anyhow::Result<()> { + call_instance!( + self, + "IDC result", + msg, + |msg, inst| inst.run_idc_callback_and_cleanup(msg), + |msg, inst| inst.run_idc_callback_and_cleanup(msg).await, + )? + } + /// Processes an ordered IDC message and returns its recorded outcome. /// /// New messages are checked against the current module before running the reducer. @@ -3814,11 +3827,13 @@ impl ModuleHost { // As in `Self::marked_closed`, `Relaxed` is sufficient because we're not synchronizing any external state. self.closed.store(true, std::sync::atomic::Ordering::Relaxed); self.scheduler().close(); + self.idc_sender().close(); self.exited().await; } pub async fn exited(&self) { - self.scheduler().closed().await; + let idc_sender = self.idc_sender(); + tokio::join!(self.scheduler().closed(), idc_sender.closed()); } pub fn inject_logs(&self, log_level: LogLevel, function_name: &str, message: &str) { @@ -4222,6 +4237,13 @@ impl ModuleHost { ModuleHostInner::Js(js) => js.module.scheduler(), } } + + fn idc_sender(&self) -> super::idc_actor::IdcActorSender { + match &*self.inner { + ModuleHostInner::Wasm(wasm) => wasm.module.idc_sender(), + ModuleHostInner::Js(js) => js.module.idc_sender(), + } + } } impl WeakModuleHost { diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index a2200c8d31a..5e39b3df5b7 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -1,6 +1,4 @@ -use super::module_host::{ - CallReducerParams, DatabaseUpdate, EventStatus, ModuleEvent, ModuleFunctionCall, WeakModuleHost, -}; +use super::module_host::{CallReducerParams, WeakModuleHost}; use super::{FunctionArgs, ModuleHost}; use crate::db::relational_db::RelationalDB; use crate::host::module_host::{CallProcedureParams, ModuleInfo}; @@ -13,7 +11,6 @@ use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{FutureExt, StreamExt}; use prometheus::IntGauge; -use spacetimedb_client_api_messages::energy::FunctionBudget; use spacetimedb_datastore::execution_context::{ExecutionContext, ReducerContext, Workload}; use spacetimedb_datastore::locking_tx_datastore::MutTxId; use spacetimedb_datastore::system_tables::{StScheduledFields, ST_SCHEDULED_ID}; @@ -604,7 +601,6 @@ fn prepare_scheduled_procedure_call( log::error!("could not determine scheduled procedure or its parameters: {err:#}"); let reschedule = id.zip(invocation.as_ref()).and_then(|(id, invocation)| { delete_scheduled_function_row( - module_info, db, (id, invocation.row_hash), Some(tx), @@ -625,7 +621,6 @@ fn prepare_scheduled_procedure_call( // so we must remove the schedule row before executing. let reschedule = id.zip(invocation.as_ref()).and_then(|(id, invocation)| { delete_scheduled_function_row( - module_info, db, (id, invocation.row_hash), Some(tx), @@ -674,7 +669,6 @@ fn call_scheduled_reducer_until_done( log::error!("could not determine scheduled reducer or its parameters: {err:#}"); let reschedule = id.zip(invocation.as_ref()).and_then(|(id, invocation)| { delete_scheduled_function_row( - module_info, db, (id, invocation.row_hash), Some(tx), @@ -780,15 +774,7 @@ fn call_scheduled_reducer_with_tx( inst_common.call_reducer_with_tx(Some(tx), params, inst) })); let reschedule = scheduled.and_then(|(id, row_hash)| { - delete_scheduled_function_row( - module_info, - db, - (id, row_hash), - None, - reschedule_from, - inst_common, - inst, - ) + delete_scheduled_function_row(db, (id, row_hash), None, reschedule_from, inst_common, inst) }); // Currently, we drop the return value from the function call. In the future, // we might want to handle it somehow. @@ -806,7 +792,6 @@ fn call_scheduled_reducer_with_tx( /// One-shot schedules are deleted and committed immediately. Interval schedules are not /// deleted here; instead their `ScheduleAt` is returned to the caller as a `Reschedule`. fn delete_scheduled_function_row( - module_info: &ModuleInfo, db: &RelationalDB, (id, row_hash): (ScheduledFunctionId, Hash), tx: Option, @@ -815,7 +800,7 @@ fn delete_scheduled_function_row( inst: &mut impl WasmInstance, ) -> Option { let tx = tx.unwrap_or_else(|| db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal)); - let schedule_at = delete_scheduled_function_row_with_tx(module_info, db, tx, id, row_hash, inst_common, inst)?; + let schedule_at = delete_scheduled_function_row_with_tx(db, tx, id, row_hash, inst_common, inst)?; let ScheduleAt::Interval(dur) = schedule_at else { return None; }; @@ -865,7 +850,6 @@ fn next_interval_tick_after(last_intended_at: Timestamp, interval: TimeDuration, /// /// Interval schedules are left in place and returned to the caller for rescheduling. fn delete_scheduled_function_row_with_tx( - module_info: &ModuleInfo, db: &RelationalDB, mut tx: MutTxId, id: ScheduledFunctionId, @@ -902,49 +886,11 @@ fn delete_scheduled_function_row_with_tx( let row_ptr = schedule_row.pointer(); db.delete(&mut tx, id.table_id, [row_ptr]); - refresh_views_then_commit_and_broadcast(tx, module_info, inst_common, inst); - None -} - -/// Refreshes stale views, commits transaction, and broadcasts subscription updates. -fn refresh_views_then_commit_and_broadcast( - tx: MutTxId, - module_info: &ModuleInfo, - inst_common: &mut InstanceCommon, - inst: &mut impl WasmInstance, -) { - let timestamp = Timestamp::now(); - let (view_result, _n_evaluated, trapped) = - inst_common.call_views_with_tx(tx, module_info.database_identity, inst, timestamp); - let mut status = match view_result.outcome { - crate::host::module_host::ViewOutcome::Success => EventStatus::Committed(DatabaseUpdate::default()), - crate::host::module_host::ViewOutcome::Failed(err) => EventStatus::FailedInternal(err), - crate::host::module_host::ViewOutcome::BudgetExceeded => EventStatus::OutOfEnergy, - }; - if trapped && matches!(status, EventStatus::Committed(_)) { - status = EventStatus::FailedInternal("The instance encountered a fatal error.".into()); - } - - let event = ModuleEvent { - timestamp, - caller_identity: module_info.database_identity, - caller_connection_id: None, - function_call: ModuleFunctionCall::default(), - status, - reducer_return_value: None, - //Keeping them 0 as it is internal transaction, not by reducer - execution_budget_used: FunctionBudget::ZERO, - host_execution_duration: Duration::from_millis(0), - request_id: None, - timer: None, - }; - - if let Err(e) = module_info - .subscriptions - .commit_and_broadcast_event(None, event, view_result.tx) - { - log::error!("Failed to broadcast deletion event: {e:#}"); + let (result, _) = inst_common.refresh_views_then_commit_and_broadcast(tx, inst); + if let Err(e) = result { + log::error!("Failed to commit scheduled-row cleanup: {e:#}"); } + None } fn reducer_call_params_for_queued_item( diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs index 8f3673e0c77..746c2469e11 100644 --- a/crates/core/src/host/v8/mod.rs +++ b/crates/core/src/host/v8/mod.rs @@ -295,6 +295,10 @@ impl JsModule { self.common.scheduler() } + pub fn idc_sender(&self) -> super::idc_actor::IdcActorSender { + self.common.idc_sender() + } + pub fn info(&self) -> Arc { self.common.info().clone() } @@ -528,6 +532,13 @@ impl JsMainInstance { self.request(ScheduledReducerRequest { params }).await } + pub(in crate::host) async fn run_idc_callback_and_cleanup( + &self, + msg: crate::host::idc_actor::DeliveredMessage, + ) -> anyhow::Result<()> { + self.request(RunIdcCallbackAndCleanupRequest { msg }).await + } + pub(in crate::host) async fn enqueue_reducer(&self, params: CallReducerParams, on_panic: JsFatalHook) { self.send_detached_request( "call_reducer", @@ -687,6 +698,12 @@ js_main_request! { } => "call_idc_reducer", Result, CallIdcReducer } +js_main_request! { + RunIdcCallbackAndCleanupRequest { + msg: crate::host::idc_actor::DeliveredMessage, + } => "run_idc_callback_and_cleanup", anyhow::Result<()>, RunIdcCallbackAndCleanup +} + js_main_request! { ClearAllClientsRequest => "clear_all_clients", anyhow::Result<()>, ClearAllClients } @@ -886,6 +903,10 @@ enum JsMainWorkerRequest { reply_tx: JsReplyTx, params: ScheduledFunctionParams, }, + RunIdcCallbackAndCleanup { + reply_tx: JsReplyTx>, + msg: crate::host::idc_actor::DeliveredMessage, + }, /// See [`JsMainInstance::call_view`]. CallView { reply_tx: JsReplyTx, @@ -1503,6 +1524,11 @@ fn handle_main_worker_request( (res, trapped) }) } + JsMainWorkerRequest::RunIdcCallbackAndCleanup { reply_tx, msg } => { + handle_worker_request("run_idc_callback_and_cleanup", reply_tx, || { + instance_common.run_idc_callback_and_cleanup(msg, inst) + }) + } JsMainWorkerRequest::CallView { reply_tx, cmd } => handle_worker_request("call_view", reply_tx, || { let (res, trapped) = instance_common.handle_cmd(cmd, inst); (res, trapped) @@ -1715,7 +1741,11 @@ where Either::Left(module) => (module.replica_ctx(), module.scheduler()), Either::Right(mcc) => (&mcc.replica_ctx, &mcc.scheduler), }; - let instance_env = InstanceEnv::new(replica_ctx.clone(), scheduler.clone()); + let idc_sender = match &generation_module_or_mcc { + Either::Left(module) => module.idc_sender(), + Either::Right(mcc) => mcc.idc_sender.clone(), + }; + let instance_env = InstanceEnv::new(replica_ctx.clone(), scheduler.clone(), idc_sender); scope.set_slot(JsInstanceEnv::new(instance_env)); let startup_result = panic::catch_unwind(AssertUnwindSafe(|| { 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 2e52679ad1d..317a9581e01 100644 --- a/crates/core/src/host/wasm_common/module_host_actor.rs +++ b/crates/core/src/host/wasm_common/module_host_actor.rs @@ -415,7 +415,7 @@ impl WasmModuleHostActor { func_names }; let uninit_instance = module.instantiate_pre()?; - let instance_env = InstanceEnv::new(mcc.replica_ctx.clone(), mcc.scheduler.clone()); + let instance_env = InstanceEnv::new(mcc.replica_ctx.clone(), mcc.scheduler.clone(), mcc.idc_sender.clone()); let mut instance = uninit_instance.instantiate(instance_env, &func_names)?; let desc = instance.extract_descriptions()?; @@ -465,13 +465,21 @@ impl WasmModuleHostActor { self.common.scheduler() } + pub fn idc_sender(&self) -> crate::host::idc_actor::IdcActorSender { + self.common.idc_sender() + } + pub fn info(&self) -> Arc { self.common.info() } pub fn create_instance(&self) -> WasmModuleInstance { let common = &self.common; - let env = InstanceEnv::new(common.replica_ctx().clone(), common.scheduler().clone()); + let env = InstanceEnv::new( + common.replica_ctx().clone(), + common.scheduler().clone(), + common.idc_sender(), + ); // this shouldn't fail, since we already called module.create_instance() // before and it didn't error, and ideally they should be deterministic let mut instance = self @@ -635,6 +643,15 @@ impl WasmModuleInstance { self.trapped = trapped; res } + + pub(in crate::host) fn run_idc_callback_and_cleanup( + &mut self, + msg: crate::host::idc_actor::DeliveredMessage, + ) -> anyhow::Result<()> { + let (result, trapped) = self.common.run_idc_callback_and_cleanup(msg, &mut self.instance); + self.trapped = trapped; + result + } } impl WasmModuleInstance { @@ -1091,7 +1108,7 @@ impl InstanceCommon { params: CallReducerParams, inst: &mut I, ) -> (ReducerCallResultWithTxOffset, bool) { - self.call_reducer_with_tx_inner(tx, params, inst, None) + self.call_reducer_with_tx_inner(tx, params, inst, None::) } pub(crate) fn call_reducer_with_success_action( @@ -1099,7 +1116,7 @@ impl InstanceCommon { tx: Option, params: CallReducerParams, inst: &mut I, - on_success: ReducerSuccessAction, + on_success: impl FnOnce(&mut MutTxId, &Option) -> anyhow::Result<()>, ) -> (ReducerCallResultWithTxOffset, bool) { self.call_reducer_with_tx_inner(tx, params, inst, Some(on_success)) } @@ -1163,7 +1180,7 @@ impl InstanceCommon { tx: Option, params: CallReducerParams, inst: &mut I, - on_success: Option, + on_success: Option) -> anyhow::Result<()>>, ) -> (ReducerCallResultWithTxOffset, bool) { let CallReducerParams { timestamp, @@ -1678,6 +1695,55 @@ impl InstanceCommon { ModuleHost::call_views_with_tx_at(tx, &mut instance, caller, timestamp) } + /// Refresh views for an internal transaction, then commit and broadcast its changes. + /// A failed refresh rolls back the transaction. The boolean reports whether the instance trapped. + pub(crate) fn refresh_views_then_commit_and_broadcast( + &mut self, + tx: MutTxId, + inst: &mut impl WasmInstance, + ) -> (anyhow::Result<()>, bool) { + let timestamp = Timestamp::now(); + let (view_result, _n_evaluated, trapped) = + self.call_views_with_tx(tx, self.info.database_identity, inst, timestamp); + let mut status = match view_result.outcome { + ViewOutcome::Success => EventStatus::Committed(DatabaseUpdate::default()), + ViewOutcome::Failed(err) => EventStatus::FailedInternal(err), + ViewOutcome::BudgetExceeded => EventStatus::OutOfEnergy, + }; + if trapped && matches!(status, EventStatus::Committed(_)) { + status = EventStatus::FailedInternal("The instance encountered a fatal error.".into()); + } + + // The commit helper returns Ok after rolling back a failed event. + // Preserve the view failure so callers do not mistake rollback for a successful commit. + let outcome = match &status { + EventStatus::Committed(_) => Ok(()), + status => Err(anyhow!("view refresh failed: {status:?}")), + }; + let event = ModuleEvent { + timestamp, + caller_identity: self.info.database_identity, + caller_connection_id: None, + function_call: ModuleFunctionCall::default(), + status, + reducer_return_value: None, + // No reducer ran for this internal transaction. + execution_budget_used: FunctionBudget::ZERO, + host_execution_duration: Duration::ZERO, + request_id: None, + timer: None, + }; + + let result = self + .info + .subscriptions + .commit_and_broadcast_event(None, event, view_result.tx) + .map(|_| ()) + .map_err(anyhow::Error::from) + .and(outcome); + (result, trapped) + } + /// Executes view calls and accumulate results. /// Returns early if any call traps or fails. fn execute_view_calls( @@ -1731,6 +1797,14 @@ impl InstanceCommon { ) -> (CallScheduledFunctionResult, bool) { crate::host::scheduler::call_scheduled_reducer(&self.info.clone(), params, self, inst) } + + pub(in crate::host) fn run_idc_callback_and_cleanup( + &mut self, + msg: crate::host::idc_actor::DeliveredMessage, + inst: &mut I, + ) -> (anyhow::Result<()>, bool) { + crate::host::idc_actor::run_callback_and_cleanup(&self.info.clone(), msg, self, inst) + } } fn collect_subscribed_view_calls( diff --git a/crates/core/src/module_host_context.rs b/crates/core/src/module_host_context.rs index 50f8c258a37..cd8293fb78c 100644 --- a/crates/core/src/module_host_context.rs +++ b/crates/core/src/module_host_context.rs @@ -1,4 +1,5 @@ use crate::energy::EnergyMonitor; +use crate::host::idc_actor::IdcActorSender; use crate::host::scheduler::Scheduler; use crate::replica_context::ReplicaContext; use spacetimedb_sats::hash::Hash; @@ -7,6 +8,7 @@ use std::sync::Arc; pub struct ModuleCreationContext { pub replica_ctx: Arc, pub scheduler: Scheduler, + pub idc_sender: IdcActorSender, pub program_hash: Hash, pub energy_monitor: Arc, } diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs index 79d9c79c708..df73b469ba6 100644 --- a/crates/datastore/src/locking_tx_datastore/datastore.rs +++ b/crates/datastore/src/locking_tx_datastore/datastore.rs @@ -3125,6 +3125,7 @@ pub(crate) mod tests { assert_eq!(stream_rows[0].next_seq, 3); assert_eq!(stream_rows[0].ack_prefix, 0); + let stream_id = stream_rows[0].stream_id; let mut outbound_rows = tx .iter(ST_OUTBOUND_MSG_ID)? .map(StOutboundMsgRow::try_from) @@ -3136,6 +3137,64 @@ pub(crate) mod tests { vec![10, 11] ); + // Reads in a write transaction include uncommitted rows and isolate each stream. + let row = to_vec(&product![20u64, Identity::ZERO]).unwrap(); + let row_ptr = datastore.insert_mut_tx(&mut tx, table_id, &row)?.1.pointer(); + let other = tx.record_outbox_insert(table_id, row_ptr)?; + assert_ne!(stream_id, other.stream_id); + assert_eq!( + StOutboundMsgRow::try_from(tx.outbound_msg_by_stream_seq(stream_id, 1)?.unwrap())?.msg_id, + 10 + ); + assert_eq!( + StOutboundMsgRow::try_from(tx.outbound_msg_by_stream_seq(other.stream_id, 1)?.unwrap())?.msg_id, + 20 + ); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 3)?.is_none()); + + // The same helpers work after commit through a read-only transaction. + datastore.commit_mut_tx(tx)?; + let tx = begin_tx(&datastore); + assert_eq!(tx.st_outbound_stream_by_id(stream_id)?.unwrap().next_seq, 3); + assert_eq!( + StOutboundMsgRow::try_from(tx.outbound_msg_by_stream_seq(stream_id, 2)?.unwrap())?.msg_id, + 11 + ); + drop(tx); + + let mut tx = begin_mut_tx(&datastore); + tx.record_outbound_idc_transport_error(stream_id, 1, "connection reset".into())?; + let row = StOutboundMsgRow::try_from(tx.outbound_msg_by_stream_seq(stream_id, 1)?.unwrap())?; + assert_eq!(row.retry_count, 1); + assert_eq!(row.last_transport_error.as_deref(), Some("connection reset")); + let payload = Bytes::from_static(b"reducer failed"); + tx.record_outbound_idc_result(stream_id, 1, StInboundMsgResultStatus::Err, payload.clone())?; + let row = StOutboundMsgRow::try_from(tx.outbound_msg_by_stream_seq(stream_id, 1)?.unwrap())?; + assert_eq!(row.result_status, Some(StInboundMsgResultStatus::Err)); + assert_eq!(row.result_payload, Some(payload)); + assert_eq!(row.last_transport_error, None); + + // Reject out-of-order acknowledgments before changing either the stream or its messages. + let stream_before = tx.st_outbound_stream_by_id(stream_id)?; + assert!(tx.ack_outbound_idc_msg(stream_id, 2).is_err()); + assert_eq!(tx.st_outbound_stream_by_id(stream_id)?, stream_before); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 1)?.is_some()); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 2)?.is_some()); + + // Repeated acknowledgments leave later messages and other streams alone. + let mut expected_stream = tx.st_outbound_stream_by_id(stream_id)?.unwrap(); + expected_stream.ack_prefix = 1; + tx.ack_outbound_idc_msg(stream_id, 1)?; + tx.ack_outbound_idc_msg(stream_id, 1)?; + assert_eq!(tx.st_outbound_stream_by_id(stream_id)?, Some(expected_stream)); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 1)?.is_none()); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 2)?.is_some()); + assert!(tx.outbound_msg_by_stream_seq(other.stream_id, 1)?.is_some()); + + tx.ack_outbound_idc_msg(stream_id, 2)?; + assert_eq!(tx.st_outbound_stream_by_id(stream_id)?.unwrap().ack_prefix, 2); + assert!(tx.outbound_msg_by_stream_seq(stream_id, 2)?.is_none()); + Ok(()) } diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index d06a0366289..95023538477 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -3634,17 +3634,17 @@ impl MutTxId { .transpose() } - /// Return whether this outbox row is already queued for delivery. - fn st_outbound_msg_exists(&self, stream_id: u64, msg_id: u64) -> Result { + /// Find the pending delivery for this outbox row. + fn st_outbound_msg(&self, stream_id: u64, msg_id: u64) -> Result> { let msg_key = AlgebraicValue::Product(product![stream_id, msg_id]); - Ok(self - .iter_by_col_eq( - ST_OUTBOUND_MSG_ID, - col_list![StOutboundMsgFields::StreamId, StOutboundMsgFields::MsgId], - &msg_key, - )? - .next() - .is_some()) + self.iter_by_col_eq( + ST_OUTBOUND_MSG_ID, + col_list![StOutboundMsgFields::StreamId, StOutboundMsgFields::MsgId], + &msg_key, + )? + .map(StOutboundMsgRow::try_from) + .next() + .transpose() } /// Reserve the next sequence number for this outbox table and receiver. @@ -3682,19 +3682,74 @@ impl MutTxId { } /// Retain the queued message until the IDC actor sends it and observes the result. - fn insert_st_outbound_msg(&mut self, stream_id: u64, msg_id: u64, seq: u64) -> Result<()> { - self.insert_via_serialize_bsatn( - ST_OUTBOUND_MSG_ID, - &StOutboundMsgRow { - stream_id, - msg_id, - seq, - retry_count: 0, - last_transport_error: None, - result_status: None, - result_payload: None, - }, - )?; + fn insert_st_outbound_msg(&mut self, stream_id: u64, msg_id: u64, seq: u64) -> Result { + let row = StOutboundMsgRow { + stream_id, + msg_id, + seq, + retry_count: 0, + last_transport_error: None, + result_status: None, + result_payload: None, + }; + self.insert_via_serialize_bsatn(ST_OUTBOUND_MSG_ID, &row)?; + Ok(row) + } + + pub fn record_outbound_idc_transport_error(&mut self, stream_id: u64, seq: u64, reason: String) -> Result<()> { + let Some(row) = self.outbound_msg_by_stream_seq(stream_id, seq)? else { + return Ok(()); + }; + let ptr = row.pointer(); + let mut current = StOutboundMsgRow::try_from(row)?; + self.delete(ST_OUTBOUND_MSG_ID, ptr)?; + current.retry_count = current.retry_count.saturating_add(1); + current.last_transport_error = Some(reason); + self.insert_via_serialize_bsatn(ST_OUTBOUND_MSG_ID, ¤t)?; + Ok(()) + } + + pub fn record_outbound_idc_result( + &mut self, + stream_id: u64, + seq: u64, + result_status: StInboundMsgResultStatus, + result_payload: Bytes, + ) -> Result<()> { + let Some(row) = self.outbound_msg_by_stream_seq(stream_id, seq)? else { + return Ok(()); + }; + let ptr = row.pointer(); + let mut current = StOutboundMsgRow::try_from(row)?; + self.delete(ST_OUTBOUND_MSG_ID, ptr)?; + current.last_transport_error = None; + current.result_status = Some(result_status); + current.result_payload = Some(result_payload); + self.insert_via_serialize_bsatn(ST_OUTBOUND_MSG_ID, ¤t)?; + Ok(()) + } + + /// Retire the next consecutive message. Repeated acknowledgments are a no-op. + pub fn ack_outbound_idc_msg(&mut self, stream_id: u64, seq: u64) -> Result<()> { + let Some(row) = self.outbound_msg_by_stream_seq(stream_id, seq)? else { + return Ok(()); + }; + let ptr = row.pointer(); + let mut stream = self + .st_outbound_stream_by_id(stream_id)? + .ok_or_else(|| anyhow!("pending IDC message has no outbound stream {stream_id}"))?; + if stream.ack_prefix.checked_add(1) != Some(seq) { + return Err(anyhow!( + "IDC stream {stream_id} cannot acknowledge sequence {seq} after acknowledgment {}", + stream.ack_prefix + ) + .into()); + } + + self.delete(ST_OUTBOUND_MSG_ID, ptr)?; + self.delete_by_row_value(ST_OUTBOUND_STREAM_ID, &stream.clone().into())?; + stream.ack_prefix = seq; + self.insert_via_serialize_bsatn(ST_OUTBOUND_STREAM_ID, &stream)?; Ok(()) } @@ -3702,18 +3757,19 @@ impl MutTxId { /// /// Module code writes only the user outbox table. This mirrors that insert into /// system tables so the host IDC actor can send messages outside the reducer tx. + /// Returns the pending message for notifying the actor without another index lookup. #[cold] #[inline(never)] - pub fn record_outbox_insert(&mut self, outbox_table_id: TableId, row_ptr: RowPointer) -> Result<()> { + pub fn record_outbox_insert(&mut self, outbox_table_id: TableId, row_ptr: RowPointer) -> Result { let (msg_id, target_identity) = self.read_outbox_msg_id_and_target_identity(outbox_table_id, row_ptr)?; let stream = self.st_outbound_stream_row(outbox_table_id, target_identity)?; // Check before advancing the stream so idempotent calls do not burn sequence numbers // or duplicate sends. if let Some(stream) = &stream - && self.st_outbound_msg_exists(stream.stream_id, msg_id)? + && let Some(row) = self.st_outbound_msg(stream.stream_id, msg_id)? { - return Ok(()); + return Ok(row); } let (stream_id, seq) = self.advance_st_outbound_stream(outbox_table_id, target_identity, stream)?; self.insert_st_outbound_msg(stream_id, msg_id, seq) diff --git a/crates/datastore/src/locking_tx_datastore/state_view.rs b/crates/datastore/src/locking_tx_datastore/state_view.rs index cc557763ef8..5192f080624 100644 --- a/crates/datastore/src/locking_tx_datastore/state_view.rs +++ b/crates/datastore/src/locking_tx_datastore/state_view.rs @@ -5,17 +5,18 @@ use crate::locking_tx_datastore::mut_tx::{IndexScanPoint, IndexScanRanged}; use crate::system_tables::{ ConnectionIdViaU128, StColumnAccessorFields, StColumnAccessorRow, StColumnFields, StColumnRow, StConnectionCredentialsFields, StConnectionCredentialsRow, StConstraintFields, StConstraintRow, StEventTableFields, - StEventTableRow, StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow, StOutboxFields, StOutboxRow, - StScheduledFields, StScheduledRow, StSequenceFields, StSequenceRow, StTableAccessorFields, StTableAccessorRow, - StTableFields, StTableRow, StViewFields, StViewRow, SystemTable, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ID, - ST_CONNECTION_CREDENTIALS_ID, ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INDEX_ACCESSOR_ID, ST_INDEX_ID, ST_OUTBOX_ID, - ST_SCHEDULED_ID, ST_SEQUENCE_ID, ST_TABLE_ACCESSOR_ID, ST_TABLE_ID, ST_VIEW_ID, + StEventTableRow, StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow, StOutboundMsgFields, + StOutboundStreamFields, StOutboundStreamRow, StOutboxFields, StOutboxRow, StScheduledFields, StScheduledRow, + StSequenceFields, StSequenceRow, StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow, + StViewFields, StViewRow, SystemTable, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ID, ST_CONNECTION_CREDENTIALS_ID, + ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INDEX_ACCESSOR_ID, ST_INDEX_ID, ST_OUTBOUND_MSG_ID, ST_OUTBOUND_STREAM_ID, + ST_OUTBOX_ID, ST_SCHEDULED_ID, ST_SEQUENCE_ID, ST_TABLE_ACCESSOR_ID, ST_TABLE_ID, ST_VIEW_ID, }; use anyhow::anyhow; use core::ops::RangeBounds; use spacetimedb_lib::ConnectionId; -use spacetimedb_primitives::{ColList, TableId}; -use spacetimedb_sats::AlgebraicValue; +use spacetimedb_primitives::{col_list, ColList, TableId}; +use spacetimedb_sats::{product, AlgebraicValue}; use spacetimedb_schema::schema::{ColumnSchema, IndexSchema, TableSchema, ViewDefInfo}; use spacetimedb_table::table::IndexScanPointIter; use spacetimedb_table::{ @@ -82,6 +83,30 @@ pub trait StateView { value: &'r AlgebraicValue, ) -> Result>; + /// Find a sender-side stream by its durable stream id. + fn st_outbound_stream_by_id(&self, stream_id: u64) -> Result> { + self.iter_by_col_eq( + ST_OUTBOUND_STREAM_ID, + StOutboundStreamFields::StreamId, + &stream_id.into(), + )? + .map(StOutboundStreamRow::try_from) + .next() + .transpose() + } + + /// Find a pending delivery without decoding its payload. + fn outbound_msg_by_stream_seq(&self, stream_id: u64, seq: u64) -> Result>> { + let key = AlgebraicValue::Product(product![stream_id, seq]); + Ok(self + .iter_by_col_eq( + ST_OUTBOUND_MSG_ID, + col_list![StOutboundMsgFields::StreamId, StOutboundMsgFields::Seq], + &key, + )? + .next()) + } + /// Look up the `st_table` row which describes `table_id`. /// /// Default method calls `iter_by_col_eq`. diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 185017da7a1..8b877bac055 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -22,7 +22,8 @@ use spacetimedb_datastore::locking_tx_datastore::{ ApplyHistoryCounters, IndexScanPointOrRange, MutTxId, TxId, ViewCallInfo, }; use spacetimedb_datastore::system_tables::{ - system_tables, StModuleRow, ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_VIEW_SUB_ID, + system_tables, StModuleRow, StOutboundStreamRow, ST_CLIENT_ID, ST_CONNECTION_CREDENTIALS_ID, ST_OUTBOUND_STREAM_ID, + ST_VIEW_SUB_ID, }; use spacetimedb_datastore::system_tables::{StFields, StVarFields, StVarName, StVarRow, ST_MODULE_ID, ST_VAR_ID}; use spacetimedb_datastore::traits::{ @@ -1554,6 +1555,52 @@ impl RelationalDB { Ok(tx.row_level_security_for_table_id(table_id)?) } + /// Read sender streams to discover pending IDC deliveries. + pub fn st_outbound_streams<'a>( + &'a self, + tx: &'a Tx, + ) -> Result> + 'a, DBError> { + Ok(self + .iter(tx, ST_OUTBOUND_STREAM_ID)? + .map(|row| StOutboundStreamRow::try_from(row).map_err(DBError::from))) + } + + /// Find a sender stream by its durable id. + pub fn st_outbound_stream_by_id( + &self, + tx: &impl StateView, + stream_id: u64, + ) -> Result, DBError> { + Ok(tx.st_outbound_stream_by_id(stream_id)?) + } + + /// Find a pending delivery by stream and sequence. + pub fn outbound_msg_by_stream_seq<'a>( + &self, + tx: &'a impl StateView, + stream_id: u64, + seq: u64, + ) -> Result>, DBError> { + Ok(tx.outbound_msg_by_stream_seq(stream_id, seq)?) + } + + /// Find the user outbox row by its message id. + pub fn outbox_row( + &self, + tx: &impl StateView, + table_id: TableId, + msg_id: u64, + ) -> Result, DBError> { + let schema = tx.schema_for_table(table_id)?; + let msg_column = schema + .primary_key + .ok_or_else(|| anyhow!("outbox table {table_id} has no primary key"))?; + Ok(tx + .iter_by_col_eq(table_id, msg_column, &msg_id.into())? + .next() + .map(|row| row.to_product_value())) + } + /// Returns an iterator, /// yielding every row in the table identified by `table_id`. pub fn iter_mut<'a>(&'a self, tx: &'a MutTx, table_id: TableId) -> Result, DBError> { diff --git a/crates/standalone/src/lib.rs b/crates/standalone/src/lib.rs index 8047eed7c87..14916d3a2f3 100644 --- a/crates/standalone/src/lib.rs +++ b/crates/standalone/src/lib.rs @@ -15,7 +15,9 @@ use spacetimedb::db; use spacetimedb::db::persistence::{DurabilityConfig, LocalPersistenceProvider}; use spacetimedb::energy::{EnergyBalance, EnergyQuanta, NullEnergyMonitor}; use spacetimedb::host::module_host::UpdateEnvironmentResult; -use spacetimedb::host::{DiskStorage, HostController, HostRuntimeConfig, MigratePlanResult, UpdateDatabaseResult}; +use spacetimedb::host::{ + DiskStorage, HostController, HostRuntimeConfig, IdcConfig, MigratePlanResult, UpdateDatabaseResult, +}; use spacetimedb::identity::{AuthCtx, Identity}; use spacetimedb::messages::control_db::{Database, HostType, Node, Replica}; use spacetimedb::metrics::ENGINE_METRICS; @@ -59,7 +61,7 @@ pub struct StandaloneEnv { client_actor_index: ClientActorIndex, metrics_registry: prometheus::Registry, _pid_file: PidFile, - auth_provider: auth::DefaultJwtAuthProvider, + auth_provider: Arc, websocket_options: WebSocketOptions, } @@ -113,7 +115,7 @@ impl StandaloneEnv { client_actor_index, metrics_registry, _pid_file, - auth_provider: auth_env, + auth_provider: Arc::new(auth_env), websocket_options: config.websocket, })) } @@ -129,6 +131,13 @@ impl StandaloneEnv { pub fn bsatn_rlb_pool(&self) -> &BsatnRowListBuilderPool { &self.host_controller.bsatn_rlb_pool } + + pub fn configure_idc(&self, http_port: u16) -> anyhow::Result<()> { + self.host_controller.set_idc_config(IdcConfig { + http_port, + token_signer: self.auth_provider.clone(), + }) + } } #[derive(Debug, thiserror::Error)] diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index 2101df47ee8..195e935c4be 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/crates/standalone/src/subcommands/start.rs @@ -263,7 +263,9 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> { "failed to bind the SpacetimeDB server to '{listen_addr}', please check that the address is valid and not already in use" ))?; socket2::SockRef::from(&tcp).set_nodelay(true)?; - log::info!("Starting SpacetimeDB listening on {}", tcp.local_addr()?); + let local_addr = tcp.local_addr()?; + ctx.configure_idc(local_addr.port())?; + log::info!("Starting SpacetimeDB listening on {local_addr}"); if let Some(pg_port) = pg_port { let server_addr = listen_addr.split(':').next().unwrap(); diff --git a/crates/testing/Cargo.toml b/crates/testing/Cargo.toml index 28d394b19fb..9d287ff99fe 100644 --- a/crates/testing/Cargo.toml +++ b/crates/testing/Cargo.toml @@ -34,6 +34,11 @@ serde.workspace = true futures.workspace = true [dev-dependencies] +spacetimedb-datastore.workspace = true +spacetimedb-primitives.workspace = true +spacetimedb-sats.workspace = true +spacetimedb-schema = { workspace = true, features = ["test"] } +scopeguard.workspace = true axum.workspace = true tower = { version = "0.5", features = ["util"] } env_logger.workspace = true diff --git a/crates/testing/src/modules.rs b/crates/testing/src/modules.rs index 7a89174be9d..f51552deb5f 100644 --- a/crates/testing/src/modules.rs +++ b/crates/testing/src/modules.rs @@ -419,6 +419,8 @@ impl CompiledModule { ) .await .unwrap(); + // These in-process module tests do not run an HTTP listener. + env.configure_idc(0).unwrap(); // TODO: Fix this when we update identity generation. let identity = Identity::ZERO; let db_identity = SpacetimeAuth::alloc(&env).await.unwrap().claims.identity; diff --git a/crates/testing/tests/environment_lifecycle.rs b/crates/testing/tests/environment_lifecycle.rs index cde647b35df..a6cbd5ea42e 100644 --- a/crates/testing/tests/environment_lifecycle.rs +++ b/crates/testing/tests/environment_lifecycle.rs @@ -64,6 +64,8 @@ fn real_module_reopen_and_environment_only_publication_preserve_values() -> anyh v8: V8Config::default(), }; let mut env = Some(StandaloneEnv::init(config, &ca, data_dir.clone(), JobCores::without_pinned_cores()).await?); + // This in-process test does not run an HTTP listener. + env.as_ref().unwrap().configure_idc(0)?; let run = std::panic::AssertUnwindSafe(async { let initial = Values::from([ ("REQUIRED".into(), "initial-required".into()), @@ -209,6 +211,7 @@ fn real_module_reopen_and_environment_only_publication_preserve_values() -> anyh drop(module); drop(env.take()); env = Some(StandaloneEnv::init(config, &ca, data_dir.clone(), JobCores::without_pinned_cores()).await?); + env.as_ref().unwrap().configure_idc(0)?; // The first request after restart is env-only, before leader lookup has started a host. assert!(env .as_ref() diff --git a/crates/testing/tests/idc_actor.rs b/crates/testing/tests/idc_actor.rs new file mode 100644 index 00000000000..16804e37faf --- /dev/null +++ b/crates/testing/tests/idc_actor.rs @@ -0,0 +1,504 @@ +//! Exercise sender delivery and cleanup with a real module host. +//! The HTTP receiver is a stub that can delay requests or return errors. +//! This test does not run a receiver reducer or an `on_result` callback. +use axum::extract::{ConnectInfo, Path, Query}; +use axum::http::{StatusCode, Version}; +use axum::routing::post; +use axum::Router; +use serde::Deserialize; +use spacetimedb::db::relational_db::RelationalDB; +use spacetimedb::host::idc_actor::{self, IdcConfig}; +use spacetimedb::host::ModuleHost; +use spacetimedb_datastore::execution_context::Workload; +use spacetimedb_datastore::system_tables::{StOutboundMsgRow, StOutboundStreamRow, ST_OUTBOUND_STREAM_ID}; +use spacetimedb_datastore::traits::IsolationLevel; +use spacetimedb_lib::db::auth::{StAccess, StTableType}; +use spacetimedb_lib::Identity; +use spacetimedb_primitives::{ColId, SequenceId, TableId}; +use spacetimedb_sats::bsatn::to_vec; +use spacetimedb_sats::{product, AlgebraicType}; +use spacetimedb_schema::def::BTreeAlgorithm; +use spacetimedb_schema::reducer_name::ReducerName; +use spacetimedb_schema::schema::{ + ColumnSchema, ConstraintSchema, IndexSchema, OutboxSchema, SequenceSchema, TableSchema, +}; +use spacetimedb_schema::table_name::TableName; +use spacetimedb_testing::modules::{CompilationMode, CompiledModule, IN_MEMORY_CONFIG}; +use std::collections::HashSet; +use std::net::SocketAddr; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::sync::mpsc; + +// Exceed one notification batch to exercise draining while deliveries are active. +const NOTIFICATION_COUNT: usize = 64; +const TEST_TIMEOUT: Duration = Duration::from_secs(5); +const SLOW_TARGET: Identity = Identity::ONE; +const FAST_TARGET: Identity = Identity::ZERO; + +#[derive(Deserialize)] +struct DeliveryQuery { + seq: u64, + ack_prefix: u64, +} + +async fn next_delivery(rx: &mut mpsc::UnboundedReceiver<(String, u64)>) -> anyhow::Result<(String, u64)> { + tokio::time::timeout(TEST_TIMEOUT, rx.recv()) + .await? + .ok_or_else(|| anyhow::anyhow!("receiver stopped before the expected delivery")) +} + +async fn expect_delivery( + rx: &mut mpsc::UnboundedReceiver<(String, u64)>, + target: Identity, + message: &StOutboundMsgRow, +) -> anyhow::Result<()> { + assert_eq!(next_delivery(rx).await?, (target.to_hex().to_string(), message.seq)); + Ok(()) +} + +/// Check startup recovery, per-stream ordering, retries, and insert notifications. +/// A blocked stream must not hold up other streams, and failed deliveries must +/// stay pending until the receiver accepts them, even after notifications close. +#[test] +#[serial_test::serial] +fn idc_actor_streams_progress_independently_and_preserve_order() { + CompiledModule::compile("module-test", CompilationMode::Debug).with_module_async( + IN_MEMORY_CONFIG, + |module| async move { + run_actor_test(module.client.module()).await.unwrap(); + }, + ); +} + +fn outbox_schema() -> TableSchema { + TableSchema::new( + TableId::SENTINEL, + TableName::for_test("outbox"), + None, + vec![ + ColumnSchema::for_test(0, "target", AlgebraicType::identity()), + ColumnSchema::for_test(1, "msg_id", AlgebraicType::U64), + ColumnSchema::for_test(2, "value", AlgebraicType::U32), + ], + vec![IndexSchema::for_test("outbox_msg_id", BTreeAlgorithm::from(1))], + vec![ConstraintSchema::unique_for_test("outbox_msg_id_unique", 1)], + vec![SequenceSchema { + sequence_id: SequenceId::SENTINEL, + sequence_name: "outbox_msg_id_seq".into(), + table_id: TableId::SENTINEL, + col_pos: ColId(1), + start: 1, + }], + StTableType::User, + StAccess::Public, + None, + Some(ColId(1)), + false, + None, + Some(OutboxSchema { + remote_reducer: ReducerName::for_test("receive_value"), + target_column: ColId(0), + arg_columns: vec![ColId(2)], + on_result_reducer: None, + signature_hash: spacetimedb_schema::def::reducer_signature_hash( + &spacetimedb_sats::ProductType::from([("value", AlgebraicType::U32)]), + &AlgebraicType::unit(), + &AlgebraicType::String, + &spacetimedb_sats::Typespace::default(), + ) + .to_hex() + .to_string(), + }), + ) +} + +fn create_outbox(db: &RelationalDB) -> anyhow::Result { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + let table_id = db.create_table(&mut tx, outbox_schema())?; + db.commit_tx(tx)?; + Ok(table_id) +} + +/// Commit a message without notifying the actor, so each scenario controls when it wakes. +fn insert_outbox_row(db: &RelationalDB, table_id: TableId, target: Identity) -> anyhow::Result { + let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); + // Zero asks the sequence to allocate the message ID, just like a module insert. + // The payload is identical so messages are distinguished by their allocated IDs. + let row = to_vec(&product![target, 0u64, 42u32])?; + let (row_ptr, insert_flags, msg_id) = { + let (_, row_ref, insert_flags) = db.insert(&mut tx, table_id, &row)?; + (row_ref.pointer(), insert_flags, row_ref.read_col::(ColId(1))?) + }; + assert_ne!(msg_id, 0); + assert!(insert_flags.is_outbox_table); + let message = tx.record_outbox_insert(table_id, row_ptr)?; + assert_eq!(message.msg_id, msg_id); + db.commit_tx(tx)?; + Ok(message) +} + +/// Wait for sender cleanup to commit, not just for the HTTP stub to receive a request. +async fn wait_for_ack(db: &RelationalDB, target: Identity, prefix: u64) -> anyhow::Result<()> { + tokio::time::timeout(TEST_TIMEOUT, async { + loop { + let tx = db.begin_tx(Workload::Internal); + let acked = db.iter(&tx, ST_OUTBOUND_STREAM_ID).unwrap().any(|row| { + let stream = StOutboundStreamRow::try_from(row).unwrap(); + stream.target_identity.0 == target && stream.ack_prefix == prefix + }); + let _ = db.release_tx(tx); + if acked { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + Ok(()) +} + +async fn run_actor_test(host: ModuleHost) -> anyhow::Result<()> { + let keys = spacetimedb::auth::JwtKeys::generate()?; + let (captured_tx, mut captured_rx) = mpsc::unbounded_channel(); + // The slow target blocks its first sequence and fails the first attempt. + // The fast target accepts immediately. + // A third target returns 403 until the test allows it to recover. + let release_slow = Arc::new(tokio::sync::Semaphore::new(0)); + let fail_first = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let rejected_target = Identity::from_byte_array([2; 32]); + let reject_auth = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let app = Router::new().route( + "/v1/database/:target_db/call-from-database/:reducer", + post({ + let release_slow = release_slow.clone(); + let reject_auth = reject_auth.clone(); + move |Path((target, _)): Path<(String, String)>, Query(query): Query| { + let captured_tx = captured_tx.clone(); + let release_slow = release_slow.clone(); + let fail_first = fail_first.clone(); + let reject_auth = reject_auth.clone(); + async move { + let seq = query.seq; + // Every new sequence must carry acknowledgment of all earlier messages. + // Retrying the same sequence must not advance that acknowledgment. + assert_eq!(query.ack_prefix, seq - 1); + let slow = target == SLOW_TARGET.to_hex().to_string(); + let rejected = + target == rejected_target.to_hex().to_string() && reject_auth.load(Ordering::Relaxed); + // Report arrival before blocking so the test can observe work still in flight. + captured_tx.send((target, seq)).unwrap(); + if rejected { + return StatusCode::FORBIDDEN; + } + if slow && seq == 1 { + release_slow.acquire().await.unwrap().forget(); + if fail_first.swap(false, Ordering::Relaxed) { + return StatusCode::SERVICE_UNAVAILABLE; + } + } + StatusCode::OK + } + } + }), + ); + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let http_port = listener.local_addr()?.port(); + let server = tokio::spawn(async move { axum::serve(listener, app).await }); + let server_abort = scopeguard::guard(server.abort_handle(), |handle| handle.abort()); + let db = host.relational_db().clone(); + // Both messages predate actor startup and have no notifications. Startup must + // discover the stream, but only send its first sequence. + let table_id = create_outbox(&db)?; + let slow_first = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + let slow_second = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + assert_eq!(slow_first.seq, 1); + assert_eq!(slow_second.seq, slow_first.seq + 1); + let (starter, notify_tx) = idc_actor::channel(); + let config = IdcConfig { + http_port, + token_signer: Arc::new(keys.private), + }; + // Use the real host to commit cleanup through the view and subscription path. + let actor = starter.start( + db.clone(), + host.info.database_identity, + idc_actor::http_client(), + config, + host.downgrade(), + )?; + expect_delivery(&mut captured_rx, SLOW_TARGET, &slow_first).await?; + + // Insert more work while one request is waiting. Other streams should still + // make progress, but the next message in the slow stream must wait. + let slow_third = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + // Repeated notifications must not start a second attempt for the busy stream. + for _ in 0..NOTIFICATION_COUNT { + notify_tx.send(slow_third.clone())?; + } + let fast_first = insert_outbox_row(&db, table_id, FAST_TARGET)?; + notify_tx.send(fast_first.clone())?; + expect_delivery(&mut captured_rx, FAST_TARGET, &fast_first).await?; + wait_for_ack(&db, FAST_TARGET, fast_first.seq).await?; + + // The fast stream is idle after cleanup. A fresh insert must wake it again. + let fast_second = insert_outbox_row(&db, table_id, FAST_TARGET)?; + notify_tx.send(fast_second.clone())?; + expect_delivery(&mut captured_rx, FAST_TARGET, &fast_second).await?; + wait_for_ack(&db, FAST_TARGET, fast_second.seq).await?; + wait_for_ack(&db, SLOW_TARGET, 0).await?; + assert!( + captured_rx.try_recv().is_err(), + "slow stream must have only one request in flight" + ); + + // Release the first attempt with 503. The actor must retry the same message, + // while the fast stream keeps making progress. + release_slow.add_permits(1); + expect_delivery(&mut captured_rx, SLOW_TARGET, &slow_first).await?; + let fast_third = insert_outbox_row(&db, table_id, FAST_TARGET)?; + notify_tx.send(fast_third.clone())?; + expect_delivery(&mut captured_rx, FAST_TARGET, &fast_third).await?; + wait_for_ack(&db, FAST_TARGET, fast_third.seq).await?; + wait_for_ack(&db, SLOW_TARGET, 0).await?; + + // Let the retry succeed. Cleanup must unlock the second message without a + // notification for it, then the third message from the earlier insert. + release_slow.add_permits(1); + expect_delivery(&mut captured_rx, SLOW_TARGET, &slow_second).await?; + expect_delivery(&mut captured_rx, SLOW_TARGET, &slow_third).await?; + wait_for_ack(&db, SLOW_TARGET, slow_third.seq).await?; + + // An auth rejection leaves the message pending. Inserts and + // duplicate notifications must not let later messages overtake it. + let first_rejected = insert_outbox_row(&db, table_id, rejected_target)?; + notify_tx.send(first_rejected.clone())?; + expect_delivery(&mut captured_rx, rejected_target, &first_rejected).await?; + let second_rejected = insert_outbox_row(&db, table_id, rejected_target)?; + for _ in 0..NOTIFICATION_COUNT { + notify_tx.send(first_rejected.clone())?; + notify_tx.send(second_rejected.clone())?; + } + expect_delivery(&mut captured_rx, rejected_target, &first_rejected).await?; + + // Closing the channel must still drain every batch and finish its deliveries. + let slow_last = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + for _ in 0..NOTIFICATION_COUNT { + notify_tx.send(slow_last.clone())?; + } + let fast_last = insert_outbox_row(&db, table_id, FAST_TARGET)?; + notify_tx.send(fast_last.clone())?; + drop(notify_tx); + // Independent streams may finish in either order. Reject duplicate deliveries, + // but allow the rejected stream to keep retrying while the others finish. + let mut expected = HashSet::from([ + (SLOW_TARGET.to_hex().to_string(), slow_last.seq), + (FAST_TARGET.to_hex().to_string(), fast_last.seq), + ]); + tokio::time::timeout(TEST_TIMEOUT, async { + while !expected.is_empty() { + let delivered = next_delivery(&mut captured_rx).await?; + if delivered.0 == rejected_target.to_hex().to_string() { + assert_eq!( + delivered.1, first_rejected.seq, + "rejected stream must keep retrying its first sequence" + ); + } else { + assert!(expected.remove(&delivered), "unexpected delivery: {delivered:?}"); + } + } + anyhow::Ok(()) + }) + .await??; + wait_for_ack(&db, SLOW_TARGET, slow_last.seq).await?; + wait_for_ack(&db, FAST_TARGET, fast_last.seq).await?; + // Only the rejected stream's rows should remain. Its 403 is a transport error, + // not a stored reducer result, and must not advance ack_prefix. + let tx = db.begin_tx(Workload::Internal); + let remaining_ids = db + .iter(&tx, table_id)? + .map(|row| row.read_col::(ColId(1))) + .collect::, _>>()?; + assert_eq!( + remaining_ids, + HashSet::from([first_rejected.msg_id, second_rejected.msg_id]) + ); + let stream = db + .st_outbound_streams(&tx)? + .collect::, _>>()? + .into_iter() + .find(|stream| stream.target_identity.0 == rejected_target) + .unwrap(); + assert_eq!(stream.ack_prefix, 0); + let row = db + .outbound_msg_by_stream_seq(&tx, stream.stream_id, first_rejected.seq)? + .unwrap(); + let row = StOutboundMsgRow::try_from(row)?; + assert!(row.result_status.is_none()); + assert!(row.retry_count >= 1); + assert!(row.last_transport_error.unwrap().contains("HTTP 403")); + let _ = db.release_tx(tx); + + // Fixing authentication resumes delivery without restarting the actor or + // sending another notification, even though the channel is now closed. + reject_auth.store(false, Ordering::Relaxed); + tokio::time::timeout(TEST_TIMEOUT, async { + while let Some((target, seq)) = captured_rx.recv().await { + assert_eq!(target, rejected_target.to_hex().to_string()); + assert!(seq == first_rejected.seq || seq == second_rejected.seq); + if seq == second_rejected.seq { + return; + } + } + panic!("recovered stream must deliver its second message"); + }) + .await?; + wait_for_ack(&db, rejected_target, second_rejected.seq).await?; + // Both formerly rejected messages have now completed cleanup. + let tx = db.begin_tx(Workload::Internal); + assert_eq!(db.iter(&tx, table_id)?.count(), 0); + let _ = db.release_tx(tx); + drop(actor); + drop(server_abort); + assert!(server.await.unwrap_err().is_cancelled()); + Ok(()) +} + +#[test] +#[serial_test::serial] +fn idc_actor_shutdown_drains_attempts_without_dispatching_more() { + CompiledModule::compile("module-test", CompilationMode::Debug).with_module_async( + IN_MEMORY_CONFIG, + |module| async move { + run_shutdown_test(module.client.module()).await.unwrap(); + }, + ); +} + +async fn run_shutdown_test(host: ModuleHost) -> anyhow::Result<()> { + let client = idc_actor::http_client(); + let mut connections = HashSet::new(); + let (requests_tx, mut requests_rx) = mpsc::unbounded_channel(); + let app = Router::new().route( + "/v1/database/:target_db/call-from-database/:reducer", + post( + move |ConnectInfo(peer): ConnectInfo, version: Version, Query(query): Query| { + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel::(); + assert_eq!(version, Version::HTTP_2); + requests_tx.send((query.seq, reply_tx, peer)).unwrap(); + async move { reply_rx.await.unwrap() } + }, + ), + ); + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let config = IdcConfig { + http_port: listener.local_addr()?.port(), + token_signer: Arc::new(spacetimedb::auth::JwtKeys::generate()?.private), + }; + let app = app.into_make_service_with_connect_info::(); + let server = tokio::spawn(async move { axum::serve(listener, app).await }); + let _server_abort = scopeguard::guard(server.abort_handle(), |handle| handle.abort()); + let db = host.relational_db().clone(); + let table_id = create_outbox(&db)?; + let first = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + let second = insert_outbox_row(&db, table_id, SLOW_TARGET)?; + + // A shutdown already requested at startup must not dispatch recovered messages. + { + let (starter, notify_tx) = idc_actor::channel(); + notify_tx.close(); + let _actor = starter.start( + db.clone(), + host.info.database_identity, + client.clone(), + config.clone(), + host.downgrade(), + )?; + tokio::time::timeout(TEST_TIMEOUT, notify_tx.closed()).await?; + assert!(requests_rx.try_recv().is_err()); + } + + // A failed attempt remains pending. Restart retries it; a successful attempt + // finishes local cleanup before shutdown returns, leaving the next row pending. + for status in [StatusCode::SERVICE_UNAVAILABLE, StatusCode::OK] { + let (starter, notify_tx) = idc_actor::channel(); + let _actor = starter.start( + db.clone(), + host.info.database_identity, + client.clone(), + config.clone(), + host.downgrade(), + )?; + let (seq, reply, peer) = tokio::time::timeout(TEST_TIMEOUT, requests_rx.recv()).await?.unwrap(); + connections.insert(peer); + assert_eq!(seq, first.seq); + { + // ModuleHost::exit signals closure through its sender and waits for + // completion. Multiple waiters must all observe the completed drain. + notify_tx.close(); + let mut shutdown = Box::pin(async { + tokio::join!(notify_tx.closed(), notify_tx.closed()); + }); + assert!( + tokio::time::timeout(Duration::from_millis(50), shutdown.as_mut()) + .await + .is_err(), + "shutdown returned before the active HTTP request completed" + ); + reply.send(status).unwrap(); + tokio::time::timeout(TEST_TIMEOUT, shutdown).await?; + } + // Repeated shutdown is harmless. + notify_tx.close(); + tokio::time::timeout(TEST_TIMEOUT, notify_tx.closed()).await?; + assert!( + requests_rx.try_recv().is_err(), + "shutdown must not dispatch another attempt" + ); + let tx = db.begin_tx(Workload::Internal); + let stream = db.st_outbound_stream_by_id(&tx, first.stream_id)?.unwrap(); + if status == StatusCode::OK { + assert_eq!(stream.ack_prefix, first.seq); + assert!(db + .outbound_msg_by_stream_seq(&tx, first.stream_id, first.seq)? + .is_none()); + } else { + assert_eq!(stream.ack_prefix, 0); + let row = db.outbound_msg_by_stream_seq(&tx, first.stream_id, first.seq)?.unwrap(); + let row = StOutboundMsgRow::try_from(row)?; + assert_eq!(row.retry_count, 1); + assert!(row.result_status.is_none()); + } + assert!(db + .outbound_msg_by_stream_seq(&tx, second.stream_id, second.seq)? + .is_some()); + let _ = db.release_tx(tx); + } + + // Startup discovers the untouched second message without a notification. + let (starter, notify_tx) = idc_actor::channel(); + let _actor = starter.start( + db.clone(), + host.info.database_identity, + client, + config, + host.downgrade(), + )?; + let (seq, reply, peer) = tokio::time::timeout(TEST_TIMEOUT, requests_rx.recv()).await?.unwrap(); + connections.insert(peer); + assert_eq!( + connections.len(), + 1, + "replacement actors must reuse the HTTP/2 connection" + ); + assert_eq!(seq, second.seq); + reply.send(StatusCode::OK).unwrap(); + wait_for_ack(&db, SLOW_TARGET, second.seq).await?; + notify_tx.close(); + tokio::time::timeout(TEST_TIMEOUT, notify_tx.closed()).await?; + tokio::time::timeout(TEST_TIMEOUT, host.exit()).await?; + tokio::time::timeout(TEST_TIMEOUT, host.exited()).await?; + Ok(()) +}