refactor: extract blocking and sequence retry helpers

Co-authored-by: Copilot App <[email protected]>

src/transcription/runtime_helpers.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable
+from collections.abc import Callable
+from typing import TypeVar
+
+from sqlalchemy.exc import IntegrityError
+
+ResultT = TypeVar("ResultT")
+
+
+async def run_blocking(func: Callable[..., ResultT], /, *args, **kwargs) -> ResultT:
+    """Run blocking CPU/filesystem work on a worker thread."""
+    return await asyncio.to_thread(func, *args, **kwargs)
+
+
+async def insert_with_sequence_retry(
+    *,
+    max_retries: int,
+    operation: Callable[[int], Awaitable[ResultT]],
+    on_conflict: Callable[[int, IntegrityError], None] | None = None,
+) -> ResultT:
+    """Retry a sequence-based insert operation on unique-key conflicts."""
+    if max_retries < 1:
+        raise ValueError("max_retries must be at least 1")
+
+    for retry in range(1, max_retries + 1):
+        try:
+            return await operation(retry)
+        except IntegrityError as exc:
+            if on_conflict is not None:
+                on_conflict(retry, exc)
+            if retry == max_retries:
+                raise
+
+    raise RuntimeError("insert_with_sequence_retry exhausted retries without returning or raising")

src/transcription/services/sources.py

@@ -47,6 +47,7 @@ from transcription.providers import TranscriptionProvider
 from transcription.providers import TranscriptionResult
 from transcription.providers import TransportEvidence
 from transcription.providers import get_transcription_provider
+from transcription.runtime_helpers import insert_with_sequence_retry
 
 from ..db.loading import orm_attribute
 from ..db.loading import selectinload
@@ -539,8 +540,8 @@ class SourceService(ServiceBase):
             software_payload = (
                 request_manifest.software.model_dump(mode="json") if request_manifest is not None else None
             )
-            attempt: ExecutionAttempt | None = None
+
-            for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1):
+            async def _insert_execution_attempt(_attempt_retry: int) -> ExecutionAttempt:
                 latest_attempt_number = (
                     await _session.exec(
                         select(func.max(ExecutionAttempt.attempt_number))
@@ -582,24 +583,25 @@ class SourceService(ServiceBase):
                     if duration_ms is not None
                     else max(0, int((finish_time - start_time).total_seconds() * 1000)),
                 )
-                try:
+                async with _session.begin_nested():
-                    async with _session.begin_nested():
+                    _session.add(candidate)
-                        _session.add(candidate)
+                    await _session.flush()
-                        await _session.flush()
+                return candidate
-                    attempt = candidate
+
-                    break
+            try:
-                except IntegrityError:
+                attempt = await insert_with_sequence_retry(
-                    logger.warning(
+                    max_retries=MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
+                    operation=_insert_execution_attempt,
+                    on_conflict=lambda attempt_retry, _exc: logger.warning(
                         "Execution attempt number conflict job_id=%s source_id=%s retry=%s/%s",
                         job_id,
                         source_id,
                         attempt_retry,
                         MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
-                    )
+                    ),
-                    continue
+                )
-
+            except IntegrityError as exc:
-            if attempt is None:
+                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id) from exc
-                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id)
 
             if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
                 source.raw_transcription = text

src/transcription/services/store.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import hashlib
 import logging
 from collections.abc import Sequence
@@ -16,6 +15,7 @@ from transcription.config import Settings
 from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
+from transcription.runtime_helpers import run_blocking
 
 from ..db.models import Document
 from ..db.models import Job
@@ -399,7 +399,7 @@ async def store_source_file(
     )
     return StoredSourceFile(
         path=stored_path,
-        file_hash=await asyncio.to_thread(_sha256_hexdigest, file_bytes),
+        file_hash=await run_blocking(_sha256_hexdigest, file_bytes),
         file_size_bytes=len(file_bytes),
     )
 

src/transcription/ui/pages/home_page.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Callable
 
 from fastapi import Request
@@ -11,6 +10,7 @@ from nicegui import ui
 
 from transcription.config import Settings
 from transcription.db.models import Photo
+from transcription.runtime_helpers import run_blocking
 from transcription.services.photos import PhotoError
 from transcription.services.photos import PhotosService
 from transcription.ui.components.app_shell import render_navigation_header
@@ -184,7 +184,7 @@ def register_page() -> None:  # noqa: PLR0915
 
             render_home_content()
 
-        homepage_markdown[0] = await asyncio.to_thread(read_homepage_markdown, settings)
+        homepage_markdown[0] = await run_blocking(read_homepage_markdown, settings)
         render_home_content.refresh()
 
     @ui.page("/homepage/edit", title="Edit Home Page")
@@ -278,7 +278,7 @@ def register_page() -> None:  # noqa: PLR0915
             ui.navigate.to("/homepage/edit")
 
         async def save_homepage() -> None:
-            await asyncio.to_thread(
+            await run_blocking(
                 save_homepage_markdown,
                 (markdown_input[0].value if markdown_input[0] is not None else "") or "",
                 settings,
@@ -300,6 +300,6 @@ def register_page() -> None:  # noqa: PLR0915
                 initial_markdown="",
             )
 
-        loaded_markdown = await asyncio.to_thread(read_homepage_markdown, settings)
+        loaded_markdown = await run_blocking(read_homepage_markdown, settings)
         if markdown_input[0] is not None:
             markdown_input[0].value = loaded_markdown

src/transcription/ui/pages/settings_page.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 from uuid import UUID
 
 from nicegui import ui
 
 from transcription.config import Settings
+from transcription.runtime_helpers import run_blocking
 from transcription.services.documents import DocumentService
 from transcription.services.people import PeopleService
 from transcription.services.prompts import PromptStore
@@ -541,8 +541,8 @@ async def _recover_prompt(prompts: PromptStore, name: str) -> None:
 
 
 async def _read_home_page_text(settings: Settings) -> str:
-    return await asyncio.to_thread(read_homepage_markdown, settings=settings)
+    return await run_blocking(read_homepage_markdown, settings=settings)
 
 
 async def _write_home_page_text(settings: Settings, markdown_text: str) -> None:
-    await asyncio.to_thread(save_homepage_markdown, markdown_text, settings=settings)
+    await run_blocking(save_homepage_markdown, markdown_text, settings=settings)