From 1433e2e8a3792cfdc6385d11ff04c7c1202790fd Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Sat, 31 Oct 2026 12:05:15 +0200 Subject: [PATCH] Fail a job whose evidence payload cannot be read, once A payload that is not JSON, or JSON that is not an object, raised inside the worker loop before process_evidence, so the job kept its lease and came back every time the lease ran out. run_job now parses it and fails the job with the evidence quarantined, as process_evidence does for any other failure. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 6 +++ .../api/app/services/processing_worker.py | 49 ++++++++++++------- services/api/tests/test_processing_worker.py | 38 +++++++++++++- 3 files changed, 75 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e2883e..af71eca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ a decision with visible consequences. ## Unreleased +- 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 + per lease. The payload is now parsed in `run_job`, which fails the job once ("evidence payload + is not readable") and quarantines the evidence, the same outcome `process_evidence` already + gives a job whose processing raises. - A read key's failed attempts are counted per known agent, and every agent id with no key shares one count per client address. The count used to be per `X-Agent-Id` header, which the caller writes: a new made-up id on each try earned five fresh attempts and added an entry diff --git a/services/api/app/services/processing_worker.py b/services/api/app/services/processing_worker.py index 007ae42..519a491 100644 --- a/services/api/app/services/processing_worker.py +++ b/services/api/app/services/processing_worker.py @@ -67,29 +67,44 @@ def claim_next_job(db, worker_id: str, *, now: datetime | None = None, db.rollback() +def _fail(db, job: ProcessingJob, error: str, evidence: EvidenceObject | None = None) -> None: + fence_job_lease(db, job.id, job.lease_owner) + job.status = "failed" + job.error = error + job.lease_owner = None + job.lease_expires_at = None + if evidence: + evidence.processing_state = "quarantined" + db.commit() + + +def run_job(db, job: ProcessingJob) -> None: + """Process one claimed job, failing it at once if its evidence is unreadable.""" + evidence = db.get(EvidenceObject, job.evidence_id) + if not evidence: + _fail(db, job, "evidence no longer exists") + return + try: + content = json.loads(evidence.normalized_payload_json or "{}").get("content") or evidence.summary + except (ValueError, AttributeError): + _fail(db, job, "evidence payload is not readable", evidence) + return + with LeaseHeartbeat( + SessionLocal, + job.id, + job.lease_owner, + lease_seconds=PROCESSING_LEASE_SECONDS, + ) as lease: + process_evidence(db, evidence, content, job, lease=lease) + + def _run() -> None: while not _stop.is_set(): db = SessionLocal() try: job = claim_next_job(db, _worker_id) if job: - evidence = db.get(EvidenceObject, job.evidence_id) - if not evidence: - fence_job_lease(db, job.id, job.lease_owner) - job.status = "failed" - job.error = "evidence no longer exists" - job.lease_owner = None - job.lease_expires_at = None - db.commit() - else: - content = json.loads(evidence.normalized_payload_json or "{}").get("content") or evidence.summary - with LeaseHeartbeat( - SessionLocal, - job.id, - job.lease_owner, - lease_seconds=PROCESSING_LEASE_SECONDS, - ) as lease: - process_evidence(db, evidence, content, job, lease=lease) + run_job(db, job) except Exception: db.rollback() finally: diff --git a/services/api/tests/test_processing_worker.py b/services/api/tests/test_processing_worker.py index fd479a2..5fdd7e0 100644 --- a/services/api/tests/test_processing_worker.py +++ b/services/api/tests/test_processing_worker.py @@ -6,7 +6,7 @@ from app.core.db import Base from app.models.evidence_object import EvidenceObject from app.models.processing_job import ProcessingJob -from app.services import runtime_pipeline +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 sqlalchemy import create_engine, select @@ -314,3 +314,39 @@ def lose_lease_and_fail(*_args): assert evidence.processing_state != "quarantined" 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) + 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=payload, + ) + db.add(evidence) + db.flush() + job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued") + db.add(job) + db.commit() + evidence_id = evidence.id + + with sessions() as db: + claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60) + processing_worker.run_job(db, claimed) + + with sessions() as db: + job = db.get(ProcessingJob, claimed.id) + evidence = db.get(EvidenceObject, evidence_id) + assert job.status == "failed" + assert job.error == "evidence payload is not readable" + assert job.lease_owner is None + assert evidence.processing_state == "quarantined" + assert claim_next_job( + db, "worker-b", now=started_at + timedelta(seconds=61), lease_seconds=60 + ) is None + finally: + engine.dispose()