From 48c53a5d7eb9ce54dd37b49afd73a348d0a3723d Mon Sep 17 00:00:00 2001 From: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Date: Fri, 26 Jun 2026 13:21:38 -0400 Subject: [PATCH] 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 | 43 +++++++++- crates/buzz-db/src/migration.rs | 14 ++++ crates/buzz-db/src/workflow.rs | 127 +++++++++++++++++++++++++++-- migrations/0001_initial_schema.sql | 33 ++++---- schema/schema.sql | 33 ++++---- 5 files changed, 212 insertions(+), 38 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2b3ce6deb..adb4c0187 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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, - /// scheduled_for)`; all other pods must skip creating a run. + /// Returns `Some` only for the first pod to claim `(community_id, + /// 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, - ) -> Result { + ) -> Result> { 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>> { + 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, + workflow_run_id: Uuid, + ) -> Result { + 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, + ) -> Result { + 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, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 84fc50a6b..26c94660e 100644 --- a/crates/buzz-db/src/migration.rs +++ b/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" diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index a35812b28..046e157ba 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -213,6 +213,23 @@ pub struct WorkflowRunRecord { pub created_at: DateTime, } +/// 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, + /// Database timestamp for when this pod won the claim. + pub claimed_at: DateTime, +} + /// 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, -) -> Result { - let result = sqlx::query( +) -> Result> { + 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>> { + 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, + workflow_run_id: Uuid, +) -> Result { + 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, +) -> Result { + 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, diff --git a/migrations/0001_initial_schema.sql b/migrations/0001_initial_schema.sql index e7f94b7f9..cbc7162c3 100644 --- a/migrations/0001_initial_schema.sql +++ b/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 ( diff --git a/schema/schema.sql b/schema/schema.sql index ccb73bf58..7c9595c5b 100644 --- a/schema/schema.sql +++ b/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 (