From ae4f48036ec5d184d85e12ae67c915f86718c3e7 Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:26:39 +0000 Subject: [PATCH] fix(email): serialize default account mutations --- core/database.py | 15 ++ routes/email_routes.py | 58 ++++- ...est_email_account_default_serialization.py | 212 ++++++++++++++++++ 3 files changed, 281 insertions(+), 4 deletions(-) create mode 100644 tests/test_email_account_default_serialization.py diff --git a/core/database.py b/core/database.py index a9ad90b8b..dee04d0ff 100644 --- a/core/database.py +++ b/core/database.py @@ -430,6 +430,21 @@ class EmailAccount(TimestampMixin, Base): ) +class EmailAccountOwnerLock(Base): + """Durable per-owner mutex for email-account default mutations. + + Row-locking databases serialize mutations by locking this row before they + inspect or stage EmailAccount changes. SQLite uses ``BEGIN IMMEDIATE`` + instead, because it ignores ``SELECT ... FOR UPDATE``; keeping the table in + the shared metadata still makes the non-SQLite path available without a + separate migration. The empty key represents unconfigured/single-user + mode, whose account queries intentionally retain their global scope. + """ + __tablename__ = "email_account_owner_locks" + + owner_key = Column(String, primary_key=True) + + class ModelEndpoint(TimestampMixin, Base): """Admin-configured model endpoints. Models are auto-discovered via /v1/models.""" __tablename__ = "model_endpoints" diff --git a/routes/email_routes.py b/routes/email_routes.py index 3c8e407bd..163b0d367 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -194,6 +194,45 @@ def _coerce_port(value, default): return None, f"Invalid port {value!r}; must be a whole number" +def _lock_email_account_owner_mutation(db, owner: str) -> None: + """Serialize one owner's account/default mutation before reading rows. + + SQLite does not implement row-level ``FOR UPDATE`` locking, so acquire its + process-safe writer reservation up front. Row-locking SQLAlchemy + databases lock a durable owner row. Creating a missing lock row inside a + savepoint safely handles two first-account requests racing for the same + owner without poisoning the outer transaction on a uniqueness conflict. + """ + from core.database import EmailAccountOwnerLock + from sqlalchemy import text + from sqlalchemy.exc import IntegrityError + + if db.get_bind().dialect.name == "sqlite": + db.execute(text("BEGIN IMMEDIATE")) + return + + owner_key = owner or "" + lock_row = db.get(EmailAccountOwnerLock, owner_key, with_for_update=True) + if lock_row is not None: + return + + inserted = False + try: + with db.begin_nested(): + db.add(EmailAccountOwnerLock(owner_key=owner_key)) + db.flush() + inserted = True + except IntegrityError: + # Another transaction created the owner row first. Lock that durable + # row below after its insert commits. + pass + + if not inserted: + db.query(EmailAccountOwnerLock).filter( + EmailAccountOwnerLock.owner_key == owner_key + ).with_for_update().one() + + def _email_tag_owner_aliases(account_id: str | None, owner: str = "") -> list[str]: aliases = [owner or ""] try: @@ -5428,6 +5467,7 @@ def setup_email_routes(): import uuid as _uuid db = SessionLocal() try: + _lock_email_account_owner_mutation(db, owner) q = db.query(EmailAccount).filter(EmailAccount.is_default == True) # noqa: E712 if owner: q = q.filter(EmailAccount.owner == owner) @@ -5552,6 +5592,7 @@ def setup_email_routes(): return {"ok": False, "error": port_err} db = SessionLocal() try: + _lock_email_account_owner_mutation(db, owner) row = EmailAccount( id=_uuid.uuid4().hex, name=name, @@ -5635,24 +5676,32 @@ def setup_email_routes(): from core.database import SessionLocal, EmailAccount db = SessionLocal() try: + _lock_email_account_owner_mutation(db, owner) row = db.get(EmailAccount, account_id) if not row: return {"ok": False, "error": "Account not found"} was_default = bool(row.is_default) db.delete(row) - db.commit() # If the deleted row was default, promote the next-oldest enabled # row owned by THIS user. Without the owner filter we'd promote # another user's account and the deleter would silently inherit # it as their default. if was_default: - promote_q = db.query(EmailAccount).filter(EmailAccount.enabled == True) # noqa: E712 + promote_q = db.query(EmailAccount).filter( + EmailAccount.id != account_id, + EmailAccount.enabled == True, # noqa: E712 + ) if owner: promote_q = promote_q.filter(EmailAccount.owner == owner) - promote = promote_q.order_by(EmailAccount.created_at.asc()).first() + promote = promote_q.order_by( + EmailAccount.created_at.asc(), EmailAccount.id.asc() + ).first() if promote: promote.is_default = True - db.commit() + # Deletion and any replacement promotion are one durable state + # transition, so another worker can never observe or race the old + # split-commit gap. + db.commit() return {"ok": True} finally: db.close() @@ -5869,6 +5918,7 @@ def setup_email_routes(): from core.database import SessionLocal, EmailAccount db = SessionLocal() try: + _lock_email_account_owner_mutation(db, owner) row = db.get(EmailAccount, account_id) if not row: return {"ok": False, "error": "Account not found"} diff --git a/tests/test_email_account_default_serialization.py b/tests/test_email_account_default_serialization.py new file mode 100644 index 000000000..04e1d1f30 --- /dev/null +++ b/tests/test_email_account_default_serialization.py @@ -0,0 +1,212 @@ +"""Regressions for process-safe email-account default mutations. + +The file-backed SQLite fixture uses a fresh connection for every Session. +That exercises the same database lock boundary used by separate web workers, +rather than relying on an in-process Python lock. +""" + +import asyncio +import threading +from unittest import mock + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import NullPool + + +@pytest.fixture +def account_db(tmp_path, monkeypatch): + from core import database as core_db + + engine = create_engine( + f"sqlite:///{tmp_path / 'accounts.db'}", + connect_args={"check_same_thread": False, "timeout": 5}, + poolclass=NullPool, + ) + core_db.Base.metadata.create_all(engine) + factory = sessionmaker( + bind=engine, + autocommit=False, + autoflush=False, + ) + monkeypatch.setattr(core_db, "SessionLocal", factory) + yield factory + engine.dispose() + + +def _endpoint(method, path): + from routes import email_routes + + with mock.patch.object(email_routes, "_start_poller"): + router = email_routes.setup_email_routes() + for route in router.routes: + if route.path == path and method in getattr(route, "methods", set()): + return route.endpoint + raise AssertionError(f"email route not found: {method} {path}") + + +def _seed_account(factory, account_id, owner, *, is_default=False, enabled=True): + from core.database import EmailAccount + + db = factory() + try: + db.add( + EmailAccount( + id=account_id, + owner=owner, + name=account_id, + is_default=is_default, + enabled=enabled, + ) + ) + db.commit() + finally: + db.close() + + +def _rows(factory): + from core.database import EmailAccount + + db = factory() + try: + return [ + (row.id, row.owner, bool(row.is_default)) + for row in db.query(EmailAccount).order_by(EmailAccount.id).all() + ] + finally: + db.close() + + +def _install_lock_pause(monkeypatch, paused_thread_name): + """Pause one worker after acquisition and observe another waiting.""" + from routes import email_routes + + real_lock = email_routes._lock_email_account_owner_mutation + first_acquired = threading.Event() + release_first = threading.Event() + contender_attempted = threading.Event() + contender_acquired = threading.Event() + + def controlled_lock(db, owner): + is_first = threading.current_thread().name == paused_thread_name + if not is_first: + contender_attempted.set() + real_lock(db, owner) + if is_first: + first_acquired.set() + assert release_first.wait(5), "timed out releasing first mutation" + else: + contender_acquired.set() + + monkeypatch.setattr( + email_routes, + "_lock_email_account_owner_mutation", + controlled_lock, + ) + return first_acquired, release_first, contender_attempted, contender_acquired + + +def test_concurrent_first_account_creates_choose_one_default(account_db, monkeypatch): + create_account = _endpoint("POST", "/api/email/accounts") + first_acquired, release_first, attempted, acquired = _install_lock_pause( + monkeypatch, "first-account" + ) + results = {} + + def create(name): + results[name] = asyncio.run( + create_account({"name": name, "is_default": False}, owner="alice") + ) + + first = threading.Thread(target=create, args=("First",), name="first-account") + second = threading.Thread(target=create, args=("Second",), name="second-account") + first.start() + assert first_acquired.wait(5) + second.start() + assert attempted.wait(5) + assert not acquired.wait(0.1), "second session bypassed the database mutation lock" + + release_first.set() + first.join(5) + second.join(5) + + assert not first.is_alive() + assert not second.is_alive() + assert results["First"]["ok"] is True + assert results["Second"]["ok"] is True + defaults = [row for row in _rows(account_db) if row[2]] + assert [(row[1], row[2]) for row in defaults] == [("alice", True)] + assert len(defaults) == 1 + + +def test_delete_promotion_and_set_default_are_one_serial_transition( + account_db, monkeypatch +): + from sqlalchemy.orm import Session as OrmSession + + _seed_account(account_db, "alice-a", "alice", is_default=True) + _seed_account(account_db, "alice-b", "alice") + _seed_account(account_db, "alice-c", "alice") + _seed_account(account_db, "bob-a", "bob", is_default=True) + + delete_account = _endpoint("DELETE", "/api/email/accounts/{account_id}") + set_default = _endpoint("POST", "/api/email/accounts/{account_id}/set-default") + first_acquired, release_first, attempted, acquired = _install_lock_pause( + monkeypatch, "delete-default" + ) + delete_commit_finished = threading.Event() + release_delete_after_commit = threading.Event() + real_commit = OrmSession.commit + results = {} + + def pause_after_delete_commit(session): + real_commit(session) + if ( + threading.current_thread().name == "delete-default" + and not delete_commit_finished.is_set() + ): + delete_commit_finished.set() + assert release_delete_after_commit.wait(5), ( + "timed out releasing delete after its first commit" + ) + + monkeypatch.setattr(OrmSession, "commit", pause_after_delete_commit) + + def delete_old_default(): + results["delete"] = asyncio.run( + delete_account("alice-a", owner="alice") + ) + + def select_new_default(): + results["set"] = asyncio.run( + set_default("alice-c", owner="alice") + ) + + delete_thread = threading.Thread(target=delete_old_default, name="delete-default") + set_thread = threading.Thread(target=select_new_default, name="set-default") + delete_thread.start() + assert first_acquired.wait(5) + set_thread.start() + assert attempted.wait(5) + assert not acquired.wait(0.1), "set-default bypassed the delete transaction" + + release_first.set() + assert delete_commit_finished.wait(5) + # The deletion transaction has committed. Let the contender complete + # before the deleting handler can continue: if promotion were still a + # second commit, it would now run after set-default and recreate two + # defaults deterministically. + assert acquired.wait(5) + set_thread.join(5) + release_delete_after_commit.set() + delete_thread.join(5) + + assert not delete_thread.is_alive() + assert not set_thread.is_alive() + assert results == {"delete": {"ok": True}, "set": {"ok": True}} + assert _rows(account_db) == [ + ("alice-b", "alice", False), + ("alice-c", "alice", True), + ("bob-a", "bob", True), + ]