From 2bbfe935eb724b4dbd897b8111c04f4e6e513c3c Mon Sep 17 00:00:00 2001 From: kesonm <106971880+kesonm@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:55:59 +0800 Subject: [PATCH 1/2] fix(scheduler): preserve catch-up and retry overdue work Fixes #6827 Generated-by: OpenAI Codex --- .../native/src/owner/scheduler/dispatch.rs | 51 ++++-- .../native/src/owner/scheduler/repository.rs | 36 ++++ .../agent/scheduler/engine/src/plan/edit.rs | 8 +- .../agent/scheduler/engine/src/plan/tests.rs | 48 +++++ products/maka/cli/tests/producer_reopen.rs | 171 +++++++++++++++++- 5 files changed, 294 insertions(+), 20 deletions(-) diff --git a/plugins/agent/native/src/owner/scheduler/dispatch.rs b/plugins/agent/native/src/owner/scheduler/dispatch.rs index 134f0172b2..bb5a73ebd2 100644 --- a/plugins/agent/native/src/owner/scheduler/dispatch.rs +++ b/plugins/agent/native/src/owner/scheduler/dispatch.rs @@ -204,6 +204,7 @@ impl Owner { let mut recover = true; let mut previous_error = None; let mut planned_wake: Option = None; + let mut retry_delay = Duration::from_secs(1); loop { if owner.stop.is_cancelled() { return; @@ -219,6 +220,7 @@ impl Owner { Ok(result) => { recover = false; result }, Err(error) => (None, Some(error)), }; + let delay = retry_wake(delay, error.is_some(), now() as i64, &mut retry_delay); if let Some(error) = &error && previous_error.as_ref() != Some(error) { eprintln!("Scheduled tasks: {error}"); } previous_error = error; @@ -243,7 +245,8 @@ impl Owner { recover: bool, ) -> Result<(Option, Option), String> { let _gate = self.scheduler.gate.lock().await; - let mut catalog = self.scheduler.load_ready(now() as i64).await?; + let loaded_at = now() as i64; + let mut catalog = self.scheduler.load_ready(loaded_at).await?; let mut dispatch_error = None; for id in catalog.plans.keys().cloned().collect::>() { if self.stop.is_cancelled() { @@ -284,18 +287,7 @@ impl Owner { } } } - Ok(( - catalog - .ready - .due - .range(( - std::ops::Bound::Excluded(now() as i64), - std::ops::Bound::Unbounded, - )) - .next() - .map(|(at, _)| *at), - dispatch_error, - )) + Ok((catalog.ready.next_after(loaded_at), dispatch_error)) } async fn dispatch_scheduled( self: &Arc, @@ -491,3 +483,36 @@ fn settlement( run_id, } } + +fn retry_wake(next: Option, failed: bool, now: i64, backoff: &mut Duration) -> Option { + if failed { + let retry_at = now.saturating_add(backoff.as_millis() as i64); + *backoff = (*backoff * 2).min(Duration::from_secs(30)); + Some(next.map_or(retry_at, |due| due.min(retry_at))) + } else { + *backoff = Duration::from_secs(1); + next + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn failed_dispatch_keeps_a_bounded_retry_wake_without_postponing_other_tasks() { + let mut backoff = Duration::from_secs(1); + let mut at = 0; + for wait in [1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000] { + let wake = retry_wake(None, true, at, &mut backoff).unwrap(); + assert_eq!(wake - at, wait); + at = wake; + } + assert_eq!( + retry_wake(Some(at + 10), true, at, &mut backoff), + Some(at + 10) + ); + assert_eq!(retry_wake(None, false, at, &mut backoff), None); + assert_eq!(retry_wake(None, true, at, &mut backoff), Some(at + 1_000)); + } +} diff --git a/plugins/agent/native/src/owner/scheduler/repository.rs b/plugins/agent/native/src/owner/scheduler/repository.rs index 7d15ebb775..e1aae4e8a4 100644 --- a/plugins/agent/native/src/owner/scheduler/repository.rs +++ b/plugins/agent/native/src/owner/scheduler/repository.rs @@ -46,6 +46,16 @@ fn due_at(plan: &Plan) -> Option { next.into_iter().chain(plan.task.expires_at).min() } impl ReadyIndex { + pub(super) fn next_after(&self, loaded_at: i64) -> Option { + // A deadline crossed during loading still needs an immediate next tick. + self.due + .range(( + std::ops::Bound::Excluded(loaded_at), + std::ops::Bound::Unbounded, + )) + .next() + .map(|(at, _)| *at) + } fn update(&mut self, id: &str, old: Option<&Plan>, new: Option<&Plan>) { self.pending.remove(id); if let Some(at) = old.and_then(due_at) @@ -570,3 +580,29 @@ impl Saved { } } } + +#[cfg(test)] +mod wake_tests { + use super::*; + + #[test] + fn deadline_crossed_during_loading_remains_the_next_wake() { + let ready = ReadyIndex { + due: BTreeMap::from([(1_001, std::collections::BTreeSet::from(["task".into()]))]), + ..Default::default() + }; + let loaded_at = 1_000; + assert!(ready.due.range(..=loaded_at).next().is_none()); + let finished_at = 1_002; + let wake = ready.next_after(loaded_at).unwrap(); + assert!( + wake <= finished_at, + "an unprocessed due task must wake immediately" + ); + assert_eq!( + ready.next_after(1_001), + None, + "do not spin on an already processed deadline" + ); + } +} diff --git a/plugins/agent/scheduler/engine/src/plan/edit.rs b/plugins/agent/scheduler/engine/src/plan/edit.rs index 37229bee9b..3e93c64818 100644 --- a/plugins/agent/scheduler/engine/src/plan/edit.rs +++ b/plugins/agent/scheduler/engine/src/plan/edit.rs @@ -17,7 +17,7 @@ * under the License. */ -use super::TaskPlan; +use super::{Misfire, TaskPlan}; use crate::{ Error, command::Update, @@ -124,7 +124,11 @@ impl TaskPlan { next.task.next_fire_at = None; } else { next.task.status = Status::Active; - if next.task.next_fire_at.is_none_or(|at| at <= now) && next.pending.is_none() { + if next.pending.is_none() + && (next.task.next_fire_at.is_none() + || (next.misfire == Misfire::Skip + && next.task.next_fire_at.is_some_and(|at| at < now))) + { next.task.next_fire_at = Some( next.task .schedule diff --git a/plugins/agent/scheduler/engine/src/plan/tests.rs b/plugins/agent/scheduler/engine/src/plan/tests.rs index b0f14120f7..532bc0ae51 100644 --- a/plugins/agent/scheduler/engine/src/plan/tests.rs +++ b/plugins/agent/scheduler/engine/src/plan/tests.rs @@ -164,3 +164,51 @@ fn reminder_pause_and_snooze_cancel_only_a_delivery_that_has_not_started() { "pause still permits reconciliation of a started original delivery" ); } + +#[test] +fn resume_obeys_misfire_policy_and_preserves_due_now() { + for misfire in [super::Misfire::Latest, super::Misfire::Skip] { + for resume_at in [2_000, 3_500] { + let mut original = plan(Delivery::Work("work".into())); + original.task.expires_at = None; + original.misfire = misfire; + let mut resumed = original.pause(1_500).resume(resume_at).unwrap(); + resumed.validate().unwrap(); + if misfire == super::Misfire::Latest || resume_at == 2_000 { + let fire = resumed.claim(resume_at).unwrap().unwrap().clone(); + assert_eq!(fire.scheduled_at, 2_000); + resumed.settle(outcome(fire.id, resume_at)).unwrap(); + assert!(resumed.claim(resume_at).unwrap().is_none()); + } else { + assert_eq!(resumed.task.next_fire_at, Some(4_000)); + assert!(resumed.claim(resume_at).unwrap().is_none()); + } + } + } +} + +#[test] +fn overdue_once_can_resume_with_latest_but_not_skip() { + let mut original = plan(Delivery::Work("work".into())); + original.task.schedule = Schedule::Once { run_at: 2_000 }; + let paused = original.pause(1_500); + assert!(paused.resume(3_000).is_err()); + let mut paused = paused; + paused.misfire = super::Misfire::Latest; + let mut resumed = paused.resume(3_000).unwrap(); + resumed.validate().unwrap(); + let fire = resumed.claim(3_000).unwrap().unwrap().clone(); + assert_eq!(fire.scheduled_at, 2_000); + resumed.settle(outcome(fire.id, 3_000)).unwrap(); + assert_eq!(resumed.task.status, Status::Completed); + assert!(resumed.claim(3_001).unwrap().is_none()); + assert_eq!(paused.resume(4_000).unwrap().task.status, Status::Expired); +} + +#[test] +fn resume_keeps_the_original_pending_fire_for_reconciliation() { + let mut original = plan(Delivery::Work("work".into())); + let fire = original.claim(2_000).unwrap().unwrap().clone(); + let mut resumed = original.pause(2_100).resume(3_500).unwrap(); + assert_eq!(resumed.claim(3_500).unwrap(), Some(&fire)); +} diff --git a/products/maka/cli/tests/producer_reopen.rs b/products/maka/cli/tests/producer_reopen.rs index fca7c77c9d..66006b83e7 100644 --- a/products/maka/cli/tests/producer_reopen.rs +++ b/products/maka/cli/tests/producer_reopen.rs @@ -714,6 +714,7 @@ fn goal_identity(current: &Value) -> Value { )] async fn body( producer: Producer, + retry_workspace: bool, sandbox: &Sandbox, hosts: &mut Vec, clients: &mut Vec, @@ -984,6 +985,20 @@ async fn body( assert_eq!(proposal["task"]["fireCount"], 0); assert_eq!(proposal["task"]["runs"], json!([])); assert!(proposal["pendingFire"].is_null()); + if retry_workspace { + let due = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 10_000; + mutate(client, "agent", "schedule", json!({"kind":"update", "taskId":id, + "revision":proposal["revision"], "input":{"timezone":"UTC","misfire":"latest", + "task":{"title":"workspace retry", "intentBody":model.marker, + "schedule":{"kind":"interval","everySeconds":1,"startAt":due}, + "effect":proposal["task"]["effect"],"maxFires":1,"expiresAt":null}} + }), Attachments::new()).await?; + } + let proposal = schedule(client, &id).await?; let approved = mutate( client, "agent", @@ -993,6 +1008,36 @@ async fn body( ) .await?; assert_eq!(approved["kind"], "committed"); + let retry_fire = if retry_workspace { + let held = sandbox.home.join("workspace-unavailable"); + tokio::fs::rename(&sandbox.workspace, &held) + .await + .map_err(|e| e.to_string())?; + let pending = phase("claim while workspace is unavailable", async { + loop { + let detail = schedule(client, &id).await?; + if let Some(fire) = detail["pendingFire"].as_str() { + break Ok(fire.to_owned()); + } + tokio::task::yield_now().await; + } + }) + .await?; + // Deliberate fault duration: keep the workspace absent across the + // claim's own notification and the first timed retry. Reads do not wake it. + tokio::time::sleep(Duration::from_secs(3)).await; + let blocked = schedule(client, &id).await?; + assert_eq!(blocked["pendingFire"], pending); + assert_eq!(blocked["task"]["fireCount"], 0); + assert_eq!(model.producer_requests.load(Ordering::SeqCst), 0); + // Restore the same directory identity without a Host mutation or wake. + tokio::fs::rename(&held, &sandbox.workspace) + .await + .map_err(|e| e.to_string())?; + Some(pending) + } else { + None + }; // No triggerNow or clock injection. Wait for actual due and Message-admitted settlement. let settled = phase("actual due and one settled Fire", async { loop { @@ -1012,6 +1057,9 @@ async fn body( let fires = task["runs"].as_array().unwrap(); assert_eq!(fires.len(), 1); let fire = &fires[0]; + if let Some(original) = retry_fire { + assert_eq!(fire["id"], original); + } assert_eq!(fire["outcome"], "ok"); assert_eq!(fire["message"], "Message admitted"); let message = format!( @@ -1219,7 +1267,7 @@ async fn body( Ok(()) } -async fn acceptance(producer: Producer) { +async fn acceptance(producer: Producer, retry_workspace: bool) { let sandbox = Sandbox::new() .await .expect("owned empty Root/HOME/workspace"); @@ -1251,7 +1299,7 @@ async fn acceptance(producer: Producer) { let mut inspectors = Vec::new(); let outcome = AssertUnwindSafe(tokio::time::timeout(Duration::from_secs(150), async { tokio::select! { - result = body(producer, &sandbox, &mut hosts, &mut clients, &mut inspectors, &model, &mut observations, &base) => result, + result = body(producer, retry_workspace, &sandbox, &mut hosts, &mut clients, &mut inspectors, &model, &mut observations, &base) => result, result = &mut server => { server_collected = true; Err(format!("model accept task ended: {result:?}")) }, } })).catch_unwind().await; @@ -1343,15 +1391,128 @@ async fn acceptance(producer: Producer) { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn goal_iteration_one_fails_original_run_without_redispatch_after_reopen() { - acceptance(Producer::Goal).await; + acceptance(Producer::Goal, false).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn scheduler_user_approved_due_fire_fails_original_run_without_redispatch_after_reopen() { - acceptance(Producer::Scheduler).await; + acceptance(Producer::Scheduler, false).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn finite_goal_budget_remains_cumulative_across_two_host_reopens() { - acceptance(Producer::GoalBudget).await; + acceptance(Producer::GoalBudget, false).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn scheduler_resumes_overdue_latest_reminders_through_public_host() { + fn wall_now() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + } + let sandbox = Sandbox::new().await.unwrap(); + let mut host = Host::spawn(&sandbox).unwrap(); + let mut clients = Vec::new(); + let result = AssertUnwindSafe(phase("overdue reminder resume", async { + let (connection, _) = host.ready().await?; + let journal = Journal::open(&sandbox.journal).await?; + clients.push(Client::new(connection, journal, "receipts".into())?); + let client = &clients[0]; + let due = wall_now() + 10_000; + let mut ids = Vec::new(); + for recurrence in [ + json!({"kind":"once","runAt":due}), + json!({"kind":"interval","everySeconds":60,"startAt":due}), + ] { + let created = mutate( + client, + "agent", + "schedule", + json!({ + "kind":"create", "input": {"timezone":"UTC", "misfire":"latest", + "task":{"title":"issue 6827 catch-up", "intentBody":"local regression", + "schedule":recurrence,"effect":{"kind":"reminder"}, + "maxFires":null,"expiresAt":null}} + }), + Attachments::new(), + ) + .await?; + let id = created["taskId"] + .as_str() + .ok_or("created taskId missing")? + .to_owned(); + let detail = schedule(client, &id).await?; + assert!( + detail["pendingFire"].is_null(), + "pause must precede the deadline" + ); + mutate( + client, + "agent", + "schedule", + json!({"kind":"pause", "taskId":id, + "revision":detail["revision"]}), + Attachments::new(), + ) + .await?; + ids.push(id); + } + // Wait for the actual schedule boundary, not a fixed readiness delay. + while wall_now() <= due { + tokio::time::sleep(Duration::from_millis(due.saturating_sub(wall_now()) + 1)).await; + } + for id in ids { + let paused = schedule(client, &id).await?; + assert_eq!(paused["task"]["status"], "paused"); + mutate( + client, + "agent", + "schedule", + json!({"kind":"resume", "taskId":id, + "revision":paused["revision"]}), + Attachments::new(), + ) + .await?; + let ready = loop { + let detail = schedule(client, &id).await?; + if detail["pendingFire"].is_string() { + break detail; + } + tokio::task::yield_now().await; + }; + assert_eq!(ready["task"]["nextFireAt"], due); + let fire = ready["pendingFire"].clone(); + mutate( + client, + "agent", + "schedule", + json!({"kind":"acknowledge", "taskId":id, + "revision":ready["revision"], "fireId":fire}), + Attachments::new(), + ) + .await?; + let settled = schedule(client, &id).await?; + assert_eq!(settled["task"]["fireCount"], 1); + assert_eq!(settled["task"]["runs"][0]["id"], fire); + assert!(settled["pendingFire"].is_null()); + } + Ok(()) + })) + .catch_unwind() + .await; + for client in clients { + client.close().await.unwrap(); + } + host.cleanup().await.unwrap(); + match result { + Ok(result) => result.unwrap(), + Err(panic) => std::panic::resume_unwind(panic), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn scheduler_retries_original_fire_when_workspace_returns_without_host_mutation() { + acceptance(Producer::Scheduler, true).await; } From 10f2480c1a9c08a9f78288d16ea132ac2c1399ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CM4n5ter=E2=80=9D?= Date: Fri, 13 Nov 2026 20:07:36 +0800 Subject: [PATCH 2/2] fix(scheduler): catch up the latest fire and enforce retry deadlines Generated-by: OpenAI Codex --- .../native/src/owner/scheduler/dispatch.rs | 162 ++++++++++++------ plugins/agent/scheduler/engine/src/cron.rs | 74 ++++++-- plugins/agent/scheduler/engine/src/plan.rs | 18 +- .../agent/scheduler/engine/src/plan/edit.rs | 3 + .../agent/scheduler/engine/src/plan/tests.rs | 12 +- .../agent/scheduler/engine/src/schedule.rs | 53 +++++- 6 files changed, 246 insertions(+), 76 deletions(-) diff --git a/plugins/agent/native/src/owner/scheduler/dispatch.rs b/plugins/agent/native/src/owner/scheduler/dispatch.rs index bb5a73ebd2..db1da911ef 100644 --- a/plugins/agent/native/src/owner/scheduler/dispatch.rs +++ b/plugins/agent/native/src/owner/scheduler/dispatch.rs @@ -19,6 +19,7 @@ use super::*; use maka_agent_state::queue::MessageResolution; +use tokio::time::Instant; impl CapturedRun { pub(in crate::owner) fn validate_background_target( &self, @@ -204,7 +205,8 @@ impl Owner { let mut recover = true; let mut previous_error = None; let mut planned_wake: Option = None; - let mut retry_delay = Duration::from_secs(1); + let mut retries = BTreeMap::new(); + let mut storage_retry = None; loop { if owner.stop.is_cancelled() { return; @@ -216,15 +218,28 @@ impl Owner { let Some(producer) = owner.producer().await else { return; }; - let (delay, error) = match owner.tick_schedules(recover).await { - Ok(result) => { recover = false; result }, - Err(error) => (None, Some(error)), + if storage_retry.is_none_or(|retry: Retry| retry.ready(Instant::now())) { + let error = match owner.tick_schedules(recover, &mut retries).await { + Ok((next, error)) => { + planned_wake = next; + recover = false; + storage_retry = None; + error + } + Err(error) => { + storage_retry = Some(Retry::after_failure(storage_retry, Instant::now())); + Some(error) + } + }; + if let Some(error) = &error + && previous_error.as_ref() != Some(error) { eprintln!("Scheduled tasks: {error}"); } + previous_error = error; + } + let retry_at = if let Some(retry) = storage_retry { + Some(retry.at) + } else { + retries.values().map(|retry| retry.retry.at).min() }; - let delay = retry_wake(delay, error.is_some(), now() as i64, &mut retry_delay); - if let Some(error) = &error - && previous_error.as_ref() != Some(error) { eprintln!("Scheduled tasks: {error}"); } - previous_error = error; - planned_wake = delay; drop(producer); tokio::select! { () = owner.stop.cancelled() => return, @@ -232,10 +247,15 @@ impl Owner { changed = changes.changed() => if changed.is_err() { return; }, changed = run_changes.changed() => if changed.is_err() { return; }, () = async { - if let Some(at) = delay { + if let Some(at) = planned_wake.filter(|_| storage_retry.is_none()) { tokio::time::sleep(Duration::from_millis((at as u64).saturating_sub(now()))).await; } else { std::future::pending::<()>().await; } } => {}, + () = async { + if let Some(at) = retry_at { + tokio::time::sleep_until(at).await; + } else { std::future::pending::<()>().await; } + } => {}, } } }); @@ -243,6 +263,7 @@ impl Owner { async fn tick_schedules( self: &Arc, recover: bool, + retries: &mut BTreeMap, ) -> Result<(Option, Option), String> { let _gate = self.scheduler.gate.lock().await; let loaded_at = now() as i64; @@ -252,7 +273,15 @@ impl Owner { if self.stop.is_cancelled() { break; } - let mut plan = catalog.plans[&id].plan.clone(); + let plan = &catalog.plans[&id].plan; + let retry = plan + .pending + .as_ref() + .and_then(|fire| retries.get(&id).and_then(|retry| retry.for_fire(&fire.id))); + if retry.is_some_and(|retry| !retry.ready(Instant::now())) { + continue; + } + let mut plan = plan.clone(); let before = serde_json::to_value(&plan).map_err(reason)?; let at = now() as i64; if recover { @@ -266,27 +295,37 @@ impl Owner { .commit(&mut catalog, &id, Some(plan.clone()), None) .await?; } - if let Some(fire) = plan.pending.clone() { - // A reminder remains durable until the user acknowledges it. - // No paint attempt or socket write is described as delivery. - if !matches!(fire.effect, ScheduledEffect::Reminder) { - match self.dispatch_scheduled(&fire).await { - Ok(Some(run)) => { - plan.settle(run).map_err(reason)?; - self.scheduler - .commit(&mut catalog, &id, Some(plan.clone()), None) - .await?; - } - Ok(None) => {} - Err(error) => { - // Keep this original fire for reconciliation without - // preventing independent tasks from making progress. - dispatch_error.get_or_insert(error); - } - } + let Some(fire) = plan.pending.as_ref() else { + continue; + }; + // A reminder remains durable until the user acknowledges it. + if matches!(fire.effect, ScheduledEffect::Reminder) { + continue; + } + match self.dispatch_scheduled(fire).await { + Ok(Some(run)) => { + plan.settle(run).map_err(reason)?; + self.scheduler + .commit(&mut catalog, &id, Some(plan), None) + .await?; + retries.remove(&id); + } + Ok(None) => { + retries.remove(&id); + } + Err(error) => { + retries.insert( + id, + FireRetry { + fire_id: fire.id.clone(), + retry: Retry::after_failure(retry, Instant::now()), + }, + ); + dispatch_error.get_or_insert(error); } } } + retries.retain(|id, _| catalog.ready.pending.contains(id)); Ok((catalog.ready.next_after(loaded_at), dispatch_error)) } async fn dispatch_scheduled( @@ -484,14 +523,36 @@ fn settlement( } } -fn retry_wake(next: Option, failed: bool, now: i64, backoff: &mut Duration) -> Option { - if failed { - let retry_at = now.saturating_add(backoff.as_millis() as i64); - *backoff = (*backoff * 2).min(Duration::from_secs(30)); - Some(next.map_or(retry_at, |due| due.min(retry_at))) - } else { - *backoff = Duration::from_secs(1); - next +struct FireRetry { + fire_id: String, + retry: Retry, +} + +impl FireRetry { + fn for_fire(&self, id: &str) -> Option { + (self.fire_id == id).then_some(self.retry) + } +} + +#[derive(Clone, Copy)] +struct Retry { + at: Instant, + delay: Duration, +} + +impl Retry { + fn ready(self, now: Instant) -> bool { + self.at <= now + } + + fn after_failure(previous: Option, now: Instant) -> Self { + let delay = previous.map_or(Duration::from_secs(1), |retry| { + (retry.delay * 2).min(Duration::from_secs(30)) + }); + Self { + at: now + delay, + delay, + } } } @@ -500,19 +561,20 @@ mod tests { use super::*; #[test] - fn failed_dispatch_keeps_a_bounded_retry_wake_without_postponing_other_tasks() { - let mut backoff = Duration::from_secs(1); - let mut at = 0; - for wait in [1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000] { - let wake = retry_wake(None, true, at, &mut backoff).unwrap(); - assert_eq!(wake - at, wait); - at = wake; + fn early_wakes_leave_only_the_failed_fire_waiting() { + let now = Instant::now(); + let failure = FireRetry { + fire_id: "failed".into(), + retry: Retry::after_failure(None, now), + }; + for elapsed in [0, 250, 999, 1_000] { + let at = now + Duration::from_millis(elapsed); + let may_dispatch = |id| failure.for_fire(id).is_none_or(|retry| retry.ready(at)); + assert!(may_dispatch("independent")); + assert_eq!(may_dispatch("failed"), elapsed == 1_000); } - assert_eq!( - retry_wake(Some(at + 10), true, at, &mut backoff), - Some(at + 10) - ); - assert_eq!(retry_wake(None, false, at, &mut backoff), None); - assert_eq!(retry_wake(None, true, at, &mut backoff), Some(at + 1_000)); + let retry = Retry::after_failure(Some(failure.retry), failure.retry.at); + assert!(!retry.ready(now + Duration::from_millis(2_999))); + assert!(retry.ready(now + Duration::from_secs(3))); } } diff --git a/plugins/agent/scheduler/engine/src/cron.rs b/plugins/agent/scheduler/engine/src/cron.rs index af539b11fa..b949bc1807 100644 --- a/plugins/agent/scheduler/engine/src/cron.rs +++ b/plugins/agent/scheduler/engine/src/cron.rs @@ -141,27 +141,69 @@ impl Cron { if date < first || !self.matches_date(date) { continue; } - let mut next = None; - for hour in self.hour.values() { - for minute in self.minute.values() { - let local = date.at(hour, minute, 0, 0); - let ambiguous = zone.to_ambiguous_zoned(local); - // Gaps have no matching wall time; folds have two - // real occurrences, selected in timestamp order. - for resolved in [ambiguous.clone().earlier()?, ambiguous.later()?] { - let at = resolved.timestamp().as_millisecond(); - if resolved.datetime() == local && at >= lower { - next = Some(next.map_or(at, |previous: i64| previous.min(at))); - } - } - } + if let Some(next) = self.on_date(date, zone, |at| at >= lower, i64::min)? { + return Ok(Some(next)); } - if next.is_some() { - return Ok(next); + } + } + } + Ok(None) + } + + pub fn latest(&self, at: i64, not_before: i64, zone: &TimeZone) -> Result, Error> { + if at < not_before { + return Ok(None); + } + let first = Timestamp::from_millisecond(not_before)? + .to_zoned(zone.clone()) + .date(); + let last = Timestamp::from_millisecond(at)? + .to_zoned(zone.clone()) + .date(); + for year in (first.year()..=last.year()).rev() { + for month in (1..=12).rev().filter(|month| self.month.contains(*month)) { + if (year == first.year() && month < first.month()) + || (year == last.year() && month > last.month()) + { + continue; + } + let month_start = Date::new(year, month, 1)?; + for day in (1..=month_start.days_in_month()).rev() { + let date = Date::new(year, month, day)?; + if date < first || date > last || !self.matches_date(date) { + continue; + } + let in_range = |candidate| (not_before..=at).contains(&candidate); + if let Some(latest) = self.on_date(date, zone, in_range, i64::max)? { + return Ok(Some(latest)); } } } } Ok(None) } + + fn on_date( + &self, + date: Date, + zone: &TimeZone, + in_range: impl Fn(i64) -> bool, + select: impl Fn(i64, i64) -> i64, + ) -> Result, Error> { + let mut selected = None; + for hour in self.hour.values() { + for minute in self.minute.values() { + let local = date.at(hour, minute, 0, 0); + let ambiguous = zone.to_ambiguous_zoned(local); + // Gaps have no matching wall time; folds have two real occurrences. + for resolved in [ambiguous.clone().earlier()?, ambiguous.later()?] { + let at = resolved.timestamp().as_millisecond(); + if resolved.datetime() == local && in_range(at) { + selected = Some(selected.map_or(at, |previous| select(previous, at))); + } + } + } + } + Ok(selected) + } } diff --git a/plugins/agent/scheduler/engine/src/plan.rs b/plugins/agent/scheduler/engine/src/plan.rs index b13817c32d..c14cabc2a3 100644 --- a/plugins/agent/scheduler/engine/src/plan.rs +++ b/plugins/agent/scheduler/engine/src/plan.rs @@ -176,15 +176,27 @@ impl TaskPlan { if self.task.expires_at.is_some_and(|expires| now >= expires) { self.task.status = Status::Expired; self.task.next_fire_at = None; - } else if self.misfire == Misfire::Skip && self.task.next_fire_at.is_some_and(|at| at < now) - { - self.advance(now)?; + } else if self.task.next_fire_at.is_some_and(|at| at < now) { + match self.misfire { + Misfire::Skip => self.advance(now)?, + Misfire::Latest => self.catch_up(now)?, + } } if before != (self.task.status, self.task.next_fire_at) { self.task.updated_at = self.task.updated_at.max(now); } Ok(()) } + fn catch_up(&mut self, now: i64) -> Result<(), Error> { + if let Some(next) = self.task.next_fire_at.filter(|at| *at < now) { + let latest = self + .task + .schedule + .latest_at(now, &TimeZone::get(&self.timezone)?)?; + self.task.next_fire_at = Some(latest.map_or(next, |at| at.max(next))); + } + Ok(()) + } /// Returns the same intent while settlement is pending. The repository must /// commit the changed Plan before the caller may dispatch the Fire. pub fn claim(&mut self, now: i64) -> Result>, Error> { diff --git a/plugins/agent/scheduler/engine/src/plan/edit.rs b/plugins/agent/scheduler/engine/src/plan/edit.rs index 3e93c64818..dc6d5bb85e 100644 --- a/plugins/agent/scheduler/engine/src/plan/edit.rs +++ b/plugins/agent/scheduler/engine/src/plan/edit.rs @@ -136,6 +136,9 @@ impl TaskPlan { .ok_or_else(|| invalid("schedule has no remaining trigger"))?, ); } + if next.pending.is_none() && next.misfire == Misfire::Latest { + next.catch_up(now)?; + } next.check_expiry()?; } next.task.updated_at = next.task.updated_at.max(now); diff --git a/plugins/agent/scheduler/engine/src/plan/tests.rs b/plugins/agent/scheduler/engine/src/plan/tests.rs index 532bc0ae51..9ca5839ec5 100644 --- a/plugins/agent/scheduler/engine/src/plan/tests.rs +++ b/plugins/agent/scheduler/engine/src/plan/tests.rs @@ -176,7 +176,8 @@ fn resume_obeys_misfire_policy_and_preserves_due_now() { resumed.validate().unwrap(); if misfire == super::Misfire::Latest || resume_at == 2_000 { let fire = resumed.claim(resume_at).unwrap().unwrap().clone(); - assert_eq!(fire.scheduled_at, 2_000); + let expected = if resume_at == 2_000 { 2_000 } else { 3_000 }; + assert_eq!(fire.scheduled_at, expected); resumed.settle(outcome(fire.id, resume_at)).unwrap(); assert!(resumed.claim(resume_at).unwrap().is_none()); } else { @@ -187,6 +188,15 @@ fn resume_obeys_misfire_policy_and_preserves_due_now() { } } +#[test] +fn reopen_catches_up_only_the_latest_missed_occurrence() { + let mut plan = plan(Delivery::Work("work".into())); + plan.task.expires_at = None; + plan.misfire = super::Misfire::Latest; + plan.recover(3_500).unwrap(); + assert_eq!(plan.claim(3_500).unwrap().unwrap().scheduled_at, 3_000); +} + #[test] fn overdue_once_can_resume_with_latest_but_not_skip() { let mut original = plan(Delivery::Work("work".into())); diff --git a/plugins/agent/scheduler/engine/src/schedule.rs b/plugins/agent/scheduler/engine/src/schedule.rs index f649d9e3a6..06ddfbd0d5 100644 --- a/plugins/agent/scheduler/engine/src/schedule.rs +++ b/plugins/agent/scheduler/engine/src/schedule.rs @@ -107,7 +107,7 @@ impl Schedule { if *anchor_at > after { Some(*anchor_at) } else { - calendar(*recurrence, *anchor_at, after, zone)? + calendar(*recurrence, *anchor_at, after, zone, false)? } } Self::Cron { @@ -117,6 +117,35 @@ impl Schedule { }; Ok(next.filter(|next| *next > after && timestamp(*next).is_ok())) } + + /// The latest occurrence at or before the supplied instant. + pub(crate) fn latest_at(&self, at: i64, zone: &TimeZone) -> Result, Error> { + self.validate()?; + timestamp(at)?; + let latest = match self { + Self::Once { run_at } => Some(*run_at), + Self::Interval { + every_seconds, + start_at, + } if *start_at <= at => { + let step = i64::from(*every_seconds) * 1000; + Some(at - (at - start_at) % step) + } + Self::Interval { .. } => None, + Self::Calendar { + recurrence, + anchor_at, + } if *anchor_at <= at => { + calendar(*recurrence, *anchor_at, at, zone, true)?.filter(|at| *at >= *anchor_at) + } + Self::Calendar { .. } => None, + Self::Cron { + expression, + start_at, + } => Cron::parse(expression)?.latest(at, *start_at, zone)?, + }; + Ok(latest.filter(|latest| *latest <= at && timestamp(*latest).is_ok())) + } } pub(crate) fn timestamp(milliseconds: i64) -> Result { if milliseconds < 0 { @@ -129,10 +158,12 @@ fn calendar( anchor: i64, after: i64, zone: &TimeZone, + backward: bool, ) -> Result, Error> { let anchor = timestamp(anchor)?.to_zoned(zone.clone()); let current = timestamp(after)?.to_zoned(zone.clone()); let date = current.date(); + let direction = if backward { -1 } else { 1 }; let mut offset = 0; loop { let candidate = match recurrence { @@ -141,9 +172,9 @@ fn calendar( Err(_) => return Ok(None), }, Recurrence::Weekly => { - let days = (anchor.weekday().to_monday_zero_offset() - - date.weekday().to_monday_zero_offset()) - .rem_euclid(7); + let difference = anchor.weekday().to_monday_zero_offset() + - date.weekday().to_monday_zero_offset(); + let days = (difference * direction).rem_euclid(7) * direction; match date.checked_add((i64::from(days) + offset * 7).days()) { Ok(date) => date, Err(_) => return Ok(None), @@ -169,10 +200,10 @@ fn calendar( .compatible()? .timestamp() .as_millisecond(); - if at > after { + if if backward { at <= after } else { at > after } { return Ok(Some(at)); } - offset += 1; + offset += i64::from(direction); } } @@ -205,6 +236,12 @@ mod tests { .unwrap(), Some(ms("2026-03-09T06:30:00Z")) ); + assert_eq!( + cron("30 1 * * *") + .latest_at(ms("2026-11-01T06:45:00Z"), &ny) + .unwrap(), + Some(ms("2026-11-01T06:30:00Z")) + ); let utc = TimeZone::UTC; assert_eq!( cron("0 0 13 * 1") @@ -266,6 +303,10 @@ mod tests { monthly.next_after(february, &ny).unwrap(), Some(ms("2026-03-31T13:00:00Z")) ); + assert_eq!( + monthly.latest_at(ms("2026-03-30T12:00:00Z"), &ny).unwrap(), + Some(february) + ); let daily = Schedule::Calendar { recurrence: Recurrence::Daily, anchor_at: ms("2026-03-07T07:30:00Z"),