diff --git a/docs/getting-started/core-concepts.md b/docs/getting-started/core-concepts.md index f1875236..696a244f 100644 --- a/docs/getting-started/core-concepts.md +++ b/docs/getting-started/core-concepts.md @@ -104,7 +104,7 @@ can be manually re-queued via `pgq requeue ` 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 ` (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`), diff --git a/docs/guides/concurrency-control.md b/docs/guides/concurrency-control.md index 5f4857a6..141475ee 100644 --- a/docs/guides/concurrency-control.md +++ b/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 diff --git a/docs/guides/job-cancellation.md b/docs/guides/job-cancellation.md index 157d2768..effc1b1c 100644 --- a/docs/guides/job-cancellation.md +++ b/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. diff --git a/docs/reference/architecture.md b/docs/reference/architecture.md index fb89d3ba..d7a8ec28 100644 --- a/docs/reference/architecture.md +++ b/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. diff --git a/pgqueuer/adapters/inmemory/jobs.py b/pgqueuer/adapters/inmemory/jobs.py index 58bb2534..776c31aa 100644 --- a/pgqueuer/adapters/inmemory/jobs.py +++ b/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) diff --git a/pgqueuer/adapters/persistence/qb.py b/pgqueuer/adapters/persistence/qb.py index bf221b7c..78de8220 100644 --- a/pgqueuer/adapters/persistence/qb.py +++ b/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 diff --git a/pgqueuer/adapters/persistence/queries.py b/pgqueuer/adapters/persistence/queries.py index a7ce45eb..b02d1814 100644 --- a/pgqueuer/adapters/persistence/queries.py +++ b/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(), diff --git a/test/integration/test_persistence_queries_sql.py b/test/integration/test_persistence_queries_sql.py index fe41b43e..53917589 100644 --- a/test/integration/test_persistence_queries_sql.py +++ b/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( diff --git a/test/test_inmemory.py b/test/test_inmemory.py index 1cb7d51b..3dc4d888 100644 --- a/test/test_inmemory.py +++ b/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)