From 858acc8ff322d908502f90abb20663f047b5fb03 Mon Sep 17 00:00:00 2001 From: Forge Date: Mon, 14 Sep 2026 16:17:29 -0400 Subject: [PATCH] fix(workflow): prevent scheduled scan gaps and starvation Signed-off-by: Forge --- Cargo.lock | 1 + crates/buzz-db/src/runtime/migration.rs | 6 +- crates/buzz-db/src/store/workflow.rs | 176 +++++++++++- crates/buzz-workflow/Cargo.toml | 3 + crates/buzz-workflow/src/lib.rs | 258 ++++++++++++++---- .../0047_workflow_schedule_scan_index.sql | 7 + schema/schema.sql | 5 + 7 files changed, 394 insertions(+), 62 deletions(-) create mode 100644 migrations/0047_workflow_schedule_scan_index.sql diff --git a/Cargo.lock b/Cargo.lock index 09d8faa9887..5778c5725c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1497,6 +1497,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "sqlx", "thiserror 2.0.18", "tokio", "tracing", diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index aebf8b9092f..30f00990810 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -703,7 +703,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 46); + assert_eq!(migrations.len(), 47); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1288,6 +1288,10 @@ mod postgres_tests { .sql .as_str() .contains("CREATE TABLE storage_accounting_snapshots")); + assert_eq!(migrations[46].version, 47); + let workflow_schedule_scan = migrations[46].sql.as_str(); + assert!(workflow_schedule_scan.contains("CREATE INDEX idx_workflows_schedule_scan")); + assert!(desired_schema.contains("CREATE INDEX idx_workflows_schedule_scan")); // schema.sql exclusion list must match the restored (pre-0041) body. assert!( desired_schema.contains("'rate_limit_violations'\n ]::TEXT[])"), diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 043ab112cfa..eda651a4a50 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -192,6 +192,32 @@ pub struct WorkflowRecord { pub updated_at: DateTime, } +/// Stable keyset cursor for the global scheduled-workflow scan. +/// +/// The scheduler must visit every enabled schedule across every community +/// without loading an unbounded result in one query. `created_at` alone is not +/// unique, and workflow UUIDs may collide across communities, so all three +/// fields participate in the ordering and cursor comparison. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ScheduledWorkflowCursor { + /// Creation timestamp of the last workflow in the previous page. + pub created_at: DateTime, + /// Server-resolved community of the last workflow in the previous page. + pub community_id: CommunityId, + /// Workflow UUID of the last workflow in the previous page. + pub workflow_id: Uuid, +} + +impl From<&WorkflowRecord> for ScheduledWorkflowCursor { + fn from(workflow: &WorkflowRecord) -> Self { + Self { + created_at: workflow.created_at, + community_id: workflow.community_id, + workflow_id: workflow.id, + } + } +} + /// A single execution of a workflow. #[derive(Debug, Clone)] pub struct WorkflowRunRecord { @@ -457,12 +483,24 @@ pub async fn list_enabled_channel_workflows( rows.into_iter().map(row_to_workflow_record).collect() } -/// List all active, enabled workflows with a `schedule` trigger across all channels. +/// List one keyset page of active, enabled schedule-triggered workflows. /// -/// Used by the cron scheduler. Filters by trigger type in SQL to avoid loading -/// event-triggered workflows that the cron loop would immediately discard. -/// Results are bounded to [`LIST_MAX_LIMIT`] rows. -pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result> { +/// Used by the cron scheduler, which exhausts pages before completing a scan. +/// Each query remains bounded to [`LIST_MAX_LIMIT`] rows without permanently +/// starving workflows ordered after the first page. `scan_through` freezes the +/// upper edge for both inserts and definition/eligibility updates, so later +/// pages cannot observe schedule changes made after the scan began. +pub async fn list_enabled_schedule_workflows_page( + pool: &PgPool, + cursor: Option, + scan_through: DateTime, + limit: i64, +) -> Result> { + let limit = limit.clamp(1, LIST_MAX_LIMIT); + let after_created_at = cursor.map(|value| value.created_at); + let after_community_id = cursor.map(|value| *value.community_id.as_uuid()); + let after_workflow_id = cursor.map(|value| value.workflow_id); + let rows = sqlx::query( r#" SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, @@ -473,11 +511,21 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result'trigger'->>'on' = 'schedule' AND c.archived_at IS NULL - ORDER BY w.created_at ASC - LIMIT $1 + AND w.created_at <= $4 + AND w.updated_at <= $4 + AND ( + $1::timestamptz IS NULL + OR (w.created_at, w.community_id, w.id) > ($1, $2::uuid, $3::uuid) + ) + ORDER BY w.created_at ASC, w.community_id ASC, w.id ASC + LIMIT $5 "#, ) - .bind(LIST_MAX_LIMIT) + .bind(after_created_at) + .bind(after_community_id) + .bind(after_workflow_id) + .bind(scan_through) + .bind(limit) .fetch_all(pool) .await?; @@ -494,7 +542,8 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result Result> { - crate::workflow::list_all_enabled_workflows(&self.pool).await + /// List one keyset page of active, enabled schedule-triggered workflows. + #[datastore_span(name = "list_enabled_schedule_workflows_page", system = "postgresql")] + pub async fn list_enabled_schedule_workflows_page( + &self, + cursor: Option, + scan_through: DateTime, + limit: i64, + ) -> Result> { + crate::workflow::list_enabled_schedule_workflows_page( + &self.pool, + cursor, + scan_through, + limit, + ) + .await } /// Claim a scheduled workflow fire for an authoritative schedule instant. @@ -2175,7 +2235,8 @@ mod postgres_tests { // 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 + // `workflow.community_id` returned by + // `list_enabled_schedule_workflows_page()`. 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. @@ -2261,6 +2322,93 @@ mod postgres_tests { (workflow_id, community) } + /// A scheduler scan must continue past the global per-query cap. Before + /// keyset pagination, the oldest 1,000 schedules were returned on every + /// tick and every newer schedule was permanently starved. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn scheduled_workflow_pages_include_rows_after_global_limit() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let owner = vec![0xc3; 32]; + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + let channel_id = make_channel(&pool, community, &owner).await; + + sqlx::query( + r#" + INSERT INTO workflows + (id, community_id, name, owner_pubkey, channel_id, definition, + definition_hash, status, enabled, created_at, updated_at) + SELECT gen_random_uuid(), $1, 'scheduled-' || n::text, $2, $3, + CASE WHEN n = 1001 + THEN '{"trigger":{"on":"message_posted"},"steps":[]}'::jsonb + ELSE '{"trigger":{"on":"schedule","cron":"* * * * *"},"steps":[]}'::jsonb + END, + decode(repeat('00', 32), 'hex'), 'active', TRUE, + CASE WHEN n = 1003 + THEN '2026-01-03T00:00:00Z'::timestamptz + ELSE '2026-01-01T00:00:00Z'::timestamptz + + n * interval '1 microsecond' + END, + CASE WHEN n = 1003 + THEN '2026-01-03T00:00:00Z'::timestamptz + ELSE '2026-01-01T00:00:00Z'::timestamptz + + n * interval '1 microsecond' + END + FROM generate_series(1, 1003) AS n + "#, + ) + .bind(community.as_uuid()) + .bind(&owner) + .bind(channel_id) + .execute(&pool) + .await + .expect("insert scheduled workflows"); + + let scan_through = chrono::DateTime::parse_from_rfc3339("2026-01-02T00:00:00Z") + .expect("scan bound") + .with_timezone(&Utc); + let first = list_enabled_schedule_workflows_page(&pool, None, scan_through, LIST_MAX_LIMIT) + .await + .expect("first page"); + assert_eq!(first.len(), LIST_MAX_LIMIT as usize); + + let cursor = first + .last() + .map(ScheduledWorkflowCursor::from) + .expect("first page cursor"); + + // Turn an older event workflow into a schedule after the snapshot + // boundary. A created_at-only fence would admit it on page two even + // though page one was evaluated against an earlier definition set. + sqlx::query( + r#" + UPDATE workflows + SET definition = '{"trigger":{"on":"schedule","cron":"* * * * *"},"steps":[]}'::jsonb, + updated_at = NOW() + WHERE community_id = $1 AND name = 'scheduled-1001' + "#, + ) + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("update workflow after scan boundary"); + + let second = + list_enabled_schedule_workflows_page(&pool, Some(cursor), scan_through, LIST_MAX_LIMIT) + .await + .expect("second page"); + + assert_eq!( + second.len(), + 1, + "only the unchanged row beyond the page cap belongs to this frozen scan" + ); + assert_eq!(second[0].name, "scheduled-1002"); + } + /// 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 diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index 7d361b1477a..70a3a617905 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/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"] diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index ee1c7467762..ce5edd83b16 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -46,13 +46,15 @@ use std::sync::OnceLock; use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION}; use buzz_core::tenant::CommunityId; -use buzz_db::workflow::RunStatus; +use buzz_db::workflow::{RunStatus, ScheduledWorkflowCursor, LIST_MAX_LIMIT}; use buzz_db::Db; use chrono::{DateTime, Utc}; use dashmap::DashMap; use tokio::sync::Semaphore; use uuid::Uuid; +const SCHEDULE_TICK_SECS: u64 = 60; + /// Runtime configuration for the workflow engine. #[derive(Clone, Debug)] pub struct WorkflowConfig { @@ -476,8 +478,9 @@ impl WorkflowEngine { /// trigger, checks whether the cron expression or interval has elapsed /// and spawns execution if so. /// - /// Uses window-based matching for cron expressions to handle tick drift: - /// `schedule.after(&(now - 60s)).next() <= now` instead of `includes(now)`. + /// Uses window-based matching for cron expressions to handle tick drift. + /// The window spans the actual elapsed time since the prior successful + /// scan, so database and execution work cannot open gaps between ticks. /// /// Interval tracking is anchored on the durable scheduled-fire claim: /// `last_fired` is an in-memory pre-filter, but the @@ -489,19 +492,51 @@ impl WorkflowEngine { pub async fn run(self: &Arc) { tracing::info!("WorkflowEngine cron loop started (60s tick)"); - loop { - tokio::time::sleep(std::time::Duration::from_secs(60)).await; + let mut previous_scan_at = Utc::now(); + loop { + // Sleep after the prior scan rather than using a fixed-rate ticker. + // This keeps interval prefilter samples at least one full tick apart + // even when a scan crosses a minute boundary. + tokio::time::sleep(std::time::Duration::from_secs(SCHEDULE_TICK_SECS)).await; let now = Utc::now(); + self.run_scheduled_scan(&mut previous_scan_at, now).await; + } + } - let workflows = match self.db.list_all_enabled_workflows().await { + /// Run one complete scheduler scan. + /// + /// Advances `previous_scan_at` only after every frozen-snapshot page has + /// been read. A failed or partial scan retains the prior watermark so the + /// next successful scan covers the entire elapsed cron window. + async fn run_scheduled_scan( + self: &Arc, + previous_scan_at: &mut DateTime, + now: DateTime, + ) -> bool { + let mut cursor = None; + let mut active_ids = std::collections::HashSet::new(); + + loop { + let workflows = match self + .db + .list_enabled_schedule_workflows_page(cursor, now, LIST_MAX_LIMIT) + .await + { Ok(wf) => wf, Err(e) => { tracing::error!("Cron tick: failed to load workflows: {e}"); - continue; + return false; } }; + if workflows.is_empty() { + break; + } + + cursor = workflows.last().map(ScheduledWorkflowCursor::from); + active_ids.extend(workflows.iter().map(|w| (w.community_id, w.id))); + for workflow in &workflows { // The same workflow UUID may exist in another community; carry // the row's owning community through fire-tracking, run creation, @@ -542,10 +577,13 @@ impl WorkflowEngine { schema::TriggerDef::Schedule { cron: Some(expr), interval: None, - } => match cron_fire_instant(expr, now, 60, workflow.id) { - Some(instant) => (instant, "cron"), - None => continue, - }, + } => { + match scheduler_cron_fire_instant(expr, *previous_scan_at, now, workflow.id) + { + Some(instant) => (instant, "cron"), + None => continue, + } + } schema::TriggerDef::Schedule { cron: None, interval: Some(dur), @@ -737,42 +775,42 @@ impl WorkflowEngine { .await; }); } - - // Fix 1: prune stale last_fired entries for workflows that are no longer - // active/enabled. Without this the DashMap grows monotonically as - // workflows are deleted or disabled. Keyed by `(community_id, id)` so - // entries are matched to the same scope they were inserted under. - let active_ids: std::collections::HashSet<(CommunityId, Uuid)> = - workflows.iter().map(|w| (w.community_id, w.id)).collect(); - self.last_fired.retain(|key, _| active_ids.contains(key)); } + + // Fix 1: prune stale last_fired entries for workflows that are no longer + // active/enabled. Without this the DashMap grows monotonically as + // workflows are deleted or disabled. Keyed by `(community_id, id)` so + // entries are matched to the same scope they were inserted under. + self.last_fired.retain(|key, _| active_ids.contains(key)); + *previous_scan_at = now; + true } } -/// Find the cron schedule instant that fired within the `window_secs`-wide -/// window ending at `now`, if any. +/// Resolve the cron instant due since the scheduler's prior successful scan. /// -/// Uses window-based matching: finds the next scheduled time after -/// `(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. 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 `None` (and logs a warning) if the expression is invalid or nothing -/// is due in the window. -fn cron_fire_instant( +/// The exact lower bound avoids both gaps and rounded overlap. If multiple +/// instants elapsed after a prolonged failed scan, only the latest is claimed; +/// replaying every missed invocation could create an unbounded side-effect +/// burst when service resumes. +fn scheduler_cron_fire_instant( expr: &str, + previous_scan_at: DateTime, now: DateTime, - window_secs: i64, workflow_id: Uuid, ) -> Option> { + if now <= previous_scan_at { + return None; + } + let normalized = schema::normalize_cron(expr); match normalized.parse::() { Ok(sched) => { - let window_start = now - chrono::Duration::seconds(window_secs); - sched.after(&window_start).next().filter(|t| *t <= now) + let search_end = now + chrono::Duration::nanoseconds(1); + sched + .after(&search_end) + .next_back() + .filter(|instant| *instant > previous_scan_at && *instant <= now) } Err(e) => { tracing::warn!( @@ -1052,14 +1090,13 @@ mod postgres_tests { #[test] 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 previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); - // The matched instant is the minute boundary 12:00:00, NOT `now`. assert_eq!( - cron_fire_instant("* * * * *", now, 60, wf_id), + scheduler_cron_fire_instant("* * * * *", previous_scan_at, now, wf_id), Some( chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:00Z") .unwrap() @@ -1072,9 +1109,10 @@ mod postgres_tests { #[test] fn cron_fire_instant_returns_none_for_invalid_expr() { let now = Utc::now(); + let previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); assert!( - cron_fire_instant("not-a-cron", now, 60, wf_id).is_none(), + scheduler_cron_fire_instant("not-a-cron", previous_scan_at, now, wf_id).is_none(), "invalid cron should return None" ); } @@ -1085,10 +1123,11 @@ mod postgres_tests { let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T14:30:00Z") .unwrap() .with_timezone(&Utc); + let previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); // "0 0 1 1 *" = midnight on Jan 1 only — June 15 is definitely outside. assert!( - cron_fire_instant("0 0 1 1 *", now, 60, wf_id).is_none(), + scheduler_cron_fire_instant("0 0 1 1 *", previous_scan_at, now, wf_id).is_none(), "Jan-1-only cron should not fire on June 15" ); } @@ -1100,9 +1139,10 @@ mod postgres_tests { let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z") .unwrap() .with_timezone(&Utc); + let previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); assert_eq!( - cron_fire_instant("0 9 * * *", now, 60, wf_id), + scheduler_cron_fire_instant("0 9 * * *", previous_scan_at, now, wf_id), Some(now), "cron should fire at exact minute boundary, anchored on 09:00:00" ); @@ -1117,9 +1157,10 @@ mod postgres_tests { let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:45Z") .unwrap() .with_timezone(&Utc); + let previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); assert_eq!( - cron_fire_instant("0 9 * * *", now, 60, wf_id), + scheduler_cron_fire_instant("0 9 * * *", previous_scan_at, now, wf_id), Some( chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z") .unwrap() @@ -1129,6 +1170,49 @@ mod postgres_tests { ); } + #[test] + fn elapsed_scan_window_covers_work_that_pushes_tick_past_sixty_seconds() { + // The prior implementation slept 60 seconds after each scan and still + // looked back exactly 60 seconds. Any loop work opened an uncovered + // gap. Here the next scan arrives 62 seconds later, and the production + // window calculation retains the 09:00 schedule instant. + let previous_scan_at = chrono::DateTime::parse_from_rfc3339("2026-06-15T08:59:59Z") + .unwrap() + .with_timezone(&Utc); + let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:01:01Z") + .unwrap() + .with_timezone(&Utc); + let scheduled = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z") + .unwrap() + .with_timezone(&Utc); + let wf_id = Uuid::new_v4(); + assert_eq!( + scheduler_cron_fire_instant("0 9 * * *", previous_scan_at, now, wf_id), + Some(scheduled), + "elapsed scan window must not leave a gap after slow loop work" + ); + } + + #[test] + fn elapsed_scan_window_coalesces_multiple_missed_instants_to_latest() { + let previous_scan_at = chrono::DateTime::parse_from_rfc3339("2026-06-15T08:59:59Z") + .unwrap() + .with_timezone(&Utc); + let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:02:20Z") + .unwrap() + .with_timezone(&Utc); + let latest = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:02:00Z") + .unwrap() + .with_timezone(&Utc); + let wf_id = Uuid::new_v4(); + + assert_eq!( + scheduler_cron_fire_instant("* * * * *", previous_scan_at, now, wf_id), + Some(latest), + "recovery must fire once at the latest due instant, not replay a burst" + ); + } + #[test] fn cron_fire_instant_returns_none_just_outside_window() { // Fixed time: 09:01:01 UTC. Cron "0 9 * * *" fires at 09:00:00. @@ -1136,9 +1220,10 @@ mod postgres_tests { let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:01:01Z") .unwrap() .with_timezone(&Utc); + let previous_scan_at = now - chrono::Duration::seconds(60); let wf_id = Uuid::new_v4(); assert!( - cron_fire_instant("0 9 * * *", now, 60, wf_id).is_none(), + scheduler_cron_fire_instant("0 9 * * *", previous_scan_at, now, wf_id).is_none(), "cron should not fire 61s after the scheduled time" ); } @@ -1888,22 +1973,51 @@ steps: // -- SEC-006: event-path regression (requires Postgres) ---------------- - async fn setup_db() -> buzz_db::Db { - let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + fn test_database_url() -> String { + std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) // Local-only test default; this is not a production credential. .unwrap_or_else(|_| { let local_test_database = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 local_test_database.to_owned() - }); + }) + } + + async fn setup_db() -> buzz_db::Db { buzz_db::Db::new(&buzz_db::DbConfig { - database_url, + database_url: test_database_url(), ..Default::default() }) .await .expect("connect test DB") } + /// A page-read failure must not offer a replacement watermark to the + /// scheduler loop. A closed lazy pool produces a deterministic local + /// failure without needing a database server. + #[tokio::test] + async fn failed_scheduler_scan_retains_prior_watermark() { + let pool = sqlx::postgres::PgPoolOptions::new() + .connect_lazy("postgres://buzz:buzz_dev@localhost:5432/buzz") // sadscan:disable np.postgres.1 + .expect("create lazy test pool"); + pool.close().await; + let engine = Arc::new(WorkflowEngine::new( + buzz_db::Db::from_pool(pool), + WorkflowConfig::default(), + )); + let mut previous_scan_at = chrono::DateTime::parse_from_rfc3339("2026-06-15T08:59:59Z") + .expect("previous watermark") + .with_timezone(&Utc); + let original_watermark = previous_scan_at; + let now = previous_scan_at + chrono::Duration::seconds(60); + + assert!( + !engine.run_scheduled_scan(&mut previous_scan_at, now).await, + "failed page reads must not advance the scheduler watermark" + ); + assert_eq!(previous_scan_at, original_watermark); + } + /// Create a community, a channel owned by `creator`, and add `member` as a /// plain member. Returns `(community, channel)`. async fn setup_channel(db: &buzz_db::Db, creator: &[u8], member: &[u8]) -> (CommunityId, Uuid) { @@ -1947,6 +2061,56 @@ steps: (community, channel_id) } + /// Exercise pagination through the production scan seam, not just the DB + /// page query. Every interval workflow is cold-start seeded, including the + /// row beyond the global page cap. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn scheduled_scan_processes_rows_after_global_page_limit() { + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + let pool = sqlx::PgPool::connect(&test_database_url()) + .await + .expect("connect direct test pool"); + + sqlx::query( + r#" + INSERT INTO workflows + (id, community_id, name, owner_pubkey, channel_id, definition, + definition_hash, status, enabled, created_at, updated_at) + SELECT gen_random_uuid(), $1, 'scan-page-' || n::text, $2, $3, + '{"name":"paged interval","trigger":{"on":"schedule","interval":"1h"},"steps":[],"enabled":true}'::jsonb, + decode(repeat('00', 32), 'hex'), 'active', TRUE, + NOW() - interval '1 second' + n * interval '1 microsecond', + NOW() - interval '1 second' + n * interval '1 microsecond' + FROM generate_series(1, 1001) AS n + "#, + ) + .bind(community.as_uuid()) + .bind(&member) + .bind(channel_id) + .execute(&pool) + .await + .expect("insert paginated interval workflows"); + + let engine = Arc::new(WorkflowEngine::new(db, WorkflowConfig::default())); + let now = Utc::now() + chrono::Duration::seconds(1); + let mut previous_scan_at = now - chrono::Duration::seconds(60); + + assert!( + engine.run_scheduled_scan(&mut previous_scan_at, now).await, + "a complete multi-page scan must return its replacement watermark" + ); + assert_eq!(previous_scan_at, now); + assert_eq!( + engine.last_fired.len(), + 1001, + "production scan must reach and seed interval rows after the first page" + ); + } + fn message_event(channel_id: Uuid) -> buzz_core::StoredEvent { let keys = nostr::Keys::generate(); let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "hello") diff --git a/migrations/0047_workflow_schedule_scan_index.sql b/migrations/0047_workflow_schedule_scan_index.sql new file mode 100644 index 00000000000..e7f3b2366d1 --- /dev/null +++ b/migrations/0047_workflow_schedule_scan_index.sql @@ -0,0 +1,7 @@ +-- Keep the global scheduled-workflow keyset scan on an ordered partial index. +-- The scheduler exhausts this index in bounded pages on every successful tick. +CREATE INDEX idx_workflows_schedule_scan + ON workflows (created_at, community_id, id) + WHERE status = 'active' + AND enabled = TRUE + AND definition->'trigger'->>'on' = 'schedule'; diff --git a/schema/schema.sql b/schema/schema.sql index d7074f359d8..506e8a32bb2 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -381,6 +381,11 @@ CREATE INDEX idx_workflows_channel_active ON workflows (community_id, channel_id -- Scheduler scans enabled schedule workflows; community_id returned per row so -- side effects run under the owning tenant's context (Lane0 contract §4a.5). CREATE INDEX idx_workflows_enabled ON workflows (enabled, status) WHERE enabled; +CREATE INDEX idx_workflows_schedule_scan + ON workflows (created_at, community_id, id) + WHERE status = 'active' + AND enabled = TRUE + AND definition->'trigger'->>'on' = 'schedule'; -- ── Workflow runs ─────────────────────────────────────────────────────────────