From 77010a0ea8c963811ea2beb234741aad745f58fb Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 12:16:32 +0000 Subject: [PATCH 1/2] prevent deferred worker items from starving ready work Order eligible worker items by visibility time, then queue ID, so expired deferrals cannot repeatedly overtake work that was already ready. Add migration 0027 and a matching index, preserving session ownership, tag filtering, locking, and attempt accounting. Add regression tests for session/non-session and tagged/untagged work, and document rollout impact. --- CHANGELOG.md | 16 +++ docs/migration-policy.md | 13 ++ .../0027_order_worker_queue_by_visibility.sql | 122 ++++++++++++++++++ tests/worker_progress_tests.rs | 86 ++++++++++++ 4 files changed, 237 insertions(+) create mode 100644 migrations/0027_order_worker_queue_by_visibility.sql create mode 100644 tests/worker_progress_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 917c21b..3dd315f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Deferred activities no longer repeatedly overtake already-ready work after + their retry delay expires. Migration `0027` orders eligible worker items by + visibility time and then queue ID, with a matching index, for both session and + non-session workers. Session ownership, tag filtering, lock handling, and + attempt accounting are unchanged. + +### Changed + +- Upgrading applies migration `0027`, which replaces `fetch_work_item` without + changing its signature and adds an index on `worker_queue(visible_at, id)`. + Index creation blocks queue writes while it runs; plan migration time for + large queues. Already-running older providers can use the updated function, + but older binaries reject the new migration version on startup. + ## [0.1.38] - 2026-10-26 ### Fixed diff --git a/docs/migration-policy.md b/docs/migration-policy.md index e5716d0..34fcceb 100644 --- a/docs/migration-policy.md +++ b/docs/migration-policy.md @@ -46,6 +46,19 @@ startup so multiple nodes do not race to apply the same migration. `ApplyAll` also rejects unknown future migration versions before it runs DDL, so an older binary will not mutate a schema that appears ahead of its code. +## Worker Queue Progress (Migration 0027) + +Migration `0027` changes worker selection to order eligible items by visibility +time, then queue ID. Deferred work moves behind items that were already ready, +even when its delay expires before the backlog drains. Session and tag routing +are unchanged. The migration adds an index on `worker_queue(visible_at, id)`; +this non-concurrent index build blocks queue writes while it runs. + +The stored procedure signature is unchanged, so already-running older providers +can continue using it. As with other migrations, older binaries cannot restart +against the upgraded schema because they reject unknown migration versions. +Coordinate migration and worker rollout accordingly. + ## Example ```rust diff --git a/migrations/0027_order_worker_queue_by_visibility.sql b/migrations/0027_order_worker_queue_by_visibility.sql new file mode 100644 index 0000000..d80fc27 --- /dev/null +++ b/migrations/0027_order_worker_queue_by_visibility.sql @@ -0,0 +1,122 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the MIT License. + +-- A deferral must move behind already-ready work even if its delay expires +-- before the backlog has been drained. +DO $migration$ +DECLARE + v_schema_name TEXT := current_schema(); +BEGIN + EXECUTE format( + 'CREATE INDEX IF NOT EXISTS idx_worker_ready_order ON %I.worker_queue(visible_at, id)', + v_schema_name + ); + + EXECUTE format($ddl$ + CREATE OR REPLACE FUNCTION %1$I.fetch_work_item( + p_now_ms BIGINT, + p_lock_timeout_ms BIGINT, + p_owner_id TEXT DEFAULT NULL, + p_session_lock_timeout_ms BIGINT DEFAULT NULL, + p_tag_filter TEXT[] DEFAULT NULL, + p_tag_mode TEXT DEFAULT 'default_only' + ) + RETURNS TABLE( + out_work_item TEXT, + out_lock_token TEXT, + out_attempt_count INTEGER + ) AS $fetch_worker$ + DECLARE + v_id BIGINT; + v_session_id TEXT; + v_session_locked_until BIGINT; + BEGIN + IF p_tag_mode = 'none' THEN + RETURN; + END IF; + + IF p_owner_id IS NOT NULL THEN + SELECT q.id, q.session_id INTO v_id, v_session_id + FROM %1$I.worker_queue q + LEFT JOIN %1$I.sessions s ON s.session_id = q.session_id AND s.locked_until > p_now_ms + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms) + AND ( + q.session_id IS NULL + OR s.worker_id = p_owner_id + OR s.session_id IS NULL + ) + AND ( + CASE p_tag_mode + WHEN 'default_only' THEN q.tag IS NULL + WHEN 'tags' THEN q.tag = ANY(p_tag_filter) + WHEN 'default_and' THEN (q.tag IS NULL OR q.tag = ANY(p_tag_filter)) + WHEN 'any' THEN TRUE + ELSE FALSE + END + ) + ORDER BY q.visible_at, q.id + LIMIT 1 + FOR UPDATE OF q SKIP LOCKED; + ELSE + SELECT q.id, q.session_id INTO v_id, v_session_id + FROM %1$I.worker_queue q + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms) + AND q.session_id IS NULL + AND ( + CASE p_tag_mode + WHEN 'default_only' THEN q.tag IS NULL + WHEN 'tags' THEN q.tag = ANY(p_tag_filter) + WHEN 'default_and' THEN (q.tag IS NULL OR q.tag = ANY(p_tag_filter)) + WHEN 'any' THEN TRUE + ELSE FALSE + END + ) + ORDER BY q.visible_at, q.id + LIMIT 1 + FOR UPDATE OF q SKIP LOCKED; + END IF; + + IF NOT FOUND THEN + RETURN; + END IF; + + out_lock_token := 'lock_' || gen_random_uuid()::TEXT; + UPDATE %1$I.worker_queue + SET lock_token = out_lock_token, + locked_until = p_now_ms + p_lock_timeout_ms, + attempt_count = attempt_count + 1 + WHERE id = v_id; + + SELECT work_item, attempt_count + INTO out_work_item, out_attempt_count + FROM %1$I.worker_queue + WHERE id = v_id; + + IF v_session_id IS NOT NULL AND p_owner_id IS NOT NULL THEN + v_session_locked_until := p_now_ms + COALESCE(p_session_lock_timeout_ms, p_lock_timeout_ms); + INSERT INTO %1$I.sessions (session_id, worker_id, locked_until, last_activity_at) + VALUES (v_session_id, p_owner_id, v_session_locked_until, p_now_ms) + ON CONFLICT (session_id) DO UPDATE + SET worker_id = p_owner_id, + locked_until = v_session_locked_until, + last_activity_at = p_now_ms + WHERE %1$I.sessions.locked_until <= p_now_ms OR %1$I.sessions.worker_id = p_owner_id; + + IF NOT FOUND THEN + UPDATE %1$I.worker_queue + SET lock_token = NULL, + locked_until = NULL, + attempt_count = attempt_count - 1 + WHERE id = v_id; + RETURN; + END IF; + END IF; + + RETURN NEXT; + END; + $fetch_worker$ LANGUAGE plpgsql; + $ddl$, v_schema_name); +END; +$migration$; diff --git a/tests/worker_progress_tests.rs b/tests/worker_progress_tests.rs new file mode 100644 index 0000000..e145e21 --- /dev/null +++ b/tests/worker_progress_tests.rs @@ -0,0 +1,86 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +use duroxide::providers::{SessionFetchConfig, TagFilter, WorkItem}; +use std::time::Duration; + +mod common; + +#[tokio::test] +async fn expired_deferrals_yield_to_already_ready_work() { + for session_bound in [false, true] { + for tagged in [false, true] { + let (store, schema) = common::create_postgres_store().await; + let session = session_bound.then(|| SessionFetchConfig { + owner_id: "progress-worker".to_string(), + lock_timeout: Duration::from_secs(30), + }); + let tag_filter = if tagged { + TagFilter::tags(["limited"]) + } else { + TagFilter::DefaultOnly + }; + for id in [1, 2] { + store + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: "progress".to_string(), + execution_id: 1, + id, + name: "Activity".to_string(), + input: String::new(), + session_id: session_bound.then(|| "progress-session".to_string()), + tag: tagged.then(|| "limited".to_string()), + }) + .await + .unwrap(); + } + let (item, token, _) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); + store + .abandon_work_item(&token, Some(Duration::from_millis(1)), true) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(20)).await; + + let (item, token, _) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!( + matches!(item, WorkItem::ActivityExecute { id: 2, .. }), + "expired deferral overtook ready work: session={session_bound}, tagged={tagged}" + ); + store.ack_work_item(&token, None).await.unwrap(); + + let (item, token, attempts) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); + assert_eq!(attempts, 1); + store.ack_work_item(&token, None).await.unwrap(); + common::cleanup_schema(&schema).await; + } + } +} From 9ae067b34abc4c1f1f1436329973fe37731edb2a Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 16:31:34 +0000 Subject: [PATCH 2/2] fix: address worker queue progress review Renumber the migration to 0029 and add its baseline-0016 diff. Centralize rollout guidance, including concurrent index preparation and the SHARE lock retained even when the build is skipped. Reject invalid prebuilt indexes without changing the migration ledger, fetch function, or queued work. Add red/green regression coverage, enforce unique increasing embedded versions in a unit test, refresh the embed marker, and reuse panic-safe test schema cleanup. Validation: 301 related unit, provider, and migration tests passed (one skipped); targeted regressions passed from fresh Cargo target directories. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 12 +- docs/migration-policy.md | 16 +- migrations/0029_diff.md | 138 +++++++++ ...0029_order_worker_queue_by_visibility.sql} | 12 + migrations/README.md | 47 ++++ src/migrations.rs | 28 +- tests/worker_progress_tests.rs | 266 +++++++++++++----- 7 files changed, 430 insertions(+), 89 deletions(-) create mode 100644 migrations/0029_diff.md rename migrations/{0027_order_worker_queue_by_visibility.sql => 0029_order_worker_queue_by_visibility.sql} (89%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3dd315f..87a7ddc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,18 +10,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Deferred activities no longer repeatedly overtake already-ready work after - their retry delay expires. Migration `0027` orders eligible worker items by + their retry delay expires. Migration `0029` orders eligible worker items by visibility time and then queue ID, with a matching index, for both session and non-session workers. Session ownership, tag filtering, lock handling, and attempt accounting are unchanged. ### Changed -- Upgrading applies migration `0027`, which replaces `fetch_work_item` without +- Upgrading applies migration `0029`, which replaces `fetch_work_item` without changing its signature and adds an index on `worker_queue(visible_at, id)`. - Index creation blocks queue writes while it runs; plan migration time for - large queues. Already-running older providers can use the updated function, - but older binaries reject the new migration version on startup. + Index creation blocks worker queue operations until the migration commits; + for busy queues, [pre-build the index concurrently](migrations/README.md#worker-queue-progress-migration-0029). + Invalid prebuilt indexes are rejected. Already-running older providers can + use the updated function, but older binaries reject the new migration + version on startup. ## [0.1.38] - 2026-10-26 diff --git a/docs/migration-policy.md b/docs/migration-policy.md index 34fcceb..bc19500 100644 --- a/docs/migration-policy.md +++ b/docs/migration-policy.md @@ -46,18 +46,10 @@ startup so multiple nodes do not race to apply the same migration. `ApplyAll` also rejects unknown future migration versions before it runs DDL, so an older binary will not mutate a schema that appears ahead of its code. -## Worker Queue Progress (Migration 0027) - -Migration `0027` changes worker selection to order eligible items by visibility -time, then queue ID. Deferred work moves behind items that were already ready, -even when its delay expires before the backlog drains. Session and tag routing -are unchanged. The migration adds an index on `worker_queue(visible_at, id)`; -this non-concurrent index build blocks queue writes while it runs. - -The stored procedure signature is unchanged, so already-running older providers -can continue using it. As with other migrations, older binaries cannot restart -against the upgraded schema because they reject unknown migration versions. -Coordinate migration and worker rollout accordingly. +## Worker Queue Progress (Migration 0029) + +See the [migration rollout notes](../migrations/README.md#worker-queue-progress-migration-0029) +for queue locking impact, concurrent index preparation, and provider compatibility. ## Example diff --git a/migrations/0029_diff.md b/migrations/0029_diff.md new file mode 100644 index 0000000..8da1748 --- /dev/null +++ b/migrations/0029_diff.md @@ -0,0 +1,138 @@ +# Diff for migration 0029 + +**Migration file:** `0029_order_worker_queue_by_visibility.sql` + +Each changed function is shown **in full** with `+`/`-` markers on changed lines. + +## Table Changes + +None. + +## New Indexes + +### `idx_worker_ready_order` +```sql +CREATE INDEX idx_worker_ready_order ON SCHEMA.worker_queue USING btree (visible_at, id) +``` + +The migration verifies that this index is valid and belongs to `worker_queue` +before replacing the function. An invalid prebuilt index aborts the transaction +without recording the migration. + +## Function Changes + +### `fetch_work_item` — Body Modified (baseline: 0016) + +Full function with diff: +```diff + CREATE OR REPLACE FUNCTION SCHEMA.fetch_work_item(p_now_ms bigint, p_lock_timeout_ms bigint, p_owner_id text DEFAULT NULL::text, p_session_lock_timeout_ms bigint DEFAULT NULL::bigint, p_tag_filter text[] DEFAULT NULL::text[], p_tag_mode text DEFAULT 'default_only'::text) + RETURNS TABLE(out_work_item text, out_lock_token text, out_attempt_count integer) + LANGUAGE plpgsql + AS $function$ + DECLARE + v_id BIGINT; + v_session_id TEXT; + v_session_locked_until BIGINT; + BEGIN +- -- none mode: return immediately with no results + IF p_tag_mode = 'none' THEN + RETURN; + END IF; + + IF p_owner_id IS NOT NULL THEN +- -- Session-aware fetch with tag filtering + SELECT q.id, q.session_id INTO v_id, v_session_id + FROM SCHEMA.worker_queue q + LEFT JOIN SCHEMA.sessions s ON s.session_id = q.session_id AND s.locked_until > p_now_ms + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms) + AND ( + q.session_id IS NULL + OR s.worker_id = p_owner_id + OR s.session_id IS NULL + ) + AND ( + CASE p_tag_mode + WHEN 'default_only' THEN q.tag IS NULL + WHEN 'tags' THEN q.tag = ANY(p_tag_filter) + WHEN 'default_and' THEN (q.tag IS NULL OR q.tag = ANY(p_tag_filter)) + WHEN 'any' THEN TRUE + ELSE FALSE + END + ) +- ORDER BY q.id ++ ORDER BY q.visible_at, q.id + LIMIT 1 + FOR UPDATE OF q SKIP LOCKED; + ELSE +- -- Non-session fetch with tag filtering + SELECT q.id, q.session_id INTO v_id, v_session_id + FROM SCHEMA.worker_queue q + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms) + AND q.session_id IS NULL + AND ( + CASE p_tag_mode + WHEN 'default_only' THEN q.tag IS NULL + WHEN 'tags' THEN q.tag = ANY(p_tag_filter) + WHEN 'default_and' THEN (q.tag IS NULL OR q.tag = ANY(p_tag_filter)) + WHEN 'any' THEN TRUE + ELSE FALSE + END + ) +- ORDER BY q.id ++ ORDER BY q.visible_at, q.id + LIMIT 1 + FOR UPDATE OF q SKIP LOCKED; + END IF; + + IF NOT FOUND THEN + RETURN; + END IF; + + out_lock_token := 'lock_' || gen_random_uuid()::TEXT; +- +- -- Increment attempt_count and lock the item + UPDATE SCHEMA.worker_queue + SET lock_token = out_lock_token, + locked_until = p_now_ms + p_lock_timeout_ms, + attempt_count = attempt_count + 1 + WHERE id = v_id; + + SELECT work_item, attempt_count + INTO out_work_item, out_attempt_count + FROM SCHEMA.worker_queue + WHERE id = v_id; + +- -- If session-bound, upsert the sessions row + IF v_session_id IS NOT NULL AND p_owner_id IS NOT NULL THEN + v_session_locked_until := p_now_ms + COALESCE(p_session_lock_timeout_ms, p_lock_timeout_ms); +- + INSERT INTO SCHEMA.sessions (session_id, worker_id, locked_until, last_activity_at) + VALUES (v_session_id, p_owner_id, v_session_locked_until, p_now_ms) + ON CONFLICT (session_id) DO UPDATE + SET worker_id = p_owner_id, + locked_until = v_session_locked_until, + last_activity_at = p_now_ms + WHERE SCHEMA.sessions.locked_until <= p_now_ms OR SCHEMA.sessions.worker_id = p_owner_id; + +- -- If upsert affected 0 rows, another worker owns this session. +- -- Roll back: clear lock so item can be retried. + IF NOT FOUND THEN + UPDATE SCHEMA.worker_queue + SET lock_token = NULL, + locked_until = NULL, + attempt_count = attempt_count - 1 + WHERE id = v_id; + RETURN; + END IF; + END IF; + + RETURN NEXT; + END; + $function$ +\ No newline at end of file + +``` + +--- diff --git a/migrations/0027_order_worker_queue_by_visibility.sql b/migrations/0029_order_worker_queue_by_visibility.sql similarity index 89% rename from migrations/0027_order_worker_queue_by_visibility.sql rename to migrations/0029_order_worker_queue_by_visibility.sql index d80fc27..d967560 100644 --- a/migrations/0027_order_worker_queue_by_visibility.sql +++ b/migrations/0029_order_worker_queue_by_visibility.sql @@ -12,6 +12,18 @@ BEGIN v_schema_name ); + -- A failed concurrent prebuild leaves an invalid index that IF NOT EXISTS skips. + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_index i + WHERE i.indexrelid = format('%I.idx_worker_ready_order', v_schema_name)::regclass + AND i.indrelid = format('%I.worker_queue', v_schema_name)::regclass + AND i.indisvalid + ) THEN + RAISE EXCEPTION + 'Index %.idx_worker_ready_order is invalid or does not index worker_queue; repair it before retrying migration 0029', + v_schema_name; + END IF; + EXECUTE format($ddl$ CREATE OR REPLACE FUNCTION %1$I.fetch_work_item( p_now_ms BIGINT, diff --git a/migrations/README.md b/migrations/README.md index 3114817..d51bf08 100644 --- a/migrations/README.md +++ b/migrations/README.md @@ -90,6 +90,53 @@ CREATE INDEX IF NOT EXISTS idx_worker_processed ON worker_queue(processed_at); Migrations are automatically applied when creating a `PostgresProvider`. Each test schema gets its own migration history, so you can test migrations in isolation. +The `embedded_migration_versions_are_unique_and_increasing` unit test checks +that bundled SQL files have distinct, increasing version numbers. Gaps are +allowed; two migrations must never share a version. + +## Worker Queue Progress (Migration 0029) + +Migration `0029` orders eligible worker items by visibility time, then queue ID. +Deferred work moves behind items that were already ready, even when its delay +expires before the backlog drains. Session ownership, tag filtering, lock +handling, and attempt accounting are unchanged. + +The migration adds `idx_worker_ready_order` on `worker_queue(visible_at, id)`. +`CREATE INDEX IF NOT EXISTS` takes a `SHARE` lock on `worker_queue`, even when +the index already exists, and holds it **until the migration transaction +commits**, not just until index creation finishes. This blocks worker fetch, +acknowledge, abandon, and lock renewal, as well as orchestration acknowledgements +that write worker rows. The migration sets no `lock_timeout`, so acquiring the +lock can also wait indefinitely. Plan a maintenance window or set an appropriate +deployment-session `lock_timeout` through PostgreSQL connection options. + +For a busy or large queue, pre-build the index **before starting the migration**, +outside any transaction, in each target schema. Replace `duroxide` below with +the provider's schema: + +```sql +CREATE INDEX CONCURRENTLY idx_worker_ready_order + ON duroxide.worker_queue (visible_at, id); + +SELECT i.indisvalid, pg_get_indexdef(i.indexrelid) +FROM pg_catalog.pg_index i +WHERE i.indexrelid = 'duroxide.idx_worker_ready_order'::regclass; +``` + +Wait for the build to finish and verify that `indisvalid` is true and the +definition matches the index above. The migration's `CREATE INDEX IF NOT +EXISTS` will then skip the expensive build, but still take the `SHARE` lock +described above. A failed or cancelled concurrent build can leave an invalid +index with the same name; `IF NOT EXISTS` does not repair it, and the migration +rejects it without recording success. Inspect such an index, then drop it with +`DROP INDEX CONCURRENTLY duroxide.idx_worker_ready_order` outside a transaction +before retrying the concurrent build. + +The `fetch_work_item` signature is unchanged, so already-running older +providers can keep using it. Older binaries cannot restart against the upgraded +schema because they reject unknown migration versions. Coordinate migration +and provider rollout accordingly. + ## Migration Execution Context - Migrations run inside transactions diff --git a/src/migrations.rs b/src/migrations.rs index 6f933f4..2810ad0 100644 --- a/src/migrations.rs +++ b/src/migrations.rs @@ -8,6 +8,7 @@ use sqlx::PgPool; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +// Update this marker when adding SQL to refresh incremental embeds: 0029. static MIGRATIONS: Dir = include_dir!("$CARGO_MANIFEST_DIR/migrations"); const BASE_TABLES: &[&str] = &[ @@ -139,7 +140,7 @@ impl MigrationRunner { // recognize (schema is ahead of the code). self.check_no_unknown_migrations(conn).await?; - let migrations = self.load_migrations()?; + let migrations = Self::load_migrations()?; let applied = self.get_applied_versions(conn).await?; self.validate_applied_prefix(&migrations, &applied)?; let existing_tables = self.get_existing_core_relations(conn).await?; @@ -185,8 +186,7 @@ impl MigrationRunner { } let applied = self.get_applied_versions(conn).await?; - let expected: std::collections::HashSet = self - .load_migrations()? + let expected: std::collections::HashSet = Self::load_migrations()? .into_iter() .map(|m| m.version) .collect(); @@ -361,7 +361,7 @@ impl MigrationRunner { } // Load migrations from filesystem - let migrations = self.load_migrations()?; + let migrations = Self::load_migrations()?; tracing::debug!( "Loaded {} migrations for schema {}", @@ -409,7 +409,7 @@ impl MigrationRunner { } /// Load migrations from the embedded migrations directory - fn load_migrations(&self) -> Result> { + fn load_migrations() -> Result> { let mut migrations = Vec::new(); // Get all files from embedded directory @@ -433,7 +433,7 @@ impl MigrationRunner { .ok_or_else(|| anyhow::anyhow!("Migration file is not valid UTF-8: {file_name}"))? .to_string(); - let version = self.parse_version(file_name)?; + let version = Self::parse_version(file_name)?; let name = file_name.to_string(); migrations.push(Migration { version, name, sql }); @@ -443,7 +443,7 @@ impl MigrationRunner { } /// Parse version number from migration filename (e.g., "0001_initial.sql" -> 1) - fn parse_version(&self, filename: &str) -> Result { + fn parse_version(filename: &str) -> Result { let version_str = filename .split('_') .next() @@ -734,6 +734,20 @@ impl MigrationRunner { mod tests { use super::*; + #[test] + fn embedded_migration_versions_are_unique_and_increasing() { + let migrations = MigrationRunner::load_migrations().expect("load embedded migrations"); + assert!(!migrations.is_empty(), "expected embedded SQL migrations"); + for pair in migrations.windows(2) { + assert!( + pair[0].version < pair[1].version, + "embedded migrations must have unique, increasing versions: {} then {}", + pair[0].name, + pair[1].name, + ); + } + } + #[test] fn migration_search_path_pins_pg_temp_last() { // The security property: pg_temp must be present and last, so temporary diff --git a/tests/worker_progress_tests.rs b/tests/worker_progress_tests.rs index e145e21..73da7fd 100644 --- a/tests/worker_progress_tests.rs +++ b/tests/worker_progress_tests.rs @@ -1,86 +1,222 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -use duroxide::providers::{SessionFetchConfig, TagFilter, WorkItem}; +use duroxide::providers::{Provider, SessionFetchConfig, TagFilter, WorkItem}; +use duroxide_pg::PostgresProvider; use std::time::Duration; -mod common; +#[path = "deadlocks/harness.rs"] +mod harness; #[tokio::test] async fn expired_deferrals_yield_to_already_ready_work() { for session_bound in [false, true] { for tagged in [false, true] { - let (store, schema) = common::create_postgres_store().await; - let session = session_bound.then(|| SessionFetchConfig { - owner_id: "progress-worker".to_string(), - lock_timeout: Duration::from_secs(30), - }); - let tag_filter = if tagged { - TagFilter::tags(["limited"]) - } else { - TagFilter::DefaultOnly - }; - for id in [1, 2] { + harness::with_schema(move |schema| async move { + let store = + PostgresProvider::new_with_schema(&harness::database_url(), Some(&schema)) + .await + .expect("create worker progress provider"); + let session = session_bound.then(|| SessionFetchConfig { + owner_id: "progress-worker".to_string(), + lock_timeout: Duration::from_secs(30), + }); + let tag_filter = if tagged { + TagFilter::tags(["limited"]) + } else { + TagFilter::DefaultOnly + }; + for id in [1, 2] { + store + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: "progress".to_string(), + execution_id: 1, + id, + name: "Activity".to_string(), + input: String::new(), + session_id: session_bound.then(|| "progress-session".to_string()), + tag: tagged.then(|| "limited".to_string()), + }) + .await + .unwrap(); + } + let (item, token, _) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); store - .enqueue_for_worker(WorkItem::ActivityExecute { - instance: "progress".to_string(), - execution_id: 1, - id, - name: "Activity".to_string(), - input: String::new(), - session_id: session_bound.then(|| "progress-session".to_string()), - tag: tagged.then(|| "limited".to_string()), - }) + .abandon_work_item(&token, Some(Duration::from_millis(1)), true) .await .unwrap(); - } - let (item, token, _) = store - .fetch_work_item( - Duration::from_secs(30), - Duration::ZERO, - session.as_ref(), - &tag_filter, - ) - .await - .unwrap() - .unwrap(); - assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); - store - .abandon_work_item(&token, Some(Duration::from_millis(1)), true) - .await - .unwrap(); - tokio::time::sleep(Duration::from_millis(20)).await; + tokio::time::sleep(Duration::from_millis(20)).await; - let (item, token, _) = store - .fetch_work_item( - Duration::from_secs(30), - Duration::ZERO, - session.as_ref(), - &tag_filter, - ) - .await - .unwrap() - .unwrap(); - assert!( + let (item, token, _) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!( matches!(item, WorkItem::ActivityExecute { id: 2, .. }), "expired deferral overtook ready work: session={session_bound}, tagged={tagged}" ); - store.ack_work_item(&token, None).await.unwrap(); + store.ack_work_item(&token, None).await.unwrap(); - let (item, token, attempts) = store - .fetch_work_item( - Duration::from_secs(30), - Duration::ZERO, - session.as_ref(), - &tag_filter, - ) - .await - .unwrap() - .unwrap(); - assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); - assert_eq!(attempts, 1); - store.ack_work_item(&token, None).await.unwrap(); - common::cleanup_schema(&schema).await; + let (item, token, attempts) = store + .fetch_work_item( + Duration::from_secs(30), + Duration::ZERO, + session.as_ref(), + &tag_filter, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); + assert_eq!(attempts, 1); + store.ack_work_item(&token, None).await.unwrap(); + }) + .await; } } } + +#[tokio::test] +async fn worker_progress_migration_validates_prebuilt_index() { + harness::with_schema(|schema| async move { + let database_url = harness::database_url(); + let store = PostgresProvider::new_with_schema(&database_url, Some(&schema)) + .await + .expect("create worker progress provider"); + let pool = store.pool(); + store + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: "progress".to_string(), + execution_id: 1, + id: 1, + name: "Activity".to_string(), + input: String::new(), + session_id: None, + tag: None, + }) + .await + .unwrap(); + + let signature = format!("{schema}.fetch_work_item(bigint,bigint,text,bigint,text[],text)"); + let current_function: String = + sqlx::query_scalar("SELECT pg_get_functiondef($1::regprocedure)") + .bind(&signature) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(current_function.matches("ORDER BY q.visible_at, q.id").count(), 2); + let baseline_function = current_function.replace("ORDER BY q.visible_at, q.id", "ORDER BY q.id"); + + // Mark this and later migrations pending to preserve the ledger prefix. + let mut tx = pool.begin().await.unwrap(); + sqlx::query(&format!("DROP INDEX {schema}.idx_worker_ready_order")) + .execute(&mut *tx) + .await + .unwrap(); + assert!( + sqlx::query(&format!("DELETE FROM {schema}._duroxide_migrations WHERE version >= 29")) + .execute(&mut *tx) + .await + .unwrap() + .rows_affected() + >= 1 + ); + sqlx::query(&baseline_function).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + + // A writer lets the concurrent build register its index, but prevents + // its first scan. Timing out leaves a real invalid index behind. + let mut writer = pool.begin().await.unwrap(); + sqlx::query(&format!("LOCK TABLE {schema}.worker_queue IN ROW EXCLUSIVE MODE")) + .execute(&mut *writer) + .await + .unwrap(); + let mut builder = pool.acquire().await.unwrap(); + sqlx::query("SET lock_timeout = '100ms'") + .execute(&mut *builder) + .await + .unwrap(); + let create_index = format!( + "CREATE INDEX CONCURRENTLY idx_worker_ready_order ON {schema}.worker_queue (visible_at, id)" + ); + let error = sqlx::query(&create_index) + .execute(&mut *builder) + .await + .expect_err("the writer should block the concurrent index build"); + assert_eq!(error.as_database_error().unwrap().code().as_deref(), Some("55P03")); + writer.rollback().await.unwrap(); + sqlx::query("RESET lock_timeout") + .execute(&mut *builder) + .await + .unwrap(); + + let index_state = format!( + "SELECT indexrelid::bigint, indisvalid FROM pg_catalog.pg_index + WHERE indexrelid = '{schema}.idx_worker_ready_order'::regclass" + ); + let (_, valid): (i64, bool) = sqlx::query_as(&index_state).fetch_one(pool).await.unwrap(); + assert!(!valid, "the failed concurrent build must leave an invalid index"); + + let snapshot = format!( + "SELECT jsonb_build_object( + 'ledger', (SELECT jsonb_agg(to_jsonb(m) ORDER BY version) + FROM {schema}._duroxide_migrations m), + 'function', pg_get_functiondef('{signature}'::regprocedure), + 'queue', (SELECT jsonb_agg(to_jsonb(q) ORDER BY id) FROM {schema}.worker_queue q), + 'index', (SELECT to_jsonb(i) FROM pg_catalog.pg_index i + WHERE indexrelid = '{schema}.idx_worker_ready_order'::regclass) + )" + ); + let before: serde_json::Value = sqlx::query_scalar(&snapshot).fetch_one(pool).await.unwrap(); + let error = match PostgresProvider::new_with_schema(&database_url, Some(&schema)).await { + Ok(_) => panic!("migration must reject an invalid prebuilt index"), + Err(error) => format!("{error:#}"), + }; + assert!( + error.contains("idx_worker_ready_order") && error.contains("invalid"), + "expected an invalid-index error, got: {error}" + ); + let after: serde_json::Value = sqlx::query_scalar(&snapshot).fetch_one(pool).await.unwrap(); + assert_eq!(after, before, "rejected startup must not change durable or catalog state"); + + sqlx::query(&format!("DROP INDEX CONCURRENTLY {schema}.idx_worker_ready_order")) + .execute(&mut *builder) + .await + .unwrap(); + sqlx::query(&create_index).execute(&mut *builder).await.unwrap(); + let prebuilt: (i64, bool) = sqlx::query_as(&index_state).fetch_one(pool).await.unwrap(); + assert!(prebuilt.1, "the completed concurrent build must be valid"); + let _upgraded = PostgresProvider::new_with_schema(&database_url, Some(&schema)) + .await + .expect("migration must accept the valid prebuilt index"); + let reused: (i64, bool) = sqlx::query_as(&index_state).fetch_one(pool).await.unwrap(); + assert_eq!(reused, prebuilt, "migration must reuse the prebuilt index"); + let after: serde_json::Value = sqlx::query_scalar(&snapshot).fetch_one(pool).await.unwrap(); + assert_eq!(after["queue"], before["queue"]); + assert_eq!(after["function"], current_function); + let migration_name: String = sqlx::query_scalar(&format!( + "SELECT name FROM {schema}._duroxide_migrations WHERE version = 29" + )) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(migration_name, "0029_order_worker_queue_by_visibility.sql"); + }) + .await; +}