Add deterministic scheduler interval regression tests

crates/core/src/host/scheduler.rs

@@ -519,7 +519,8 @@ fn prepare_scheduled_procedure_call(
             log::error!("could not determine scheduled procedure or its parameters: {err:#}");
             let reschedule = id.and_then(|id| {
                 let reschedule_from = (Timestamp::now(), Instant::now());
-                delete_scheduled_function_row(module_info, db, id, Some(tx), reschedule_from, inst_common, inst)
+                delete_scheduled_function_row(module_info, db, id, Some(tx), inst_common, inst)
+                    .and_then(|schedule_at| calculate_reschedule(schedule_at, reschedule_from))
             });
             return ScheduledProcedureStep::Done(CallScheduledFunctionResult { reschedule }, false);
         }
@@ -528,7 +529,8 @@ fn prepare_scheduled_procedure_call(
     // For scheduled procedures, it's incorrect to retry them if execution aborts midway,
     // so we must remove the schedule row before executing.
     let reschedule = id.and_then(|id| {
-        delete_scheduled_function_row(module_info, db, id, Some(tx), (timestamp, instant), inst_common, inst)
+        delete_scheduled_function_row(module_info, db, id, Some(tx), inst_common, inst)
+            .and_then(|schedule_at| calculate_reschedule(schedule_at, (timestamp, instant)))
     });
     ScheduledProcedureStep::Procedure { params, reschedule }
 }
@@ -555,7 +557,8 @@ fn call_scheduled_reducer_until_done(
             log::error!("could not determine scheduled reducer or its parameters: {err:#}");
             let reschedule = id.and_then(|id| {
                 let reschedule_from = (Timestamp::now(), Instant::now());
-                delete_scheduled_function_row(module_info, db, id, Some(tx), reschedule_from, inst_common, inst)
+                delete_scheduled_function_row(module_info, db, id, Some(tx), inst_common, inst)
+                    .and_then(|schedule_at| calculate_reschedule(schedule_at, reschedule_from))
             });
             return (CallScheduledFunctionResult { reschedule }, false);
         }
@@ -604,11 +607,13 @@ fn call_scheduled_reducer_with_tx(
     // as it might be, so catch it so we can handle it "gracefully". Panics will
     // print their message and backtrace when they occur, so we don't need to do
     // anything with the error payload.
-    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
-        inst_common.call_reducer_with_tx(Some(tx), params, inst)
-    }));
-    let reschedule =
-        id.and_then(|id| delete_scheduled_function_row(module_info, db, id, None, reschedule_from, inst_common, inst));
+    let (result, reschedule) = execute_then_reschedule(reschedule_from, || {
+        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
+            inst_common.call_reducer_with_tx(Some(tx), params, inst)
+        }));
+        let schedule_at = id.and_then(|id| delete_scheduled_function_row(module_info, db, id, None, inst_common, inst));
+        (result, schedule_at)
+    });
     // Currently, we drop the return value from the function call. In the future,
     // we might want to handle it somehow.
     let trapped = match result {
@@ -623,28 +628,41 @@ fn call_scheduled_reducer_with_tx(
 /// open and otherwise creating an internal transaction for the cleanup.
 ///
 /// 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`.
+/// deleted here; instead their `ScheduleAt` is returned to the caller for rescheduling.
 fn delete_scheduled_function_row(
     module_info: &ModuleInfo,
     db: &RelationalDB,
     id: ScheduledFunctionId,
     tx: Option<MutTxId>,
-    reschedule_from: (Timestamp, Instant),
     inst_common: &mut InstanceCommon,
     inst: &mut impl WasmInstance,
-) -> Option<Reschedule> {
-    let (timestamp, instant) = reschedule_from;
+) -> Option<ScheduleAt> {
     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, inst_common, inst)?;
+    delete_scheduled_function_row_with_tx(module_info, db, tx, id, inst_common, inst)
+}
+
+fn calculate_reschedule(schedule_at: ScheduleAt, reschedule_from: (Timestamp, Instant)) -> Option<Reschedule> {
     let ScheduleAt::Interval(dur) = schedule_at else {
         return None;
     };
+    let (timestamp, instant) = reschedule_from;
     Some(Reschedule {
         at_ts: schedule_at.to_timestamp_from(timestamp),
         at_real: instant + dur.to_duration_abs(),
     })
 }
 
+/// Executes a scheduled reducer and its cleanup while retaining the call time captured
+/// before execution. This keeps reducer runtime out of the next interval calculation.
+fn execute_then_reschedule<T>(
+    reschedule_from: (Timestamp, Instant),
+    execute: impl FnOnce() -> (T, Option<ScheduleAt>),
+) -> (T, Option<Reschedule>) {
+    let (result, schedule_at) = execute();
+    let reschedule = schedule_at.and_then(|schedule_at| calculate_reschedule(schedule_at, reschedule_from));
+    (result, reschedule)
+}
+
 /// Deletes a scheduled-row entry inside an existing mutable transaction.
 ///
 /// If the row describes a one-shot schedule, this also refreshes any stale views and
@@ -844,3 +862,31 @@ fn read_schedule_at(row: &RowRef<'_>, at_column: ColId) -> anyhow::Result<Schedu
     let schedule_at_av: AlgebraicValue = row.read_col(at_column)?;
     ScheduleAt::try_from(schedule_at_av).map_err(|e| anyhow!("Failed to convert 'scheduled_at' to ScheduleAt: {e:?}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn reducer_reschedules_from_call_time_not_completion_time() {
+        let interval = Duration::from_secs(5);
+        let call_timestamp = Timestamp::from_micros_since_unix_epoch(1_000_000);
+        let call_instant = Instant::now();
+        let completion_timestamp = call_timestamp + Duration::from_secs(30);
+        let completion_instant = call_instant + Duration::from_secs(30);
+
+        let (completion, reschedule) = execute_then_reschedule((call_timestamp, call_instant), || {
+            (
+                (completion_timestamp, completion_instant),
+                Some(ScheduleAt::Interval(interval.into())),
+            )
+        });
+        let reschedule = reschedule.unwrap();
+
+        assert_eq!(completion, (completion_timestamp, completion_instant));
+        assert_eq!(reschedule.at_ts, call_timestamp + interval);
+        assert_eq!(reschedule.at_real, call_instant + interval);
+        assert_ne!(reschedule.at_ts, completion_timestamp + interval);
+        assert_ne!(reschedule.at_real, completion_instant + interval);
+    }
+}