diff --git a/apps/services/gateway/gateway/routes/email/automation/cleanup.py b/apps/services/gateway/gateway/routes/email/automation/cleanup.py index 13103919..50cf2cb5 100644 --- a/apps/services/gateway/gateway/routes/email/automation/cleanup.py +++ b/apps/services/gateway/gateway/routes/email/automation/cleanup.py @@ -557,6 +557,13 @@ async def sweep_uncategorized( except HTTPException: summary["error"] = "account not found" return summary + # Close the read transaction the loader just opened before the + # network round-trip. A session left `idle in transaction` across a + # provider call holds ACCESS SHARE on everything it touched, which + # is how a migration's ALTER TABLE ends up queued behind a sweep — + # see the 2026-08-06 write-up. Nothing is pending, so this only + # ends the transaction. + await db.commit() if not await provider.authenticate(): # ABORT — do NOT continue with provider=None. A local-only label # is logged APPLIED but Outlook (categories-authoritative) wipes diff --git a/apps/services/gateway/gateway/routes/email/automation/replyzero.py b/apps/services/gateway/gateway/routes/email/automation/replyzero.py index 25f095bb..2563eb88 100644 --- a/apps/services/gateway/gateway/routes/email/automation/replyzero.py +++ b/apps/services/gateway/gateway/routes/email/automation/replyzero.py @@ -1152,6 +1152,28 @@ async def _maybe_classify_threads(account_id: str) -> None: # CC send. Capped per cycle; threads past the cap are left # UNWRITTEN (not blind-AWAITING) so they retry next cycle. if sent_handled < _REPLY_DETERMINE_CAP: + # End OUR transaction before the model call. + # + # `_mark_thread_replied` opens its own session and spends a + # full LLM determination in it. Ours has been in a + # transaction since the reads above, so without this commit + # it sits `idle in transaction` — holding ACCESS SHARE on + # every table it touched — for the length of that call, + # times up to _REPLY_DETERMINE_CAP consecutive sent threads. + # + # That is the exact shape that took production down on + # 2026-08-06: a session parked mid-LLM-call, a migration's + # ALTER TABLE queued behind its lock, and Postgres's FIFO + # lock queue then stalling every later reader of that table. + # + # The acb_llm wall-clock ceiling and the 600s + # idle_in_transaction_session_timeout bound that damage; + # this removes its cause. The ceiling alone is NOT enough + # here — 40 capped calls idle well past 600s, at which point + # Postgres kills this session and the upserts accumulated + # below are lost with it. Committing first is what makes the + # backstop and this loop safe together. + await db.commit() await _mark_thread_replied(account_id, r.thread_id) sent_handled += 1 # else: overflow — leave it for the next cycle, never guess. diff --git a/apps/services/gateway/gateway/routes/email/automation/runner.py b/apps/services/gateway/gateway/routes/email/automation/runner.py index f257fd0d..74591be0 100644 --- a/apps/services/gateway/gateway/routes/email/automation/runner.py +++ b/apps/services/gateway/gateway/routes/email/automation/runner.py @@ -1503,6 +1503,11 @@ async def _process_past_emails_job( store = get_key_store() creds = json.loads(store.decrypt(acc.credentials_encrypted)) provider = _instantiate_provider(acc.provider, creds) + # End the transaction the SELECT above opened before the network + # round-trip — a session parked `idle in transaction` across a + # provider call is what queues a migration's ALTER TABLE behind + # it (2026-08-06). Nothing pending; this only ends it. + await db.commit() if not await provider.authenticate(): provider = None diff --git a/tests/unit/test_email_reply_zero.py b/tests/unit/test_email_reply_zero.py index 334d4d2b..a34c9590 100644 --- a/tests/unit/test_email_reply_zero.py +++ b/tests/unit/test_email_reply_zero.py @@ -612,3 +612,57 @@ async def fake_upsert(_db, _aid, tid, status, _mid, _mat, reason, **kw): assert cap["preserve_done"] is True # inbound must not clobber DONE assert str(cap["reason"]).endswith("· auto") # fallback tagged for re-check assert det.await_args.kwargs["user_sent_last"] is False + + +async def test_backfill_commits_before_every_model_call() -> None: + """The backfill must not sit `idle in transaction` across an LLM call. + + `_mark_thread_replied` opens its OWN session and spends a full LLM + determination in it. This function's session has been in a transaction since + the reads at the top, so without an explicit commit it sits `idle in + transaction` — holding ACCESS SHARE on every table it touched — for the + length of that call, times up to `_REPLY_DETERMINE_CAP` sent threads. + + That is the shape that took production down on 2026-08-06: a parked session, + a migration's ALTER TABLE queued behind its lock, and Postgres's FIFO lock + queue then stalling every later reader of the table — including the one the + send path makes. The acb_llm ceiling and the 600s + idle_in_transaction_session_timeout bound the damage; committing first + removes the cause. The ceiling alone is NOT sufficient here: 40 capped calls + idle well past 600s, at which point Postgres kills the session and the + upserts accumulated in this loop are lost with it. + + Asserts ORDER, not merely that commit was called — a commit after the model + call would satisfy a presence check while leaving the lock held throughout. + """ + latest = [ + _row("t1", "m1", "me@x.com", "sent"), + _row("t2", "m2", "me@x.com", "sent"), + ] + db = _backfill_db(latest, []) + order: list[str] = [] + + async def _commit() -> None: + order.append("commit") + + async def _mark(*_a, **_kw) -> None: + order.append("model-call") + + db.commit = AsyncMock(side_effect=_commit) + + with patch.object(_rz, "_get_db", AsyncMock(return_value=db)), \ + patch.object(_rz, "_load_assistant_about", + AsyncMock(return_value=("", ""))), \ + patch.object(_rz, "_mark_thread_replied", + AsyncMock(side_effect=_mark)), \ + patch.object(_rz, "_upsert_thread_status", AsyncMock()), \ + patch.object(_rz, "_reconcile_thread_labels", AsyncMock()): + await m._maybe_classify_threads("acc-1") + + assert "model-call" in order, "the sent-thread path never ran" + for i, event in enumerate(order): + if event == "model-call": + assert i > 0 and order[i - 1] == "commit", ( + f"model call at position {i} is not immediately preceded by a " + f"commit — the transaction is held across it. Order: {order}" + )