diff --git a/.github/instructions/services.instructions.md b/.github/instructions/services.instructions.md index 0a711b6..9b942ab 100644 --- a/.github/instructions/services.instructions.md +++ b/.github/instructions/services.instructions.md @@ -33,14 +33,12 @@ applyTo: 'src/transcription/services/*.py' ## Transaction Finalization -When a service method accepts an optional `session` kwarg, write methods must follow this rule: +When a service method accepts an optional `session` kwarg, write methods must use `self._finalize` to finalize the transaction properly according to whether or not they are sharing a session. - If `session` is `None`: the method owns the transaction and should `commit()`. - If `session` is provided: the method must **not** commit; it should `flush()` so IDs and FK values are available to the caller's transaction. - Use `refresh()` on returned ORM objects when the caller needs DB-populated values (defaults, triggers, merged state). -Prefer implementing this once in `ServiceBase` (for example a `_finalize_write(...)` helper) so CRUD methods stay small and consistent. - Recommended helper behavior: - Inputs: active session object, original `session` arg (or a boolean ownership flag), and an optional list of objects to refresh. @@ -48,6 +46,30 @@ Recommended helper behavior: This keeps orchestration functions atomic: they can pass one shared session across multiple services and commit exactly once at the workflow boundary. +## Workflow Transaction Boundaries + +For multi-step job lifecycles (for example queued transcription jobs), orchestration functions must use explicit transaction phases. + +Required boundary model: + +- **Transaction A (claim):** transition `JobStatus.QUEUED -> JobStatus.PROCESSING` and commit immediately. +- Perform provider/network work **outside** database transactions. +- **Transaction B (terminal success):** write transcript content and set `JobStatus.TRANSCRIBED` in the same shared-session commit. +- **Transaction B (terminal failure):** write transcript error detail and set `JobStatus.FAILED` in the same shared-session commit. +- **Transaction C (retry path):** write transcript error detail, increment retry count, and set `JobStatus.QUEUED` in one shared-session commit. + +Atomicity rules: + +- Never commit transcript updates separately from the paired terminal/retry job status change. +- Terminal state (`TRANSCRIBED` or `FAILED`) and transcript row changes must succeed or roll back together. +- Retry persistence (`QUEUED` + retry increment + error detail) must succeed or roll back together. + +Separation of concerns: + +- Worker modules should stay lightweight and delegate lifecycle transitions to service/workflow orchestration functions. +- 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. + # Service Composition Some operations, like uploading a picutre, require modifications to multiple tables, which can be done by composing methods from the service object into a separate function. diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index 70ea616..a75e727 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -1,4 +1,6 @@ from collections.abc import Sequence +from datetime import UTC +from datetime import datetime from uuid import UUID from sqlalchemy.orm import selectinload @@ -96,10 +98,43 @@ class JobService(ServiceBase): session: AsyncSession | None = None, ) -> Job: """Mark a job with a new status.""" + return await self.update_job_state(job_id=job_id, status=status, session=session) + + async def update_job_state( + self, + *, + job_id: UUID, + status: JobStatus, + retry_count_increment: int = 0, + session: AsyncSession | None = None, + ) -> Job: + """Update a job's lifecycle fields. + + When ``session`` is provided, this method flushes so callers can commit + once at an orchestration boundary. + """ async with self._session_scope(session) as _session: job = await _session.get(Job, job_id) if job is None: raise ValueError(f"Job with id {job_id} not found") job.status = status + if retry_count_increment: + job.retry_count += retry_count_increment + job.updated_at = datetime.now(UTC) await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job + + async def read_next_queued_job( + self, + *, + session: AsyncSession | None = None, + ) -> Job | None: + """Read the next queued job ordered by creation time.""" + async with self._session_scope(session) as _session: + query = ( + select(Job) + .options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] + .where(Job.status == JobStatus.QUEUED) + .order_by(Job.created_at) # pyright: ignore[reportArgumentType] + ) + return (await _session.exec(query)).first() diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 6a6e1c8..00af127 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -10,6 +10,7 @@ from uuid import UUID from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import selectinload +from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from transcription.config import Settings @@ -109,6 +110,36 @@ class TranscriptionService(ServiceBase): ) await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session) + async def upsert_transcript_by_job( + self, + *, + job_id: UUID, + text: str | None, + error_detail: str | None, + provider: str | None = None, + prompt_name: str = DEFAULT_PROMPT_FILE, + session: AsyncSession | None = None, + ) -> Transcript: + """Create or update a transcript for a job id.""" + async with self._session_scope(session) as _session: + transcript = (await _session.exec(select(Transcript).where(Transcript.job_id == job_id))).first() + if transcript is None: + transcript = Transcript( + job_id=job_id, + provider=provider or self.settings.provider.value, + prompt_name=prompt_name, + ) + + transcript.text = text + transcript.error_detail = error_detail + if provider is not None: + transcript.provider = provider + transcript.prompt_name = prompt_name + + _session.add(transcript) + await self._finalize(session=_session, caller_session=session, refresh=(transcript,)) + return transcript + async def transcribe_document_image( image_path: str | Path, diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 5564cc0..2ab1da2 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -1,14 +1,20 @@ +import asyncio import logging -from functools import partial 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 from ..models import JobStatus +from ..providers import TranscriptionResult from . import ServiceBundle -from .transcription import TranscriptionError +from .transcription import DEFAULT_PROMPT_FILE +from .transcription import transcribe_document_image logger = logging.getLogger(__name__) @@ -17,18 +23,19 @@ async def advance_job( job: Job, services: ServiceBundle, settings: Settings | None = None, - session: AsyncSession | None = None, ) -> Job | None: - """Process a single job using the service bundle.""" + """Advance a single job by lifecycle status.""" settings = settings or get_settings() match job.status: case JobStatus.QUEUED: - processed_job = await _process_queued(job, services, settings=settings, session=session) + return await process_job(job=job, services=services, settings=settings) case JobStatus.FAILED: if job.retry_count < settings.worker_max_retries: - job.retry_count += 1 - job.status = JobStatus.QUEUED - logger.info(f"Queuing job {job.id} for retry {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, + ) else: logger.error(f"Job {job.id} has failed and reached max retries.") return @@ -36,33 +43,227 @@ async def advance_job( return -async def _process_queued( +async def process_job( + *, job: Job, services: ServiceBundle, settings: Settings | None = None, session: AsyncSession | None = None, ) -> Job | None: - """Process a queued job using the service bundle.""" - settings = settings or get_settings() - job_id = job.id - updater = partial(services.jobs.mark_job_status, job_id, session=session) - + """Process a queued job with workflow-owned transaction boundaries.""" + runtime_settings = settings or get_settings() if job.status != JobStatus.QUEUED: logger.warning(f"Job {job.id} is not queued. Current status: {job.status}") return - job = await updater(JobStatus.PROCESSING) - try: - await services.transcriptions.transcribe_document(job.document.file_path, job.id) - await updater(JobStatus.TRANSCRIBED) - except TranscriptionError as exc: - job.status = JobStatus.FAILED - job.error_message = str(exc) + # Transaction A: claim job for processing. + if session is None: + job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING) + else: + 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 document_id=%s error=%s", + "Job failed operation=worker.process_job job_id=%s error_id=%s category=%s", job.id, - job.document.id, - exc, + error.error_id, + error.category.value, ) - finally: return job + + try: + result = await transcribe_document_image(document.file_path) + 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, + document.id, + 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 + + 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, + document.id, + error.error_id, + error.category.value, + ) + return job + + +async def process_next_queued_job( + *, + services: ServiceBundle, + settings: Settings | None = None, + session: AsyncSession | None = None, +) -> bool: + """Process the next queued job if one exists.""" + job = await services.jobs.read_next_queued_job(session=session) + if job is None: + return False + + await process_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, + services: ServiceBundle, + result: TranscriptionResult, + session: AsyncSession | None = None, +) -> Job: + """Transaction B: transcript + TRANSCRIBED in one commit.""" + if session is None: + async with services.jobs._session_scope() as local_session: + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=result.text, + error_detail=None, + provider=result.provider, + prompt_name=result.prompt_name, + session=local_session, + ) + updated_job = await services.jobs.mark_job_status( + job.id, + JobStatus.TRANSCRIBED, + session=local_session, + ) + await local_session.commit() + return updated_job + + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=result.text, + error_detail=None, + provider=result.provider, + prompt_name=result.prompt_name, + session=session, + ) + updated_job = await services.jobs.mark_job_status( + job.id, + JobStatus.TRANSCRIBED, + session=session, + ) + await session.commit() + return updated_job + + +async def _finalize_retry( + *, + job: Job, + services: ServiceBundle, + error: AppError, + settings: Settings, + session: AsyncSession | None = None, +) -> Job: + """Transaction C: transcript error + QUEUED + retry increment in one commit.""" + if session is None: + async with services.jobs._session_scope() as local_session: + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=None, + error_detail=format_error_detail(error), + prompt_name=DEFAULT_PROMPT_FILE, + session=local_session, + ) + updated_job = await services.jobs.update_job_state( + job_id=job.id, + status=JobStatus.QUEUED, + retry_count_increment=1, + session=local_session, + ) + await local_session.commit() + else: + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=None, + error_detail=format_error_detail(error), + prompt_name=DEFAULT_PROMPT_FILE, + session=session, + ) + updated_job = await services.jobs.update_job_state( + job_id=job.id, + status=JobStatus.QUEUED, + retry_count_increment=1, + session=session, + ) + await session.commit() + + if settings.worker_retry_backoff_seconds > 0: + await asyncio.sleep(settings.worker_retry_backoff_seconds) + return updated_job + + +async def _finalize_failed( + *, + job: Job, + services: ServiceBundle, + error: AppError, + session: AsyncSession | None = None, +) -> Job: + """Transaction B: transcript error + FAILED in one commit.""" + if session is None: + async with services.jobs._session_scope() as local_session: + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=None, + error_detail=format_error_detail(error), + prompt_name=DEFAULT_PROMPT_FILE, + session=local_session, + ) + updated_job = await services.jobs.mark_job_status( + job.id, + JobStatus.FAILED, + session=local_session, + ) + await local_session.commit() + return updated_job + + await services.transcriptions.upsert_transcript_by_job( + job_id=job.id, + text=None, + error_detail=format_error_detail(error), + prompt_name=DEFAULT_PROMPT_FILE, + session=session, + ) + updated_job = await services.jobs.mark_job_status( + job.id, + JobStatus.FAILED, + session=session, + ) + await session.commit() + return updated_job diff --git a/src/transcription/worker.py b/src/transcription/worker.py index dd10347..c80ccaa 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -8,30 +8,21 @@ from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from contextlib import contextmanager from contextlib import suppress -from datetime import UTC -from datetime import datetime from uuid import UUID from sqlalchemy.ext.asyncio import async_sessionmaker -from sqlalchemy.orm import selectinload -from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession -from transcription.config import Settings -from transcription.config import get_settings from transcription.db import get_session from transcription.errors import AppError -from transcription.errors import ErrorCategory from transcription.errors import classify_unexpected_error -from transcription.errors import format_error_detail -from transcription.models import Job -from transcription.models import JobStatus -from transcription.models import Transcript -from transcription.services.transcription import transcribe_document_image 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 process_next_queued_job as process_next_queued_job_workflow logger = logging.getLogger(__name__) @@ -108,121 +99,17 @@ async def process_next_queued_job( Returns True when a job was processed, False when no queued job exists. """ + if session_factory is None: + services = ServiceBundle() + else: + services = ServiceBundle( + documents=DocumentService(session_factory=session_factory), + jobs=JobService(session_factory=session_factory), + transcriptions=TranscriptionService(session_factory=session_factory), + ) + if session is None: async with get_session(session_factory=session_factory) as local_session: - return await _process_next_queued_job(session=local_session) - return await _process_next_queued_job(session=session) + return await process_next_queued_job_workflow(services=services, session=local_session) - -async def _process_next_queued_job(*, session: AsyncSession) -> bool: - query = ( - select(Job) - .options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] - .where(Job.status == JobStatus.QUEUED) - .order_by(Job.created_at) # pyright: ignore[reportArgumentType] - ) - job = (await session.exec(query)).first() - - if job is None: - return False - - logger.info("Picked queued job operation=worker.pick job_id=%s", job.id) - job.status = JobStatus.PROCESSING - job.updated_at = datetime.now(UTC) - session.add(job) - await session.commit() - await session.refresh(job) - - 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(session=session, job=job, error=error) - 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 True - - try: - # Provider SDK calls are synchronous and should not block the event loop. - result = await asyncio.to_thread(transcribe_document_image, document.file_path) - await _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None) - job.status = JobStatus.TRANSCRIBED - job.updated_at = datetime.now(UTC) - session.add(job) - await session.commit() - logger.info( - "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s", - job.id, - document.id, - result.provider, - ) - except Exception as exc: - error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job") - settings = get_settings() - if _should_retry(job=job, error=error, settings=settings): - await _requeue_for_retry(session=session, job=job, error=error, settings=settings) - 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, - ) - else: - await _finalize_failed_job(session=session, job=job, error=error) - logger.exception( - "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s", - job.id, - document.id, - error.error_id, - error.category.value, - ) - - return True - - -async def _upsert_transcript( - *, session: AsyncSession, job_id, text: str | None, error_detail: str | None -) -> Transcript: - transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job_id))).first() - if transcript is None: - transcript = Transcript(job_id=job_id) - - transcript.text = text - transcript.error_detail = error_detail - session.add(transcript) - await session.commit() - await session.refresh(transcript) - return transcript - - -def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool: - return error.retriable and job.retry_count < settings.worker_max_retries - - -async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None: - await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) - job.retry_count += 1 - job.status = JobStatus.QUEUED - job.updated_at = datetime.now(UTC) - session.add(job) - await session.commit() - if settings.worker_retry_backoff_seconds > 0: - await asyncio.sleep(settings.worker_retry_backoff_seconds) - - -async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None: - await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) - job.status = JobStatus.FAILED - job.updated_at = datetime.now(UTC) - session.add(job) - await session.commit() + return await process_next_queued_job_workflow(services=services, session=session)