jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 14:31:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflow): wire community-scoped durable claim into scheduler

Browse Source 
The scheduled-workflow fire path never called the durable claim
primitives (`claim_scheduled_workflow_fire` had zero callers outside the
DB wrappers and tests), so with N relay pods every pod that saw a due
cron/interval row created a run and executed the side effect — a
multi-pod duplicate-run bug. And the claim primitive itself was still
keyed by bare `workflow_id` (`WHERE w.id = $1`) despite the schema's
`(community_id, id)` workflow identity: with duplicate workflow UUIDs
across communities (which the schema explicitly allows and the Issue-4
confinement tests pin), a single `INSERT ... SELECT` matched every
community's row and fanned one claim across all of them.

Scope the claim to its community and wire the claim->run boundary into
the scheduler:

- `claim_scheduled_workflow_fire` takes `CommunityId` and binds
  `WHERE w.community_id = $1 AND w.id = $2`. The community is server
  provenance — the `workflow.community_id` from the global scan, never
  client input. This reverses the earlier S1 "resolve-from-id-alone"
  lock, which was written against a globally-unique-`workflow_id`
  assumption the final `(community_id, id)` schema does not hold (and
  which is unimplementable on that key). The surviving invariant is
  "the claim community is server provenance, not client-controlled."

- `WorkflowEngine::run` now claims before `create_workflow_run`; the
  loser skips before any run creation or side effect. The claim anchor
  `scheduled_for` is deterministic across pods: the cron's own scheduled
  instant (`cron_fire_instant`, not `now`) or the interval bucket
  boundary (`interval_fire_instant`, floor to the interval). The
  interval anchor is seeded from `latest_scheduled_workflow_fire` on the
  first tick after restart so a process bounce can't double-fire within
  an interval. `attach_scheduled_workflow_run` links the won claim to
  its run for ops/audit forensics.

Tests:
- Rewrite the stale S1 comment block and `claim_for_workflow_in_other_
  community_no_ops` (which encoded the now-false globally-unique
  assumption) into `claim_confined_to_its_community`: a dup workflow
  UUID in A and B claims independently (claiming A/id leaves B/id
  claimable). Proven RED on the bare-`id` regression.
- `concurrent_same_window_claims_exactly_one_wins` and the prune-anchor
  test updated to the scoped signature.
- New `cron_fire_instant` / `interval_fire_instant` unit tests pin the
  deterministic, drift-stable claim anchors.

Full `cargo test -p buzz-db -p buzz-workflow -p buzz-relay` green vs
live Postgres (serial); clippy clean on the trio.

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
2026-06-27 14:16:00 -04:00
parent 5e62ba8cb8
commit 4dc65b6ee5
3 changed files with 324 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
crates/buzz-db/src/lib.rs
View File

@@ -1586,15 +1586,20 @@ impl Db {
/// Claim a scheduled workflow fire for an authoritative schedule instant.
///
/// Returns `Some` only for the first pod to claim `(workflow_id,
/// scheduled_for)`; all other pods must skip creating a run. The claim SQL
/// resolves `community_id` from the workflow row; callers never supply it.
/// Returns `Some` only for the first pod to claim `(community_id,
/// workflow_id, scheduled_for)`; all other pods must skip creating a run.
/// `community_id` is server provenance (the workflow row's own community
/// from the scheduler scan), never client-supplied — `workflows` is keyed
/// `(community_id, id)`, so the claim must bind both to avoid fanning
/// across communities that share the workflow UUID.
pub async fn claim_scheduled_workflow_fire(
&self,
community_id: CommunityId,
workflow_id: Uuid,
scheduled_for: chrono::DateTime<chrono::Utc>,
) -> Result<Option<workflow::ScheduledWorkflowFireClaim>> {
workflow::claim_scheduled_workflow_fire(&self.pool, workflow_id, scheduled_for).await
workflow::claim_scheduled_workflow_fire(&self.pool, community_id, workflow_id, scheduled_for)
.await
}
/// Fetch the latest claimed schedule instant for interval trigger anchoring.

170
crates/buzz-db/src/workflow.rs
View File

@@ -423,27 +423,38 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRec
/// Claim a scheduled workflow fire for an authoritative schedule instant.
///
/// Returns `Some` only for the first pod that claims `(workflow_id,
/// scheduled_for)`. All other pods receive `None` and must skip creating a
/// workflow run. The `scheduled_for` value must come from an external
/// schedule anchor (cron expression) or DB-authoritative interval anchor; a
/// per-pod in-memory timestamp is not safe because different pods can compute
/// different claim keys.
/// Returns `Some` only for the first pod that claims `(community_id,
/// workflow_id, scheduled_for)`. All other pods receive `None` and must skip
/// creating a workflow run. The `scheduled_for` value must come from an
/// external schedule anchor (cron expression) or DB-authoritative interval
/// anchor; a per-pod in-memory timestamp is not safe because different pods
/// can compute different claim keys.
///
/// `community_id` is server provenance — for the global scheduler scan it is
/// the `workflow.community_id` returned by [`list_all_enabled_workflows`], not
/// any client-supplied value. It is required because `workflows` is keyed
/// `(community_id, id)`: duplicate workflow UUIDs across communities are
/// allowed, so resolving the owning community from `id` alone is ambiguous and
/// would fan a single claim across every community holding that UUID. Binding
/// `(community_id, id)` confines the claim — and its `SELECT`/`INSERT` row — to
/// exactly the intended tenant.
pub async fn claim_scheduled_workflow_fire(
pool: &PgPool,
community_id: CommunityId,
workflow_id: Uuid,
scheduled_for: DateTime<Utc>,
) -> Result<Option<ScheduledWorkflowFireClaim>> {
let row = sqlx::query(
r#"
INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for)
SELECT w.community_id, w.id, $2
SELECT w.community_id, w.id, $3
FROM workflows w
WHERE w.id = $1
WHERE w.community_id = $1 AND w.id = $2
ON CONFLICT (community_id, workflow_id, scheduled_for) DO NOTHING
RETURNING community_id, workflow_id, scheduled_for, claimed_at
"#,
)
.bind(community_id.as_uuid())
.bind(workflow_id)
.bind(scheduled_for)
.fetch_optional(pool)
@@ -1541,27 +1552,39 @@ mod tests {
assert_eq!(cloned.status, ApprovalStatus::Granted);
}
// -- F1 / S1 attack surface for scheduled workflow claims ------------------
// -- Scheduled workflow claim confinement ---------------------------------
//
// These tests pin the locked spec from Eva [13] / Mari [12]:
// RECONCILED spec (supersedes the earlier S1 lock; Eva/Max 2026-06-27).
//
// The earlier S1 lock asserted "`workflow_id` is globally unique, so the
// claim resolves community server-side from `workflow_id` alone and the
// caller never names it." The final schema does NOT have that property:
// `workflows` PK is `(community_id, id)` and `scheduled_workflow_fires` is
// keyed/FK'd by `(community_id, workflow_id, scheduled_for)`. Duplicate
// workflow UUIDs across communities are explicitly allowed (and pinned by
// the Issue-4 confinement tests below). So resolve-from-id-alone is both
// unimplementable and unsafe: `WHERE w.id = $1` matches every community
// holding that UUID and fans one claim across all of them.
//
// The invariant that survives is NOT "the claim never receives community";
// it is "the community used for the claim is server provenance, never
// client-controlled." For the global scheduler scan that provenance is the
// `workflow.community_id` returned by `list_all_enabled_workflows()`. The
// claim therefore takes `community_id` and binds
// `WHERE w.community_id = $1 AND w.id = $2`, confining the claim row to the
// intended tenant.
//
// 1. `workflows.community_id` is row-owned, NOT NULL, immutable.
// 2. Claim resolves `community_id` server-side via `workflow_id`.
// 3. Claim uniqueness is `(workflow_id, scheduled_for)` — `workflow_id`
// is globally unique, so adding `community_id` to the key weakens it.
// 4. `latest_scheduled_workflow_fire` drops caller-supplied community.
// 2. The claim binds `(community_id, workflow_id)` of the workflow row.
// 3. Claim uniqueness is `(community_id, workflow_id, scheduled_for)`.
// 4. `latest_scheduled_workflow_fire` / `attach_scheduled_workflow_run`
// are already community-scoped; `claim` now matches.
//
// Until that lands, `claim_for_workflow_in_other_community_no_ops` is the
// S1 regression lock: today the schema permits a caller in community A to
// claim a workflow owned by community B and have `claimed.community_id`
// come back as A. That is a cross-tenant write surface (Theorem S1,
// `docs/multi-tenant-relay.md:248`: "the claimed community never appears
// in this function — only the resolved one").
//
// The other two tests are characterization guards: same-window race must
// yield exactly one claim winner, and the retention primitive's docstring
// caveat (pruning below the largest interval breaks `latest_*`) is
// load-bearing for the §5c deployment-config rule Sami flagged.
// `claim_confined_to_its_community` is the confinement lock: a dup workflow
// UUID in A and B must claim independently (claiming A/id leaves B/id
// claimable). The other two tests are characterization guards: same-window
// race must yield exactly one winner, and pruning below the largest
// interval breaks `latest_*` (the §5c retention rule Sami flagged).
use crate::user::ensure_user;
@@ -1632,58 +1655,73 @@ mod tests {
(workflow_id, community)
}
/// F1 attack: a caller in community A must NOT be able to claim a fire
/// for a workflow owned by community B and have the claim resolve under
/// A's tenant. The resolved community on the returned claim row MUST
/// equal the workflow's actual tenant (B).
/// Confinement: a duplicate workflow UUID existing in both community A and
/// community B must claim independently. Claiming `(A, id, t)` must NOT
/// consume `(B, id, t)` — B's identical instant stays claimable, and the
/// A-claim's resolved community is A (server provenance), never B.
///
/// Post-fix (`1fa3d837f`) the claim signature no longer accepts a caller
/// tenant — the SQL resolves `community_id` from the `workflows` row via
/// `INSERT ... SELECT w.community_id, w.id, $2 FROM workflows w WHERE
/// w.id = $1`. This test still creates an "attacker_community" that the
/// caller is *not* permitted to name on the wire; the assertion is that
/// the resolved tenant equals the workflow owner's community, never the
/// attacker's. With the pre-fix signature this test was RED (the row's
/// `community_id` came back as the attacker's); under the locked spec it
/// must be GREEN.
/// This is the reconciliation of the old S1 lock with the real
/// `(community_id, id)` schema: because `id` is not globally unique, the
/// claim binds `WHERE w.community_id = $1 AND w.id = $2`. With the old
/// bare-`id` SQL (`WHERE w.id = $1`), a single `INSERT ... SELECT` matched
/// BOTH workflow rows and fanned the claim across A and B — this test goes
/// RED on that regression (B/id is no longer independently claimable).
#[tokio::test]
#[ignore = "requires Postgres"]
async fn claim_for_workflow_in_other_community_no_ops() {
async fn claim_confined_to_its_community() {
let pool = setup_pool().await;
// The attacker community exists in the schema but the caller has no
// way to pass it to the claim API anymore — that *is* the S1 fix.
// Keeping the row in this test makes the no-influence invariant
// explicit: even with two real tenants in play, the resolved
// community is the workflow's owner.
let _attacker_community = make_community(&pool).await;
let owner_community = make_community(&pool).await;
let (workflow_id, expected_community) = make_workflow_in(&pool, owner_community).await;
let community_a = make_community(&pool).await;
let community_b = make_community(&pool).await;
// Same workflow UUID + same channel UUID in both communities — the PK
// is `(community_id, id)`, so the collision is structurally allowed.
let workflow_id = Uuid::new_v4();
let channel_id = Uuid::new_v4();
insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "sched-a").await;
insert_workflow_with_ids(&pool, community_b, workflow_id, channel_id, "sched-b").await;
let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 0, 0).unwrap();
let claim = claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for)
// Claim A/id/t.
let claim_a = claim_scheduled_workflow_fire(&pool, community_a, workflow_id, scheduled_for)
.await
.expect("claim should not error")
.expect("claim should succeed exactly once");
.expect("claim A should not error")
.expect("claim A should win");
assert_eq!(
claim.community_id,
expected_community,
"claim must resolve community from workflow_id (server-side); \
resolved={resolved:?} expected={expected_community:?}",
resolved = claim.community_id,
claim_a.community_id, community_a,
"A-claim must resolve to community A (server provenance)"
);
assert_eq!(claim_a.workflow_id, workflow_id);
assert_eq!(claim_a.scheduled_for, scheduled_for);
// B/id/t must still be claimable — A's claim did not touch B's row.
let claim_b = claim_scheduled_workflow_fire(&pool, community_b, workflow_id, scheduled_for)
.await
.expect("claim B should not error")
.expect("claim B must still win — A's claim must not have consumed B's instant");
assert_eq!(
claim_b.community_id, community_b,
"B-claim must resolve to community B"
);
// And a second A-claim for the same instant must now lose (dedup holds
// within the community).
let claim_a_again =
claim_scheduled_workflow_fire(&pool, community_a, workflow_id, scheduled_for)
.await
.expect("second A-claim should not error");
assert!(
claim_a_again.is_none(),
"the same (A, id, t) instant must not be claimable twice"
);
assert_eq!(claim.workflow_id, workflow_id);
assert_eq!(claim.scheduled_for, scheduled_for);
}
/// Same `(workflow_id, scheduled_for)` claimed concurrently by N tasks
/// must yield exactly one `Some` winner. Post-fix the PK is
/// `(workflow_id, scheduled_for)` (`workflow_id` is globally unique,
/// `community_id` is a scoped/audit label only) — exactly the locked
/// spec. Characterization guard: protects the dedup boundary against
/// regressions in the claim SQL.
/// Same `(community_id, workflow_id, scheduled_for)` claimed concurrently by
/// N tasks must yield exactly one `Some` winner. Post-reconciliation the
/// claim key is `(community_id, workflow_id, scheduled_for)`; `community_id`
/// is server provenance, not a client-named label. Characterization guard:
/// protects the dedup boundary against regressions in the claim SQL.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn concurrent_same_window_claims_exactly_one_wins() {
@@ -1698,7 +1736,7 @@ mod tests {
for _ in 0..N {
let pool = pool.clone();
handles.push(tokio::spawn(async move {
claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for).await
claim_scheduled_workflow_fire(&pool, community, workflow_id, scheduled_for).await
}));
}
@@ -1738,7 +1776,7 @@ mod tests {
let (workflow_id, _) = make_workflow_in(&pool, community).await;
let scheduled_for = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for)
claim_scheduled_workflow_fire(&pool, community, workflow_id, scheduled_for)
.await
.expect("claim ok")
.expect("first claim wins");

263
crates/buzz-workflow/src/lib.rs
View File

@@ -350,8 +350,13 @@ impl WorkflowEngine {
/// Uses window-based matching for cron expressions to handle tick drift:
/// `schedule.after(&(now - 60s)).next() <= now` instead of `includes(now)`.
///
/// Interval tracking is in-memory (`last_fired` DashMap). Lost on restart —
/// missed fires during downtime are not replayed.
/// Interval tracking is anchored on the durable scheduled-fire claim:
/// `last_fired` is an in-memory pre-filter, but the
/// `(community_id, workflow_id, scheduled_for)` claim row is the
/// at-most-once boundary across pods and restarts. On the first tick after
/// a restart the interval anchor is seeded from
/// `latest_scheduled_workflow_fire` so a process bounce cannot double-fire
/// within an interval.
pub async fn run(self: &Arc<Self>) {
tracing::info!("WorkflowEngine cron loop started (60s tick)");
@@ -399,30 +404,78 @@ impl WorkflowEngine {
continue;
};
let (should_fire, trigger_type) = match &def.trigger {
// Resolve the *deterministic* schedule instant this tick is
// firing for. `scheduled_for` is computed identically on every
// pod (cron's own scheduled time, or the interval bucket
// boundary) so all pods collide on a single durable claim —
// never `now`, which is per-pod and would let every pod fire.
let (scheduled_for, trigger_type) = match &def.trigger {
schema::TriggerDef::Schedule {
cron: Some(expr),
interval: None,
} => {
// Fix 7: delegate to pure helper for testability.
(cron_should_fire(expr, now, 60, workflow.id), "cron")
}
} => match cron_fire_instant(expr, now, 60, workflow.id) {
Some(instant) => (instant, "cron"),
None => continue,
},
schema::TriggerDef::Schedule {
cron: None,
interval: Some(dur),
} => {
// Fix 7: delegate to pure helper for testability.
let last = self.last_fired.get(&(community_id, workflow.id)).map(|t| *t);
(
interval_should_fire(dur, last, now, workflow.id),
"interval",
)
// Cheap pre-filter: skip the claim attempt when the
// in-memory clock says we're clearly mid-interval. The
// durable claim below is the real at-most-once boundary;
// this only avoids a DB write every tick. Seed the
// anchor from the DB on the first tick after restart so
// a process bounce can't double-fire within an interval.
let last = match self.last_fired.get(&(community_id, workflow.id)) {
Some(t) => Some(*t),
None => self
.db
.latest_scheduled_workflow_fire(community_id, workflow.id)
.await
.unwrap_or(None),
};
if !interval_should_fire(dur, last, now, workflow.id) {
continue;
}
match interval_fire_instant(dur, now, workflow.id) {
Some(instant) => (instant, "interval"),
None => continue,
}
}
_ => (false, ""), // Non-schedule triggers handled by on_event()
_ => continue, // Non-schedule triggers handled by on_event()
};
if !should_fire {
continue;
// Durable at-most-once claim — the cross-pod fire boundary.
// The loser receives `None` and skips BEFORE any run creation or
// side effect. `community_id` is the workflow row's own
// community (server provenance from the scan), never client
// input; the claim binds `(community_id, workflow_id,
// scheduled_for)` so a duplicate workflow UUID in another
// community claims independently.
match self
.db
.claim_scheduled_workflow_fire(community_id, workflow.id, scheduled_for)
.await
{
Ok(Some(_)) => {}
Ok(None) => {
// Another pod (or an earlier tick this pod) already
// claimed this instant. Still advance the in-memory
// interval clock so we don't re-attempt the claim every
// tick for the rest of the interval.
if trigger_type == "interval" {
self.last_fired.insert((community_id, workflow.id), now);
}
continue;
}
Err(e) => {
tracing::error!(
workflow_id = %workflow.id,
"Cron tick: scheduled-fire claim failed: {e}"
);
continue;
}
}
// Fix 5: handle serialization errors explicitly rather than silently
@@ -459,14 +512,34 @@ impl WorkflowEngine {
workflow_id = %workflow.id,
"Cron tick: failed to create workflow run: {e}"
);
// The claim is held but the run failed to create. The
// claim row intentionally stays (its `workflow_run_id`
// NULL) so this instant is not re-fired: at-most-once is
// preserved over exactly-once on transient run-insert
// failures.
continue;
}
};
// Update last_fired AFTER successful DB insert so that a
// failed insert doesn't suppress the next tick for the full interval.
// Only needed for interval triggers — cron uses window-based matching
// which already prevents double-fire within the same minute.
// Link the won claim to its run for ops/audit forensics. The
// claim row already guarantees dedupe; this is best-effort.
if let Err(e) = self
.db
.attach_scheduled_workflow_run(community_id, workflow.id, scheduled_for, run_id)
.await
{
tracing::warn!(
workflow_id = %workflow.id,
run_id = %run_id,
"Cron tick: failed to attach run to scheduled-fire claim: {e}"
);
}
// Update last_fired AFTER a successful claim+insert so that a
// failure doesn't suppress the next tick for the full interval.
// Only needed for interval triggers — cron uses window-based
// matching which already prevents double-fire within the same
// minute, and the durable claim backstops both.
if trigger_type == "interval" {
self.last_fired.insert((community_id, workflow.id), now);
}
@@ -501,32 +574,70 @@ impl WorkflowEngine {
}
}
/// Check whether a cron expression should fire within the `window_secs`-wide
/// window ending at `now`.
/// Find the cron schedule instant that fired within the `window_secs`-wide
/// window ending at `now`, if any.
///
/// Uses window-based matching: finds the next scheduled time after
/// `(now - window_secs)` and checks whether it falls at or before `now`.
/// `(now - window_secs)` and returns it when it falls at or before `now`.
/// This tolerates tick drift gracefully — a 61s tick won't miss a
/// minute-granularity cron expression.
/// minute-granularity cron expression. The returned instant is the cron's own
/// scheduled time (not `now`), so every pod evaluating the same expression in
/// the same window computes the *same* value — making it a safe, deterministic
/// claim anchor for cross-pod at-most-once firing.
///
/// Returns `false` (and logs a warning) if the expression is invalid.
fn cron_should_fire(expr: &str, now: DateTime<Utc>, window_secs: i64, workflow_id: Uuid) -> bool {
/// Returns `None` (and logs a warning) if the expression is invalid or nothing
/// is due in the window.
fn cron_fire_instant(
expr: &str,
now: DateTime<Utc>,
window_secs: i64,
workflow_id: Uuid,
) -> Option<DateTime<Utc>> {
let normalized = schema::normalize_cron(expr);
match normalized.parse::<cron::Schedule>() {
Ok(sched) => {
let window_start = now - chrono::Duration::seconds(window_secs);
sched
.after(&window_start)
.next()
.map(|t| t <= now)
.unwrap_or(false)
sched.after(&window_start).next().filter(|t| *t <= now)
}
Err(e) => {
tracing::warn!(
workflow_id = %workflow_id,
"Cron tick: invalid cron expression '{expr}': {e}"
);
false
None
}
}
}
/// Quantize `now` to the interval bucket boundary, yielding a deterministic
/// claim anchor that every pod computes identically within the same bucket.
///
/// The boundary is `floor(now / interval) * interval` from the Unix epoch.
/// Because the scheduler ticks every 60s and interval schedules are minutes or
/// longer, bounded cross-pod clock skew keeps all pods inside the same bucket,
/// so they collide on one `(community, workflow, scheduled_for)` claim — only
/// one wins and creates the run. Returns `None` if the duration is unparseable
/// or non-positive (the caller skips firing).
fn interval_fire_instant(dur: &str, now: DateTime<Utc>, workflow_id: Uuid) -> Option<DateTime<Utc>> {
match executor::parse_duration_secs(dur) {
Ok(interval_secs) if interval_secs > 0 => {
let secs = interval_secs as i64;
let bucket = (now.timestamp().div_euclid(secs)) * secs;
DateTime::from_timestamp(bucket, 0)
}
Ok(_) => {
tracing::warn!(
workflow_id = %workflow_id,
"Cron tick: interval duration is zero — skipping"
);
None
}
Err(e) => {
tracing::warn!(
workflow_id = %workflow_id,
"Cron tick: invalid interval '{dur}': {e}"
);
None
}
}
}
@@ -730,30 +841,36 @@ mod tests {
use super::*;
#[test]
fn cron_should_fire_matches_within_window() {
fn cron_fire_instant_matches_within_window() {
// "every minute" cron — should always fire within a 60s window.
let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:30Z")
.unwrap()
.with_timezone(&Utc);
let wf_id = Uuid::new_v4();
assert!(
cron_should_fire("* * * * *", now, 60, wf_id),
"every-minute cron should fire within 60s window"
// The matched instant is the minute boundary 12:00:00, NOT `now`.
assert_eq!(
cron_fire_instant("* * * * *", now, 60, wf_id),
Some(
chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:00Z")
.unwrap()
.with_timezone(&Utc)
),
"every-minute cron should return the minute boundary as the anchor"
);
}
#[test]
fn cron_should_fire_returns_false_for_invalid_expr() {
fn cron_fire_instant_returns_none_for_invalid_expr() {
let now = Utc::now();
let wf_id = Uuid::new_v4();
assert!(
!cron_should_fire("not-a-cron", now, 60, wf_id),
"invalid cron should return false"
cron_fire_instant("not-a-cron", now, 60, wf_id).is_none(),
"invalid cron should return None"
);
}
#[test]
fn cron_should_fire_returns_false_outside_window() {
fn cron_fire_instant_returns_none_outside_window() {
// Fixed time: 2026-06-15 14:30:00 UTC (a Sunday in June)
let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T14:30:00Z")
.unwrap()
@@ -761,41 +878,49 @@ mod tests {
let wf_id = Uuid::new_v4();
// "0 0 1 1 *" = midnight on Jan 1 only — June 15 is definitely outside.
assert!(
!cron_should_fire("0 0 1 1 *", now, 60, wf_id),
cron_fire_instant("0 0 1 1 *", now, 60, wf_id).is_none(),
"Jan-1-only cron should not fire on June 15"
);
}
#[test]
fn cron_should_fire_at_exact_minute_boundary() {
fn cron_fire_instant_at_exact_minute_boundary() {
// Fixed time: exactly 09:00:00 UTC. Cron "0 9 * * *" fires at 09:00.
// Window [08:59:00, 09:00:00] should contain the fire time.
let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z")
.unwrap()
.with_timezone(&Utc);
let wf_id = Uuid::new_v4();
assert!(
cron_should_fire("0 9 * * *", now, 60, wf_id),
"cron should fire at exact minute boundary"
assert_eq!(
cron_fire_instant("0 9 * * *", now, 60, wf_id),
Some(now),
"cron should fire at exact minute boundary, anchored on 09:00:00"
);
}
#[test]
fn cron_should_fire_within_drift_window() {
fn cron_fire_instant_within_drift_window_anchors_on_scheduled_time() {
// Fixed time: 09:00:45 UTC (45s drift). Cron "0 9 * * *" fires at 09:00.
// Window [08:59:45, 09:00:45] should still contain 09:00:00.
// Window [08:59:45, 09:00:45] should still contain 09:00:00. Critically,
// the anchor is the *scheduled* 09:00:00 — not the drifted `now` — so a
// second pod ticking at 09:00:50 computes the identical claim key.
let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:45Z")
.unwrap()
.with_timezone(&Utc);
let wf_id = Uuid::new_v4();
assert!(
cron_should_fire("0 9 * * *", now, 60, wf_id),
"cron should fire even with 45s drift"
assert_eq!(
cron_fire_instant("0 9 * * *", now, 60, wf_id),
Some(
chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z")
.unwrap()
.with_timezone(&Utc)
),
"cron anchor must be the scheduled instant, stable across pod tick drift"
);
}
#[test]
fn cron_should_fire_returns_false_just_outside_window() {
fn cron_fire_instant_returns_none_just_outside_window() {
// Fixed time: 09:01:01 UTC. Cron "0 9 * * *" fires at 09:00:00.
// Window [09:00:01, 09:01:01] does NOT contain 09:00:00.
let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:01:01Z")
@@ -803,11 +928,45 @@ mod tests {
.with_timezone(&Utc);
let wf_id = Uuid::new_v4();
assert!(
!cron_should_fire("0 9 * * *", now, 60, wf_id),
cron_fire_instant("0 9 * * *", now, 60, wf_id).is_none(),
"cron should not fire 61s after the scheduled time"
);
}
#[test]
fn interval_fire_instant_quantizes_to_bucket_boundary() {
// Two pods ticking at different sub-interval offsets must compute the
// *same* bucket boundary so they collide on one claim. 1h interval,
// epoch-aligned: 12:34:56 and 12:59:01 both floor to 12:00:00.
let wf_id = Uuid::new_v4();
let a = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:34:56Z")
.unwrap()
.with_timezone(&Utc);
let b = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:59:01Z")
.unwrap()
.with_timezone(&Utc);
let bucket = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:00Z")
.unwrap()
.with_timezone(&Utc);
assert_eq!(interval_fire_instant("1h", a, wf_id), Some(bucket));
assert_eq!(interval_fire_instant("1h", b, wf_id), Some(bucket));
// Next hour is a distinct bucket.
let c = chrono::DateTime::parse_from_rfc3339("2026-06-15T13:00:10Z")
.unwrap()
.with_timezone(&Utc);
let next_bucket = chrono::DateTime::parse_from_rfc3339("2026-06-15T13:00:00Z")
.unwrap()
.with_timezone(&Utc);
assert_eq!(interval_fire_instant("1h", c, wf_id), Some(next_bucket));
}
#[test]
fn interval_fire_instant_returns_none_for_invalid_duration() {
let now = Utc::now();
let wf_id = Uuid::new_v4();
assert!(interval_fire_instant("not-a-duration", now, wf_id).is_none());
}
#[test]
fn interval_should_fire_returns_false_on_first_tick() {
// When last_fired is None (first tick), defaults to now → elapsed = 0 → false.