Phase 4: measure only the provider call in duration_ms

Review log [55]. Three historical local_timeout rows recorded 0.4-2.0s more
than the configured budget because the measurement window opened before the
provider call.

The plan named two causes, and both were already gone. Diffed against
f86c0ff~1: at V4.6 the window held resolve_provider_input (async;
normalization + artifact write + DB work) and a session.commit(). Phase 1
deleted both. What remains between the clock and the wait_for is
build_provider_input, now pure field copying because normalization moved to
ingest and file_hash is already stored: 6.2 us per call, zero awaits, so it
cannot yield to the event loop.

A third cause was still there and is not in the plan. The regression test
below measured 890ms where ~200ms was expected. services.sources.provider is
a lazy property that appears as an argument expression to _call_transcriber,
so it is evaluated after the clock starts but before wait_for begins timing.
Constructing OpenRouterTranscriptionProvider costs 475ms on first access and
0.001ms after, so the first attempt of every worker process booked half a
second of HTTP client construction as provider latency. That plausibly
accounts for the low end of the historical overshoot.

workflows.py
  - Re-capture monotonic_started_at immediately before the wait_for, reusing
    the same variable. The pre-loop assignment stays as the fallback: binding
    a new name inside the try would leave the general-exception handler
    referencing an unbound variable when build_provider_input raises. All
    three duration write sites (success, TimeoutError, general failure) then
    measure the correct window with no further change.
  - Hoist the provider property above the per-source loop. It is
    loop-invariant, so this also removes the repeated lookup from the two
    evidence-capture sites.

tests/services/test_workflows_reliability.py
  test_timeout_duration_excludes_pre_call_setup simulates 400ms of blocking
  setup against a 200ms budget and asserts the recorded duration sits near
  the budget and well clear of budget+setup. Confirmed to fail on the pre-fix
  code (assert 625 < 540) and pass after, so it guards behaviour rather than
  restating it. This is the plan's verification criterion as a test.

ui/pages/sources_page.py
  _format_duration renders >=1s as "27.6 s" and below that as "612 ms",
  replacing the raw "27612 ms". No test asserted the old format.

Plan task 3 (record preprocessing as its own value) declined and logged as a
deviation: after Phase 1 there is no preprocessing left to record, and a
preprocessing_ms column to measure 6 us of attribute copying is complexity
without a reader.

Verified: 293 passed, 4 skipped, 0 ruff, 0 ty.

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
zoltan57
2026-08-18 16:02:58 -05:00
co-authored by Copilot App
parent 7dd0d2c9bf
commit 110f40a28b
3 changed files with 96 additions and 4 deletions
+13 -3
View File
@@ -205,6 +205,10 @@ async def process_queued_job( # noqa: PLR0915
externally_stopped = False externally_stopped = False
prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings) prompt_execution = _resolve_job_prompt_execution(source_job=source_job, settings=runtime_settings)
# Resolve the lazy provider property once, outside the timed region. First access
# constructs the HTTP client (~0.5s), which would otherwise be booked as provider
# latency on the first attempt of every worker process (review log [55]).
provider = services.sources.provider
for source in sources: for source in sources:
if await _job_no_longer_processing(job_id=job.id, services=services, session=session): if await _job_no_longer_processing(job_id=job.id, services=services, session=session):
@@ -212,6 +216,8 @@ async def process_queued_job( # noqa: PLR0915
break break
started_at = datetime.now(UTC) started_at = datetime.now(UTC)
# Fallback start for failures raised before the provider call; reset to the
# true call boundary immediately before the wait_for below.
monotonic_started_at = asyncio.get_running_loop().time() monotonic_started_at = asyncio.get_running_loop().time()
result: TranscriptionResult | None = None result: TranscriptionResult | None = None
provider_input = None provider_input = None
@@ -225,12 +231,16 @@ async def process_queued_job( # noqa: PLR0915
media_type=provider_input.media_type, media_type=provider_input.media_type,
page_number=source.page_number, page_number=source.page_number,
) )
# Restart the clock so duration_ms covers only what the wait_for below
# governs. The pre-loop assignment stays as the fallback for failures
# raised before this point, which would otherwise leave it unbound.
monotonic_started_at = asyncio.get_running_loop().time()
result = await asyncio.wait_for( result = await asyncio.wait_for(
_call_transcriber( _call_transcriber(
input_path=provider_input.path, input_path=provider_input.path,
prompt_execution=prompt_execution, prompt_execution=prompt_execution,
settings=runtime_settings, settings=runtime_settings,
provider=services.sources.provider, provider=provider,
source_reference=source_reference, source_reference=source_reference,
requested_model=source_job.model, requested_model=source_job.model,
), ),
@@ -283,8 +293,8 @@ async def process_queued_job( # noqa: PLR0915
0, 0,
int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000), int((asyncio.get_running_loop().time() - monotonic_started_at) * 1000),
), ),
request_manifest=services.sources.provider.current_request_manifest, request_manifest=provider.current_request_manifest,
transport_evidence=services.sources.provider.current_transport_evidence, transport_evidence=provider.current_transport_evidence,
failure_phase="local_timeout", failure_phase="local_timeout",
) )
failed_pages.append(page_outcome) failed_pages.append(page_outcome)
+8 -1
View File
@@ -366,6 +366,13 @@ def _render_source_job_metadata_zone(
_render_provider_evidence(latest_attempt=latest_attempt) _render_provider_evidence(latest_attempt=latest_attempt)
def _format_duration(duration_ms: int) -> str:
"""Render an attempt duration with a unit that suits its magnitude."""
if duration_ms >= 1000:
return f"{duration_ms / 1000:.1f} s"
return f"{duration_ms} ms"
def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None: def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None) -> None:
ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3") ui.label("Provider Evidence").classes("text-xs font-semibold ui-text-primary mt-3")
if latest_attempt is None: if latest_attempt is None:
@@ -374,7 +381,7 @@ def _render_provider_evidence(*, latest_attempt: LatestExecutionAttempt | None)
attempt = latest_attempt.attempt attempt = latest_attempt.attempt
metadata_row("Attempt:", str(attempt.attempt_number)) metadata_row("Attempt:", str(attempt.attempt_number))
metadata_row("Duration:", f"{attempt.duration_ms} ms") metadata_row("Duration:", _format_duration(attempt.duration_ms))
_render_json_evidence("Request Manifest", attempt.request_manifest) _render_json_evidence("Request Manifest", attempt.request_manifest)
_render_json_evidence("Transport Response", _transport_display(latest_attempt)) _render_json_evidence("Transport Response", _transport_display(latest_attempt))
_render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot) _render_json_evidence("OpenRouter SDK Response Snapshot", attempt.sdk_response_snapshot)
@@ -1,6 +1,7 @@
"""Reliability tests for worker workflow timeout behavior.""" """Reliability tests for worker workflow timeout behavior."""
import asyncio import asyncio
import time
from pathlib import Path from pathlib import Path
from uuid import uuid4 from uuid import uuid4
@@ -18,6 +19,7 @@ from transcription.db.models import JobStatus
from transcription.db.models import Source from transcription.db.models import Source
from transcription.providers.base import TranscriptionResult from transcription.providers.base import TranscriptionResult
from transcription.services import ServiceBundle from transcription.services import ServiceBundle
from transcription.services import workflows as workflows_module
from transcription.services.workflows import process_queued_job from transcription.services.workflows import process_queued_job
@@ -103,6 +105,79 @@ class TestWorkflowReliability:
assert "timed out" in error_detail.lower() assert "timed out" in error_detail.lower()
assert "20.0s" in error_detail 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 @pytest.mark.asyncio
async def test_completed_page_is_committed_before_next_provider_call_finishes( async def test_completed_page_is_committed_before_next_provider_call_finishes(
self, self,