fix(queries): keep a picked row when a job is cancelled

docs/getting-started/core-concepts.md

@@ -104,7 +104,7 @@ can be manually re-queued via `pgq requeue <id>` or
 | `successful` | Handler completed without raising an exception |
 | `exception` | Handler raised an unhandled exception (traceback is logged) |
 | `failed` | Job held for manual review after terminal failure. Inspect with `pgq failed` and re-queue with `pgq requeue <id>` (see [Holding Failed Jobs](../guides/hold-failed-jobs.md)) |
-| `canceled` | Job was canceled via `mark_job_as_cancelled()` |
+| `canceled` | Queued job removed by `mark_job_as_cancelled()`, or a picked job stopped by its worker after a cancel request |
 | `deleted` | Job was removed before being processed |
 
 Once a job reaches a terminal state (`successful`, `exception`, `canceled`, `deleted`),

docs/guides/concurrency-control.md

@@ -35,7 +35,8 @@ async def process_shared_resource(job: Job) -> None:
 ```
 
 Setting `concurrency_limit=1` guarantees that only one job of this entrypoint runs
-at a time across all workers.
+at a time across all workers. Cancelling the running job does not free the slot
+until the worker logs the outcome.
 
 ## Global concurrency limit
 

docs/guides/job-cancellation.md

@@ -27,8 +27,11 @@ job_ids = queries.enqueue("task_entrypoint", b"Job data", priority=5)
 await queries.mark_job_as_cancelled(job_ids)
 ```
 
-`mark_job_as_cancelled` accepts a list of job IDs. Jobs that have already completed are
-unaffected.
+`mark_job_as_cancelled` accepts a list of job IDs. A job that is still `queued` is
+removed at once and logged as `canceled`. A job that is already `picked` stays in the
+queue until its worker logs the outcome, so it keeps its `concurrency_limit` slot and
+its `dedupe_key` while the handler runs. The NOTIFY still asks the handler to stop.
+Jobs that have already completed are unaffected.
 
 ## Handling cancellation in job code
 
@@ -47,5 +50,6 @@ async def process_job(job: Job, ctx: Context) -> None:
 
 ## Job status after cancellation
 
-Cancelled jobs transition to the `canceled` status and are logged in the statistics table.
-See [Architecture](../reference/architecture.md) for the full status lifecycle.
+A queued job is logged `canceled` immediately. A picked job is logged `canceled` when
+its worker sees the request and the handler returns; until that log, the row stays
+`picked`. See [Architecture](../reference/architecture.md) for the full status lifecycle.

docs/reference/architecture.md

@@ -83,7 +83,8 @@ The lifecycle of a job flows through these statuses:
   stored for later inspection.
 - **`failed`**: Job held in the queue for manual review when `on_failure="hold"` is set.
   Can be re-queued or deleted manually.
-- **`canceled`**: Jobs canceled before completion receive this status and are logged.
+- **`canceled`**: A queued job canceled with `mark_job_as_cancelled` is logged at once.
+  A picked job stays `picked` until its worker logs `canceled`.
 - **`deleted`**: Used when jobs are removed from the queue without running, such as during
   manual cleanup operations.
 

pgqueuer/adapters/inmemory/jobs.py

@@ -440,7 +440,7 @@ def owned_by(self, job: models.Job) -> bool:
 
     def finish(self, job: models.Job, status: JOB_STATUS, now: datetime) -> bool:
         job_id = int(job.id)
-        # A job cancelled while it ran already has its terminal log row.
+        # Gone (cleared) or owned by a newer picker: leave the existing log alone.
         if job_id not in self.rows or not self.owned_by(job):
             return False
         row = self.rows[job_id]
@@ -507,9 +507,11 @@ def apply_requeue(self, row: JobRow, now: datetime) -> None:
     def cancel(self, ids: list[JobId]) -> list[JobRow]:
         removed: list[JobRow] = []
         for raw_id in ids:
-            row = self.rows.pop(int(raw_id), None)
-            if row is None:
+            row = self.rows.get(int(raw_id))
+            # A picked row keeps its slot and dedupe key until the worker logs.
+            if row is None or row.status == "picked":
                 continue
+            self.rows.pop(row.id)
             self.release(row)
             row.ready_token += 1
             self.dedupe.forget(row.id)

pgqueuer/adapters/persistence/qb.py

@@ -689,8 +689,8 @@ def build_queue_size_query(self) -> str:
     """
 
     def build_log_job_query(self) -> str:
-        # The owner check sits on the target row so a concurrent re-pick is re-checked
-        # under READ COMMITTED; a NULL owner (admin cancellation) is unfenced.
+        # Owner check is re-read under READ COMMITTED. A NULL owner (admin cancel)
+        # removes queued and failed rows only; a picked row stays for its worker.
         return f"""WITH job_status AS (
             SELECT
                 UNNEST($1::bigint[])                            AS id,
@@ -702,7 +702,10 @@ def build_log_job_query(self) -> str:
             USING job_status js
             WHERE q.id = js.id
                 AND js.status != 'failed'
-                AND (js.queue_manager_id IS NULL OR q.queue_manager_id = js.queue_manager_id)
+                AND (
+                    q.queue_manager_id = js.queue_manager_id
+                    OR (js.queue_manager_id IS NULL AND q.status <> 'picked')
+                )
             RETURNING q.id, q.entrypoint, q.priority
         ), held AS (
             UPDATE {self.qualified.queue_table} q

pgqueuer/adapters/persistence/queries.py

@@ -421,7 +421,12 @@ async def clear_queue(self, entrypoint: str | list[str] | None = None) -> None:
             await self.driver.execute(self.qbq.build_truncate_queue_query())
 
     async def mark_job_as_cancelled(self, ids: list[types.JobId]) -> None:
-        """Log *ids* as 'canceled' and emit a cancellation NOTIFY."""
+        """Cancel *ids* and emit a cancellation NOTIFY.
+
+        A ``queued`` or ``failed`` row is logged ``canceled`` and removed. A
+        ``picked`` row stays until its worker logs the outcome, so the job keeps
+        its concurrency slot and dedupe key while the handler runs.
+        """
         await asyncio.gather(
             self.driver.execute(
                 self.qbq.build_log_job_query(),

test/integration/test_persistence_queries_sql.py

@@ -18,6 +18,7 @@
 
 from pgqueuer.adapters.persistence.queries import Queries
 from pgqueuer.db import AsyncpgDriver
+from pgqueuer.domain.errors import DuplicateJobError
 from pgqueuer.domain.models import CronExpressionEntrypoint, TracebackRecord
 from pgqueuer.domain.types import (
     JOB_STATUS,
@@ -234,40 +235,74 @@ async def test_on_conflict_skip_shape_preservation(
     assert jids[4] is not None
 
 
-async def test_mark_job_cancellation(
-    apgdriver: AsyncpgDriver,
-) -> None:
-    """mark_job_as_cancelled: job marked canceled, log entry created."""
+async def test_cancel_keeps_picked_row_until_worker_logs(apgdriver: AsyncpgDriver) -> None:
+    """Cancel of a running job keeps its slot and dedupe key until the worker logs (#1031)."""
     q = Queries(apgdriver)
-
-    jids = await q.enqueue(["ep"], [None], [0])
-    qm_id = QueueManagerId(uuid.uuid4())
-
-    jobs = await q.dequeue(
-        batch_size=1,
-        entrypoints={QueueEntrypoint("ep"): EntrypointExecutionParameter(concurrency_limit=1)},
-        queue_manager_id=qm_id,
-        global_concurrency_limit=None,
-        heartbeat_timeout=timedelta(seconds=30),
+    params = {QueueEntrypoint("serial"): EntrypointExecutionParameter(concurrency_limit=1)}
+    owner = QueueManagerId(uuid.uuid4())
+    timeout = timedelta(seconds=30)
+    (running_id,) = await q.enqueue("serial", None, dedupe_key="a")
+    (waiting_id,) = await q.enqueue("serial", None, dedupe_key="b")
+    (queued_id,) = await q.enqueue("serial", None, dedupe_key="c")
+
+    (running,) = await q.dequeue(1, params, owner, None, timeout)
+    assert running.id == running_id
+
+    await q.mark_job_as_cancelled([running_id, queued_id])
+
+    assert await q.queue_job_by_id(queued_id) is None
+    assert [entry.status for entry in await q.job_log_history(queued_id)] == ["queued", "canceled"]
+    held = await q.queue_job_by_id(running_id)
+    assert held is not None and held.status == "picked"
+    assert [entry.status for entry in await q.job_log_history(running_id)] == ["queued", "picked"]
+    assert await q.dequeue(1, params, owner, None, timeout) == []
+    with pytest.raises(DuplicateJobError):
+        await q.enqueue("serial", None, dedupe_key="a")
+
+    await q.log_jobs([(running, "canceled", None)])
+
+    assert await q.queue_job_by_id(running_id) is None
+    assert [entry.status for entry in await q.job_log_history(running_id)] == [
+        "queued",
+        "picked",
+        "canceled",
+    ]
+    (next_job,) = await q.dequeue(1, params, owner, None, timeout)
+    assert next_job.id == waiting_id
+    (replacement,) = await q.enqueue("serial", None, dedupe_key="a")
+    assert replacement is not None
+
+
+async def test_cancel_removes_failed_job(apgdriver: AsyncpgDriver) -> None:
+    """A held failed job is still removed and logged canceled."""
+    q = Queries(apgdriver)
+    (jid,) = await q.enqueue("ep", None)
+    (job,) = await q.dequeue(
+        1,
+        {QueueEntrypoint("ep"): EntrypointExecutionParameter(0)},
+        QueueManagerId(uuid.uuid4()),
+        None,
+        timedelta(seconds=30),
     )
-    assert len(jobs) == 1
+    await q.log_jobs([(job, "failed", None)])
 
-    await q.mark_job_as_cancelled(jids)
+    await q.mark_job_as_cancelled([jid])
 
-    logs = await q.queue_log()
-    cancel_logs = [log for log in logs if log.job_id == int(jids[0]) and log.status == "canceled"]
-    assert len(cancel_logs) == 1
+    assert await q.queue_job_by_id(jid) is None
+    assert [entry.status for entry in await q.job_log_history(jid)] == [
+        "queued",
+        "picked",
+        "failed",
+        "canceled",
+    ]
 
 
 @pytest.mark.parametrize("outcome", ["successful", "canceled", "exception", "failed"])
-@pytest.mark.parametrize(("remove", "terminal"), [("cancel", "canceled"), ("clear", "deleted")])
-async def test_log_jobs_skips_job_no_longer_in_queue(
+async def test_log_jobs_skips_job_cleared_while_it_ran(
     apgdriver: AsyncpgDriver,
-    remove: str,
-    terminal: str,
     outcome: JOB_STATUS,
 ) -> None:
-    """log_jobs writes nothing for a job cancelled or cleared while it ran."""
+    """log_jobs writes nothing for a job cleared while it ran."""
     q = Queries(apgdriver)
     (jid,) = await q.enqueue("ep", None)
     (job,) = await q.dequeue(
@@ -277,16 +312,13 @@ async def test_log_jobs_skips_job_no_longer_in_queue(
         global_concurrency_limit=None,
         heartbeat_timeout=timedelta(seconds=30),
     )
-    if remove == "cancel":
-        await q.mark_job_as_cancelled([jid])
-    else:
-        await q.clear_queue("ep")
+    await q.clear_queue("ep")
 
     await q.log_jobs([(job, outcome, None)])
 
     history = await q.job_log_history(jid)
-    assert [e.status for e in history] == ["queued", "picked", terminal]
-    assert await q.job_status([jid]) == [(jid, terminal)]
+    assert [e.status for e in history] == ["queued", "picked", "deleted"]
+    assert await q.job_status([jid]) == [(jid, "deleted")]
 
 
 async def test_traceback_jsonb_roundtrip(

test/test_inmemory.py

@@ -612,14 +612,11 @@ async def test_log_jobs_adds_to_log(queries: InMemoryQueries) -> None:
 
 
 @pytest.mark.parametrize("outcome", ["successful", "canceled", "exception", "failed"])
-@pytest.mark.parametrize(("remove", "terminal"), [("cancel", "canceled"), ("clear", "deleted")])
-async def test_log_jobs_skips_job_no_longer_in_queue(
+async def test_log_jobs_skips_job_cleared_while_it_ran(
     queries: InMemoryQueries,
-    remove: str,
-    terminal: str,
     outcome: JOB_STATUS,
 ) -> None:
-    """A job cancelled or cleared while it ran keeps its terminal status, as in PostgreSQL."""
+    """A job cleared while it ran keeps status deleted, as in PostgreSQL."""
     (jid,) = await queries.enqueue("ep", None)
     (job,) = await queries.dequeue(
         10,
@@ -628,16 +625,13 @@ async def test_log_jobs_skips_job_no_longer_in_queue(
         None,
         heartbeat_timeout=timedelta(seconds=30),
     )
-    if remove == "cancel":
-        await queries.mark_job_as_cancelled([jid])
-    else:
-        await queries.clear_queue("ep")
+    await queries.clear_queue("ep")
 
     await queries.log_jobs([(job, outcome, None)])
 
     history = await queries.job_log_history(jid)
-    assert [e.status for e in history] == ["queued", "picked", terminal]
-    assert await queries.job_status([jid]) == [(jid, terminal)]
+    assert [e.status for e in history] == ["queued", "picked", "deleted"]
+    assert await queries.job_status([jid]) == [(jid, "deleted")]
 
 
 async def test_log_jobs_logs_job_still_in_queue(queries: InMemoryQueries) -> None:
@@ -662,6 +656,72 @@ async def test_log_jobs_logs_job_still_in_queue(queries: InMemoryQueries) -> Non
 # ---------------------------------------------------------------------------
 
 
+async def test_cancel_keeps_picked_row_until_worker_logs(queries: InMemoryQueries) -> None:
+    """Cancel of a running job keeps its slot and dedupe key until the worker logs (#1031)."""
+    params = {QueueEntrypoint("serial"): EntrypointExecutionParameter(concurrency_limit=1)}
+    owner = QueueManagerId(uuid.uuid4())
+    timeout = timedelta(seconds=30)
+    (running_id,) = await queries.enqueue("serial", None, dedupe_key="a")
+    (waiting_id,) = await queries.enqueue("serial", None, dedupe_key="b")
+    (queued_id,) = await queries.enqueue("serial", None, dedupe_key="c")
+
+    (running,) = await queries.dequeue(1, params, owner, None, timeout)
+    assert running.id == running_id
+
+    await queries.mark_job_as_cancelled([running_id, queued_id])
+
+    assert await queries.queue_job_by_id(queued_id) is None
+    assert [entry.status for entry in await queries.job_log_history(queued_id)] == [
+        "queued",
+        "canceled",
+    ]
+    held = await queries.queue_job_by_id(running_id)
+    assert held is not None and held.status == "picked"
+    assert [entry.status for entry in await queries.job_log_history(running_id)] == [
+        "queued",
+        "picked",
+    ]
+    assert await queries.dequeue(1, params, owner, None, timeout) == []
+    with pytest.raises(DuplicateJobError):
+        await queries.enqueue("serial", None, dedupe_key="a")
+
+    await queries.log_jobs([(running, "canceled", None)])
+
+    assert await queries.queue_job_by_id(running_id) is None
+    assert [entry.status for entry in await queries.job_log_history(running_id)] == [
+        "queued",
+        "picked",
+        "canceled",
+    ]
+    (next_job,) = await queries.dequeue(1, params, owner, None, timeout)
+    assert next_job.id == waiting_id
+    (replacement,) = await queries.enqueue("serial", None, dedupe_key="a")
+    assert replacement is not None
+
+
+async def test_cancel_removes_failed_job(queries: InMemoryQueries) -> None:
+    """A held failed job is still removed and logged canceled."""
+    (jid,) = await queries.enqueue("ep", None)
+    (job,) = await queries.dequeue(
+        1,
+        {QueueEntrypoint("ep"): EntrypointExecutionParameter(0)},
+        QueueManagerId(uuid.uuid4()),
+        None,
+        timedelta(seconds=30),
+    )
+    await queries.log_jobs([(job, "failed", None)])
+
+    await queries.mark_job_as_cancelled([jid])
+
+    assert await queries.queue_job_by_id(jid) is None
+    assert [entry.status for entry in await queries.job_log_history(jid)] == [
+        "queued",
+        "picked",
+        "failed",
+        "canceled",
+    ]
+
+
 async def test_mark_job_as_cancelled(queries: InMemoryQueries, driver: InMemoryDriver) -> None:
     notifications: list[str] = []
     await driver.add_listener(queries.qbq.settings.channel, notifications.append)