diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index a64a39aa9a2..78d8b414d12 100644 --- a/crates/core/src/host/scheduler.rs +++ b/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, - reschedule_from: (Timestamp, Instant), inst_common: &mut InstanceCommon, inst: &mut impl WasmInstance, -) -> Option { - let (timestamp, instant) = reschedule_from; +) -> 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, 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 { 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( + reschedule_from: (Timestamp, Instant), + execute: impl FnOnce() -> (T, Option), +) -> (T, Option) { + 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