fix: release db sessions during long-running jobs

backend/config.py

@@ -192,6 +192,26 @@ class Settings(BaseSettings):
     # is rate-limited; 3 has been safe in UAT). Env: MOODBOARD_IMAGE_CONCURRENCY.
     moodboard_image_concurrency: int = 3
 
+    # ── Postgres connection pool (UAT pool-exhaustion hardening) ──
+    # QueuePool knobs applied in database.py for non-SQLite engines (SQLite
+    # keeps its own default pool). Defaults preserve the historical QueuePool
+    # shape (5 connections + 10 overflow = 15 max) — the real fix for the UAT
+    # "QueuePool limit reached" timeouts is the session-lifetime work in the
+    # long-running jobs (no transaction held across external awaits), not a
+    # bigger pool. Raise these in .env only after that proves insufficient.
+    # Env: DB_POOL_SIZE / DB_MAX_OVERFLOW / DB_POOL_TIMEOUT / DB_POOL_RECYCLE / DB_POOL_USE_LIFO.
+    db_pool_size: int = 5
+    db_max_overflow: int = 10
+    db_pool_timeout: float = 30.0
+    # Recycle connections every 30 min — comfortably under Postgres's default
+    # idle timeout so a pooled connection that sat idle is never reaped by the
+    # server mid-checkout (pool_pre_ping handles the transient case; recycling
+    # avoids the churn entirely).
+    db_pool_recycle: int = 1800
+    # LIFO checkout keeps the most-recently-used connection hot under load,
+    # which avoids connect-storms when the pool is contended.
+    db_pool_use_lifo: bool = True
+
     # ── Deployment ──
     cors_origins: str = ""           # comma-separated extra frontend origins (prod / LAN)
     enable_scraper: bool = True      # set false in pilot deploy (no browser deps needed)

backend/database.py

@@ -1,21 +1,53 @@
+from contextlib import contextmanager
+
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, DeclarativeBase
+
 from config import settings
 
-# `check_same_thread=False` is required for SQLite (FastAPI serves requests from a
-# thread pool) but is NOT a valid argument for the PostgreSQL driver (psycopg2) —
-# passing it unconditionally makes every Postgres connection raise TypeError.
-# Only apply it when the URL is a SQLite one.
-connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
 
-engine = create_engine(
-    settings.database_url,
+def _pool_kwargs(url: str) -> dict:
+    """QueuePool settings for non-SQLite engines.
+
+    SQLite keeps its own default pool (SingletonThreadPool for file DBs) which
+    does not accept QueuePool kwargs — and has no pool exhaustion problem since
+    it is single-connection-per-thread. Values are clamped so a bad env value
+    can never shrink the pool to zero or make the timeout negative.
+    """
+    if url.startswith("sqlite"):
+        return {}
+    return {
+        "pool_size": max(1, int(settings.db_pool_size)),
+        "max_overflow": max(0, int(settings.db_max_overflow)),
+        "pool_timeout": max(0.0, float(settings.db_pool_timeout)),
+        "pool_recycle": max(0, int(settings.db_pool_recycle)),
+        "pool_use_lifo": bool(settings.db_pool_use_lifo),
+    }
+
+
+def build_engine(url: str | None = None, **kwargs):
+    """Create the app engine with the configured pool settings.
+
+    ``url`` defaults to settings.database_url. Extra kwargs (e.g. pool_size for
+    tests) override the configured values. `check_same_thread=False` is applied
+    only for SQLite (FastAPI serves requests from a thread pool); passing it to
+    the PostgreSQL driver (psycopg2) raises TypeError.
+    """
+    url = url or settings.database_url
+    connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {}
+    return create_engine(
+        url,
         connect_args=connect_args,
-    # Ping the connection before checkout so idle PostgreSQL connections that were
-    # reaped by the server get transparently replaced instead of raising
+        # Ping the connection before checkout so idle PostgreSQL connections that
+        # were reaped by the server get transparently replaced instead of raising
         # "server closed the connection unexpectedly". No-op for SQLite.
         pool_pre_ping=True,
-)
+        **_pool_kwargs(url),
+        **kwargs,
+    )
+
+
+engine = build_engine()
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 
@@ -23,6 +55,23 @@ class Base(DeclarativeBase):
     pass
 
 
+@contextmanager
+def db_session():
+    """Short-lived session for background jobs and scripts.
+
+    Yields a Session that is ALWAYS closed (rolling back any uncommitted
+    transaction) when the block exits — the job must not hold a Session across
+    long external awaits (Gemini/embedding/HTTP calls) because the session's
+    open transaction pins a pooled connection for the whole call. Prefer this
+    over ``db = SessionLocal()`` in long-running code.
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
 def get_db():
     db = SessionLocal()
     try:

backend/routers/ingest.py

@@ -171,6 +171,10 @@ async def _run_ingest_job(
 
             try:
                 job["current_page_status"] = "tagging"
+                # Close the read transaction opened by the existence check above
+                # BEFORE the (potentially slow, retrying) Mistral tagging call —
+                # an open transaction pins a pooled connection for the whole call.
+                db.commit()
                 tag_result = await _tag_with_retry(str(page_path), i)
                 auto_save = tag_result["auto_save"]
                 needs_review = tag_result["needs_review"]

backend/services/moodboard_pdf_service.py

@@ -1092,6 +1092,10 @@ def _match_library_per_item(brief: dict, embedding_svc: EmbeddingService | None,
         return {}
 
     item_keys = list(queries.keys())
+    # Release the read transaction opened by the _fabric_status lookups above
+    # BEFORE the blocking embedding call — otherwise the pooled connection stays
+    # checked out for the entire embed (seconds to minutes of Gemini latency).
+    db.commit()
     try:
         embeddings = embedding_svc.embed_texts(list(queries.values()))
     except Exception as e:
@@ -1129,6 +1133,10 @@ async def run_decode_job(decode_id: str, pdf_bytes: bytes, embedding_svc: Embedd
             logger.error(f"Moodboard decode {decode_id} not found")
             return
 
+        # End the read transaction before the (potentially slow) PDF extraction:
+        # an open transaction pins the pooled connection for the whole stage.
+        db.commit()
+
         images_dir = Path(settings.images_path)
         page_paths: list = []
         page_texts: list[str] = []
@@ -1378,6 +1386,10 @@ async def run_synthesis_job(decode_id: str, embedding_svc: EmbeddingService | No
         with _stage_timer(decode, db, "ground") as st:
             try:
                 grounding = _build_sapl_grounding(brief, db)
+                # Release the read transaction opened by _build_sapl_grounding
+                # BEFORE the blocking deepseek call — otherwise the pooled
+                # connection stays checked out for the whole LLM round-trip.
+                db.commit()
                 grounding["summary"] = await _ground_summary(grounding, decode.buyer_name, decode.season)
             except Exception as e:
                 logger.warning(f"Moodboard synthesis {decode_id}: SAPL grounding failed: {e}")
@@ -1463,6 +1475,9 @@ async def run_synthesis_job(decode_id: str, embedding_svc: EmbeddingService | No
     except Exception as e:
         logger.error(f"Moodboard synthesis {decode_id} failed: {e}", exc_info=True)
         try:
+            # Roll back the failed transaction first — re-querying a session
+            # left in a failed state raises PendingRollbackError.
+            db.rollback()
             d = db.query(MoodboardDecode).filter(MoodboardDecode.id == decode_id).first()
             if d is not None:
                 if d.status != "completed":
@@ -1499,6 +1514,11 @@ async def _run_visuals_job(decode_id: str, steer: dict) -> None:
         vt = set(steer.get("visual_types") or ["concepts", "prints", "fabrics"])
         wants_garments = bool(steer.get("generate_visuals", True)) and "concepts" in vt
 
+        # End the read transaction from the decode lookup BEFORE the external
+        # deepseek/image calls — an open transaction pins the pooled connection
+        # for the whole stage.
+        db.commit()
+
         # Per-hero-piece construction specs (deepseek) — drafted here so the
         # async path returns the brief to the designer ASAP.
         specs: list = []
@@ -1553,6 +1573,9 @@ async def _run_visuals_job(decode_id: str, steer: dict) -> None:
     except Exception as e:
         logger.error(f"Moodboard visuals {decode_id} failed: {e}", exc_info=True)
         try:
+            # Roll back the failed transaction first — re-querying a session
+            # left in a failed state raises PendingRollbackError.
+            db.rollback()
             d = db.query(MoodboardDecode).filter(MoodboardDecode.id == decode_id).first()
             if d:
                 d.visuals_status = "failed"

backend/services/stitch_service.py

@@ -393,6 +393,11 @@ async def _enrich_from_integration(db: Session) -> dict:
             # this order already has owners — nothing new to learn this cycle.
             if not fresh and any(m.owner_name for m in milestones):
                 continue
+            # Release the read transaction (and its pooled connection) BEFORE the
+            # per-order HTTP call + pacing sleep: with dozens of orders per cycle,
+            # holding the transaction across these awaits pins one connection for
+            # the whole enrichment loop and exhausts the pool under concurrency.
+            db.commit()
             owners = await _fetch_style_tna_owners(ic, order.catalog_style_id)
             owner_api_calls += 1
             if not owners:

backend/tests/test_embedding_optimization.py

@@ -30,6 +30,11 @@ def _zero() -> list[float]:
     return [0.0] * EMBEDDING_DIM
 
 
+# _match_library_per_item now ends its read transaction (db.commit()) before the
+# embedding call; tests that stub out enrichment pass a minimal session stand-in.
+_STUB_DB = SimpleNamespace(commit=lambda: None)
+
+
 class FakeEmbedModels:
     """Mimics client.models.embed_content: accepts a str (per-item path) or a
     list of Content objects (batch path) and returns a configurable response
@@ -321,7 +326,7 @@ def test_match_library_per_item_uses_single_batched_query(monkeypatch, set_knobs
         "print_direction": {"families": ["Florals"], "key_patterns": []},
         "key_silhouettes": [{"name": "A-Line", "description": "flared skirt"}],
     }
-    results = mps._match_library_per_item(brief, svc, db=None)
+    results = mps._match_library_per_item(brief, svc, db=_STUB_DB)
 
     expected_keys = [
         "fabric_direction.fancies.0",
@@ -363,7 +368,7 @@ def test_match_library_per_item_skips_zero_vector_slots(monkeypatch, set_knobs):
         "print_direction": {},
         "key_silhouettes": [],
     }
-    results = mps._match_library_per_item(brief, svc, db=None)
+    results = mps._match_library_per_item(brief, svc, db=_STUB_DB)
     # Zero-vector slot produced no garbage match; the healthy item still matched.
     assert "fabric_direction.fancies.0" in results
     assert "fabric_direction.fancies.1" not in results

backend/tests/test_pool_session_release.py

@@ -0,0 +1,182 @@
+"""Focused tests for the UAT pool-exhaustion hardening.
+
+These verify the session-lifetime rule: long-running jobs must release their
+DB transaction (``db.commit()``) BEFORE an external await (HTTP/LLM/embedding),
+so the pooled connection is returned for the duration of the call instead of
+being pinned for the whole stage.
+"""
+import asyncio
+import types
+
+import pytest
+from sqlalchemy import text
+
+from services import stitch_service as svc
+from database import _pool_kwargs, build_engine
+
+
+# ── helpers ────────────────────────────────────────────────────────────────
+
+class _FakeQuery:
+    """Minimal chainable query stub: filter() is a no-op, all() returns rows."""
+
+    def __init__(self, rows):
+        self._rows = rows
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def all(self):
+        return self._rows
+
+
+class _RecSession:
+    """Records every commit() so a test can assert WHERE the commits happened."""
+
+    def __init__(self, orders, milestones):
+        self._orders = orders
+        self._milestones = milestones
+        self.commits = 0
+        self.commits_at_owner_fetch = None  # set by the fake owner fetch
+
+    def query(self, model):
+        if model is svc.StitchOrder:
+            return _FakeQuery(self._orders)
+        return _FakeQuery(self._milestones)  # StitchMilestone (single-order tests)
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        pass
+
+
+async def _async_noop(*args, **kwargs):
+    return None
+
+
+# ── _enrich_from_integration: transaction released before external await ───
+
+def _order(plan_id="P1", style_id=None):
+    return types.SimpleNamespace(
+        id="O1", plan_id=plan_id, catalog_style_id=style_id,
+        order_qty=None, buyer_id=None, buyer_name=None, last_synced=None,
+    )
+
+
+def _milestone(task_name="A1", owner_name=""):
+    return types.SimpleNamespace(
+        order_id="O1", task_name=task_name, owner_name=owner_name, owner_id="",
+    )
+
+
+def test_enrich_commits_before_owner_fetch_await(monkeypatch):
+    """The read transaction must be committed before the per-order HTTP await."""
+    order = _order()
+    session = _RecSession([order], [_milestone()])
+
+    async def fake_reference(ic):
+        return {"P1": {"style_id": "S1", "qty": 100, "buyer_id": "B1"}}, {"B1": "Zara"}, True
+
+    async def fake_owners(ic, style_id):
+        # This runs while the coroutine is awaited — assert the transaction was
+        # already released (at least one commit happened) before we got here.
+        session.commits_at_owner_fetch = session.commits
+        return {"A1": {"name": "X", "id": "1"}}
+
+    monkeypatch.setattr(svc, "_get_integration_reference", fake_reference)
+    monkeypatch.setattr(svc, "_fetch_style_tna_owners", fake_owners)
+    monkeypatch.setattr(asyncio, "sleep", _async_noop)
+    monkeypatch.setattr(svc.settings, "stitch_api_key", "test-key")
+
+    result = asyncio.run(svc._enrich_from_integration(session))
+
+    # The owner fetch must have observed a commit BEFORE the await — otherwise
+    # the transaction (and pooled connection) was held across the HTTP call.
+    assert session.commits_at_owner_fetch is not None
+    assert session.commits_at_owner_fetch >= 1
+    # The loop still persists everything (final commit after the loop).
+    assert session.commits >= 2
+    # Behaviour preserved: enrichment applied and reported.
+    assert order.catalog_style_id == "S1"
+    assert order.buyer_name == "Zara"
+    assert result == {"enriched_orders": 1, "owners_set": 1, "owner_api_calls": 1}
+
+
+def test_enrich_warm_cache_skips_owner_fetch(monkeypatch):
+    """TTL-warm + already-owned orders make zero HTTP calls and no mid-loop commit."""
+    order = _order()
+    session = _RecSession([order], [_milestone(owner_name="X")])
+    owner_fetch_calls = []
+
+    async def fake_reference(ic):
+        return {"P1": {"style_id": "S1", "qty": 100, "buyer_id": "B1"}}, {"B1": "Zara"}, False
+
+    async def fake_owners(ic, style_id):
+        owner_fetch_calls.append(style_id)
+        return {}
+
+    monkeypatch.setattr(svc, "_get_integration_reference", fake_reference)
+    monkeypatch.setattr(svc, "_fetch_style_tna_owners", fake_owners)
+    monkeypatch.setattr(asyncio, "sleep", _async_noop)
+    monkeypatch.setattr(svc.settings, "stitch_api_key", "test-key")
+
+    result = asyncio.run(svc._enrich_from_integration(session))
+
+    assert owner_fetch_calls == []
+    assert session.commits == 1  # only the final commit
+    assert result["owners_set"] == 0
+
+
+# ── pool knobs (database.py) ───────────────────────────────────────────────
+
+def test_pool_kwargs_sqlite_is_empty():
+    # SQLite keeps its own pool and must NOT receive QueuePool kwargs.
+    assert _pool_kwargs("sqlite:///tmp/x.db") == {}
+    assert _pool_kwargs("sqlite:///:memory:") == {}
+
+
+def test_pool_kwargs_postgres_uses_configured_values():
+    kw = _pool_kwargs("postgresql://u:p@localhost/db")
+    assert kw["pool_size"] == 5
+    assert kw["max_overflow"] == 10
+    assert kw["pool_timeout"] == 30.0
+    assert kw["pool_recycle"] == 1800
+    assert kw["pool_use_lifo"] is True
+
+
+def test_pool_kwargs_clamps_bad_env_values(monkeypatch):
+    from config import settings
+    monkeypatch.setattr(settings, "db_pool_size", 0)
+    monkeypatch.setattr(settings, "db_max_overflow", -5)
+    monkeypatch.setattr(settings, "db_pool_timeout", -1)
+    monkeypatch.setattr(settings, "db_pool_recycle", -3)
+    kw = _pool_kwargs("postgresql://u:p@localhost/db")
+    assert kw["pool_size"] == 1     # never zero
+    assert kw["max_overflow"] == 0  # never negative
+    assert kw["pool_timeout"] == 0.0
+    assert kw["pool_recycle"] == 0
+
+
+def test_build_engine_postgres_uses_queuepool_with_knobs():
+    engine = build_engine("postgresql://u:p@localhost/sapl")
+    try:
+        pool = engine.pool
+        assert type(pool).__name__ == "QueuePool"
+        assert pool.size() == 5
+        assert pool._max_overflow == 10
+        assert pool._timeout == 30.0
+        assert pool._recycle == 1800
+    finally:
+        engine.dispose()
+
+
+def test_build_engine_kwargs_override_and_sqlite_works(tmp_path):
+    db_path = tmp_path / "test.db"
+    engine = build_engine(f"sqlite:///{db_path}", pool_size=2)
+    try:
+        # SQLite default pool (no QueuePool kwargs forced on it) still connects.
+        with engine.connect() as conn:
+            assert conn.execute(text("SELECT 1")).scalar() == 1
+    finally:
+        engine.dispose()

backend/tests/test_visuals_helpers.py

@@ -256,6 +256,9 @@ class _FakeSession:
     def commit(self):
         pass
 
+    def rollback(self):
+        pass
+
     def close(self):
         pass