Track scheduled function delay

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(50);
+
 #[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<CallProcedureParams>,
         reschedule: Option<Reschedule>,
+        delay: Option<(Arc<str>, 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);
@@ -530,7 +544,11 @@ 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)
     });
-    ScheduledProcedureStep::Procedure { params, reschedule }
+    ScheduledProcedureStep::Procedure {
+        params: Box::new(params),
+        reschedule,
+        delay,
+    }
 }
 
 fn call_scheduled_reducer_until_done(
@@ -540,6 +558,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);
@@ -561,7 +580,12 @@ fn call_scheduled_reducer_until_done(
         }
     };
 
-    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);
+    }
+    result
 }
 
 fn scheduled_item_id(item: &QueueItem) -> Option<ScheduledFunctionId> {
@@ -571,6 +595,37 @@ fn scheduled_item_id(item: &QueueItem) -> Option<ScheduledFunctionId> {
     }
 }
 
+fn scheduled_function_delay_for_item(item: &QueueItem) -> Option<(Arc<str>, 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)
+}
+
 #[allow(clippy::too_many_arguments)]
 fn call_scheduled_reducer_with_tx(
     module_info: &ModuleInfo,

crates/core/src/worker_metrics/mod.rs

@@ -534,6 +534,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)]