Fix #21: start the fetch lease when the instance is claimed; run the queue-race validations

.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)
 

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

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 }

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.