From 76b6293fced25536d3b0c8aa1aa2983a60ea7f56 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Tue, 21 Jul 2026 15:08:37 +0530 Subject: [PATCH 1/3] schedule reducer from timestamp --- crates/core/src/host/scheduler.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index c5ec94796d5..4a3d2b0d864 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -548,11 +548,11 @@ fn call_scheduled_reducer_until_done( let tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal); let params = reducer_call_params_for_queued_item(module_info, db, &tx, item); - let params = match params { + let (timestamp, instant, params) = match params { // If the function was already deleted, leave the `ScheduledFunction` // in the database for when the module restarts. Ok(None) => return (CallScheduledFunctionResult { reschedule: None }, false), - Ok(Some((_timestamp, _instant, params))) => params, + Ok(Some(params)) => params, Err(err) => { // All we can do here is log an error. log::error!("could not determine scheduled reducer or its parameters: {err:#}"); @@ -561,7 +561,16 @@ fn call_scheduled_reducer_until_done( } }; - call_scheduled_reducer_with_tx(module_info, db, id, tx, params, inst_common, inst) + call_scheduled_reducer_with_tx( + module_info, + db, + id, + tx, + Some((timestamp, instant)), + params, + inst_common, + inst, + ) } fn scheduled_item_id(item: &QueueItem) -> Option { @@ -576,6 +585,7 @@ fn call_scheduled_reducer_with_tx( db: &RelationalDB, id: Option, mut tx: MutTxId, + reschedule_from: Option<(Timestamp, Instant)>, params: CallReducerParams, inst_common: &mut InstanceCommon, inst: &mut impl WasmInstance, @@ -605,7 +615,7 @@ fn call_scheduled_reducer_with_tx( let result = panic::catch_unwind(panic::AssertUnwindSafe(|| { inst_common.call_reducer_with_tx(Some(tx), params, inst) })); - let reschedule = delete_scheduled_function_row(module_info, db, id, None, None, inst_common, inst); + let reschedule = delete_scheduled_function_row(module_info, db, id, None, reschedule_from, inst_common, inst); // Currently, we drop the return value from the function call. In the future, // we might want to handle it somehow. let trapped = match result { From cc03604069dc6124c2adbd99518d9ec88b2547ab Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Tue, 21 Jul 2026 15:33:31 +0530 Subject: [PATCH 2/3] clippy --- crates/core/src/host/scheduler.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index 4a3d2b0d864..a57216765ad 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -580,6 +580,7 @@ fn scheduled_item_id(item: &QueueItem) -> Option { } } +#[allow(clippy::too_many_arguments)] fn call_scheduled_reducer_with_tx( module_info: &ModuleInfo, db: &RelationalDB, From c1cd04db80aa8e90590b971eb80fe7bbbcf87ae4 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Mon, 27 Jul 2026 15:31:56 +0530 Subject: [PATCH 3/3] Not use Option in function signature --- crates/core/src/host/scheduler.rs | 62 +++++++++++++------------------ 1 file changed, 26 insertions(+), 36 deletions(-) diff --git a/crates/core/src/host/scheduler.rs b/crates/core/src/host/scheduler.rs index a57216765ad..a64a39aa9a2 100644 --- a/crates/core/src/host/scheduler.rs +++ b/crates/core/src/host/scheduler.rs @@ -517,22 +517,19 @@ fn prepare_scheduled_procedure_call( Err(err) => { // All we can do here is log an error. log::error!("could not determine scheduled procedure or its parameters: {err:#}"); - let reschedule = delete_scheduled_function_row(module_info, db, id, Some(tx), None, inst_common, inst); + 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) + }); return ScheduledProcedureStep::Done(CallScheduledFunctionResult { reschedule }, false); } }; // For scheduled procedures, it's incorrect to retry them if execution aborts midway, // so we must remove the schedule row before executing. - let reschedule = delete_scheduled_function_row( - module_info, - db, - id, - Some(tx), - Some((timestamp, instant)), - inst_common, - inst, - ); + 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 } } @@ -556,21 +553,15 @@ fn call_scheduled_reducer_until_done( Err(err) => { // All we can do here is log an error. log::error!("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); + 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) + }); return (CallScheduledFunctionResult { reschedule }, false); } }; - call_scheduled_reducer_with_tx( - module_info, - db, - id, - tx, - Some((timestamp, instant)), - params, - inst_common, - inst, - ) + call_scheduled_reducer_with_tx(module_info, db, id, tx, (timestamp, instant), params, inst_common, inst) } fn scheduled_item_id(item: &QueueItem) -> Option { @@ -586,7 +577,7 @@ fn call_scheduled_reducer_with_tx( db: &RelationalDB, id: Option, mut tx: MutTxId, - reschedule_from: Option<(Timestamp, Instant)>, + reschedule_from: (Timestamp, Instant), params: CallReducerParams, inst_common: &mut InstanceCommon, inst: &mut impl WasmInstance, @@ -616,7 +607,8 @@ fn call_scheduled_reducer_with_tx( let result = panic::catch_unwind(panic::AssertUnwindSafe(|| { inst_common.call_reducer_with_tx(Some(tx), params, inst) })); - let reschedule = delete_scheduled_function_row(module_info, db, id, None, reschedule_from, inst_common, inst); + let reschedule = + id.and_then(|id| delete_scheduled_function_row(module_info, db, id, None, reschedule_from, inst_common, inst)); // Currently, we drop the return value from the function call. In the future, // we might want to handle it somehow. let trapped = match result { @@ -635,23 +627,21 @@ fn call_scheduled_reducer_with_tx( fn delete_scheduled_function_row( module_info: &ModuleInfo, db: &RelationalDB, - id: Option, + id: ScheduledFunctionId, tx: Option, - timestamp: Option<(Timestamp, Instant)>, + reschedule_from: (Timestamp, Instant), inst_common: &mut InstanceCommon, inst: &mut impl WasmInstance, ) -> Option { - id.and_then(|id| { - let (timestamp, instant) = timestamp.unwrap_or_else(|| (Timestamp::now(), Instant::now())); - 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)?; - let ScheduleAt::Interval(dur) = schedule_at else { - return None; - }; - Some(Reschedule { - at_ts: schedule_at.to_timestamp_from(timestamp), - at_real: instant + dur.to_duration_abs(), - }) + let (timestamp, instant) = reschedule_from; + 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)?; + let ScheduleAt::Interval(dur) = schedule_at else { + return None; + }; + Some(Reschedule { + at_ts: schedule_at.to_timestamp_from(timestamp), + at_real: instant + dur.to_duration_abs(), }) }