transaction boundaries

.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.

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()

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,

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
+                return await services.jobs.update_job_state(
-                job.status = JobStatus.QUEUED
+                    job_id=job.id,
-                logger.info(f"Queuing job {job.id} for retry {job.retry_count}/{settings.worker_max_retries}")
+                    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."""
+    """Process a queued job with workflow-owned transaction boundaries."""
-    settings = settings or get_settings()
+    runtime_settings = settings or get_settings()
-    job_id = job.id
-    updater = partial(services.jobs.mark_job_status, job_id, session=session)
-
     if job.status != JobStatus.QUEUED:
         logger.warning(f"Job {job.id} is not queued. Current status: {job.status}")
         return
 
-    job = await updater(JobStatus.PROCESSING)
+    # Transaction A: claim job for processing.
-    try:
+    if session is None:
-        await services.transcriptions.transcribe_document(job.document.file_path, job.id)
+        job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING)
-        await updater(JobStatus.TRANSCRIBED)
+    else:
-    except TranscriptionError as exc:
+        job = await services.jobs.mark_job_status(job.id, JobStatus.PROCESSING, session=session)
-        job.status = JobStatus.FAILED
+        await session.commit()
-        job.error_message = str(exc)
+
+    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,
+            error.error_id,
-            exc,
+            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

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_workflow(services=services, session=local_session)
-    return await _process_next_queued_job(session=session)
 
-
+    return await process_next_queued_job_workflow(services=services, session=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()