Skip to content

Commit 3beda87

Browse files
Scheduled reducer: use timestamp from reducer params for next run (#5574)
# Description of Changes We are scheduling using `Timestamp::now()` instead of last run `timestamp`. It causes any interval reducer which has non-negligible execution time to gradually drift. I fixed it once back :| - #3657. # API and ABI breaking changes NA # Expected complexity level and risk 1 # Testing I want to add a test for it, but I don't know how to add non-flaky test for it. --------- Co-authored-by: joshua-spacetime <josh@clockworklabs.io>
1 parent 0935d32 commit 3beda87

1 file changed

Lines changed: 29 additions & 28 deletions

File tree

crates/core/src/host/scheduler.rs

Lines changed: 29 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -517,22 +517,19 @@ fn prepare_scheduled_procedure_call(
517517
Err(err) => {
518518
// All we can do here is log an error.
519519
log::error!("could not determine scheduled procedure or its parameters: {err:#}");
520-
let reschedule = delete_scheduled_function_row(module_info, db, id, Some(tx), None, inst_common, inst);
520+
let reschedule = id.and_then(|id| {
521+
let reschedule_from = (Timestamp::now(), Instant::now());
522+
delete_scheduled_function_row(module_info, db, id, Some(tx), reschedule_from, inst_common, inst)
523+
});
521524
return ScheduledProcedureStep::Done(CallScheduledFunctionResult { reschedule }, false);
522525
}
523526
};
524527

525528
// For scheduled procedures, it's incorrect to retry them if execution aborts midway,
526529
// so we must remove the schedule row before executing.
527-
let reschedule = delete_scheduled_function_row(
528-
module_info,
529-
db,
530-
id,
531-
Some(tx),
532-
Some((timestamp, instant)),
533-
inst_common,
534-
inst,
535-
);
530+
let reschedule = id.and_then(|id| {
531+
delete_scheduled_function_row(module_info, db, id, Some(tx), (timestamp, instant), inst_common, inst)
532+
});
536533
ScheduledProcedureStep::Procedure { params, reschedule }
537534
}
538535

@@ -548,20 +545,23 @@ fn call_scheduled_reducer_until_done(
548545
let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
549546

550547
let params = reducer_call_params_for_queued_item(module_info, db, &tx, item);
551-
let params = match params {
548+
let (timestamp, instant, params) = match params {
552549
// If the function was already deleted, leave the `ScheduledFunction`
553550
// in the database for when the module restarts.
554551
Ok(None) => return (CallScheduledFunctionResult { reschedule: None }, false),
555-
Ok(Some((_timestamp, _instant, params))) => params,
552+
Ok(Some(params)) => params,
556553
Err(err) => {
557554
// All we can do here is log an error.
558555
log::error!("could not determine scheduled reducer or its parameters: {err:#}");
559-
let reschedule = delete_scheduled_function_row(module_info, db, id, Some(tx), None, inst_common, inst);
556+
let reschedule = id.and_then(|id| {
557+
let reschedule_from = (Timestamp::now(), Instant::now());
558+
delete_scheduled_function_row(module_info, db, id, Some(tx), reschedule_from, inst_common, inst)
559+
});
560560
return (CallScheduledFunctionResult { reschedule }, false);
561561
}
562562
};
563563

564-
call_scheduled_reducer_with_tx(module_info, db, id, tx, params, inst_common, inst)
564+
call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst)
565565
}
566566

567567
fn scheduled_item_id(item: &QueueItem) -> Option<ScheduledFunctionId> {
@@ -571,11 +571,13 @@ fn scheduled_item_id(item: &QueueItem) -> Option<ScheduledFunctionId> {
571571
}
572572
}
573573

574+
#[allow(clippy::too_many_arguments)]
574575
fn call_scheduled_reducer_with_tx(
575576
module_info: &ModuleInfo,
576577
db: &RelationalDB,
577578
id: Option<ScheduledFunctionId>,
578579
mut tx: MutTxId,
580+
reschedule_from: (Timestamp, Instant),
579581
params: CallReducerParams,
580582
inst_common: &mut InstanceCommon,
581583
inst: &mut impl WasmInstance,
@@ -605,7 +607,8 @@ fn call_scheduled_reducer_with_tx(
605607
let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
606608
inst_common.call_reducer_with_tx(Some(tx), params, inst)
607609
}));
608-
let reschedule = delete_scheduled_function_row(module_info, db, id, None, None, inst_common, inst);
610+
let reschedule =
611+
id.and_then(|id| delete_scheduled_function_row(module_info, db, id, None, reschedule_from, inst_common, inst));
609612
// Currently, we drop the return value from the function call. In the future,
610613
// we might want to handle it somehow.
611614
let trapped = match result {
@@ -624,23 +627,21 @@ fn call_scheduled_reducer_with_tx(
624627
fn delete_scheduled_function_row(
625628
module_info: &ModuleInfo,
626629
db: &RelationalDB,
627-
id: Option<ScheduledFunctionId>,
630+
id: ScheduledFunctionId,
628631
tx: Option<MutTxId>,
629-
timestamp: Option<(Timestamp, Instant)>,
632+
reschedule_from: (Timestamp, Instant),
630633
inst_common: &mut InstanceCommon,
631634
inst: &mut impl WasmInstance,
632635
) -> Option<Reschedule> {
633-
id.and_then(|id| {
634-
let (timestamp, instant) = timestamp.unwrap_or_else(|| (Timestamp::now(), Instant::now()));
635-
let tx = tx.unwrap_or_else(|| db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal));
636-
let schedule_at = delete_scheduled_function_row_with_tx(module_info, db, tx, id, inst_common, inst)?;
637-
let ScheduleAt::Interval(dur) = schedule_at else {
638-
return None;
639-
};
640-
Some(Reschedule {
641-
at_ts: schedule_at.to_timestamp_from(timestamp),
642-
at_real: instant + dur.to_duration_abs(),
643-
})
636+
let (timestamp, instant) = reschedule_from;
637+
let tx = tx.unwrap_or_else(|| db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal));
638+
let schedule_at = delete_scheduled_function_row_with_tx(module_info, db, tx, id, inst_common, inst)?;
639+
let ScheduleAt::Interval(dur) = schedule_at else {
640+
return None;
641+
};
642+
Some(Reschedule {
643+
at_ts: schedule_at.to_timestamp_from(timestamp),
644+
at_real: instant + dur.to_duration_abs(),
644645
})
645646
}
646647

0 commit comments

Comments
 (0)