I have a table called
queue with items that need to be processed:
CREATE TABLE public.queue ( id serial NOT NULL CONSTRAINT queue_pkey PRIMARY KEY );
process represents processed items from the queue (e.g. a report has been generated). In reality, there are more such tables (there are more processes that need to be performed on an item). There is one-to-one relation between
process – each item can be processed just once.
CREATE TABLE public.process ( id serial NOT NULL CONSTRAINT process_pkey PRIMARY KEY, queue_item_id integer NOT NULL CONSTRAINT process_queue_item_id_key UNIQUE CONSTRAINT process_queue_item_id_8953ec7b_fk_datastore REFERENCES public.queue DEFERRABLE INITIALLY DEFERRED );
Here are some test data:
BEGIN; TRUNCATE TABLE queue, process RESTART IDENTITY CASCADE; INSERT INTO queue SELECT GENERATE_SERIES(1, 10000); COMMIT;
The worker that processes the items is implemented as follows. The code is in Django (Python framework), but I am convinced that my error is not caused by Django or its ORM.
(For simplicity, there is no termination condition.)
while True: with transaction.atomic(): queue_items = Queue.objects \ .filter(process=None) \ .order_by() \ .select_for_update(skip_locked=True, of=('self',))[:8] print('Generating report...') time.sleep(0.1) Process.objects.bulk_create( (Process(queue_item=q) for q in queue_items) )
Here is a transcript of the SQL queries that travel to the database:
-- while True: BEGIN; SELECT queue."id" FROM queue LEFT OUTER JOIN "process" ON (queue."id" = "process"."queue_item_id") WHERE "process"."id" IS NULL LIMIT 8 FOR UPDATE OF queue SKIP LOCKED; -- print('Generating report...') -- time.sleep(0.5) INSERT INTO "process" ("queue_item_id") VALUES (1), (2), (3), (4), (5), (6), (7), (8) RETURNING "process"."id"; COMMIT;
If I start one worker, the queue is processed perfectly fine. If I run two or more workers, I start getting this error:
duplicate key value violates unique constraint "process_queue_item_id_key" DETAIL: Key (queue_item_id)=(**) already exists.
How another transaction could create rows in
process for items in a
queue when those rows are locked?
What I tried:
- I tried to rewrite SELECT query with EXISTS:
SELECT "queue"."id" FROM "queue" WHERE NOT (EXISTS(SELECT U0."id", U0."queue_item_id" FROM "process" U0 WHERE U0."queue_item_id" = "queue"."id")) LIMIT 8 FOR UPDATE OF "queue" SKIP LOCKED
without success, the same error occurs.
- If I arrange the rows randomly, the error occurs much later (almost at the end of the queue).
SELECT "queue"."id" FROM "queue" LEFT OUTER JOIN "process" ON ("queue"."id" = "process"."queue_item_id") WHERE "process"."id" IS NULL ORDER BY RANDOM() LIMIT 8 FOR UPDATE OF "queue" SKIP LOCKED
- I put a breakpoint in the middle of the transaction and in another transaction I checked that the row locking in my opinion works correctly:
SELECT id FROM queue WHERE id NOT IN ( SELECT id FROM queue FOR UPDATE SKIP LOCKED );
- My Postgres version:
PostgreSQL 13.1, compiled by Visual C++ build 1914, 64-bit.
- Each worker has its own connection to Postgres with default isolation level (read committed).