Add cancel() method to QueueJobStore

Allow transitioning queued or in_progress jobs to the cancelled
state, matching the JobStatus enum value that was previously
unreachable. Includes tests for both valid transitions and the
rejection of cancelling already-completed jobs.

Author: jonathansick

src/docverse/storage/queue_job_store.py

@@ -201,6 +201,29 @@ async def fail(
         await self._session.refresh(row)
         return QueueJob.model_validate(row, from_attributes=True)
 
+    async def cancel(self, job_id: int) -> QueueJob:
+        """Mark job cancelled, set date_completed=now().
+
+        Raises
+        ------
+        InvalidJobStateError
+            If the job is not in queued or in_progress status.
+        """
+        row = await self._get_row(job_id)
+        allowed = {JobStatus.queued.value, JobStatus.in_progress.value}
+        if row.status not in allowed:
+            msg = (
+                f"Cannot cancel job {job_id}: "
+                f"expected 'queued'/'in_progress', "
+                f"got '{row.status}'"
+            )
+            raise InvalidJobStateError(msg)
+        row.status = JobStatus.cancelled.value
+        row.date_completed = datetime.now(tz=UTC)
+        await self._session.flush()
+        await self._session.refresh(row)
+        return QueueJob.model_validate(row, from_attributes=True)
+
     async def _get_row(self, job_id: int) -> SqlQueueJob:
         """Fetch a SqlQueueJob row by id, raising if not found."""
         result = await self._session.execute(

tests/storage/queue_job_store_test.py

@@ -152,6 +152,47 @@ async def test_fail_job(
     assert failed.errors == {"message": "something went wrong"}
 
 
+@pytest.mark.asyncio
+async def test_cancel_queued_job(
+    db_session: async_scoped_session[AsyncSession],
+    store: QueueJobStore,
+) -> None:
+    async with db_session.begin():
+        job = await store.create(kind=JobKind.build_processing, org_id=1)
+        cancelled = await store.cancel(job.id)
+        await db_session.commit()
+    assert cancelled.status == JobStatus.cancelled
+    assert cancelled.date_completed is not None
+
+
+@pytest.mark.asyncio
+async def test_cancel_in_progress_job(
+    db_session: async_scoped_session[AsyncSession],
+    store: QueueJobStore,
+) -> None:
+    async with db_session.begin():
+        job = await store.create(kind=JobKind.build_processing, org_id=1)
+        await store.start(job.id)
+        cancelled = await store.cancel(job.id)
+        await db_session.commit()
+    assert cancelled.status == JobStatus.cancelled
+    assert cancelled.date_completed is not None
+
+
+@pytest.mark.asyncio
+async def test_cancel_completed_job_raises(
+    db_session: async_scoped_session[AsyncSession],
+    store: QueueJobStore,
+) -> None:
+    async with db_session.begin():
+        job = await store.create(kind=JobKind.build_processing, org_id=1)
+        await store.start(job.id)
+        await store.complete(job.id)
+        with pytest.raises(InvalidJobStateError):
+            await store.cancel(job.id)
+        await db_session.commit()
+
+
 @pytest.mark.asyncio
 async def test_get_by_public_id(
     db_session: async_scoped_session[AsyncSession],