From 773a66a389a003be1455a2845f7542e219bacaa4 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Sun, 1 Nov 2026 17:16:25 +0200 Subject: [PATCH 1/2] Retry job refuses when worker holds live lease - Add requeue helper that only clears lease when job is not processing, has no lease, or lease is expired - Update retry_job route to use requeue and return 409 when live lease exists - Add tests for live lease refusal, failed job requeue, expired lease requeue, and missing job handling - Prevents double-processing of evidence when retry is called during active job Fixes race condition where retry could queue same job for multiple workers simultaneously, violating lease protection. --- CHANGELOG.md | 3 + services/api/app/routes/runtime.py | 9 +- .../api/app/services/processing_worker.py | 24 +++++ services/api/tests/test_processing_worker.py | 102 +++++++++++++++++- 4 files changed, 131 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 25a7cd3..5f94695 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ a decision with visible consequences. ## Unreleased +- Retrying a job that a worker is still running is refused with a 409 instead of being queued + again, which could cause the same evidence to be processed twice. The `requeue` helper only + clears the lease when the job is not `processing`, has no lease, or the lease is expired. - A job whose stored evidence payload cannot be read - not JSON, or JSON that is not an object - used to raise inside the worker loop, leaving the job `processing` with its lease; when the lease ran out the next poll claimed it again and failed the same way, forever, one wasted claim diff --git a/services/api/app/routes/runtime.py b/services/api/app/routes/runtime.py index 4219b54..ca338cf 100644 --- a/services/api/app/routes/runtime.py +++ b/services/api/app/routes/runtime.py @@ -30,6 +30,7 @@ from app.services.memory_truth import mark_unsupported from app.services.ollama_service import answer_with_context, ollama_health from app.services.pi_inference import CHOSEN_IN, last_answer, model_choice, pi_configured +from app.services.processing_worker import requeue from app.services.qdrant_store import INDEX_UNREACHABLE, search_memory_embeddings, semantic_status from app.services.scoring import memory_rank_bonus from fastapi import APIRouter, Depends, Header, HTTPException, Request @@ -370,12 +371,8 @@ def retry_job(job_id: str, agent_id: str = Depends(get_agent_id), _: str = Depen evidence = db.get(EvidenceObject, job.evidence_id) if not evidence: raise HTTPException(404, "Evidence no longer exists") - job.status = "pending" - job.stage = "retry_queued" - job.error = "" - job.lease_owner = None - job.lease_expires_at = None - db.commit() + if not requeue(db, job.id): + raise HTTPException(409, "Processing job is running; retry after it ends") return {"status": "queued", "job_id": job.id} finally: db.close() diff --git a/services/api/app/services/processing_worker.py b/services/api/app/services/processing_worker.py index 519a491..34d82dc 100644 --- a/services/api/app/services/processing_worker.py +++ b/services/api/app/services/processing_worker.py @@ -67,6 +67,30 @@ def claim_next_job(db, worker_id: str, *, now: datetime | None = None, db.rollback() +def requeue(db, job_id: str, *, now: datetime | None = None) -> bool: + """Queue a job again unless a worker holds a live lease on it; True when queued.""" + now = now or datetime.now(UTC) + not_running = or_( + ProcessingJob.status != "processing", + ProcessingJob.lease_expires_at.is_(None), + ProcessingJob.lease_expires_at <= now, + ) + result = db.execute( + update(ProcessingJob) + .where(ProcessingJob.id == job_id, not_running) + .values( + status="pending", + stage="retry_queued", + error="", + lease_owner=None, + lease_expires_at=None, + ) + .execution_options(synchronize_session=False) + ) + db.commit() + return result.rowcount == 1 + + def _fail(db, job: ProcessingJob, error: str, evidence: EvidenceObject | None = None) -> None: fence_job_lease(db, job.id, job.lease_owner) job.status = "failed" diff --git a/services/api/tests/test_processing_worker.py b/services/api/tests/test_processing_worker.py index 5fdd7e0..527bfdb 100644 --- a/services/api/tests/test_processing_worker.py +++ b/services/api/tests/test_processing_worker.py @@ -8,7 +8,7 @@ from app.models.processing_job import ProcessingJob from app.services import processing_worker, runtime_pipeline from app.services.processing_lease import LeaseHeartbeat, LeaseLostError -from app.services.processing_worker import claim_next_job +from app.services.processing_worker import claim_next_job, requeue from sqlalchemy import create_engine, select from sqlalchemy.orm import sessionmaker @@ -316,6 +316,106 @@ def lose_lease_and_fail(*_args): engine.dispose() +def test_retry_refuses_a_job_with_a_live_lease(tmp_path): + engine, sessions = _database(tmp_path) + try: + started_at = datetime.now(UTC) + with sessions() as db: + evidence = EvidenceObject( + agent_id="owner", source_id="source-1", source_key="listener", + source_type="listener", summary="A bounded event", + normalized_payload_json='{"content":"A bounded event"}', + ) + db.add(evidence) + db.flush() + job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued") + db.add(job) + db.commit() + + with sessions() as db: + claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60) + assert claimed is not None + + with sessions() as db: + result = requeue(db, claimed.id, now=started_at + timedelta(seconds=10)) + assert result is False + job = db.get(ProcessingJob, claimed.id) + assert job.status == "processing" + assert job.lease_owner == "worker-a" + # Compare with timezone-aware datetime + expected_expiry = started_at + timedelta(seconds=60) + assert job.lease_expires_at.replace(tzinfo=UTC) == expected_expiry + finally: + engine.dispose() + + +def test_retry_requeues_a_failed_job_and_an_expired_lease(tmp_path): + engine, sessions = _database(tmp_path) + try: + # Test (a): failed job with error + with sessions() as db: + evidence = EvidenceObject( + agent_id="owner", source_id="source-1", source_key="listener", + source_type="listener", summary="A bounded event", + normalized_payload_json='{"content":"A bounded event"}', + ) + db.add(evidence) + db.flush() + failed_job = ProcessingJob( + agent_id="owner", evidence_id=evidence.id, status="failed", + stage="analysis", error="some error" + ) + db.add(failed_job) + db.commit() + + with sessions() as db: + result = requeue(db, failed_job.id) + assert result is True + job = db.get(ProcessingJob, failed_job.id) + assert job.status == "pending" + assert job.stage == "retry_queued" + assert job.error == "" + assert job.lease_owner is None + assert job.lease_expires_at is None + + # Test (b): job with expired lease + started_at = datetime.now(UTC) + with sessions() as db: + evidence = EvidenceObject( + agent_id="owner", source_id="source-2", source_key="listener", + source_type="listener", summary="Another event", + normalized_payload_json='{"content":"Another event"}', + ) + db.add(evidence) + db.flush() + claimed_job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued") + db.add(claimed_job) + db.commit() + + with sessions() as db: + claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60) + assert claimed is not None + + with sessions() as db: + result = requeue(db, claimed.id, now=started_at + timedelta(seconds=61)) + assert result is True + job = db.get(ProcessingJob, claimed.id) + assert job.lease_owner is None + assert job.lease_expires_at is None + finally: + engine.dispose() + + +def test_requeue_of_a_missing_job_is_false(tmp_path): + engine, sessions = _database(tmp_path) + try: + with sessions() as db: + result = requeue(db, "no-such-job") + assert result is False + finally: + engine.dispose() + + @pytest.mark.parametrize("payload", ["not json", "[1, 2]"], ids=["not-json", "json-list"]) def test_an_unreadable_payload_fails_the_job_once(tmp_path, payload): engine, sessions = _database(tmp_path) From 173e061374dbbced618405e56f32486c17cfb684 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Sun, 1 Nov 2026 17:27:45 +0200 Subject: [PATCH 2/2] Changelog: say what a retry of a running job did before Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f94695..017b08f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,9 +6,9 @@ a decision with visible consequences. ## Unreleased -- Retrying a job that a worker is still running is refused with a 409 instead of being queued - again, which could cause the same evidence to be processed twice. The `requeue` helper only - clears the lease when the job is not `processing`, has no lease, or the lease is expired. +- Retrying a job that a worker is still running (its lease has not expired) is refused with a 409. + It used to be queued again and cleared the lease, so a second worker could process the same + evidence while the first was still on it. - A job whose stored evidence payload cannot be read - not JSON, or JSON that is not an object - used to raise inside the worker loop, leaving the job `processing` with its lease; when the lease ran out the next poll claimed it again and failed the same way, forever, one wasted claim