From a6c043e35bf80745a5a1f9808818e6b417aea8e8 Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 5 Sep 2026 01:56:55 +0200 Subject: [PATCH 1/3] core: bound the wait for a procedure instance slot `ModuleInstanceManager::get_instance` acquired a slot in the bounded procedure pool with a bare `Semaphore::acquire_owned().await`. Once every slot was held (a handful per database by default, one per core), every later procedure call, HTTP handler call and scheduled procedure on that database waited forever, and a websocket caller got no result envelope, error or timeout at all. Add `procedure-queue-timeout` to the `[wasm]` and `[v8]` config sections (default 30 seconds, 0 waits forever as before). When it elapses, `get_instance` fails with `InstancePoolTimeout` before anything was started, so the caller can retry. The error reaches: - websocket procedure calls as a `ProcedureStatus::InternalError` result for the request id, instead of silence - HTTP `/call` and module HTTP handlers as 503 Service Unavailable - the scheduler, which re-queues the item instead of dropping it `enqueue_with_procedure_instance` now takes an already acquired lease so the caller can report the timeout while it still owns the result target. --- crates/client-api/src/routes/database.rs | 7 + crates/core/src/config.rs | 20 +++ crates/core/src/host/module_host.rs | 189 ++++++++++++++++----- crates/core/src/host/scheduler.rs | 26 ++- crates/core/src/host/v8/mod.rs | 8 +- crates/core/src/host/wasmtime/mod.rs | 1 + crates/standalone/config.toml | 6 + crates/standalone/src/subcommands/start.rs | 7 + 8 files changed, 222 insertions(+), 42 deletions(-) diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs index f170f9b290e..274e6094517 100644 --- a/crates/client-api/src/routes/database.rs +++ b/crates/client-api/src/routes/database.rs @@ -139,6 +139,10 @@ fn map_procedure_error(e: ProcedureCallError, procedure: &str) -> (StatusCode, S log::info!("Procedure {procedure} could not run because the module is out of energy"); StatusCode::PAYMENT_REQUIRED } + ProcedureCallError::PoolTimeout(_) => { + log::info!("Procedure {procedure} could not run because no procedure instance became free in time"); + StatusCode::SERVICE_UNAVAILABLE + } ProcedureCallError::InternalError(_) => { // TODO: May need to split this from module errors vs host errors log::info!("Internal error while invoking procedure {procedure}: {e:#}"); @@ -298,6 +302,9 @@ async fn handle_http_route_impl( Err(spacetimedb::host::module_host::HttpHandlerCallError::NoSuchModule(_)) => { return Err(NO_SUCH_DATABASE.into()); } + Err(err @ spacetimedb::host::module_host::HttpHandlerCallError::PoolTimeout(_)) => { + return Err((StatusCode::SERVICE_UNAVAILABLE, err.to_string()).into()); + } Err(spacetimedb::host::module_host::HttpHandlerCallError::InternalError(err)) => { return Err((StatusCode::INTERNAL_SERVER_ERROR, err).into()); } diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index 8dc9ceb7273..b4d5dde24ef 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -171,9 +171,11 @@ impl<'de> serde::Deserialize<'de> for ConfigFile { module_http: config.module_http, wasm: WasmConfig { procedure_instance_pool_size: config.wasm.procedure_instance_pool_size, + procedure_queue_timeout: config.wasm.procedure_queue_timeout, }, v8: V8Config { procedure_instance_pool_size: config.v8.procedure_instance_pool_size, + procedure_queue_timeout: config.v8.procedure_queue_timeout, heap_policy: config.v8_heap_policy, }, }) @@ -233,12 +235,16 @@ impl Default for ModuleHttpConfig { #[derive(Clone, Copy, Debug)] pub struct WasmConfig { pub procedure_instance_pool_size: NonZeroUsize, + /// How long a call may wait for a free procedure instance before it fails. + /// `None` waits forever. + pub procedure_queue_timeout: Option, } impl Default for WasmConfig { fn default() -> Self { Self { procedure_instance_pool_size: default_wasm_procedure_instance_pool_size(), + procedure_queue_timeout: default_procedure_queue_timeout(), } } } @@ -251,12 +257,15 @@ struct WasmConfigToml { deserialize_with = "de_nz_usize" )] pub procedure_instance_pool_size: NonZeroUsize, + #[serde(default = "default_procedure_queue_timeout", deserialize_with = "de_nz_duration")] + pub procedure_queue_timeout: Option, } impl Default for WasmConfigToml { fn default() -> Self { Self { procedure_instance_pool_size: default_wasm_procedure_instance_pool_size(), + procedure_queue_timeout: default_procedure_queue_timeout(), } } } @@ -264,6 +273,9 @@ impl Default for WasmConfigToml { #[derive(Clone, Copy, Debug)] pub struct V8Config { pub procedure_instance_pool_size: NonZeroUsize, + /// How long a call may wait for a free procedure instance before it fails. + /// `None` waits forever. + pub procedure_queue_timeout: Option, pub heap_policy: V8HeapPolicyConfig, } @@ -271,6 +283,7 @@ impl Default for V8Config { fn default() -> Self { Self { procedure_instance_pool_size: default_v8_procedure_instance_pool_size(), + procedure_queue_timeout: default_procedure_queue_timeout(), heap_policy: V8HeapPolicyConfig::default(), } } @@ -291,12 +304,15 @@ struct V8ConfigToml { deserialize_with = "de_nz_usize" )] pub procedure_instance_pool_size: NonZeroUsize, + #[serde(default = "default_procedure_queue_timeout", deserialize_with = "de_nz_duration")] + pub procedure_queue_timeout: Option, } impl Default for V8ConfigToml { fn default() -> Self { Self { procedure_instance_pool_size: default_v8_procedure_instance_pool_size(), + procedure_queue_timeout: default_procedure_queue_timeout(), } } } @@ -382,6 +398,10 @@ fn default_wasm_procedure_instance_pool_size() -> NonZeroUsize { std::thread::available_parallelism().unwrap_or_else(|_| NonZeroUsize::new(1).unwrap()) } +fn default_procedure_queue_timeout() -> Option { + Some(Duration::from_secs(30)) +} + fn de_nz_usize<'de, D>(deserializer: D) -> Result where D: serde::Deserializer<'de>, diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index 25eb09e6382..0009b6820dd 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -392,6 +392,7 @@ pub enum ModuleWithInstance { core: AllocatedJobCore, init_inst: Box, procedure_instance_pool_size: NonZeroUsize, + procedure_queue_timeout: Option, }, Js { module: super::v8::JsModule, @@ -491,18 +492,22 @@ impl WasmtimeModuleHost { }); } - async fn enqueue_with_procedure_instance( + /// Runs `wasm` on the procedure instance held by `lease`, + /// which the caller acquires with [`ModuleInstanceManager::get_instance`] + /// so that a pool timeout can be reported before any work is enqueued. + fn enqueue_with_procedure_instance( &self, label: &str, on_panic: Arc, timer_guard: CallTimerGuard, + lease: ModuleInstanceLease>, arg: A, wasm: impl AsyncFnOnce(A, &mut ModuleInstance) + Send + 'static, ) where A: Send + 'static, { let instance_manager = self.procedure_instances.clone(); - let ModuleInstanceLease { instance, slot } = instance_manager.get_instance().await; + let ModuleInstanceLease { instance, slot } = lease; let label = label.to_owned(); self.executor.enqueue_async_job(async move || { scopeguard::defer_on_unwind!({ @@ -1256,6 +1261,9 @@ struct ModuleInstanceManager { module: M, metrics: InstanceManagerMetrics, instance_slots: Option>, + /// How long [`Self::get_instance`] waits for a slot in a bounded pool. + /// `None` waits forever. + slot_timeout: Option, } struct ModuleInstanceLease { @@ -1402,8 +1410,9 @@ impl ModuleInstanceManager { init_inst: Option, metrics: InstanceManagerMetrics, max_instances: NonZeroUsize, + slot_timeout: Option, ) -> Self { - Self::new_inner(module, init_inst, metrics, Some(max_instances)) + Self::new_inner(module, init_inst, metrics, Some(max_instances), slot_timeout) } fn new_inner( @@ -1411,6 +1420,7 @@ impl ModuleInstanceManager { init_inst: Option, metrics: InstanceManagerMetrics, max_instances: Option, + slot_timeout: Option, ) -> Self { let mut instances = VecDeque::new(); instances.extend(init_inst); @@ -1430,25 +1440,35 @@ impl ModuleInstanceManager { module, metrics, instance_slots: max_instances.map(|max_instances| Arc::new(Semaphore::new(max_instances.get()))), + slot_timeout, } } - async fn with_instance(&self, f: impl AsyncFnOnce(M::Instance) -> (R, M::Instance)) -> R { - let ModuleInstanceLease { instance, slot } = self.get_instance().await; + async fn with_instance( + &self, + f: impl AsyncFnOnce(M::Instance) -> (R, M::Instance), + ) -> Result { + let ModuleInstanceLease { instance, slot } = self.get_instance().await?; let (res, instance) = f(instance).await; self.return_instance(ModuleInstanceLease { instance, slot }); - res + Ok(res) } - async fn get_instance(&self) -> ModuleInstanceLease { + /// Checks out an instance, waiting for a free slot if the pool is bounded. + /// + /// Fails with [`InstancePoolTimeout`] if no slot frees up within `slot_timeout`, + /// in which case nothing was started on behalf of the caller. + async fn get_instance(&self) -> Result, InstancePoolTimeout> { let slot = if let Some(instance_slots) = &self.instance_slots { - Some( - instance_slots - .clone() - .acquire_owned() - .await - .expect("module instance slot semaphore should not close"), - ) + let acquire = instance_slots.clone().acquire_owned(); + let permit = match self.slot_timeout { + Some(timeout) => match tokio::time::timeout(timeout, acquire).await { + Ok(permit) => permit, + Err(_elapsed) => return Err(InstancePoolTimeout(timeout)), + }, + None => acquire.await, + }; + Some(permit.expect("module instance slot semaphore should not close")) } else { None }; @@ -1467,7 +1487,7 @@ impl ModuleInstanceManager { res }; - ModuleInstanceLease { instance, slot } + Ok(ModuleInstanceLease { instance, slot }) } fn return_instance(&self, lease: ModuleInstanceLease) { @@ -1651,12 +1671,30 @@ pub enum ViewCallError { InternalError(String), } +/// No procedure instance became free within the configured `procedure-queue-timeout`. +/// +/// The call was not started, so it is safe to retry. +#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)] +#[error("Timed out after {0:?} waiting for a free procedure instance; the call was not started")] +pub struct InstancePoolTimeout(pub Duration); + +/// Errors from checking out a pooled procedure instance to run a call on. +#[derive(thiserror::Error, Debug)] +pub enum PooledCallError { + #[error(transparent)] + NoSuchModule(#[from] NoSuchModule), + #[error(transparent)] + PoolTimeout(#[from] InstancePoolTimeout), +} + #[derive(thiserror::Error, Debug)] pub enum ProcedureCallError { #[error(transparent)] Args(#[from] InvalidProcedureArguments), #[error(transparent)] NoSuchModule(#[from] NoSuchModule), + #[error(transparent)] + PoolTimeout(#[from] InstancePoolTimeout), #[error("No such procedure")] NoSuchProcedure, #[error("Procedure terminated due to insufficient budget")] @@ -1665,16 +1703,36 @@ pub enum ProcedureCallError { InternalError(String), } +impl From for ProcedureCallError { + fn from(err: PooledCallError) -> Self { + match err { + PooledCallError::NoSuchModule(err) => Self::NoSuchModule(err), + PooledCallError::PoolTimeout(err) => Self::PoolTimeout(err), + } + } +} + #[derive(thiserror::Error, Debug)] pub enum HttpHandlerCallError { #[error(transparent)] NoSuchModule(#[from] NoSuchModule), + #[error(transparent)] + PoolTimeout(#[from] InstancePoolTimeout), #[error("no such http handler")] NoSuchHandler, #[error("The module instance encountered a fatal error: {0}")] InternalError(String), } +impl From for HttpHandlerCallError { + fn from(err: PooledCallError) -> Self { + match err { + PooledCallError::NoSuchModule(err) => Self::NoSuchModule(err), + PooledCallError::PoolTimeout(err) => Self::PoolTimeout(err), + } + } +} + #[derive(thiserror::Error, Debug)] pub enum InitDatabaseError { #[error(transparent)] @@ -1777,6 +1835,7 @@ impl ModuleHost { core, init_inst, procedure_instance_pool_size, + procedure_queue_timeout, } => { info = module.info(); let module = Arc::new(module); @@ -1790,6 +1849,7 @@ impl ModuleHost { None, metrics, procedure_instance_pool_size, + procedure_queue_timeout, )); Arc::new(ModuleHostInner::Wasm(Box::new(WasmtimeModuleHost { module, @@ -1801,6 +1861,7 @@ impl ModuleHost { info = module.info(); let metrics = module.metrics(); let procedure_instance_pool_size = module.procedure_instance_pool_size(); + let procedure_queue_timeout = module.procedure_queue_timeout(); let host_module = module.clone(); let main_instance = SharedJsMainInstanceManager::new(init_inst, metrics.clone()); let procedure_instances = ModuleInstanceManager::new_bounded_with_metrics( @@ -1808,6 +1869,7 @@ impl ModuleHost { None, metrics, procedure_instance_pool_size, + procedure_queue_timeout, ); Arc::new(ModuleHostInner::Js(Box::new(V8ModuleHost { module: host_module, @@ -1931,7 +1993,7 @@ impl ModuleHost { arg: A, wasm: impl AsyncFnOnce(A, &mut ModuleInstance) -> R + Send + 'static, js: impl AsyncFnOnce(A, &JsProcedureInstance) -> R, - ) -> Result + ) -> Result where R: Send + 'static, A: Send + 'static, @@ -1958,7 +2020,7 @@ impl ModuleHost { }) .await }) - .await + .await? } ModuleHostInner::Js(host) => { host.procedure_instances @@ -1967,7 +2029,7 @@ impl ModuleHost { let res = js(arg, &inst).await; (res, inst) }) - .await + .await? } }) } @@ -2653,7 +2715,10 @@ impl ModuleHost { match &*self.inner { ModuleHostInner::Js(host) => { - let lease = host.procedure_instances.get_instance().await; + let lease = match host.procedure_instances.get_instance().await { + Ok(lease) => lease, + Err(err) => return self.send_procedure_error(&procedure_name, timer, target, err.into()), + }; let call = lease.instance.enqueue_procedure(params).await; let module = self.clone(); tokio::spawn(async move { @@ -2679,25 +2744,25 @@ impl ModuleHost { let target_for_job = target.clone(); let timer_guard = self.start_call_timer(&procedure_name); let on_panic = self.on_panic.clone(); - wasm_host - .enqueue_with_procedure_instance( - &procedure_name, - on_panic, - timer_guard, - params, - async move |params, inst| { - let ret = inst.call_procedure(params).await; - if let Err(err) = module.log_and_send_procedure_result( - &procedure_name_for_job, - timer, - target_for_job, - ret, - ) { - log::warn!("Procedure call failed: {err:#}"); - } - }, - ) - .await; + let lease = match wasm_host.procedure_instances.get_instance().await { + Ok(lease) => lease, + Err(err) => return self.send_procedure_error(&procedure_name, timer, target, err.into()), + }; + wasm_host.enqueue_with_procedure_instance( + &procedure_name, + on_panic, + timer_guard, + lease, + params, + async move |params, inst| { + let ret = inst.call_procedure(params).await; + if let Err(err) = + module.log_and_send_procedure_result(&procedure_name_for_job, timer, target_for_job, ret) + { + log::warn!("Procedure call failed: {err:#}"); + } + }, + ); Ok(()) } } @@ -2867,7 +2932,7 @@ impl ModuleHost { &self, name: &str, params: CallProcedureParams, - ) -> Result { + ) -> Result { call_pooled_instance!( self, name, @@ -3647,12 +3712,16 @@ fn args_error_log_message(function_kind: &str, function_name: &str) -> String { #[cfg(test)] mod tests { - use super::ModuleHost; + use super::{ + GenericModule, GenericModuleInstance, InstanceManagerMetrics, InstancePoolTimeout, ModuleHost, + ModuleInstanceManager, + }; use crate::client::{ ClientActorId, ClientConfig, ClientConnectionReceiver, ClientConnectionSender, OutboundMessage, Protocol, WsVersion, }; use crate::db::relational_db::tests_utils::{insert, with_auto_commit, TestDB}; + use crate::messages::control_db::HostType; use crate::subscription::module_subscription_actor::ModuleSubscriptions; use spacetimedb_client_api_messages::websocket::{common::RowListLen as _, v1 as ws_v1, v2 as ws_v2}; use spacetimedb_lib::identity::AuthCtx; @@ -3678,6 +3747,46 @@ mod tests { (Arc::new(sender), receiver) } + struct TestModule; + struct TestInstance; + + impl GenericModuleInstance for TestInstance { + fn trapped(&self) -> bool { + false + } + } + + impl GenericModule for TestModule { + type Instance = TestInstance; + async fn create_instance(&self) -> TestInstance { + TestInstance + } + fn host_type(&self) -> HostType { + HostType::Wasm + } + } + + #[tokio::test] + async fn bounded_pool_times_out_waiting_for_a_slot() { + let timeout = std::time::Duration::from_millis(20); + let pool = ModuleInstanceManager::new_bounded_with_metrics( + TestModule, + None, + InstanceManagerMetrics::new(HostType::Wasm, Identity::ZERO), + std::num::NonZeroUsize::new(1).expect("1 is non-zero"), + Some(timeout), + ); + + let held = pool + .get_instance() + .await + .expect("the first checkout gets the only slot"); + assert_eq!(pool.get_instance().await.err(), Some(InstancePoolTimeout(timeout))); + + pool.return_instance(held); + assert!(pool.get_instance().await.is_ok()); + } + #[test] fn one_off_query_v2_returns_rows() -> anyhow::Result<()> { let runtime = tokio::runtime::Runtime::new()?; diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index a07dd43597e..0ef14e1f4f6 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -3,7 +3,7 @@ use super::module_host::{ }; use super::{FunctionArgs, ModuleHost}; use crate::db::relational_db::RelationalDB; -use crate::host::module_host::{CallProcedureParams, ModuleInfo}; +use crate::host::module_host::{CallProcedureParams, InstancePoolTimeout, ModuleInfo, PooledCallError}; use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance}; use crate::host::{InvalidProcedureArguments, InvalidReducerArguments, NoSuchModule}; use anyhow::anyhow; @@ -321,6 +321,17 @@ impl ScheduledFunctionParams { pub(crate) enum CallScheduledFunctionError { #[error(transparent)] NoSuchModule(#[from] NoSuchModule), + #[error(transparent)] + PoolTimeout(#[from] InstancePoolTimeout), +} + +impl From for CallScheduledFunctionError { + fn from(err: PooledCallError) -> Self { + match err { + PooledCallError::NoSuchModule(err) => Self::NoSuchModule(err), + PooledCallError::PoolTimeout(err) => Self::PoolTimeout(err), + } + } } #[cfg(target_pointer_width = "64")] @@ -413,6 +424,19 @@ impl SchedulerActor { // If the module already exited, leave the `ScheduledFunction` in // the database for when the module restarts. Err(CallScheduledFunctionError::NoSuchModule(_)) => {} + // No procedure instance freed up in time, so the call never started. + // Put the item back rather than silently dropping the schedule. + Err(CallScheduledFunctionError::PoolTimeout(err)) => { + let function_name: &str = match &item { + QueueItem::Id { function_name, .. } => function_name, + QueueItem::VolatileNonatomicImmediate { function_name, .. } => function_name, + }; + log::warn!("scheduled procedure {function_name} did not start: {err}; retrying"); + let key = self.queue.insert(item, Duration::ZERO); + if let Some(id) = id { + self.key_map.insert(id, key); + } + } Ok(CallScheduledFunctionResult { reschedule: None }) => { // nothing to do } diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs index 6f357962bbb..8f96668727b 100644 --- a/crates/core/src/host/v8/mod.rs +++ b/crates/core/src/host/v8/mod.rs @@ -111,7 +111,7 @@ use std::num::NonZeroUsize; use std::os::raw::c_void; use std::panic::{self, AssertUnwindSafe}; use std::sync::{Arc, LazyLock}; -use std::time::Instant; +use std::time::{Duration, Instant}; use tokio::sync::{mpsc, oneshot}; use v8::script_compiler::{compile_module, Source}; use v8::{ @@ -265,6 +265,7 @@ impl V8RuntimeInner { load_balance_guard, core_pinner, procedure_instance_pool_size: config.procedure_instance_pool_size, + procedure_queue_timeout: config.procedure_queue_timeout, heap_policy: config.heap_policy, metrics, }; @@ -280,6 +281,7 @@ pub struct JsModule { load_balance_guard: Arc, core_pinner: CorePinner, procedure_instance_pool_size: NonZeroUsize, + procedure_queue_timeout: Option, heap_policy: V8HeapPolicyConfig, metrics: InstanceManagerMetrics, } @@ -305,6 +307,10 @@ impl JsModule { self.procedure_instance_pool_size } + pub(in crate::host) fn procedure_queue_timeout(&self) -> Option { + self.procedure_queue_timeout + } + async fn create_procedure_instance(&self) -> JsProcedureInstance { let program = self.program.clone(); let common = self.common.clone(); diff --git a/crates/core/src/host/wasmtime/mod.rs b/crates/core/src/host/wasmtime/mod.rs index 8deff0a532a..1ab3303bfd1 100644 --- a/crates/core/src/host/wasmtime/mod.rs +++ b/crates/core/src/host/wasmtime/mod.rs @@ -187,6 +187,7 @@ impl WasmtimeRuntime { core, init_inst: Box::new(init_inst), procedure_instance_pool_size: self.config.procedure_instance_pool_size, + procedure_queue_timeout: self.config.procedure_queue_timeout, }) } } diff --git a/crates/standalone/config.toml b/crates/standalone/config.toml index fdf85987338..8d4a0748188 100644 --- a/crates/standalone/config.toml +++ b/crates/standalone/config.toml @@ -27,11 +27,17 @@ directives = [ # Maximum number of WASM procedure instances per database. Omit to use the # number of cores reported by the OS. # procedure-instance-pool-size = 8 +# How long a call may wait for a free procedure instance before it fails with +# an error. Accepts a duration string or a number of seconds. Set to 0 to wait +# forever. Omit to use 30 seconds. +# procedure-queue-timeout = "30s" [v8] # Maximum number of JS procedure isolates per database. Omit to use the number # of cores reported by the OS. # procedure-instance-pool-size = 8 +# Same as `wasm.procedure-queue-timeout`, for JS procedure isolates. +# procedure-queue-timeout = "30s" [v8-heap-policy] # Check the V8 heap after this many requests. Set to 0 to disable. diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index c3a274048ac..4001ad50fd4 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/crates/standalone/src/subcommands/start.rs @@ -531,9 +531,11 @@ mod tests { [wasm] procedure-instance-pool-size = 4 + procedure-queue-timeout = "10s" [v8] procedure-instance-pool-size = 3 + procedure-queue-timeout = 0 [v8-heap-policy] heap-check-request-interval = 0 @@ -560,6 +562,11 @@ mod tests { assert!(!config.common.module_http.enabled); assert_eq!(config.common.wasm.procedure_instance_pool_size.get(), 4); assert_eq!(config.common.v8.procedure_instance_pool_size.get(), 3); + assert_eq!( + config.common.wasm.procedure_queue_timeout, + Some(Duration::from_secs(10)) + ); + assert_eq!(config.common.v8.procedure_queue_timeout, None); assert_eq!(config.common.v8.heap_policy.heap_check_request_interval, None); assert_eq!( config.common.v8.heap_policy.heap_check_time_interval, From 339cbbad748043f07245a3e42cc530efa73cd234 Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 5 Sep 2026 01:57:52 +0200 Subject: [PATCH 2/3] core: wait for a procedure instance off the websocket receive loop `enqueue_procedure` awaited the pool slot before returning, and the websocket receive task handles one message at a time. With a full pool this stalled every later message from that connection (subscribe, reducer calls, one-off queries) until a slot freed up, or now until the queue timeout elapsed. Only already established subscriptions kept flowing, since the send task is separate. Spawn the checkout and the call as a task once the call is validated, so the receive loop only pays for argument validation. Validation errors are still reported synchronously, and the result or the timeout error is delivered to the same target as before. --- crates/core/src/host/module_host.rs | 36 ++++++++++++++++++++++++----- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index 0009b6820dd..a175990bc92 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -2703,16 +2703,40 @@ impl ModuleHost { }; let procedure_name = name; - let guard_procedure_name = procedure_name.clone(); - scopeguard::defer_on_unwind!({ - log::error!("websocket procedure operation {guard_procedure_name} panicked"); - (self.on_panic)(); - }); - if let Err(err) = self.guard_closed() { return self.send_procedure_error(&procedure_name, timer, target, err.into()); } + // Waiting for a free procedure instance must not happen on the caller's + // websocket receive loop: that loop handles one message at a time, + // so a full pool would stall every later message from the connection. + let module = self.clone(); + tokio::spawn(async move { + let guard_procedure_name = procedure_name.clone(); + scopeguard::defer_on_unwind!({ + log::error!("websocket procedure operation {guard_procedure_name} panicked"); + (module.on_panic)(); + }); + + if let Err(err) = module + .run_procedure_from_pool(procedure_name, timer, params, target) + .await + { + log::warn!("failed to send procedure result: {err:#}"); + } + }); + Ok(()) + } + + /// Checks out a pooled procedure instance, runs `params` on it + /// and delivers the result (or the checkout error) to `target`. + async fn run_procedure_from_pool( + &self, + procedure_name: String, + timer: Option, + params: CallProcedureParams, + target: ProcedureResultTarget, + ) -> Result<(), BroadcastError> { match &*self.inner { ModuleHostInner::Js(host) => { let lease = match host.procedure_instances.get_instance().await { From 19abb54f697b5f1cbd7b4a28e89bdbdbbcbe4ce6 Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 5 Sep 2026 02:39:02 +0200 Subject: [PATCH 3/3] core: count procedure calls that time out waiting for the pool Add `spacetime_procedure_instance_pool_timeouts_total`, labelled by database and module type, so a saturated procedure pool shows up in metrics rather than only as a stalled client. The wait itself is already visible for WASM procedures through `spacetime_reducer_wait_time_sec` (the call timer starts before the checkout) and `spacetime_worker_instance_operation_queue_length`. --- crates/core/src/host/module_host.rs | 55 ++++++++++++++++++++++++++- crates/core/src/worker_metrics/mod.rs | 5 +++ 2 files changed, 58 insertions(+), 2 deletions(-) diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index a175990bc92..7905d7c4ac2 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -35,7 +35,7 @@ use derive_more::From; use indexmap::IndexSet; use itertools::Itertools; use parking_lot::Mutex; -use prometheus::{Histogram, HistogramTimer, IntGauge}; +use prometheus::{Histogram, HistogramTimer, IntCounter, IntGauge}; use rustc_hash::FxHashMap; use scopeguard::ScopeGuard; use smallvec::SmallVec; @@ -1286,6 +1286,42 @@ struct SharedJsMainInstanceManager { pub(in crate::host) struct InstanceManagerMetrics { module_instances: ModuleInstancesMetric, create_instance_time: CreateInstanceTimeMetric, + pool_timeouts: PoolTimeoutsMetric, +} + +/// Handle on the `spacetime_procedure_instance_pool_timeouts_total` label for a particular database +/// which calls `remove_label_values` to clean up on drop. +#[derive(Clone)] +struct PoolTimeoutsMetric { + inner: Arc, +} + +struct PoolTimeoutsMetricInner { + metric: IntCounter, + host_type: HostType, + database_identity: Identity, +} + +impl Drop for PoolTimeoutsMetricInner { + fn drop(&mut self) { + let _ = WORKER_METRICS + .procedure_instance_pool_timeouts + .remove_label_values(&self.database_identity, &self.host_type); + } +} + +impl PoolTimeoutsMetric { + fn new(host_type: HostType, database_identity: Identity) -> Self { + Self { + inner: Arc::new(PoolTimeoutsMetricInner { + metric: WORKER_METRICS + .procedure_instance_pool_timeouts + .with_label_values(&database_identity, &host_type), + host_type, + database_identity, + }), + } + } } /// Handle on the `spacetime_module_create_instance_time_seconds` label for a particular database @@ -1336,9 +1372,14 @@ impl InstanceManagerMetrics { Self { module_instances: ModuleInstancesMetric::new(host_type, database_identity), create_instance_time: CreateInstanceTimeMetric::new(host_type, database_identity), + pool_timeouts: PoolTimeoutsMetric::new(host_type, database_identity), } } + fn track_pool_timeout(&self) { + self.pool_timeouts.inner.metric.inc(); + } + pub(in crate::host) fn observe_instance_created(&self, duration: std::time::Duration) { self.create_instance_time.observe(duration); self.module_instances.inc(); @@ -1464,7 +1505,10 @@ impl ModuleInstanceManager { let permit = match self.slot_timeout { Some(timeout) => match tokio::time::timeout(timeout, acquire).await { Ok(permit) => permit, - Err(_elapsed) => return Err(InstancePoolTimeout(timeout)), + Err(_elapsed) => { + self.metrics.track_pool_timeout(); + return Err(InstancePoolTimeout(timeout)); + } }, None => acquire.await, }; @@ -3747,6 +3791,7 @@ mod tests { use crate::db::relational_db::tests_utils::{insert, with_auto_commit, TestDB}; use crate::messages::control_db::HostType; use crate::subscription::module_subscription_actor::ModuleSubscriptions; + use crate::worker_metrics::WORKER_METRICS; use spacetimedb_client_api_messages::websocket::{common::RowListLen as _, v1 as ws_v1, v2 as ws_v2}; use spacetimedb_lib::identity::AuthCtx; use spacetimedb_lib::{AlgebraicType, Identity}; @@ -3801,11 +3846,17 @@ mod tests { Some(timeout), ); + let timeouts = WORKER_METRICS + .procedure_instance_pool_timeouts + .with_label_values(&Identity::ZERO, &HostType::Wasm); + let timeouts_before = timeouts.get(); + let held = pool .get_instance() .await .expect("the first checkout gets the only slot"); assert_eq!(pool.get_instance().await.err(), Some(InstancePoolTimeout(timeout))); + assert_eq!(timeouts.get(), timeouts_before + 1); pool.return_instance(held); assert!(pool.get_instance().await.is_ok()); diff --git a/crates/core/src/worker_metrics/mod.rs b/crates/core/src/worker_metrics/mod.rs index 2777e270621..2e33fb8c40b 100644 --- a/crates/core/src/worker_metrics/mod.rs +++ b/crates/core/src/worker_metrics/mod.rs @@ -685,6 +685,11 @@ metrics_group!( #[buckets(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10, 50, 100)] pub module_create_instance_time_seconds: HistogramVec, + #[name = spacetime_procedure_instance_pool_timeouts_total] + #[help = "Calls that gave up waiting for a free procedure instance (see `procedure-queue-timeout`)"] + #[labels(db: Identity, module_type: HostType)] + pub procedure_instance_pool_timeouts: IntCounterVec, + #[name = spacetime_subscription_rows_examined] #[help = "Distribution of rows examined per subscription query"] #[labels(db: Identity, scan_type: str, table: str, unindexed_columns: str)]