feat(db): anchor scheduled workflow claims

Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>

crates/buzz-db/src/lib.rs

@@ -1248,16 +1248,16 @@ impl Db {
         workflow::list_all_enabled_workflows(&self.pool).await
     }
 
-    /// Claim a scheduled workflow fire for a quantized schedule window.
+    /// Claim a scheduled workflow fire for an authoritative schedule instant.
     ///
-    /// Returns `true` only for the first pod to claim `(workflow_id,
+    /// Returns `Some` only for the first pod to claim `(community_id,
-    /// scheduled_for)`; all other pods must skip creating a run.
+    /// workflow_id, scheduled_for)`; all other pods must skip creating a run.
     pub async fn claim_scheduled_workflow_fire(
         &self,
         community_id: CommunityId,
         workflow_id: Uuid,
         scheduled_for: chrono::DateTime<chrono::Utc>,
-    ) -> Result<bool> {
+    ) -> Result<Option<workflow::ScheduledWorkflowFireClaim>> {
         workflow::claim_scheduled_workflow_fire(
             &self.pool,
             community_id,
@@ -1267,6 +1267,41 @@ impl Db {
         .await
     }
 
+    /// Fetch the latest claimed schedule instant for interval trigger anchoring.
+    pub async fn latest_scheduled_workflow_fire(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+    ) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
+        workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await
+    }
+
+    /// Attach the workflow run id created from a won scheduled-fire claim.
+    pub async fn attach_scheduled_workflow_run(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        scheduled_for: chrono::DateTime<chrono::Utc>,
+        workflow_run_id: Uuid,
+    ) -> Result<bool> {
+        workflow::attach_scheduled_workflow_run(
+            &self.pool,
+            community_id,
+            workflow_id,
+            scheduled_for,
+            workflow_run_id,
+        )
+        .await
+    }
+
+    /// Delete old scheduled workflow fire claims before a retention cutoff.
+    pub async fn prune_scheduled_workflow_fires_before(
+        &self,
+        older_than: chrono::DateTime<chrono::Utc>,
+    ) -> Result<u64> {
+        workflow::prune_scheduled_workflow_fires_before(&self.pool, older_than).await
+    }
+
     /// Update a workflow's name, definition, and hash.
     pub async fn update_workflow(
         &self,

crates/buzz-db/src/migration.rs

@@ -145,6 +145,20 @@ mod tests {
                 .contains("CREATE TABLE scheduled_workflow_fires"),
             "initial schema migration should include workflow cron claim table"
         );
+        assert!(
+            migrations[0]
+                .sql
+                .as_str()
+                .contains("workflow_run_id UUID REFERENCES workflow_runs"),
+            "workflow cron claim table should optionally link to the run it created"
+        );
+        assert!(
+            migrations[0]
+                .sql
+                .as_str()
+                .contains("PRIMARY KEY (community_id, workflow_id, scheduled_for)"),
+            "workflow cron claim uniqueness must include the community label"
+        );
         assert!(
             migrations[0].sql.as_str().contains("CREATE TABLE channels"),
             "initial schema migration should include Buzz core tables"

crates/buzz-db/src/workflow.rs

@@ -213,6 +213,23 @@ pub struct WorkflowRunRecord {
     pub created_at: DateTime<Utc>,
 }
 
+/// A winning scheduled workflow fire claim.
+///
+/// The primary identity is `(community_id, workflow_id, scheduled_for)`. The
+/// database also returns `claimed_at` so the workflow scheduler can log and audit
+/// the exact claim row it won without relying on a per-pod clock.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ScheduledWorkflowFireClaim {
+    /// Community that owns this scheduled fire.
+    pub community_id: CommunityId,
+    /// Workflow definition that should run.
+    pub workflow_id: Uuid,
+    /// Authoritative schedule instant this claim represents.
+    pub scheduled_for: DateTime<Utc>,
+    /// Database timestamp for when this pod won the claim.
+    pub claimed_at: DateTime<Utc>,
+}
+
 /// A pending or resolved approval gate for a workflow step.
 #[derive(Debug, Clone)]
 pub struct ApprovalRecord {
@@ -378,33 +395,131 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRec
     rows.into_iter().map(row_to_workflow_record).collect()
 }
 
-/// Claim a scheduled workflow fire for a quantized schedule window.
+/// Claim a scheduled workflow fire for an authoritative schedule instant.
 ///
-/// Returns `true` only for the first pod that claims `(workflow_id,
+/// Returns `Some` only for the first pod that claims `(community_id,
-/// scheduled_for)`. All other pods receive `false` and must skip creating a
+/// workflow_id, scheduled_for)`. All other pods receive `None` and must skip
-/// workflow run. This is the restart-safe cross-pod dedupe boundary for cron.
+/// 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.
 pub async fn claim_scheduled_workflow_fire(
     pool: &PgPool,
     community_id: CommunityId,
     workflow_id: Uuid,
     scheduled_for: DateTime<Utc>,
-) -> Result<bool> {
+) -> Result<Option<ScheduledWorkflowFireClaim>> {
-    let result = sqlx::query(
+    let row = sqlx::query(
         r#"
         INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for)
         VALUES ($1, $2, $3)
         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)
+    .await?;
+
+    row.map(|row| {
+        let community_id: Uuid = row.try_get("community_id")?;
+        Ok(ScheduledWorkflowFireClaim {
+            community_id: CommunityId::from_uuid(community_id),
+            workflow_id: row.try_get("workflow_id")?,
+            scheduled_for: row.try_get("scheduled_for")?,
+            claimed_at: row.try_get("claimed_at")?,
+        })
+    })
+    .transpose()
+}
+
+/// Fetch the greatest claimed schedule instant for a workflow.
+///
+/// Interval schedulers use this as their DB-authoritative `last_fired` anchor.
+/// It makes all pods compute the same next interval instant after a successful
+/// claim, and preserves the interval clock across pod restarts. This intentionally
+/// reads from `scheduled_workflow_fires`, not `workflow_runs`, because the claim
+/// row is the source of truth for schedule deduplication.
+pub async fn latest_scheduled_workflow_fire(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+) -> Result<Option<DateTime<Utc>>> {
+    let row = sqlx::query(
+        r#"
+        SELECT MAX(scheduled_for) AS scheduled_for
+        FROM scheduled_workflow_fires
+        WHERE community_id = $1
+          AND workflow_id = $2
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .fetch_one(pool)
+    .await?;
+
+    row.try_get("scheduled_for").map_err(Into::into)
+}
+
+/// 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
+/// boundary. If run creation succeeds, callers should attach the run id before
+/// spawning execution. If run creation fails, leaving `workflow_run_id` NULL is
+/// intentional: the schedule instant was claimed and must not duplicate later.
+pub async fn attach_scheduled_workflow_run(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+    scheduled_for: DateTime<Utc>,
+    workflow_run_id: Uuid,
+) -> Result<bool> {
+    let result = sqlx::query(
+        r#"
+        UPDATE scheduled_workflow_fires
+        SET workflow_run_id = $4
+        WHERE community_id = $1
+          AND workflow_id = $2
+          AND scheduled_for = $3
+          AND workflow_run_id IS NULL
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .bind(scheduled_for)
+    .bind(workflow_run_id)
     .execute(pool)
     .await?;
 
     Ok(result.rows_affected() == 1)
 }
 
+/// Delete old scheduled workflow fire claims for retention.
+///
+/// Schedule claim rows are correctness metadata, but they grow with every fire.
+/// The relay/ops janitor should retain enough history for audits and interval
+/// anchoring: the cutoff must be older than the largest interval schedule the
+/// deployment supports, or interval workflows can lose their DB-authoritative
+/// anchor after pruning.
+pub async fn prune_scheduled_workflow_fires_before(
+    pool: &PgPool,
+    older_than: DateTime<Utc>,
+) -> Result<u64> {
+    let result = sqlx::query(
+        r#"
+        DELETE FROM scheduled_workflow_fires
+        WHERE claimed_at < $1
+        "#,
+    )
+    .bind(older_than)
+    .execute(pool)
+    .await?;
+
+    Ok(result.rows_affected())
+}
+
 /// Update a workflow's name, definition, and definition_hash.
 pub async fn update_workflow(
     pool: &PgPool,

migrations/0001_initial_schema.sql

@@ -223,20 +223,6 @@ CREATE TABLE workflows (
 
 CREATE INDEX idx_workflows_channel_active ON workflows (channel_id, status, enabled);
 
--- Restart-safe horizontal-scaling claim table for scheduled workflow fires.
--- Each schedule tick must win this insert before creating a workflow run; this
--- replaces per-pod in-memory last-fired state as the deduplication boundary.
-CREATE TABLE scheduled_workflow_fires (
-    community_id    UUID NOT NULL REFERENCES communities(id),
-    workflow_id     UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
-    scheduled_for   TIMESTAMPTZ NOT NULL,
-    claimed_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    PRIMARY KEY (community_id, workflow_id, scheduled_for)
-);
-
-CREATE INDEX idx_scheduled_workflow_fires_scheduled_for
-    ON scheduled_workflow_fires (community_id, scheduled_for);
-
 -- ── Workflow runs ─────────────────────────────────────────────────────────────
 
 CREATE TABLE workflow_runs (
@@ -256,6 +242,25 @@ CREATE TABLE workflow_runs (
 CREATE INDEX idx_workflow_runs_workflow ON workflow_runs (workflow_id);
 CREATE INDEX idx_workflow_runs_status ON workflow_runs (status);
 
+-- Restart-safe horizontal-scaling claim table for scheduled workflow fires.
+-- Each schedule tick must win this insert before creating a workflow run; this
+-- replaces per-pod in-memory last-fired state as the deduplication boundary.
+CREATE TABLE scheduled_workflow_fires (
+    community_id    UUID NOT NULL REFERENCES communities(id),
+    workflow_id     UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
+    scheduled_for   TIMESTAMPTZ NOT NULL,
+    claimed_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    workflow_run_id UUID REFERENCES workflow_runs(id) ON DELETE SET NULL,
+    PRIMARY KEY (community_id, workflow_id, scheduled_for)
+);
+
+CREATE INDEX idx_scheduled_workflow_fires_scheduled_for
+    ON scheduled_workflow_fires (community_id, scheduled_for);
+
+CREATE UNIQUE INDEX idx_scheduled_workflow_fires_run
+    ON scheduled_workflow_fires (workflow_run_id)
+    WHERE workflow_run_id IS NOT NULL;
+
 -- ── Workflow approvals ────────────────────────────────────────────────────────
 
 CREATE TABLE workflow_approvals (

schema/schema.sql

@@ -228,20 +228,6 @@ CREATE TABLE workflows (
 
 CREATE INDEX idx_workflows_channel_active ON workflows (channel_id, status, enabled);
 
--- Restart-safe horizontal-scaling claim table for scheduled workflow fires.
--- Each schedule tick must win this insert before creating a workflow run; this
--- replaces per-pod in-memory last-fired state as the deduplication boundary.
-CREATE TABLE scheduled_workflow_fires (
-    community_id    UUID NOT NULL REFERENCES communities(id),
-    workflow_id     UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
-    scheduled_for   TIMESTAMPTZ NOT NULL,
-    claimed_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    PRIMARY KEY (community_id, workflow_id, scheduled_for)
-);
-
-CREATE INDEX idx_scheduled_workflow_fires_scheduled_for
-    ON scheduled_workflow_fires (community_id, scheduled_for);
-
 -- ── Workflow runs ─────────────────────────────────────────────────────────────
 
 CREATE TABLE workflow_runs (
@@ -261,6 +247,25 @@ CREATE TABLE workflow_runs (
 CREATE INDEX idx_workflow_runs_workflow ON workflow_runs (workflow_id);
 CREATE INDEX idx_workflow_runs_status ON workflow_runs (status);
 
+-- Restart-safe horizontal-scaling claim table for scheduled workflow fires.
+-- Each schedule tick must win this insert before creating a workflow run; this
+-- replaces per-pod in-memory last-fired state as the deduplication boundary.
+CREATE TABLE scheduled_workflow_fires (
+    community_id    UUID NOT NULL REFERENCES communities(id),
+    workflow_id     UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE,
+    scheduled_for   TIMESTAMPTZ NOT NULL,
+    claimed_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    workflow_run_id UUID REFERENCES workflow_runs(id) ON DELETE SET NULL,
+    PRIMARY KEY (community_id, workflow_id, scheduled_for)
+);
+
+CREATE INDEX idx_scheduled_workflow_fires_scheduled_for
+    ON scheduled_workflow_fires (community_id, scheduled_for);
+
+CREATE UNIQUE INDEX idx_scheduled_workflow_fires_run
+    ON scheduled_workflow_fires (workflow_run_id)
+    WHERE workflow_run_id IS NOT NULL;
+
 -- ── Workflow approvals ────────────────────────────────────────────────────────
 
 CREATE TABLE workflow_approvals (