mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
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 <tlongwell@block.xyz> Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
co-authored by
Tyler Longwell
parent
9e0534eecc
commit
ea687ccb33
@@ -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<DateTime<Utc>>,
|
||||
now: DateTime<Utc>,
|
||||
) -> 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<Utc>>,
|
||||
community_id: CommunityId,
|
||||
workflow_id: Uuid,
|
||||
dur: &str,
|
||||
last: Option<DateTime<Utc>>,
|
||||
now: DateTime<Utc>,
|
||||
) -> 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<Utc>> = 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<Utc>> = 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<Utc>> = 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();
|
||||
|
||||
Reference in New Issue
Block a user