diff --git a/CHANGELOG.md b/CHANGELOG.md index 917c21b..87a7ddc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,24 @@ 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 `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 `0029`, which replaces `fetch_work_item` without + changing its signature and adds an index on `worker_queue(visible_at, id)`. + 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 ### Fixed diff --git a/docs/migration-policy.md b/docs/migration-policy.md index e5716d0..bc19500 100644 --- a/docs/migration-policy.md +++ b/docs/migration-policy.md @@ -46,6 +46,11 @@ 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 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 ```rust 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/0029_order_worker_queue_by_visibility.sql b/migrations/0029_order_worker_queue_by_visibility.sql new file mode 100644 index 0000000..d967560 --- /dev/null +++ b/migrations/0029_order_worker_queue_by_visibility.sql @@ -0,0 +1,134 @@ +-- 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 + ); + + -- 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, + 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/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 new file mode 100644 index 0000000..73da7fd --- /dev/null +++ b/tests/worker_progress_tests.rs @@ -0,0 +1,222 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +use duroxide::providers::{Provider, SessionFetchConfig, TagFilter, WorkItem}; +use duroxide_pg::PostgresProvider; +use std::time::Duration; + +#[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] { + 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 + .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(); + }) + .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; +}