generated from john/python-template
zoltan57andCopilot App
fca959fa5d
Review log [8]. classify_unexpected_error already returned retriable=False and the verdict was logged and then thrown away. Measured across src/: retriable was assigned in 9 places and read in none. The plan asks for a test that a programming error "does not silently retry". Probing with an injected AttributeError showed that is not what happens, and the two real failure modes need different fixes. Mode A, raised after the claim commits (inside advance_job): raised exactly once, job left at PROCESSING, retry_count 0, never re-claimed, because claim_next_queued_job filters status == QUEUED. A permanently stranded job with one swallowed log line, not a retry. advance_job's PROCESSING branch, commented "Recover mid-flight jobs", is unreachable from the worker for the same reason. Mode B, raised before or during the claim: 20 raises in 1.2s, an unbounded hot spin at the poll interval. It never reaches the per-job retry machinery, so WORKER_MAX_RETRIES does not cap it and the plan's 60s worst case understates this path. services/workflows.py _advance_job_with_containment wraps advance_job. Any escaping exception is classified and the job driven to terminal FAILED, which is visible in the UI and resubmittable. The caller session is rolled back first and the terminal write runs in its own transaction, so it stays atomic even when the failure left that session dirty (plan task 3). The loop continues, so one poison job cannot halt transcription for every other job. worker.py handle_worker_exceptions re-raises non-retriable faults rather than suppressing them; retriable ones are still suppressed so transient conditions do not stop work. run_worker_loop catches that, logs CRITICAL and returns cleanly. Returning rather than propagating matters: the exception would otherwise surface only at app shutdown, through the wait_for in worker_consumer_lifespan. tests test_run_worker_loop_survives_process_next_exception asserted the loop SURVIVES a RuntimeError and continues, which is the Mode B defect written down as an expectation. Replaced by test_run_worker_loop_stops_on_non_retriable_exception, with a new test_run_worker_loop_survives_retriable_exception so suppression of genuinely transient faults stays covered, and test_error_after_claim_fails_the_job_instead_of_stranding_it for Mode A. All three were verified to fail on pre-fix code. The Mode B guard fails by timing out, which is the infinite spin made visible. Verified: 295 passed, 4 skipped, 0 ruff, 0 ty. Co-authored-by: Copilot App <[email protected]>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""Reliability tests for worker workflow timeout behavior."""
|
|
|
|
import asyncio
|
|
import time
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlmodel import col
|
|
from sqlmodel import select
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import ExecutionAttempt
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import JobStatus
|
|
from transcription.db.models import Source
|
|
from transcription.providers.base import TranscriptionResult
|
|
from transcription.services import ServiceBundle
|
|
from transcription.services import workflows as workflows_module
|
|
from transcription.services.workflows import process_queued_job
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestWorkflowReliability:
|
|
"""Verify timeout and terminal-state reliability behavior."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_queued_job_timeout_marks_job_failed(self, default_session_factory, monkeypatch):
|
|
"""Provider timeout transitions a queued job to failed with error detail."""
|
|
services = ServiceBundle()
|
|
object.__setattr__(
|
|
services,
|
|
"documents",
|
|
services.documents.__class__(session_factory=default_session_factory),
|
|
)
|
|
object.__setattr__(
|
|
services,
|
|
"jobs",
|
|
services.jobs.__class__(session_factory=default_session_factory),
|
|
)
|
|
object.__setattr__(
|
|
services,
|
|
"sources",
|
|
services.sources.__class__(session_factory=default_session_factory),
|
|
)
|
|
|
|
async with services.jobs._session_scope() as session:
|
|
document = Document(id=uuid4(), name="timeout-doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
session.add(job)
|
|
await session.flush()
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="timeout.jpg",
|
|
filename="timeout.jpg",
|
|
file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
|
|
file_hash="c" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
|
|
await session.commit()
|
|
|
|
loaded = await services.jobs.read_job(job_id=job.id, session=session)
|
|
|
|
async def _never_returns(
|
|
image_path,
|
|
*,
|
|
prompt_name="transcribe_document.md",
|
|
prompt_text=None,
|
|
temperature=None,
|
|
top_p=None,
|
|
settings=None,
|
|
provider=None,
|
|
):
|
|
_ = (image_path, prompt_name, prompt_text, temperature, top_p, settings, provider)
|
|
raise TimeoutError("simulated provider timeout")
|
|
|
|
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns)
|
|
|
|
timeout_settings = Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=20.0)
|
|
result = await process_queued_job(job=loaded, services=services, settings=timeout_settings)
|
|
|
|
assert result is not None
|
|
assert result.status == JobStatus.FAILED
|
|
|
|
async with services.jobs._session_scope() as session:
|
|
attempts = (
|
|
await session.execute(
|
|
select(ExecutionAttempt).where(
|
|
col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
|
|
)
|
|
)
|
|
).scalars().all()
|
|
error_detail = next(attempt.error_detail for attempt in attempts if attempt.error_detail is not None)
|
|
assert "timed out" in error_detail.lower()
|
|
assert "20.0s" in error_detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout_duration_excludes_pre_call_setup(self, default_session_factory, monkeypatch):
|
|
"""duration_ms covers only the provider call, not the setup preceding it.
|
|
|
|
Regression guard for review log [55]: three historical ``local_timeout`` rows
|
|
recorded 0.4-2.0 s more than the configured budget because the measurement
|
|
window opened before payload resolution. Blocking setup is simulated here so
|
|
the assertion fails if that window ever reopens.
|
|
"""
|
|
services = ServiceBundle.from_session_factory(default_session_factory)
|
|
async with services.jobs._session_scope() as session:
|
|
document = Document(id=uuid4(), name="window-doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
session.add(job)
|
|
await session.flush()
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="window.jpg",
|
|
filename="window.jpg",
|
|
file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
|
|
file_hash="d" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
|
|
await session.commit()
|
|
loaded = await services.jobs.read_job(job_id=job.id, session=session)
|
|
|
|
setup_seconds = 0.40
|
|
budget_seconds = 0.20
|
|
real_build = workflows_module.build_provider_input
|
|
|
|
def _slow_build(source_arg):
|
|
time.sleep(setup_seconds)
|
|
return real_build(source_arg)
|
|
|
|
async def _never_returns(*args, **kwargs):
|
|
_ = (args, kwargs)
|
|
await asyncio.sleep(budget_seconds * 20)
|
|
|
|
monkeypatch.setattr("transcription.services.workflows.build_provider_input", _slow_build)
|
|
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _never_returns)
|
|
|
|
result = await process_queued_job(
|
|
job=loaded,
|
|
services=services,
|
|
settings=Settings(openrouter_api_key="test-key", worker_provider_timeout_seconds=budget_seconds),
|
|
)
|
|
assert result is not None
|
|
assert result.status == JobStatus.FAILED
|
|
|
|
async with services.jobs._session_scope() as session:
|
|
attempts = (
|
|
(
|
|
await session.exec(
|
|
select(ExecutionAttempt).where(
|
|
col(ExecutionAttempt.job_source_id).in_([js.id for js in result.job_sources])
|
|
)
|
|
)
|
|
)
|
|
.all()
|
|
)
|
|
assert len(attempts) == 1
|
|
duration_ms = attempts[0].duration_ms
|
|
|
|
# At or just above the budget, and well clear of budget + setup.
|
|
assert duration_ms >= int(budget_seconds * 1000 * 0.9)
|
|
assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_after_claim_fails_the_job_instead_of_stranding_it(
|
|
self,
|
|
default_session_factory,
|
|
monkeypatch,
|
|
):
|
|
"""A non-retriable fault after the claim drives the job terminal, not stuck.
|
|
|
|
Regression guard for review log [8]. The claim commits PROCESSING before any
|
|
provider work, and claim_next_queued_job only ever selects QUEUED, so an
|
|
exception escaping advance_job used to strand the job in PROCESSING forever
|
|
with one swallowed log line. Measured before the fix: raised once, job left
|
|
processing, retry_count 0, never re-claimed.
|
|
"""
|
|
services = ServiceBundle.from_session_factory(default_session_factory)
|
|
async with services.jobs._session_scope() as session:
|
|
document = Document(id=uuid4(), name="strand-doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
session.add(job)
|
|
await session.flush()
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="strand.jpg",
|
|
filename="strand.jpg",
|
|
file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
|
|
file_hash="e" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING))
|
|
await session.commit()
|
|
job_id = job.id
|
|
|
|
async def _succeeds(*args, **kwargs):
|
|
_ = (args, kwargs)
|
|
return TranscriptionResult(text="page text", provider="test", model="test-model")
|
|
|
|
async def _boom(**kwargs):
|
|
_ = kwargs
|
|
raise AttributeError("deliberate programming error")
|
|
|
|
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds)
|
|
monkeypatch.setattr("transcription.services.workflows._finalize_batch_outcome", _boom)
|
|
|
|
processed = await workflows_module.process_next_queued_job(services=services)
|
|
|
|
assert processed is True
|
|
async with services.jobs._session_scope() as session:
|
|
final = await session.get(Job, job_id)
|
|
assert final is not None
|
|
# Terminal and resubmittable, rather than stranded in PROCESSING.
|
|
assert final.status == JobStatus.FAILED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_completed_page_is_committed_before_next_provider_call_finishes(
|
|
self,
|
|
default_session_factory,
|
|
monkeypatch,
|
|
):
|
|
services = ServiceBundle.from_session_factory(default_session_factory)
|
|
async with services.jobs._session_scope() as session:
|
|
document = Document(id=uuid4(), name="durability-doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
job = Job(document_id=document.id, status=JobStatus.QUEUED)
|
|
session.add(job)
|
|
await session.flush()
|
|
for page_number in (1, 2):
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=page_number,
|
|
upload_name=f"page-{page_number}.jpg",
|
|
filename=f"page-{page_number}.jpg",
|
|
file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")),
|
|
file_hash=str(page_number) * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
session.add(JobSource(job_id=job.id, source_id=source.id))
|
|
await session.commit()
|
|
loaded = await services.jobs.read_job(job_id=job.id, session=session)
|
|
|
|
second_started = asyncio.Event()
|
|
release_second = asyncio.Event()
|
|
call_count = 0
|
|
|
|
async def _transcribe(image_path, **kwargs):
|
|
nonlocal call_count
|
|
_ = (image_path, kwargs)
|
|
call_count += 1
|
|
if call_count == 2:
|
|
second_started.set()
|
|
await release_second.wait()
|
|
return TranscriptionResult(text=f"page {call_count}", provider="fixture", model="model")
|
|
|
|
monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _transcribe)
|
|
task = asyncio.create_task(process_queued_job(job=loaded, services=services))
|
|
await asyncio.wait_for(second_started.wait(), timeout=2)
|
|
|
|
attempts = await services.evidence.list_execution_attempts(job_id=job.id)
|
|
assert len(attempts) == 1
|
|
assert attempts[0].raw_transcription == "page 1"
|
|
|
|
release_second.set()
|
|
result = await task
|
|
assert result is not None
|
|
assert result.status == JobStatus.TRANSCRIBED
|