fix(scheduler): preserve catch-up and retry overdue work

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,6 +205,8 @@ impl Owner {
             let mut recover = true;
             let mut previous_error = None;
             let mut planned_wake: Option<i64> = None;
+            let mut retries = BTreeMap::new();
+            let mut storage_retry = None;
             loop {
                 if owner.stop.is_cancelled() {
                     return;
@@ -215,41 +218,70 @@ 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()
                 };
-                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,
                     () = owner.scheduler.wake.notified() => {},
                     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; }
+                    } => {},
                 }
             }
         });
     }
     async fn tick_schedules(
         self: &Arc<Self>,
         recover: bool,
+        retries: &mut BTreeMap<String, FireRetry>,
     ) -> Result<(Option<i64>, Option<String>), 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::<Vec<_>>() {
             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 {
@@ -263,39 +295,38 @@ 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);
                 }
             }
         }
-        Ok((
-            catalog
-                .ready
-                .due
-                .range((
-                    std::ops::Bound::Excluded(now() as i64),
-                    std::ops::Bound::Unbounded,
-                ))
-                .next()
-                .map(|(at, _)| *at),
-            dispatch_error,
-        ))
+        retries.retain(|id, _| catalog.ready.pending.contains(id));
+        Ok((catalog.ready.next_after(loaded_at), dispatch_error))
     }
     async fn dispatch_scheduled(
         self: &Arc<Self>,
@@ -492,3 +523,59 @@ fn settlement(
         run_id,
     }
 }
+
+struct FireRetry {
+    fire_id: String,
+    retry: Retry,
+}
+
+impl FireRetry {
+    fn for_fire(&self, id: &str) -> Option<Retry> {
+        (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<Self>, 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,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    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);
+        }
+        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)));
+    }
+}

plugins/agent/native/src/owner/scheduler/repository.rs

@@ -46,6 +46,16 @@ fn due_at(plan: &Plan) -> Option<i64> {
     next.into_iter().chain(plan.task.expires_at).min()
 }
 impl ReadyIndex {
+    pub(super) fn next_after(&self, loaded_at: i64) -> Option<i64> {
+        // 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"
+        );
+    }
+}

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<Option<i64>, 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<Option<i64>, 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)
+    }
 }

plugins/agent/scheduler/engine/src/plan.rs

@@ -176,15 +176,27 @@ impl<E: TaskEffect, A: Clone> TaskPlan<E, A> {
         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<Option<&Fire<E, A>>, Error> {