From 4e4bf502193f6a042b13c3fa96750f44257c5aa4 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sun, 28 Jun 2026 12:32:22 -0500 Subject: [PATCH] boundaries --- .github/instructions/services.instructions.md | 2 + src/transcription/models.py | 5 ++ src/transcription/services/jobs.py | 32 ++++++---- src/transcription/services/workflows.py | 64 ++++++------------- src/transcription/worker.py | 4 +- 5 files changed, 46 insertions(+), 61 deletions(-) diff --git a/.github/instructions/services.instructions.md b/.github/instructions/services.instructions.md index 9b942ab..84803a9 100644 --- a/.github/instructions/services.instructions.md +++ b/.github/instructions/services.instructions.md @@ -67,6 +67,8 @@ Atomicity rules: Separation of concerns: - Worker modules should stay lightweight and delegate lifecycle transitions to service/workflow orchestration functions. +- In `workflows.py`, `process_queued_job` should own one complete attempt lifecycle: `QUEUED -> PROCESSING -> TRANSCRIBED|FAILED`. +- In `workflows.py`, `advance_job` should coordinate broader status progression around attempts (for example retry scheduling from `FAILED -> QUEUED`). - Services should expose session-aware write helpers (flush on caller-owned session) so orchestration controls commit boundaries. - Backoff/sleep behavior must run outside transactional scopes. diff --git a/src/transcription/models.py b/src/transcription/models.py index 65c8283..60eb089 100644 --- a/src/transcription/models.py +++ b/src/transcription/models.py @@ -49,6 +49,11 @@ class Job(SQLModel, table=True): document: Document = Relationship(back_populates="jobs") transcript: Optional["Transcript"] = Relationship(back_populates="job") + @property + def filename(self) -> str: + """Return the filename of the associated document.""" + return self.document.filename if self.document else "unknown" + class Transcript(SQLModel, table=True): """The output of a transcription job.""" diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index a75e727..e287b38 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -29,16 +29,17 @@ class JobService(ServiceBase): async def read_job(self, job_id: UUID, session: AsyncSession | None = None) -> Job: """Read an existing job from the database. - The selectinload option is used to eagerly load the related document for the job, which makes the full Document - model object available in the return Job object. + The related document is always eagerly loaded so callers can safely + access ``job.document`` in async contexts without triggering lazy-load IO. """ async with self._session_scope(session) as _session: - job = await _session.get( - Job, - job_id, - # Makes the full Document model object available in the return Job object - options=(selectinload(Job.document),), # pyright: ignore[reportArgumentType] + query = ( + select(Job) + .options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] + .where(Job.id == job_id) + .execution_options(populate_existing=True) ) + job = (await _session.exec(query)).first() if job is None: raise ValueError(f"Job with id {job_id} not found") return job @@ -67,7 +68,7 @@ class JobService(ServiceBase): ) -> Sequence[Job]: """Query jobs from the database based on provided filters.""" async with self._session_scope(session) as _session: - query = select(Job) + query = select(Job).options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] if status is not None: query = query.where(Job.status == status) if filename is not None: @@ -81,11 +82,10 @@ class JobService(ServiceBase): load_docs: bool = False, session: AsyncSession | None = None, ) -> Sequence[Job]: - """List all jobs in the database.""" + """List all jobs in the database with eagerly loaded documents.""" + _ = load_docs async with self._session_scope(session) as _session: - query = select(Job) - if load_docs: - query = query.options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] + query = select(Job).options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] result = await _session.exec(query) return result.all() @@ -114,7 +114,13 @@ class JobService(ServiceBase): once at an orchestration boundary. """ async with self._session_scope(session) as _session: - job = await _session.get(Job, job_id) + query = ( + select(Job) + .options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] + .where(Job.id == job_id) + .execution_options(populate_existing=True) + ) + job = (await _session.exec(query)).first() if job is None: raise ValueError(f"Job with id {job_id} not found") job.status = status diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 2ab1da2..368222b 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -6,7 +6,6 @@ from sqlmodel.ext.asyncio.session import AsyncSession from ..config import Settings from ..config import get_settings from ..errors import AppError -from ..errors import ErrorCategory from ..errors import classify_unexpected_error from ..errors import format_error_detail from ..models import Job @@ -23,18 +22,20 @@ async def advance_job( job: Job, services: ServiceBundle, settings: Settings | None = None, + session: AsyncSession | None = None, ) -> Job | None: """Advance a single job by lifecycle status.""" settings = settings or get_settings() match job.status: case JobStatus.QUEUED: - return await process_job(job=job, services=services, settings=settings) + return await process_queued_job(job=job, services=services, session=session) case JobStatus.FAILED: if job.retry_count < settings.worker_max_retries: return await services.jobs.update_job_state( job_id=job.id, status=JobStatus.QUEUED, retry_count_increment=1, + session=session, ) else: logger.error(f"Job {job.id} has failed and reached max retries.") @@ -43,15 +44,13 @@ async def advance_job( return -async def process_job( +async def process_queued_job( *, job: Job, services: ServiceBundle, - settings: Settings | None = None, session: AsyncSession | None = None, ) -> Job | None: - """Process a queued job with workflow-owned transaction boundaries.""" - runtime_settings = settings or get_settings() + """Process one complete transcription attempt for a queued job.""" if job.status != JobStatus.QUEUED: logger.warning(f"Job {job.id} is not queued. Current status: {job.status}") return @@ -60,28 +59,19 @@ async def process_job( if session is None: job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING) else: + # If we're sharing the session, need to make sure setting the Job to PROCESSING is committed before we start + # the transcription, otherwise other workers may see the job as still QUEUED and try to process it. job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session) await session.commit() document = job.document - if document is None: - error = AppError( - "Document not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Re-upload the source document and retry processing.", - ) - await _finalize_failed(job=job, services=services, error=error, session=session) - logger.error( - "Job failed operation=worker.process_job job_id=%s error_id=%s category=%s", - job.id, - error.error_id, - error.category.value, - ) - return job + assert document is not None, ( + f"Job {job.id} has no associated document or the document failed to be loaded by the job service." + ) try: result = await transcribe_document_image(document.file_path) - await _finalize_transcribed(job=job, services=services, result=result, session=session) + job = await _finalize_transcribed(job=job, services=services, result=result, session=session) logger.info( "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s", job.id, @@ -89,27 +79,13 @@ async def process_job( result.provider, ) except Exception as exc: # noqa: BLE001 - error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job") - if _should_retry(job=job, error=error, settings=runtime_settings): - await _finalize_retry( - job=job, - services=services, - error=error, - settings=runtime_settings, - session=session, - ) - logger.warning( - "Job retried operation=worker.process_job " - "job_id=%s document_id=%s retry_count=%s error_id=%s category=%s", - job.id, - document.id, - job.retry_count, - error.error_id, - error.category.value, - ) - return job + match exc: + case AppError() as error: + pass + case _: + error = classify_unexpected_error(exc, operation="worker.process_job") - await _finalize_failed(job=job, services=services, error=error, session=session) + job = await _finalize_failed(job=job, services=services, error=error, session=session) logger.error( "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s", job.id, @@ -131,14 +107,10 @@ async def process_next_queued_job( if job is None: return False - await process_job(job=job, services=services, settings=settings, session=session) + await advance_job(job=job, services=services, settings=settings, session=session) return True -def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool: - return error.retriable and job.retry_count < settings.worker_max_retries - - async def _finalize_transcribed( *, job: Job, diff --git a/src/transcription/worker.py b/src/transcription/worker.py index c80ccaa..e4ef785 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -21,7 +21,7 @@ from .services import ServiceBundle from .services.documents import DocumentService from .services.jobs import JobService from .services.transcription import TranscriptionService -from .services.workflows import process_job +from .services.workflows import advance_job from .services.workflows import process_next_queued_job as process_next_queued_job_workflow logger = logging.getLogger(__name__) @@ -37,7 +37,7 @@ async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Ev with handle_worker_exceptions(): async with _get_queue_item(queue) as job_id: job = await service.read_job(job_id) - asyncio.create_task(process_job(job=job, services=ServiceBundle())) + asyncio.create_task(advance_job(job=job, services=ServiceBundle())) @contextmanager