Skip to content

Commit ce54414

Browse files
committed
update documentation
1 parent 996c28a commit ce54414

File tree

2 files changed

+10
-10
lines changed

2 files changed

+10
-10
lines changed

README.md

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -114,17 +114,16 @@ It uses row level locks of postgres to mimic the atomic pop and atomic push of r
114114

115115
```sql
116116
UPDATE task_queue
117-
SET processing = true,
118-
deadline =
119-
current_timestamp + CAST(lease_timeout || ' seconds' AS INTERVAL)
117+
SET started_at = current_timestamp
120118
WHERE id = (
121119
SELECT id
122120
FROM task_queue
123121
WHERE completed_at IS NULL
124-
AND processing = false
122+
AND started_at IS NULL
125123
AND queue_name = <your_queue_name>
126124
AND ttl > 0
127-
ORDER BY created_at
125+
AND can_start_at <= current_timestamp
126+
ORDER BY can_start_at
128127
FOR UPDATE SKIP LOCKED
129128
LIMIT 1
130129
)
@@ -137,10 +136,11 @@ Let's say two workers try to get a new task at the same time, assuming that they
137136
SELECT id
138137
FROM task_queue
139138
WHERE completed_at IS NULL
140-
AND processing = false
139+
AND started_at IS NULL
141140
AND queue_name = <your_queue_name>
142141
AND ttl > 0
143-
ORDER BY created_at
142+
AND can_start_at <= current_timestamp
143+
ORDER BY can_start_at
144144
```
145145

146146
The first worker locks the row with the `FOR UPDATE` clause until the update is completed and committed. If we hadn't used the `SKIP LOCKED` clause, the second worker would have seen the same row and waited for the first worker to finish the update. However, since the first worker already updated it, the subquery would no longer be valid, and the second worker would return zero rows because `WHERE id = NULL`.

postgrestq/task_queue.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -273,7 +273,7 @@ def get(self) -> Tuple[
273273
"""Get a task from the task queue (non-blocking).
274274
275275
This statement marks the next available task in the queue as
276-
"processing" and returns its ID and task details. The query
276+
started (being processed) and returns its ID and task details. The query
277277
uses a FOR UPDATE SKIP LOCKED clause to lock the selected
278278
task so that other workers can't select the same task simultaneously.
279279
@@ -291,7 +291,7 @@ def get(self) -> Tuple[
291291
>>> taskqueue.complete(task_id)
292292
293293
After some time (i.e. `lease_timeout`) tasks expire and are
294-
marked as not processing and the TTL is decreased by
294+
marked as not being processed and the TTL is decreased by
295295
one. If TTL is still > 0 the task will be retried.
296296
297297
Note, this method is non-blocking, i.e. it returns immediately
@@ -525,7 +525,7 @@ def get_updated_expired_task(
525525
) -> Tuple[Optional[str], Optional[int]]:
526526
"""
527527
Given the id of an expired task, it tries to reschedule it by
528-
marking it as not processing, resetting the deadline
528+
marking it as not started, resetting the deadline
529529
and decreasing TTL by one. It returns None if the task is
530530
already updated (or being updated) by another worker.
531531

0 commit comments

Comments
 (0)