From ea687ccb3380771ead8aa014ec437f2b19aedfd0 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Sat, 27 Jun 2026 14:40:23 -0400 Subject: [PATCH] fix(workflow): seed interval cold-start anchor so new schedules fire A brand-new interval workflow on a cold engine has no in-memory last_fired entry and no prior durable claim, so the scheduler resolves last = None. interval_should_fire then reads last = now and suppresses the tick (correct: wait a full interval), but the in-memory anchor is only written AFTER a won claim, and no claim is attempted until the prefilter passes. Every subsequent tick repeats with last = None, so the workflow suppresses forever. Extract interval_prefilter_should_fire (free fn over the last_fired map + a thin &self wrapper): on the cold-start None suppress path it seeds the anchor to now so the next tick counts from a real anchor and fires after one interval. It seeds ONLY when last was None; an existing Some anchor is mid-interval and must elapse on its own, so it is never advanced. A due/firing tick passes through without seeding (the post-claim path owns that write). Unit tests (no Db/Postgres; pure in-memory anchor state): - cold start seeds then fires after one interval - mid-interval suppress does not advance an existing anchor - a due fire passes through without seeding Caught by Max in cold review of the scheduled-workflow lane; predates this branch's claim work but lives on the exact lane being cleared. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-workflow/src/lib.rs | 152 +++++++++++++++++++++++++++++++- 1 file changed, 151 insertions(+), 1 deletion(-) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 6754d5f91..1bd6f2aca 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -341,6 +341,41 @@ impl WorkflowEngine { Ok(()) } + /// Interval prefilter: decide whether the interval workflow should fire this + /// tick, applying the cold-start anchor seed as a side effect. + /// + /// `last` is the resolved anchor (in-memory entry if present, else the + /// durable `latest_scheduled_workflow_fire` read). Returns `true` to proceed + /// to the durable claim, `false` to suppress this tick. + /// + /// Cold-start liveness: a brand-new interval workflow has no in-memory entry + /// AND no prior claim, so `last` is `None`. `interval_should_fire` then reads + /// `last = now` and suppresses — correct for the first tick (wait a full + /// interval), but the in-memory anchor is only written *after* a successful + /// claim, and no claim is attempted until the prefilter passes. Without + /// seeding, every subsequent tick repeats with `last = None` and the workflow + /// suppresses forever. So on the `None` suppress path we seed `now`: the next + /// tick counts from a real anchor and the workflow fires after one interval. + /// We seed ONLY when `last` was `None`; when `last` is `Some` we are correctly + /// mid-interval and must not advance the anchor, or it would never elapse. + fn interval_prefilter_should_fire( + &self, + community_id: CommunityId, + workflow_id: Uuid, + dur: &str, + last: Option>, + now: DateTime, + ) -> bool { + interval_prefilter_should_fire( + &self.last_fired, + community_id, + workflow_id, + dur, + last, + now, + ) + } + /// Background loop for scheduled (cron/interval) triggers. /// /// Ticks every 60 seconds. For each active workflow with a `Schedule` @@ -452,7 +487,13 @@ impl WorkflowEngine { } }, }; - if !interval_should_fire(dur, last, now, workflow.id) { + if !self.interval_prefilter_should_fire( + community_id, + workflow.id, + dur, + last, + now, + ) { continue; } match interval_fire_instant(dur, now, workflow.id) { @@ -688,6 +729,31 @@ fn interval_should_fire( } } +/// Interval prefilter decision + cold-start anchor seed. See the +/// [`WorkflowEngine::interval_prefilter_should_fire`] wrapper for the liveness +/// rationale. Free function over the `last_fired` map so it is unit-testable +/// without a `Db`/Postgres: the only state it touches is the in-memory anchor. +/// +/// Returns `true` to fire, `false` to suppress. On the cold-start `None` suppress +/// path it seeds `now` so the next tick has a real anchor; it never advances an +/// existing (`Some`) anchor, which is mid-interval and must elapse on its own. +fn interval_prefilter_should_fire( + last_fired: &DashMap<(CommunityId, Uuid), DateTime>, + community_id: CommunityId, + workflow_id: Uuid, + dur: &str, + last: Option>, + now: DateTime, +) -> bool { + if interval_should_fire(dur, last, now, workflow_id) { + return true; + } + if last.is_none() { + last_fired.insert((community_id, workflow_id), now); + } + false +} + /// Check emoji and filter-expression conditions that determine whether a /// matched workflow should actually fire. Extracted from `on_event` to keep /// the per-workflow loop body small. @@ -1041,6 +1107,90 @@ mod tests { ); } + // ── Interval cold-start liveness (Max's blocker on the scheduled lane) ── + // A brand-new interval workflow has no in-memory anchor and no prior durable + // claim, so the prefilter resolves `last = None`. Without seeding, every tick + // reads `None`, suppresses, and writes nothing — the workflow never fires. + // `interval_prefilter_should_fire` must seed `now` on that first suppress so a + // real anchor exists for the next tick. + + #[test] + fn interval_cold_start_seeds_anchor_then_fires_after_one_interval() { + let map: DashMap<(CommunityId, Uuid), DateTime> = DashMap::new(); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let wf = Uuid::new_v4(); + let t0 = Utc::now(); + + // Tick 1 (cold start): no in-memory entry, DB anchor is None → last = None. + let fired_1 = interval_prefilter_should_fire(&map, community, wf, "1h", None, t0); + assert!(!fired_1, "first tick must suppress (wait a full interval)"); + let seeded = map.get(&(community, wf)).map(|v| *v); + assert_eq!( + seeded, + Some(t0), + "first suppressed tick must seed the anchor to `now`, else it suppresses forever" + ); + + // Tick 2, mid-interval: caller now passes the seeded anchor as `last`. + let t1 = t0 + chrono::Duration::minutes(30); + let last = map.get(&(community, wf)).map(|v| *v); + let fired_2 = interval_prefilter_should_fire(&map, community, wf, "1h", last, t1); + assert!(!fired_2, "still mid-interval → suppress"); + assert_eq!( + map.get(&(community, wf)).map(|v| *v), + Some(t0), + "mid-interval suppress must NOT advance the anchor (or it would never elapse)" + ); + + // Tick 3, one interval elapsed → fire. + let t2 = t0 + chrono::Duration::hours(1); + let last = map.get(&(community, wf)).map(|v| *v); + let fired_3 = interval_prefilter_should_fire(&map, community, wf, "1h", last, t2); + assert!( + fired_3, + "after one full interval the cold-started workflow must fire" + ); + } + + #[test] + fn interval_prefilter_does_not_advance_existing_anchor_on_suppress() { + // Regression for the inverse bug: if a `Some` anchor were re-seeded to + // `now` on every suppressed tick, the interval would never elapse. + let map: DashMap<(CommunityId, Uuid), DateTime> = DashMap::new(); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let wf = Uuid::new_v4(); + let now = Utc::now(); + let anchor = now - chrono::Duration::minutes(10); // 10m into a 1h interval + map.insert((community, wf), anchor); + + let fired = interval_prefilter_should_fire(&map, community, wf, "1h", Some(anchor), now); + assert!(!fired, "mid-interval suppress"); + assert_eq!( + map.get(&(community, wf)).map(|v| *v), + Some(anchor), + "existing anchor must be preserved exactly, not advanced to now" + ); + } + + #[test] + fn interval_prefilter_passes_through_a_due_fire_without_touching_anchor() { + // When the interval has elapsed the prefilter returns true and leaves the + // anchor to the post-claim update path (which writes `now` only on a won + // claim), so the prefilter must not seed here. + let map: DashMap<(CommunityId, Uuid), DateTime> = DashMap::new(); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let wf = Uuid::new_v4(); + let now = Utc::now(); + let anchor = now - chrono::Duration::hours(2); // overdue on a 1h interval + + let fired = interval_prefilter_should_fire(&map, community, wf, "1h", Some(anchor), now); + assert!(fired, "overdue interval must fire"); + assert!( + map.get(&(community, wf)).is_none(), + "a firing tick must not seed via the prefilter; the post-claim path owns the write" + ); + } + #[test] fn workflow_config_defaults() { let cfg = WorkflowConfig::default();