diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index eb3143f..8b8547f 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -13,7 +13,7 @@ This is a **PostgreSQL provider** for [Duroxide](https://github.com/microsoft/du ### Key Design Decisions 1. **Stored Procedures over inline SQL** - All database operations use schema-qualified stored procedures in `0002_create_stored_procedures.sql` for atomic transactions and deadlock prevention -2. **Rust-provided timestamps** - All procedures receive `p_now_ms` from Rust (not database `NOW()`) to avoid clock skew (see migration `0006`) +2. **Rust-provided timestamps** - All procedures receive `p_now_ms` from Rust (not database `NOW()`) to avoid clock skew (see migration `0006`). Exception: `fetch_orchestration_item` computes the lease as `GREATEST(p_now_ms + time spent in the call, clock_timestamp()) + timeout`, so a lease is never already expired when it is returned (migration `0025`) 3. **Schema isolation** - Multi-tenant support via custom PostgreSQL schemas (`new_with_schema()`) 4. **Two-phase locking in fetch operations** - Prevents B-tree index deadlocks (peek query → advisory lock → FOR UPDATE verification) diff --git a/CHANGELOG.md b/CHANGELOG.md index fa9ea29..588e08b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **`fetch_orchestration_item` no longer returns an already-expired lock lease + (migration `0025`, #21, microsoft/duroxide#46).** The lease was + `p_now_ms + lock timeout`, with `p_now_ms` read by the caller before the call, + before any connection-pool wait and before the advisory-lock wait inside the + call. When those waits were longer than the lock timeout, every ack, including + the `max_attempts` poison ack, failed with `Invalid lock token` and the message + was retried forever. The lease now starts when the instance is claimed: + `GREATEST(p_now_ms + time spent in the call, clock_timestamp()) + lock timeout`. + Nothing else in the procedure changes. Regression tests: + `tests/fetch_lease_regression_tests.rs`. + +### Changed + +- **Bump `duroxide` dependency** — `0.1.30` → `0.1.32` (0.1.31 was reserved + and not published). The core 0.1.32 release fixes queue, signal and + continue-as-new races, returns KV keys in key order, and makes + `Runtime::shutdown` stop every task the runtime spawned. No provider trait + changes. +- Update the `pg-stress` companion crate to use `duroxide` 0.1.32 as well. +- Run the new upstream provider validations on PostgreSQL: race replay, + continue-as-new transition delivery, duplicate starts, hidden-start attempt + preservation and delayed-abandon locked rows. + ## [0.1.35] - 2026-08-24 ### Performance diff --git a/Cargo.toml b/Cargo.toml index f81c33d..f9503e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,7 @@ exclude = [ ] [dependencies] -duroxide = { version = "0.1.30", features = ["provider-test"] } +duroxide = { version = "0.1.32", features = ["provider-test"] } async-trait = "0.1" tokio = { version = "1", features = ["full"] } sqlx = { version = "0.8", features = ["runtime-tokio-native-tls", "postgres", "chrono"], default-features = false } diff --git a/migrations/0025_diff.md b/migrations/0025_diff.md new file mode 100644 index 0000000..c657c5d --- /dev/null +++ b/migrations/0025_diff.md @@ -0,0 +1,233 @@ +# Diff for migration 0025 + +**Migration file:** `0025_fetch_orchestration_lease_at_claim.sql` + +Each changed function is shown **in full** with `+`/`-` markers on changed lines. + +## Table Changes + +None. + +## New Indexes + +None. + +## Function Changes + +### `fetch_orchestration_item` — body modified (baseline: 0019) + +Full function with diff: +```diff + CREATE OR REPLACE FUNCTION SCHEMA.fetch_orchestration_item(p_now_ms bigint, p_lock_timeout_ms bigint, p_min_version_packed bigint DEFAULT NULL::bigint, p_max_version_packed bigint DEFAULT NULL::bigint) + RETURNS TABLE(out_instance_id text, out_orchestration_name text, out_orchestration_version text, out_execution_id bigint, out_history jsonb, out_messages jsonb, out_lock_token text, out_attempt_count integer, out_kv_snapshot jsonb) + LANGUAGE plpgsql + AS $function$ + DECLARE + v_instance_id TEXT; + v_lock_token TEXT; + v_locked_until BIGINT; ++ v_lease_base_ms BIGINT; + v_orchestration_name TEXT; + v_orchestration_version TEXT; + v_current_execution_id BIGINT; + v_history JSONB; + v_messages JSONB; + v_kv_snapshot JSONB; + v_lock_acquired INTEGER; + v_max_attempt_count INTEGER; + BEGIN + LOOP + v_instance_id := NULL; + + -- Phase 1: Find a candidate instance (no FOR UPDATE yet) + -- When version filter is provided, join to instances+executions to filter + IF p_min_version_packed IS NOT NULL THEN + SELECT q.instance_id INTO v_instance_id + FROM SCHEMA.orchestrator_queue q + LEFT JOIN SCHEMA.instances i ON i.instance_id = q.instance_id + LEFT JOIN SCHEMA.executions e ON e.instance_id = i.instance_id + AND e.execution_id = i.current_execution_id + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM SCHEMA.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + AND ( + e.duroxide_version_major IS NULL + OR (e.duroxide_version_major * 1000000 + e.duroxide_version_minor * 1000 + e.duroxide_version_patch) + BETWEEN p_min_version_packed AND p_max_version_packed + ) + ORDER BY q.visible_at, q.id + LIMIT 1; + ELSE + SELECT q.instance_id INTO v_instance_id + FROM SCHEMA.orchestrator_queue q + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM SCHEMA.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + ORDER BY q.visible_at, q.id + LIMIT 1; + END IF; + + IF NOT FOUND THEN + RETURN; + END IF; + + -- Phase 2: Acquire instance-level advisory lock + PERFORM pg_advisory_xact_lock(hashtext(v_instance_id)); + + -- Phase 3: Re-verify the instance is still available with FOR UPDATE. + -- If contention invalidated the candidate, keep searching. + IF p_min_version_packed IS NOT NULL THEN + SELECT q.instance_id INTO v_instance_id + FROM SCHEMA.orchestrator_queue q + LEFT JOIN SCHEMA.instances i ON i.instance_id = q.instance_id + LEFT JOIN SCHEMA.executions e ON e.instance_id = i.instance_id + AND e.execution_id = i.current_execution_id + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM SCHEMA.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + AND ( + e.duroxide_version_major IS NULL + OR (e.duroxide_version_major * 1000000 + e.duroxide_version_minor * 1000 + e.duroxide_version_patch) + BETWEEN p_min_version_packed AND p_max_version_packed + ) + FOR UPDATE OF q SKIP LOCKED; + ELSE + SELECT q.instance_id INTO v_instance_id + FROM SCHEMA.orchestrator_queue q + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM SCHEMA.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + FOR UPDATE OF q SKIP LOCKED; + END IF; + + IF NOT FOUND THEN + CONTINUE; + END IF; + +- -- Step 2: Generate lock token and acquire instance lock ++ -- Step 2: Generate lock token and acquire instance lock. ++ -- The lease starts now: the later of the caller clock advanced by the ++ -- time spent in this call and the database clock. p_now_ms was read ++ -- before the call and any wait, so it alone can give an expired lease. ++ v_lease_base_ms := GREATEST( ++ p_now_ms + (EXTRACT(EPOCH FROM (clock_timestamp() - statement_timestamp())) * 1000)::BIGINT, ++ (EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::BIGINT ++ ); + v_lock_token := 'lock_' || gen_random_uuid()::TEXT; +- v_locked_until := p_now_ms + p_lock_timeout_ms; ++ v_locked_until := v_lease_base_ms + p_lock_timeout_ms; + + INSERT INTO SCHEMA.instance_locks (instance_id, lock_token, locked_until, locked_at) + VALUES (v_instance_id, v_lock_token, v_locked_until, p_now_ms) + ON CONFLICT(instance_id) DO UPDATE + SET lock_token = EXCLUDED.lock_token, + locked_until = EXCLUDED.locked_until, + locked_at = EXCLUDED.locked_at + WHERE SCHEMA.instance_locks.locked_until <= p_now_ms; + + GET DIAGNOSTICS v_lock_acquired = ROW_COUNT; + + IF v_lock_acquired = 0 THEN + CONTINUE; + END IF; + + -- Step 3: Mark all visible messages with our lock and increment attempt_count + UPDATE SCHEMA.orchestrator_queue q + SET lock_token = v_lock_token, + locked_until = v_locked_until, + attempt_count = q.attempt_count + 1 + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms); + + -- Step 4: Fetch all locked messages and get max attempt_count + SELECT COALESCE(JSONB_AGG(q.work_item::JSONB ORDER BY q.id), '[]'::JSONB), + COALESCE(MAX(q.attempt_count), 1) + INTO v_messages, v_max_attempt_count + FROM SCHEMA.orchestrator_queue q + WHERE q.lock_token = v_lock_token; + + -- Step 5: Load instance metadata + SELECT i.orchestration_name, i.orchestration_version, i.current_execution_id + INTO v_orchestration_name, v_orchestration_version, v_current_execution_id + FROM SCHEMA.instances i + WHERE i.instance_id = v_instance_id; + + -- Step 6: Load history or implement fallback + IF FOUND THEN + SELECT COALESCE(JSONB_AGG(h.event_data::JSONB ORDER BY h.event_id), '[]'::JSONB) + INTO v_history + FROM SCHEMA.history h + WHERE h.instance_id = v_instance_id AND h.execution_id = v_current_execution_id; + + v_orchestration_version := COALESCE(v_orchestration_version, 'unknown'); + ELSE + SELECT COALESCE(JSONB_AGG(h.event_data::JSONB ORDER BY h.execution_id, h.event_id), '[]'::JSONB) + INTO v_history + FROM SCHEMA.history h + WHERE h.instance_id = v_instance_id; + + IF JSONB_ARRAY_LENGTH(v_history) > 0 AND v_history->0 ? 'OrchestrationStarted' THEN + v_orchestration_name := v_history->0->'OrchestrationStarted'->>'name'; + v_orchestration_version := v_history->0->'OrchestrationStarted'->>'version'; + v_current_execution_id := 1; + ELSIF JSONB_ARRAY_LENGTH(v_messages) > 0 AND v_messages->0 ? 'StartOrchestration' THEN + v_orchestration_name := v_messages->0->'StartOrchestration'->>'orchestration'; + v_orchestration_version := COALESCE(v_messages->0->'StartOrchestration'->>'version', 'unknown'); + v_current_execution_id := COALESCE((v_messages->0->'StartOrchestration'->>'execution_id')::BIGINT, 1); + ELSIF JSONB_ARRAY_LENGTH(v_messages) > 0 AND v_messages->0 ? 'ContinueAsNew' THEN + v_orchestration_name := v_messages->0->'ContinueAsNew'->>'orchestration'; + v_orchestration_version := COALESCE(v_messages->0->'ContinueAsNew'->>'version', 'unknown'); + v_current_execution_id := 1; + ELSE + v_orchestration_name := 'Unknown'; + v_orchestration_version := 'unknown'; + v_current_execution_id := 1; + END IF; + END IF; + + -- Load KV snapshot for this instance + SELECT COALESCE( + jsonb_object_agg( + ks.key, + jsonb_build_object('value', ks.value, 'last_updated_at_ms', ks.last_updated_at_ms) + ), + '{}'::jsonb + ) + INTO v_kv_snapshot + FROM SCHEMA.kv_store ks + WHERE ks.instance_id = v_instance_id; + + RETURN QUERY SELECT + v_instance_id, + v_orchestration_name, + v_orchestration_version, + v_current_execution_id, + v_history, + v_messages, + v_lock_token, + v_max_attempt_count, + v_kv_snapshot; + RETURN; + END LOOP; + END; + $function$ + +``` + +## Notes + +- Generated with `scripts/generate_migration_diff.sh 25` against PostgreSQL 15.19. +- Only the lease changes. Signature and result shape are unchanged, so the + migration uses `CREATE OR REPLACE`; the function keeps its OID and `EXECUTE` + grants. diff --git a/migrations/0025_fetch_orchestration_lease_at_claim.sql b/migrations/0025_fetch_orchestration_lease_at_claim.sql new file mode 100644 index 0000000..44e4f18 --- /dev/null +++ b/migrations/0025_fetch_orchestration_lease_at_claim.sql @@ -0,0 +1,249 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the MIT License. + +-- Migration 0025: never hand out an expired orchestration lock lease +-- +-- 0022 and 0024 are used by open pull requests; the runner applies pending +-- migrations by version, so the gap is harmless. +-- +-- fetch_orchestration_item (baseline: 0019) set the lease to +-- p_now_ms + p_lock_timeout_ms. The caller reads p_now_ms before the call, +-- before any connection-pool wait and before the advisory-lock wait inside the +-- call. When the waits were longer than the lock timeout, the item was returned +-- with a lease that had already expired, so every ack, including the +-- max_attempts poison ack, failed with 'Invalid lock token' and the message was +-- retried forever (microsoft/duroxide-pg#21, microsoft/duroxide#46). +-- +-- The lease now starts when the instance is claimed: +-- +-- GREATEST(p_now_ms + time spent in this call, database clock_timestamp()) +-- + p_lock_timeout_ms +-- +-- The first term covers waits inside the call; the second covers a p_now_ms +-- that was already stale on arrival. GREATEST can only make the lease longer. +-- Nothing else changes: the same signature, result shape, locking, candidate +-- selection and predicates as 0019. CREATE OR REPLACE keeps the function OID +-- and EXECUTE grants. + +DO $$ +DECLARE + v_schema_name TEXT := current_schema(); +BEGIN + EXECUTE format($fmt$ + CREATE OR REPLACE FUNCTION %1$I.fetch_orchestration_item( + p_now_ms BIGINT, + p_lock_timeout_ms BIGINT, + p_min_version_packed BIGINT DEFAULT NULL, + p_max_version_packed BIGINT DEFAULT NULL + ) + RETURNS TABLE( + out_instance_id TEXT, + out_orchestration_name TEXT, + out_orchestration_version TEXT, + out_execution_id BIGINT, + out_history JSONB, + out_messages JSONB, + out_lock_token TEXT, + out_attempt_count INTEGER, + out_kv_snapshot JSONB + ) AS $fetch_orch$ + DECLARE + v_instance_id TEXT; + v_lock_token TEXT; + v_locked_until BIGINT; + v_lease_base_ms BIGINT; + v_orchestration_name TEXT; + v_orchestration_version TEXT; + v_current_execution_id BIGINT; + v_history JSONB; + v_messages JSONB; + v_kv_snapshot JSONB; + v_lock_acquired INTEGER; + v_max_attempt_count INTEGER; + BEGIN + LOOP + v_instance_id := NULL; + + -- Phase 1: Find a candidate instance (no FOR UPDATE yet) + -- When version filter is provided, join to instances+executions to filter + IF p_min_version_packed IS NOT NULL THEN + SELECT q.instance_id INTO v_instance_id + FROM %1$I.orchestrator_queue q + LEFT JOIN %1$I.instances i ON i.instance_id = q.instance_id + LEFT JOIN %1$I.executions e ON e.instance_id = i.instance_id + AND e.execution_id = i.current_execution_id + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + AND ( + e.duroxide_version_major IS NULL + OR (e.duroxide_version_major * 1000000 + e.duroxide_version_minor * 1000 + e.duroxide_version_patch) + BETWEEN p_min_version_packed AND p_max_version_packed + ) + ORDER BY q.visible_at, q.id + LIMIT 1; + ELSE + SELECT q.instance_id INTO v_instance_id + FROM %1$I.orchestrator_queue q + WHERE q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + ORDER BY q.visible_at, q.id + LIMIT 1; + END IF; + + IF NOT FOUND THEN + RETURN; + END IF; + + -- Phase 2: Acquire instance-level advisory lock + PERFORM pg_advisory_xact_lock(hashtext(v_instance_id)); + + -- Phase 3: Re-verify the instance is still available with FOR UPDATE. + -- If contention invalidated the candidate, keep searching. + IF p_min_version_packed IS NOT NULL THEN + SELECT q.instance_id INTO v_instance_id + FROM %1$I.orchestrator_queue q + LEFT JOIN %1$I.instances i ON i.instance_id = q.instance_id + LEFT JOIN %1$I.executions e ON e.instance_id = i.instance_id + AND e.execution_id = i.current_execution_id + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + AND ( + e.duroxide_version_major IS NULL + OR (e.duroxide_version_major * 1000000 + e.duroxide_version_minor * 1000 + e.duroxide_version_patch) + BETWEEN p_min_version_packed AND p_max_version_packed + ) + FOR UPDATE OF q SKIP LOCKED; + ELSE + SELECT q.instance_id INTO v_instance_id + FROM %1$I.orchestrator_queue q + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND NOT EXISTS ( + SELECT 1 FROM %1$I.instance_locks il + WHERE il.instance_id = q.instance_id AND il.locked_until > p_now_ms + ) + FOR UPDATE OF q SKIP LOCKED; + END IF; + + IF NOT FOUND THEN + CONTINUE; + END IF; + + -- Step 2: Generate lock token and acquire instance lock. + -- The lease starts now: the later of the caller clock advanced by the + -- time spent in this call and the database clock. p_now_ms was read + -- before the call and any wait, so it alone can give an expired lease. + v_lease_base_ms := GREATEST( + p_now_ms + (EXTRACT(EPOCH FROM (clock_timestamp() - statement_timestamp())) * 1000)::BIGINT, + (EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::BIGINT + ); + v_lock_token := 'lock_' || gen_random_uuid()::TEXT; + v_locked_until := v_lease_base_ms + p_lock_timeout_ms; + + INSERT INTO %1$I.instance_locks (instance_id, lock_token, locked_until, locked_at) + VALUES (v_instance_id, v_lock_token, v_locked_until, p_now_ms) + ON CONFLICT(instance_id) DO UPDATE + SET lock_token = EXCLUDED.lock_token, + locked_until = EXCLUDED.locked_until, + locked_at = EXCLUDED.locked_at + WHERE %1$I.instance_locks.locked_until <= p_now_ms; + + GET DIAGNOSTICS v_lock_acquired = ROW_COUNT; + + IF v_lock_acquired = 0 THEN + CONTINUE; + END IF; + + -- Step 3: Mark all visible messages with our lock and increment attempt_count + UPDATE %1$I.orchestrator_queue q + SET lock_token = v_lock_token, + locked_until = v_locked_until, + attempt_count = q.attempt_count + 1 + WHERE q.instance_id = v_instance_id + AND q.visible_at <= TO_TIMESTAMP(p_now_ms / 1000.0) + AND (q.lock_token IS NULL OR q.locked_until <= p_now_ms); + + -- Step 4: Fetch all locked messages and get max attempt_count + SELECT COALESCE(JSONB_AGG(q.work_item::JSONB ORDER BY q.id), '[]'::JSONB), + COALESCE(MAX(q.attempt_count), 1) + INTO v_messages, v_max_attempt_count + FROM %1$I.orchestrator_queue q + WHERE q.lock_token = v_lock_token; + + -- Step 5: Load instance metadata + SELECT i.orchestration_name, i.orchestration_version, i.current_execution_id + INTO v_orchestration_name, v_orchestration_version, v_current_execution_id + FROM %1$I.instances i + WHERE i.instance_id = v_instance_id; + + -- Step 6: Load history or implement fallback + IF FOUND THEN + SELECT COALESCE(JSONB_AGG(h.event_data::JSONB ORDER BY h.event_id), '[]'::JSONB) + INTO v_history + FROM %1$I.history h + WHERE h.instance_id = v_instance_id AND h.execution_id = v_current_execution_id; + + v_orchestration_version := COALESCE(v_orchestration_version, 'unknown'); + ELSE + SELECT COALESCE(JSONB_AGG(h.event_data::JSONB ORDER BY h.execution_id, h.event_id), '[]'::JSONB) + INTO v_history + FROM %1$I.history h + WHERE h.instance_id = v_instance_id; + + IF JSONB_ARRAY_LENGTH(v_history) > 0 AND v_history->0 ? 'OrchestrationStarted' THEN + v_orchestration_name := v_history->0->'OrchestrationStarted'->>'name'; + v_orchestration_version := v_history->0->'OrchestrationStarted'->>'version'; + v_current_execution_id := 1; + ELSIF JSONB_ARRAY_LENGTH(v_messages) > 0 AND v_messages->0 ? 'StartOrchestration' THEN + v_orchestration_name := v_messages->0->'StartOrchestration'->>'orchestration'; + v_orchestration_version := COALESCE(v_messages->0->'StartOrchestration'->>'version', 'unknown'); + v_current_execution_id := COALESCE((v_messages->0->'StartOrchestration'->>'execution_id')::BIGINT, 1); + ELSIF JSONB_ARRAY_LENGTH(v_messages) > 0 AND v_messages->0 ? 'ContinueAsNew' THEN + v_orchestration_name := v_messages->0->'ContinueAsNew'->>'orchestration'; + v_orchestration_version := COALESCE(v_messages->0->'ContinueAsNew'->>'version', 'unknown'); + v_current_execution_id := 1; + ELSE + v_orchestration_name := 'Unknown'; + v_orchestration_version := 'unknown'; + v_current_execution_id := 1; + END IF; + END IF; + + -- Load KV snapshot for this instance + SELECT COALESCE( + jsonb_object_agg( + ks.key, + jsonb_build_object('value', ks.value, 'last_updated_at_ms', ks.last_updated_at_ms) + ), + '{}'::jsonb + ) + INTO v_kv_snapshot + FROM %1$I.kv_store ks + WHERE ks.instance_id = v_instance_id; + + RETURN QUERY SELECT + v_instance_id, + v_orchestration_name, + v_orchestration_version, + v_current_execution_id, + v_history, + v_messages, + v_lock_token, + v_max_attempt_count, + v_kv_snapshot; + RETURN; + END LOOP; + END; + $fetch_orch$ LANGUAGE plpgsql; +$fmt$, v_schema_name); +END $$; diff --git a/pg-stress/Cargo.toml b/pg-stress/Cargo.toml index afe2bc8..e9f5b64 100644 --- a/pg-stress/Cargo.toml +++ b/pg-stress/Cargo.toml @@ -9,7 +9,7 @@ name = "pg-stress" path = "src/bin/pg-stress.rs" [dependencies] -duroxide = { version = "0.1.30", features = ["provider-test"] } +duroxide = { version = "0.1.32", features = ["provider-test"] } duroxide-pg = { path = ".." } tokio = { version = "1", features = ["full"] } tracing = "0.1" diff --git a/tests/fetch_lease_regression_tests.rs b/tests/fetch_lease_regression_tests.rs new file mode 100644 index 0000000..0f7a1f2 --- /dev/null +++ b/tests/fetch_lease_regression_tests.rs @@ -0,0 +1,454 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Regression tests: `fetch_orchestration_item` must never hand out an +//! orchestration lock lease that is already expired. +//! +//! References: and +//! . +//! +//! Before migration 0025 the procedure wrote +//! `locked_until = p_now_ms + p_lock_timeout_ms`, where `p_now_ms` was taken by +//! the caller before the call and before any wait. A wait longer than the lock +//! timeout produced a lease that had expired before the item was returned, so +//! every ack, including the poison ack that `max_attempts` relies on, failed +//! with `Invalid lock token` and the message was retried forever. + +use duroxide::providers::{ExecutionMetadata, Provider, ProviderError, WorkItem}; +use duroxide::{ErrorDetails, Event, EventKind, PoisonMessageType}; +use duroxide_pg::PostgresProvider; +use sqlx::postgres::PgPoolOptions; +use sqlx::PgPool; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +const ORCHESTRATION: &str = "FetchLeaseRegression"; +const VERSION: &str = "1.0.0"; +/// The runtime default for `RuntimeOptions::max_attempts`. +const MAX_ATTEMPTS: u32 = 10; +/// Lock timeout used by the contention tests. +const LOCK_TIMEOUT: Duration = Duration::from_secs(2); +/// How long a competing session holds the instance advisory lock. Longer than +/// `LOCK_TIMEOUT`, so a fetch that waits it out used to return an expired lease. +const HOLD: Duration = Duration::from_secs(4); + +fn get_database_url() -> String { + dotenvy::dotenv().ok(); + std::env::var("DATABASE_URL").expect("DATABASE_URL must be set") +} + +fn unique_schema_name() -> String { + let guid = uuid::Uuid::new_v4().to_string(); + let suffix = &guid[guid.len() - 8..]; + format!("fetch_lease_test_{suffix}") +} + +/// Instance ids are unique per test run: advisory-lock keys are database-wide, +/// so a key held by one test must never collide with another test. +fn unique_instance(prefix: &str) -> String { + let guid = uuid::Uuid::new_v4().to_string(); + format!("{prefix}-{}", &guid[..8]) +} + +fn rust_now_ms() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system clock before epoch") + .as_millis() as i64 +} + +async fn raw_pool(database_url: &str) -> PgPool { + PgPoolOptions::new() + .max_connections(4) + .connect(database_url) + .await + .expect("Failed to connect raw pool") +} + +async fn drop_schema(pool: &PgPool, schema: &str) { + sqlx::query(&format!("DROP SCHEMA IF EXISTS {schema} CASCADE")) + .execute(pool) + .await + .expect("Failed to drop schema"); +} + +async fn enqueue_start(provider: &PostgresProvider, instance: &str) { + provider + .enqueue_for_orchestrator( + WorkItem::StartOrchestration { + instance: instance.to_string(), + orchestration: ORCHESTRATION.to_string(), + input: "{}".to_string(), + version: Some(VERSION.to_string()), + parent_instance: None, + parent_id: None, + parent_execution_id: None, + execution_id: 1, + }, + None, + ) + .await + .unwrap_or_else(|e| panic!("failed to enqueue {instance}: {e:?}")); +} + +/// Remaining lease time in milliseconds for `lock_token`, measured against the +/// database clock and against this process's clock. Negative means expired. +async fn lease_remaining_ms(pool: &PgPool, schema: &str, lock_token: &str) -> (i64, i64) { + let (locked_until, db_now_ms): (i64, i64) = sqlx::query_as(&format!( + "SELECT locked_until, (EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::BIGINT + FROM {schema}.instance_locks + WHERE lock_token = $1" + )) + .bind(lock_token) + .fetch_one(pool) + .await + .expect("an instance_locks row must exist for the returned lock token"); + (locked_until - db_now_ms, locked_until - rust_now_ms()) +} + +/// Holds the advisory locks that a fetch of each instance takes, in one +/// transaction for `hold`. +fn hold_instance_locks( + pool: PgPool, + instances: Vec, + hold: Duration, +) -> ( + tokio::task::JoinHandle<()>, + tokio::sync::oneshot::Receiver<()>, +) { + let (taken_tx, taken_rx) = tokio::sync::oneshot::channel(); + let holder = tokio::spawn(async move { + let mut tx = pool.begin().await.expect("holder begin failed"); + for instance in &instances { + sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1))") + .bind(instance) + .execute(&mut *tx) + .await + .expect("holder advisory lock failed"); + } + let _ = taken_tx.send(()); + tokio::time::sleep(hold).await; + tx.rollback().await.expect("holder rollback failed"); + }); + (holder, taken_rx) +} + +fn started_event(instance: &str) -> Event { + Event::with_event_id( + 1, + instance, + 1, + None, + EventKind::OrchestrationStarted { + name: ORCHESTRATION.to_string(), + version: VERSION.to_string(), + input: "{}".to_string(), + parent_instance: None, + parent_id: None, + parent_execution_id: None, + carry_forward_events: None, + initial_custom_status: None, + }, + ) +} + +async fn ack_started( + provider: &PostgresProvider, + instance: &str, + lock_token: &str, +) -> Result<(), ProviderError> { + provider + .ack_orchestration_item( + lock_token, + 1, + vec![started_event(instance)], + vec![], + vec![], + ExecutionMetadata { + orchestration_name: Some(ORCHESTRATION.to_string()), + orchestration_version: Some(VERSION.to_string()), + ..Default::default() + }, + vec![], + ) + .await +} + +/// The ack the runtime commits when a message exceeds `max_attempts` +/// (`fail_orchestration_as_poison` in duroxide 0.1.30): the execution is +/// marked Failed with `ErrorDetails::Poison`, keyed on the fetched lock token. +async fn poison_ack( + provider: &PostgresProvider, + instance: &str, + lock_token: &str, + attempt_count: u32, +) -> Result<(), ProviderError> { + let details = ErrorDetails::Poison { + attempt_count, + max_attempts: MAX_ATTEMPTS, + message_type: PoisonMessageType::Orchestration { + instance: instance.to_string(), + execution_id: 1, + }, + message: "[]".to_string(), + }; + let failed = Event::with_event_id( + 2, + instance, + 1, + None, + EventKind::OrchestrationFailed { + details: details.clone(), + }, + ); + provider + .ack_orchestration_item( + lock_token, + 1, + vec![started_event(instance), failed], + vec![], + vec![], + ExecutionMetadata { + status: Some("Failed".to_string()), + output: Some(details.display_message()), + orchestration_name: Some(ORCHESTRATION.to_string()), + orchestration_version: Some(VERSION.to_string()), + ..Default::default() + }, + vec![], + ) + .await +} + +async fn queued_messages(pool: &PgPool, schema: &str, instance: &str) -> i64 { + sqlx::query_scalar(&format!( + "SELECT COUNT(*) FROM {schema}.orchestrator_queue WHERE instance_id = $1" + )) + .bind(instance) + .fetch_one(pool) + .await + .expect("queue count failed") +} + +/// Issue #46 reproduction A: a caller timestamp that is already 10 s stale +/// when the procedure runs (what a 10 s pool or advisory-lock wait produces). +/// The lease must still be live when it is returned, and the ack must commit. +/// Both procedure branches (with and without the version filter) are covered. +#[tokio::test] +async fn stale_caller_timestamp_never_yields_an_expired_lease() { + let schema = unique_schema_name(); + let database_url = get_database_url(); + let provider = PostgresProvider::new_with_schema(&database_url, Some(&schema)) + .await + .expect("Failed to create provider"); + let pool = raw_pool(&database_url).await; + + const STALE_MS: i64 = 10_000; + const TIMEOUT_MS: i64 = 5_000; + + for version_filter in [None, Some((0_i64, 999_999_999_i64))] { + let instance = unique_instance("stale"); + enqueue_start(&provider, &instance).await; + // The message must be visible to the stale clock as well; otherwise the + // procedure finds no candidate and the defect is not exercised. + sqlx::query(&format!( + "UPDATE {schema}.orchestrator_queue + SET visible_at = visible_at - INTERVAL '60 seconds' + WHERE instance_id = $1" + )) + .bind(&instance) + .execute(&pool) + .await + .expect("backdate failed"); + + let (min_packed, max_packed) = match version_filter { + Some((min, max)) => (Some(min), Some(max)), + None => (None, None), + }; + let row: Option<(String, String, i32)> = sqlx::query_as(&format!( + "SELECT out_instance_id, out_lock_token, out_attempt_count + FROM {schema}.fetch_orchestration_item( + (EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::BIGINT - $1, + $2, $3, $4)" + )) + .bind(STALE_MS) + .bind(TIMEOUT_MS) + .bind(min_packed) + .bind(max_packed) + .fetch_optional(&pool) + .await + .expect("fetch_orchestration_item failed"); + let (fetched, lock_token, attempt_count) = + row.expect("the backdated message must be fetched"); + assert_eq!(fetched, instance); + assert_eq!(attempt_count, 1); + + let (db_remaining, rust_remaining) = lease_remaining_ms(&pool, &schema, &lock_token).await; + println!( + "filter={version_filter:?}: lease remaining db={db_remaining}ms rust={rust_remaining}ms" + ); + assert!( + db_remaining > TIMEOUT_MS - 2_000, + "lease returned with {db_remaining}ms left by the database clock \ + (lock timeout {TIMEOUT_MS}ms); it was computed from the stale caller timestamp" + ); + assert!( + rust_remaining > 0, + "lease returned already expired by {}ms by the caller clock", + -rust_remaining + ); + + let queue_lease: i64 = sqlx::query_scalar(&format!( + "SELECT MIN(locked_until) FROM {schema}.orchestrator_queue WHERE lock_token = $1" + )) + .bind(&lock_token) + .fetch_one(&pool) + .await + .expect("queue lease read failed"); + let lock_lease: i64 = sqlx::query_scalar(&format!( + "SELECT locked_until FROM {schema}.instance_locks WHERE lock_token = $1" + )) + .bind(&lock_token) + .fetch_one(&pool) + .await + .expect("instance lease read failed"); + assert_eq!( + queue_lease, lock_lease, + "queue rows and instance lock carry the same lease" + ); + + ack_started(&provider, &instance, &lock_token) + .await + .unwrap_or_else(|e| panic!("ack with the returned lease failed: {e}")); + assert_eq!(queued_messages(&pool, &schema, &instance).await, 0); + } + + drop_schema(&pool, &schema).await; +} + +/// Issue #46 reproduction B, made deterministic: one hot instance whose +/// message has reached `max_attempts`, a session that holds the instance for +/// longer than the lock timeout, and 16 concurrent dispatchers polling it. +/// The dispatchers still wait for the holder, but every lease handed out must +/// be live, and the poison ack must commit so the message leaves the queue. +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn hot_instance_contention_never_yields_an_expired_lease_and_poison_commits() { + const DISPATCHERS: usize = 16; + + let schema = unique_schema_name(); + let database_url = get_database_url(); + let provider = Arc::new( + PostgresProvider::new_with_schema(&database_url, Some(&schema)) + .await + .expect("Failed to create provider"), + ); + let pool = raw_pool(&database_url).await; + + let hot = unique_instance("hot"); + enqueue_start(&provider, &hot).await; + // The message has already been attempted max_attempts times; the next + // fetch returns attempt_count = max_attempts + 1 and must be poisoned. + sqlx::query(&format!( + "UPDATE {schema}.orchestrator_queue SET attempt_count = $2 WHERE instance_id = $1" + )) + .bind(&hot) + .bind(MAX_ATTEMPTS as i32) + .execute(&pool) + .await + .expect("attempt_count setup failed"); + + let (holder, taken) = hold_instance_locks(pool.clone(), vec![hot.clone()], HOLD); + taken.await.expect("holder did not take the locks"); + + let done = Arc::new(AtomicBool::new(false)); + let deadline = Instant::now() + HOLD + Duration::from_secs(6); + let mut dispatchers = Vec::new(); + for _ in 0..DISPATCHERS { + let provider = provider.clone(); + let pool = pool.clone(); + let schema = schema.clone(); + let done = done.clone(); + dispatchers.push(tokio::spawn(async move { + let mut slowest = Duration::ZERO; + let mut claims = Vec::new(); + while !done.load(Ordering::SeqCst) && Instant::now() < deadline { + let started = Instant::now(); + let fetched = provider + .fetch_orchestration_item(LOCK_TIMEOUT, Duration::ZERO, None) + .await + .expect("fetch_orchestration_item failed"); + slowest = slowest.max(started.elapsed()); + match fetched { + Some((item, lock_token, attempt_count)) => { + let remaining = lease_remaining_ms(&pool, &schema, &lock_token).await; + let acked = if attempt_count > MAX_ATTEMPTS { + poison_ack(&provider, &item.instance, &lock_token, attempt_count).await + } else { + ack_started(&provider, &item.instance, &lock_token).await + }; + if acked.is_ok() { + done.store(true, Ordering::SeqCst); + } + claims.push((item.instance, attempt_count, remaining, acked)); + } + None => tokio::time::sleep(Duration::from_millis(25)).await, + } + } + (slowest, claims) + })); + } + + let mut slowest = Duration::ZERO; + let mut claims = Vec::new(); + for dispatcher in dispatchers { + let (task_slowest, task_claims) = dispatcher.await.expect("dispatcher panicked"); + slowest = slowest.max(task_slowest); + claims.extend(task_claims); + } + holder.await.expect("holder panicked"); + + println!("slowest fetch {slowest:?}; claims:"); + for (instance, attempt_count, remaining, acked) in &claims { + println!( + " {instance} attempt={attempt_count} lease_remaining(db,rust)={remaining:?} ack={acked:?}" + ); + } + + for (instance, _, (db_remaining, rust_remaining), _) in &claims { + assert!( + *db_remaining > 0 && *rust_remaining > 0, + "{instance} was returned with an already-expired lease \ + (remaining db={db_remaining}ms rust={rust_remaining}ms)" + ); + } + assert_eq!( + claims.len(), + 1, + "exactly one dispatcher must claim the hot instance: {claims:?}" + ); + let (instance, attempt_count, _, acked) = &claims[0]; + assert_eq!(instance, &hot); + assert_eq!(*attempt_count, MAX_ATTEMPTS + 1); + if let Err(e) = acked { + panic!("the poison ack failed: {e}"); + } + + assert_eq!( + queued_messages(&pool, &schema, &hot).await, + 0, + "the poisoned message must leave the queue" + ); + let history = provider.read(&hot).await.expect("read history failed"); + assert!( + history.iter().any(|e| matches!( + &e.kind, + EventKind::OrchestrationFailed { + details: ErrorDetails::Poison { .. } + } + )), + "history must record the poison failure: {history:?}" + ); + + drop_schema(&pool, &schema).await; +} diff --git a/tests/postgres_provider_test.rs b/tests/postgres_provider_test.rs index 73be146..9bd6947 100644 --- a/tests/postgres_provider_test.rs +++ b/tests/postgres_provider_test.rs @@ -8,7 +8,7 @@ use duroxide::provider_validation::{ error_handling, instance_creation, instance_locking, kv_store, lock_expiration, management, multi_execution, prune, queue_semantics, sessions, tag_filtering, }; -use duroxide::provider_validations::ProviderFactory; +use duroxide::provider_validations::{race_replay, ProviderFactory}; use duroxide::providers::Provider; use duroxide_pg::PostgresProvider; use sqlx::{postgres::PgPoolOptions, Executor}; @@ -179,6 +179,36 @@ macro_rules! provider_validation_test { }; } +mod race_replay_tests { + use super::*; + + provider_validation_test!(race_replay::test_duplicate_start_preserves_pinned_handler); + provider_validation_test!(race_replay::test_continue_as_new_unregistered_backoff); + provider_validation_test!(race_replay::test_continue_as_new_duplicate_start); + provider_validation_test!( + race_replay::test_continue_as_new_poisoned_successor_is_own_execution + ); + provider_validation_test!(race_replay::test_queue_race_cancellation_replay); + provider_validation_test!(race_replay::test_continue_as_new_queue_race_replay); + provider_validation_test!(race_replay::test_queue_replay_version_stamp_roundtrip); + provider_validation_test!(race_replay::test_positional_wait_race_replay); + provider_validation_test!(race_replay::test_legacy_queue_race_decision_preserved); + + #[tokio::test] + async fn test_continue_as_new_transition_delivery_legacy_stamp() { + let factory = PostgresProviderFactory::new(); + race_replay::test_continue_as_new_transition_delivery(&factory, "0.1.30").await; + factory.cleanup_schema().await; + } + + #[tokio::test] + async fn test_continue_as_new_transition_delivery_new_stamp() { + let factory = PostgresProviderFactory::new(); + race_replay::test_continue_as_new_transition_delivery(&factory, "0.1.31").await; + factory.cleanup_schema().await; + } +} + mod atomicity_tests { use super::*; @@ -296,6 +326,10 @@ mod poison_message_tests { provider_validation_test!(poison_message::attempt_count_is_per_message); provider_validation_test!(poison_message::abandon_work_item_ignore_attempt_decrements); provider_validation_test!(poison_message::abandon_orchestration_item_ignore_attempt_decrements); + provider_validation_test!(poison_message::orchestration_ignore_attempt_preserves_hidden_start); + provider_validation_test!( + poison_message::orchestration_delayed_abandon_preserves_unlocked_rows + ); provider_validation_test!(poison_message::ignore_attempt_never_goes_negative); provider_validation_test!(poison_message::max_attempt_count_across_message_batch); }