Skip to content

Prevent deferred worker items from starving ready work #51

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Thom Chiovoloni (thomcc-work) wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Prevent deferred worker items from starving ready work #51

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
77010a0
prevent deferred worker items from starving ready work
thomcc-work Oct 30, 2026
9ae067b
fix: address worker queue progress review
thomcc-work Oct 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions docs/migration-policy.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
138 changes: 138 additions & 0 deletions migrations/0029_diff.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

```

---
134 changes: 134 additions & 0 deletions migrations/0029_order_worker_queue_by_visibility.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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$;
47 changes: 47 additions & 0 deletions migrations/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading