fix(workflow): catch up missed cron fires

crates/buzz-db/src/runtime/migration.rs

@@ -702,7 +702,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 44);
+        assert_eq!(migrations.len(), 45);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -2470,6 +2470,7 @@ mod postgres_tests {
             "events",
             "channels",
             "scheduled_workflow_fires",
+            "workflow_schedule_cursors",
             "audit_log",
         ] {
             let exists = sqlx::query_scalar::<_, bool>(

crates/buzz-db/src/store/deletion.rs

@@ -82,12 +82,14 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "users",
     "workflow_approvals",
     "workflow_runs",
+    "workflow_schedule_cursors",
     "workflows",
 ];
 
 /// Foreign-key-safe child-before-parent order for the PostgreSQL purge.
 pub const PURGE_SCOPED_TABLES: &[&str] = &[
     "workflow_approvals",
+    "workflow_schedule_cursors",
     "scheduled_workflow_fires",
     "workflow_runs",
     "push_wake_outbox",

crates/buzz-db/src/store/workflow.rs

@@ -559,6 +559,85 @@ pub async fn latest_scheduled_workflow_fire(
     row.try_get("scheduled_for").map_err(Into::into)
 }
 
+/// Read the durable evaluated-through cursor for a scheduled workflow.
+///
+/// If the cursor is unexpectedly absent, seed it at `now` and return `None` so
+/// the caller skips historical replay on this tick. Normal creation, update,
+/// enablement, and status transitions maintain this row with a database trigger.
+pub async fn get_or_seed_workflow_schedule_cursor(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+    now: DateTime<Utc>,
+) -> Result<Option<DateTime<Utc>>> {
+    let inserted = sqlx::query(
+        r#"
+        INSERT INTO workflow_schedule_cursors
+            (community_id, workflow_id, evaluated_through)
+        SELECT community_id, id, $3
+        FROM workflows
+        WHERE community_id = $1 AND id = $2
+        ON CONFLICT (community_id, workflow_id) DO NOTHING
+        RETURNING evaluated_through
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .bind(now)
+    .fetch_optional(pool)
+    .await?;
+
+    if inserted.is_some() {
+        return Ok(None);
+    }
+
+    let cursor = sqlx::query_scalar(
+        r#"
+        SELECT evaluated_through
+        FROM workflow_schedule_cursors
+        WHERE community_id = $1 AND workflow_id = $2
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .fetch_optional(pool)
+    .await?;
+
+    Ok(cursor)
+}
+
+/// Advance a workflow's evaluated-through cursor with compare-and-swap.
+///
+/// Multiple scheduler pods may scan the same window. Only a pod that still
+/// observes `expected` advances it; fire claims independently deduplicate work.
+pub async fn advance_workflow_schedule_cursor(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+    expected: DateTime<Utc>,
+    evaluated_through: DateTime<Utc>,
+) -> Result<bool> {
+    let affected = sqlx::query(
+        r#"
+        UPDATE workflow_schedule_cursors
+        SET evaluated_through = $4, updated_at = NOW()
+        WHERE community_id = $1
+          AND workflow_id = $2
+          AND evaluated_through = $3
+          AND evaluated_through < $4
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .bind(expected)
+    .bind(evaluated_through)
+    .execute(pool)
+    .await?
+    .rows_affected();
+
+    Ok(affected == 1)
+}
+
 /// Link a won scheduled-fire claim to the workflow run it created.
 ///
 /// This is for ops/audit forensics only; the claim row remains the dedupe
@@ -1563,6 +1642,42 @@ impl Db {
         crate::workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await
     }
 
+    /// Read or safely seed a workflow's durable schedule evaluation cursor.
+    #[datastore_span(name = "get_or_seed_workflow_schedule_cursor", system = "postgresql")]
+    pub async fn get_or_seed_workflow_schedule_cursor(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        now: chrono::DateTime<chrono::Utc>,
+    ) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
+        crate::workflow::get_or_seed_workflow_schedule_cursor(
+            &self.pool,
+            community_id,
+            workflow_id,
+            now,
+        )
+        .await
+    }
+
+    /// Compare-and-swap a workflow's durable schedule evaluation cursor.
+    #[datastore_span(name = "advance_workflow_schedule_cursor", system = "postgresql")]
+    pub async fn advance_workflow_schedule_cursor(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        expected: chrono::DateTime<chrono::Utc>,
+        evaluated_through: chrono::DateTime<chrono::Utc>,
+    ) -> Result<bool> {
+        crate::workflow::advance_workflow_schedule_cursor(
+            &self.pool,
+            community_id,
+            workflow_id,
+            expected,
+            evaluated_through,
+        )
+        .await
+    }
+
     /// Attach the workflow run id created from a won scheduled-fire claim.
     #[datastore_span(name = "attach_scheduled_workflow_run", system = "postgresql")]
     pub async fn attach_scheduled_workflow_run(

crates/buzz-workflow/Cargo.toml

@@ -27,5 +27,8 @@ tracing     = { workspace = true }
 thiserror   = { workspace = true }
 reqwest     = { workspace = true, optional = true }
 
+[dev-dependencies]
+sqlx = { workspace = true }
+
 [features]
 reqwest = ["dep:reqwest"]

crates/buzz-workflow/src/error.rs

@@ -54,9 +54,10 @@ pub enum WorkflowError {
     #[error("database error: {0}")]
     Database(String),
 
-    /// The workflow's owner is not currently authorized to run it (removed
-    /// from the channel, insufficient role for the definition's actions, or
-    /// the authority lookup failed — all deny, fail-closed).
+    /// The workflow's owner is confirmed not to be authorized to run it
+    /// (removed from the channel or insufficient role for the definition's
+    /// actions). Authority lookup failures use [`Self::Database`] so callers
+    /// can fail closed without mistaking an outage for confirmed revocation.
     #[error("unauthorized: {0}")]
     Unauthorized(String),