Bug: concurrent receive() for same consumer can double-deliver same events

[NikolayS]

Summary

Deep raw SQL testing found that two concurrent pgque.receive(queue, same_consumer) calls can both receive the same messages, with different batch_ids.

This appears to be a race in batch acquisition: the subscription row is read/updated without a strong enough lock or conditional update, so two sessions can both observe no active batch and allocate separate batches over the same tick range.

Impact

For the same queue + consumer, concurrent workers can process the same event twice.

That violates the expected single-consumer cursor semantics for a registered consumer. Fan-out across different consumers is expected; duplicate delivery to the same consumer due to concurrent receive() is not.

Evidence

Using a local two-session repro with an artificial sleep to widen the race, both sessions received the same msg_ids with different batch_ids:

('T1', 1.03, [(1, 7, 'p1'), (2, 7, 'p2')])
('T2', 1.02, [(1, 6, 'p1'), (2, 6, 'p2')])
subscription [(7, 2, 1)]

Both sessions got msg_id 1 and 2. Final subscription state retained only one of the batch ids.

Expected

For a given (queue, consumer):

• at most one concurrent receiver should acquire the next batch
• a second concurrent receiver should either see the same active batch in a controlled/idempotent way, block, or return no new batch depending on intended semantics
• it should not allocate a distinct batch id for the same event range

Likely fix direction

In the lower-level batch acquisition path (next_batch_custom / related subscription update):

• lock the subscription row (FOR UPDATE) before checking/updating sub_batch, or
• use a conditional update like WHERE sub_batch IS NULL and verify exactly one row updated, or
• use advisory lock keyed by subscription id

The invariant is: only one session can transition a subscription from no active batch to active batch.

Test to add

Add a concurrency regression test with two sessions calling receive() for the same consumer at the same time. It should assert that the same msg_id is not delivered under two different batch ids.

Environment

Tested on main at 9b3f89f.

[NikolayS]

Investigation + fix summary

(a) Intended model: single-worker-per-consumer

Evidence (file + line + quote):

• sql/pgque.sql line 178–179: constraint subscription_pkey primary key (sub_queue, sub_consumer) + constraint subscription_batch_idx unique (sub_batch) — one cursor row per (queue, consumer) pair; at most one active batch at any time.
• pgq/functions/pgq.next_batch.sql lines 168–171: -- has already active batch / if batch_id is not null then / return; / end if; — PgQ's own guard: if a batch is already open, return it unchanged. Single-cursor design.
• blueprints/SPECx.md (architecture section): PgQ was designed for Skytools daemons where one process owns one consumer registration.

The intended model is one worker per consumer name. Fan-out uses distinct consumer names — each gets an independent cursor over all events.

(b) Fix: FOR UPDATE lock (not docs-only)

The bug was a missing row-level lock in pgque.next_batch_custom(). Two concurrent sessions could both read sub_batch = NULL before either committed, allocate distinct batch_ids from the sequence, and both UPDATE the subscription row. The second UPDATE overwrote the first (the WHERE clause matched on (sub_queue, sub_consumer), not on sub_batch IS NULL). Both sessions then called get_batch_events() for the same tick range: double-delivery.

Fix: add FOR UPDATE OF s to the SELECT in next_batch_custom(). The second session blocks on the row lock until the first commits; on unblocking it re-reads sub_batch != NULL and returns via the existing early-return guard — never allocating a new batch_id.

Also documented the single-worker-per-consumer contract in docs/reference.md and in the pgque.receive() function comment.

(c) Red commit: 13e222b

test: concurrent receive() regression for issue #97 — adds tests/test_concurrent_receive.sql with four assertions (T1–T4). A true concurrent race requires two physical sessions and cannot be reproduced deterministically in a single SQL session; these tests verify the sequential invariant and serve as regression sentinels.

PASS T1: sequential idempotency -- same batch_id on repeated receive()
PASS T2: next_batch_custom idempotent for active subscription
PASS T3: no double-delivery -- receive() reuses active batch
PASS T4: subscription cursor cleared after ack

(d) Green commit: 527f4a8

fix: FOR UPDATE in next_batch_custom prevents #97 — adds FOR UPDATE OF s to the subscription SELECT in next_batch_custom(), plus documentation in docs/reference.md and sql/pgque-api/receive.sql.

All existing tests (test_api_receive, test_api_send, test_core_consumer, test_core_events, test_core_lifecycle) pass unchanged.

(e) PR

#125

[NikolayS]

Deferred to v0.2.1+ — the PR #125 fix touches PgQ core engine (next_batch_custom) which is flagged sacred by CLAUDE.md. Maintainer is holding the change for more rigorous validation before merge.