From 61dfc95cc17c2f206edd1bd96cbf7a4553443999 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Tue, 28 Jul 2026 14:15:55 +0530 Subject: [PATCH 1/5] schedule function metrics --- crates/core/src/host/scheduler.rs | 66 ++++++++++++++++++++++++--- crates/core/src/worker_metrics/mod.rs | 6 +++ 2 files changed, 66 insertions(+), 6 deletions(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index c5ec94796d5..93385687ce7 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -6,6 +6,7 @@ use crate::db::relational_db::RelationalDB; use crate::host::module_host::{CallProcedureParams, ModuleInfo}; use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance}; use crate::host::{InvalidProcedureArguments, InvalidReducerArguments, NoSuchModule}; +use crate::worker_metrics::WORKER_METRICS; use anyhow::anyhow; use core::time::Duration; use futures::{FutureExt, StreamExt}; @@ -190,6 +191,9 @@ const MAX_SCHEDULE_DELAY: Duration = Duration::from_millis( (1 << (6 * 6)) - 1, ); +/// Record when a scheduled function starts more than this long after it was due. +const RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD: Duration = Duration::from_millis(20); + #[derive(thiserror::Error, Debug)] pub enum ScheduleError { #[error("Unable to schedule with long delay at {0:?}")] @@ -447,8 +451,9 @@ struct Reschedule { enum ScheduledProcedureStep { Done(CallScheduledFunctionResult, bool), Procedure { - params: CallProcedureParams, + params: Box, reschedule: Option, + delay: Option<(Arc, Duration)>, }, } @@ -465,9 +470,17 @@ pub(super) async fn call_scheduled_procedure( // even though it has been already moved during `delete_scheduled_function_row` call. match next_step { ScheduledProcedureStep::Done(result, trapped) => (result, trapped), - ScheduledProcedureStep::Procedure { params, reschedule } => { + ScheduledProcedureStep::Procedure { + params, + reschedule, + delay, + } => { + if let Some((function_name, delay)) = delay.as_ref() { + record_scheduled_function_delay(module_info, function_name, *delay); + } + // Execute the procedure. See above for commentary on `catch_unwind()`. - let result = panic::AssertUnwindSafe(inst_common.call_procedure(params, inst)) + let result = panic::AssertUnwindSafe(inst_common.call_procedure(*params, inst)) .catch_unwind() .await; @@ -504,6 +517,7 @@ fn prepare_scheduled_procedure_call( inst: &mut impl WasmInstance, ) -> ScheduledProcedureStep { let ScheduledFunctionParams(item) = params; + let delay = scheduled_function_delay_for_item(&item); let id = scheduled_item_id(&item); let db = &**module_info.relational_db(); let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); @@ -533,7 +547,11 @@ fn prepare_scheduled_procedure_call( inst_common, inst, ); - ScheduledProcedureStep::Procedure { params, reschedule } + ScheduledProcedureStep::Procedure { + params: Box::new(params), + reschedule, + delay, + } } fn call_scheduled_reducer_until_done( @@ -543,6 +561,7 @@ fn call_scheduled_reducer_until_done( inst: &mut impl WasmInstance, ) -> (CallScheduledFunctionResult, bool) { let ScheduledFunctionParams(item) = params; + let delay = scheduled_function_delay_for_item(&item); let id = scheduled_item_id(&item); let db = &**module_info.relational_db(); let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); @@ -555,13 +574,17 @@ fn call_scheduled_reducer_until_done( Ok(Some((_timestamp, _instant, params))) => params, Err(err) => { // All we can do here is log an error. - log::error!("could not determine scheduled reducer or its parameters: {err:#}"); + log::warn!("could not determine scheduled reducer or its parameters: {err:#}"); let reschedule = delete_scheduled_function_row(module_info, db, id, Some(tx), None, inst_common, inst); return (CallScheduledFunctionResult { reschedule }, false); } }; - call_scheduled_reducer_with_tx(module_info, db, id, tx, params, inst_common, inst) + let result = call_scheduled_reducer_with_tx(module_info, db, id, tx, params, inst_common, inst); + if let Some((function_name, delay)) = delay.as_ref() { + record_scheduled_function_delay(module_info, function_name, *delay); + } + result } fn scheduled_item_id(item: &QueueItem) -> Option { @@ -571,6 +594,37 @@ fn scheduled_item_id(item: &QueueItem) -> Option { } } +fn scheduled_function_delay_for_item(item: &QueueItem) -> Option<(Arc, Duration)> { + match item { + QueueItem::Id { function_name, at, .. } => { + Some((function_name.clone(), scheduled_function_delay(Timestamp::now(), *at))) + } + QueueItem::VolatileNonatomicImmediate { .. } => None, + } +} + +fn record_scheduled_function_delay(module_info: &ModuleInfo, function_name: &str, delay: Duration) { + if delay <= RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD { + return; + } + WORKER_METRICS + .scheduled_function_delay + .with_label_values(&module_info.database_identity, function_name) + .observe(delay.as_secs_f64()); + + log::warn!( + "scheduled function `{}` for database {} is delayed by {:.3}s, exceeding the {:.3}s threshold", + function_name, + module_info.database_identity, + delay.as_secs_f64(), + RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD.as_secs_f64(), + ); +} + +fn scheduled_function_delay(actual: Timestamp, requested: Timestamp) -> Duration { + actual.duration_since(requested).unwrap_or(Duration::ZERO) +} + fn call_scheduled_reducer_with_tx( module_info: &ModuleInfo, db: &RelationalDB, diff --git a/crates/core/src/worker_metrics/mod.rs b/crates/core/src/worker_metrics/mod.rs index 1b8de9efba7..0c78fb12f8d 100644 --- a/crates/core/src/worker_metrics/mod.rs +++ b/crates/core/src/worker_metrics/mod.rs @@ -510,6 +510,12 @@ metrics_group!( #[buckets(100e-6, 500e-6, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10)] pub reducer_wait_time: HistogramVec, + #[name = spacetime_scheduled_function_delay_seconds] + #[help = "The amount of time (in seconds) between when a scheduled function was due and when the scheduler began invoking it"] + #[labels(db: Identity, function: str)] + #[buckets(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60, 300)] + pub scheduled_function_delay: HistogramVec, + #[name = spacetime_worker_wasm_instance_errors_total] #[help = "The number of fatal WASM instance errors, such as reducer panics."] #[labels(database_identity: Identity, module_hash: Hash, reducer_symbol: str)] From 999263fadbe8848e79c7e9d3a5de3a4217acd25d Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Tue, 28 Jul 2026 14:18:20 +0530 Subject: [PATCH 2/5] threshould to be 50ms --- crates/core/src/host/scheduler.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index 93385687ce7..585d167088c 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -192,7 +192,7 @@ const MAX_SCHEDULE_DELAY: Duration = Duration::from_millis( ); /// Record when a scheduled function starts more than this long after it was due. -const RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD: Duration = Duration::from_millis(20); +const RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD: Duration = Duration::from_millis(50); #[derive(thiserror::Error, Debug)] pub enum ScheduleError { From 9b1c804fbd472130364c0be842bb986f8e5cd80a Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Wed, 29 Jul 2026 14:58:30 +0530 Subject: [PATCH 3/5] lint --- crates/core/src/host/scheduler.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index 61c79b3eb19..36cb6a4efc0 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -580,7 +580,8 @@ fn call_scheduled_reducer_until_done( } }; - let result = call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst); + let result = + call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst); if let Some((function_name, delay)) = delay.as_ref() { record_scheduled_function_delay(module_info, function_name, *delay); } From 4b4974690515c3e932340ccc973159d3983bd32f Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 30 Jul 2026 18:28:44 +0530 Subject: [PATCH 4/5] memoize metrics & use timestamp from params --- crates/core/src/host/module_host.rs | 46 +++++++++++++++++++++++++++++ crates/core/src/host/scheduler.rs | 42 +++++++++++++------------- 2 files changed, 67 insertions(+), 21 deletions(-) diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs index 35f88434330..a84e55876e6 100644 --- a/crates/core/src/host/module_host.rs +++ b/crates/core/src/host/module_host.rs @@ -36,6 +36,7 @@ use indexmap::IndexSet; use itertools::Itertools; use parking_lot::Mutex; use prometheus::{Histogram, HistogramTimer, IntGauge}; +use rustc_hash::FxHashMap; use scopeguard::ScopeGuard; use smallvec::SmallVec; use spacetimedb_auth::identity::ConnectionAuthCtx; @@ -254,6 +255,45 @@ pub struct ModuleMetrics { pub request_round_trip_subscribe: Histogram, pub request_round_trip_unsubscribe: Histogram, pub request_round_trip_sql: Histogram, + scheduled_function_delay: ScheduledFunctionDelayMetrics, +} + +#[derive(Debug)] +struct ScheduledFunctionDelayMetrics { + database_identity: Identity, + metrics_by_function: Mutex, Histogram>>, +} + +impl ScheduledFunctionDelayMetrics { + fn new(database_identity: &Identity) -> Self { + Self { + database_identity: *database_identity, + metrics_by_function: Mutex::new(FxHashMap::default()), + } + } + + fn observe(&self, function_name: &str, delay: Duration) { + let mut metrics_by_function = self.metrics_by_function.lock(); + let metric = metrics_by_function + .entry(Box::::from(function_name)) + .or_insert_with(|| { + WORKER_METRICS + .scheduled_function_delay + .with_label_values(&self.database_identity, function_name) + }); + metric.observe(delay.as_secs_f64()); + } +} + +impl Drop for ScheduledFunctionDelayMetrics { + fn drop(&mut self) { + let metrics_by_function = std::mem::take(self.metrics_by_function.get_mut()); + for function_name in metrics_by_function.keys() { + let _ = WORKER_METRICS + .scheduled_function_delay + .remove_label_values(&self.database_identity, function_name); + } + } } impl ModuleMetrics { @@ -272,6 +312,7 @@ impl ModuleMetrics { let request_round_trip_sql = WORKER_METRICS .request_round_trip .with_label_values(&WorkloadType::Sql, db, ""); + let scheduled_function_delay = ScheduledFunctionDelayMetrics::new(db); Self { connected_clients, ws_clients_spawned, @@ -279,8 +320,13 @@ impl ModuleMetrics { request_round_trip_subscribe, request_round_trip_unsubscribe, request_round_trip_sql, + scheduled_function_delay, } } + + pub(in crate::host) fn observe_scheduled_function_delay(&self, function_name: &str, delay: Duration) { + self.scheduled_function_delay.observe(function_name, delay); + } } impl ModuleInfo { diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index 36cb6a4efc0..146f082c733 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -6,7 +6,6 @@ use crate::db::relational_db::RelationalDB; use crate::host::module_host::{CallProcedureParams, ModuleInfo}; use crate::host::wasm_common::module_host_actor::{InstanceCommon, WasmInstance}; use crate::host::{InvalidProcedureArguments, InvalidReducerArguments, NoSuchModule}; -use crate::worker_metrics::WORKER_METRICS; use anyhow::anyhow; use core::time::Duration; use futures::{FutureExt, StreamExt}; @@ -191,8 +190,8 @@ const MAX_SCHEDULE_DELAY: Duration = Duration::from_millis( (1 << (6 * 6)) - 1, ); -/// Record when a scheduled function starts more than this long after it was due. -const RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD: Duration = Duration::from_millis(50); +/// Warn when a scheduled function starts more than this long after it was due. +const SCHEDULED_FUNCTION_DELAY_WARNING_THRESHOLD: Duration = Duration::from_millis(10); #[derive(thiserror::Error, Debug)] pub enum ScheduleError { @@ -517,7 +516,7 @@ fn prepare_scheduled_procedure_call( inst: &mut impl WasmInstance, ) -> ScheduledProcedureStep { let ScheduledFunctionParams(item) = params; - let delay = scheduled_function_delay_for_item(&item); + let delay = scheduled_function_delay_context_for_item(&item); let id = scheduled_item_id(&item); let db = &**module_info.relational_db(); let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); @@ -544,6 +543,8 @@ fn prepare_scheduled_procedure_call( let reschedule = id.and_then(|id| { delete_scheduled_function_row(module_info, db, id, Some(tx), (timestamp, instant), inst_common, inst) }); + let delay = + delay.map(|(function_name, requested_at)| (function_name, scheduled_function_delay(timestamp, requested_at))); ScheduledProcedureStep::Procedure { params: Box::new(params), reschedule, @@ -558,7 +559,7 @@ fn call_scheduled_reducer_until_done( inst: &mut impl WasmInstance, ) -> (CallScheduledFunctionResult, bool) { let ScheduledFunctionParams(item) = params; - let delay = scheduled_function_delay_for_item(&item); + let delay = scheduled_function_delay_context_for_item(&item); let id = scheduled_item_id(&item); let db = &**module_info.relational_db(); let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); @@ -580,12 +581,11 @@ fn call_scheduled_reducer_until_done( } }; - let result = - call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst); - if let Some((function_name, delay)) = delay.as_ref() { - record_scheduled_function_delay(module_info, function_name, *delay); + if let Some((function_name, requested_at)) = delay { + let delay = scheduled_function_delay(timestamp, requested_at); + record_scheduled_function_delay(module_info, &function_name, delay); } - result + call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst) } fn scheduled_item_id(item: &QueueItem) -> Option { @@ -595,30 +595,28 @@ fn scheduled_item_id(item: &QueueItem) -> Option { } } -fn scheduled_function_delay_for_item(item: &QueueItem) -> Option<(Arc, Duration)> { +fn scheduled_function_delay_context_for_item(item: &QueueItem) -> Option<(Arc, Timestamp)> { match item { - QueueItem::Id { function_name, at, .. } => { - Some((function_name.clone(), scheduled_function_delay(Timestamp::now(), *at))) - } + QueueItem::Id { function_name, at, .. } => Some((function_name.clone(), *at)), QueueItem::VolatileNonatomicImmediate { .. } => None, } } fn record_scheduled_function_delay(module_info: &ModuleInfo, function_name: &str, delay: Duration) { - if delay <= RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD { + module_info + .metrics + .observe_scheduled_function_delay(function_name, delay); + + if delay <= SCHEDULED_FUNCTION_DELAY_WARNING_THRESHOLD { return; } - WORKER_METRICS - .scheduled_function_delay - .with_label_values(&module_info.database_identity, function_name) - .observe(delay.as_secs_f64()); log::warn!( "scheduled function `{}` for database {} is delayed by {:.3}s, exceeding the {:.3}s threshold", function_name, module_info.database_identity, delay.as_secs_f64(), - RECORD_SCHEDULED_FUNCTION_DELAY_THRESHOLD.as_secs_f64(), + SCHEDULED_FUNCTION_DELAY_WARNING_THRESHOLD.as_secs_f64(), ); } @@ -810,7 +808,9 @@ fn call_params_for_queued_item( ) -> anyhow::Result<(Timestamp, Instant, T)>, ) -> anyhow::Result> { Ok(Some(match item { - QueueItem::Id { id, function_name, at } => { + QueueItem::Id { + id, function_name, at, .. + } => { let Some(schedule_row) = get_schedule_row_mut(tx, db, id)? else { // If the row is not found, it means the schedule is cancelled by the user. return Ok(None); From eb4b7979151ccbe5e5e21d348763ad3267436313 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Thu, 30 Jul 2026 18:32:48 +0530 Subject: [PATCH 5/5] warning threshold --- crates/core/src/host/scheduler.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index 146f082c733..c7d58287331 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -191,7 +191,7 @@ const MAX_SCHEDULE_DELAY: Duration = Duration::from_millis( ); /// Warn when a scheduled function starts more than this long after it was due. -const SCHEDULED_FUNCTION_DELAY_WARNING_THRESHOLD: Duration = Duration::from_millis(10); +const SCHEDULED_FUNCTION_DELAY_WARNING_THRESHOLD: Duration = Duration::from_millis(30); #[derive(thiserror::Error, Debug)] pub enum ScheduleError {