diff --git a/core/database.py b/core/database.py index dee04d0ff..6eb529949 100644 --- a/core/database.py +++ b/core/database.py @@ -5,7 +5,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Optional from urllib.parse import unquote, urlparse -from sqlalchemy import event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, text +from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, inspect, text from sqlalchemy.engine import Engine, make_url from sqlalchemy.types import TypeDecorator from sqlalchemy.ext.declarative import declarative_base, declared_attr @@ -437,14 +437,86 @@ class EmailAccountOwnerLock(Base): 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. + separate migration. The empty key represents the normalized legacy / + unconfigured scope shared by ``owner IS NULL`` and ``owner = ''`` rows. """ __tablename__ = "email_account_owner_locks" owner_key = Column(String, primary_key=True) +_EMAIL_ACCOUNT_DEFAULT_INDEX = "ux_email_accounts_one_default_per_owner" +_EMAIL_ACCOUNT_DEFAULT_INDEX_DDL = { + "sqlite": ( + f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} " + "ON email_accounts (COALESCE(owner, '')) WHERE is_default = 1" + ), + "postgresql": ( + f"CREATE UNIQUE INDEX IF NOT EXISTS {_EMAIL_ACCOUNT_DEFAULT_INDEX} " + "ON email_accounts ((COALESCE(owner, ''))) WHERE is_default IS TRUE" + ), +} + + +# SQLAlchemy cannot express one portable partial, functional index across the +# two supported database families. Register dialect-specific DDL so fresh +# databases get the invariant as part of create_all(); the startup migration +# below installs the same index on existing databases after normalizing legacy +# duplicate rows. +for _dialect_name, _index_ddl in _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.items(): + event.listen( + EmailAccount.__table__, + "after_create", + DDL(_index_ddl).execute_if(dialect=_dialect_name), + ) + + +def lock_email_account_owner_mutations(db, *owners: str) -> None: + """Lock normalized email-account owner scopes in canonical order. + + ``NULL`` and the empty string are one legacy/single-user owner partition, + matching the unique default-account index. SQLite has only a database + writer reservation, while row-locking databases use durable mutex rows. + Sorting all requested owner keys keeps multi-owner operations such as user + rename from deadlocking with another mutation that requests the same keys + in the opposite order. + """ + from sqlalchemy.exc import IntegrityError + + owner_keys = sorted({owner or "" for owner in owners} or {""}) + if db.get_bind().dialect.name == "sqlite": + db.execute(text("BEGIN IMMEDIATE")) + return + + for owner_key in owner_keys: + lock_row = db.get( + EmailAccountOwnerLock, + owner_key, + with_for_update=True, + ) + if lock_row is not None: + continue + + inserted = False + try: + with db.begin_nested(): + db.add(EmailAccountOwnerLock(owner_key=owner_key)) + db.flush() + inserted = True + except IntegrityError: + # A competing transaction created the mutex row first. Once its + # insert commits, lock that durable row before touching accounts. + pass + + if not inserted: + ( + db.query(EmailAccountOwnerLock) + .filter(EmailAccountOwnerLock.owner_key == owner_key) + .with_for_update() + .one() + ) + + class ModelEndpoint(TimestampMixin, Base): """Admin-configured model endpoints. Models are auto-discovered via /v1/models.""" __tablename__ = "model_endpoints" @@ -1827,72 +1899,142 @@ class Integration(TimestampMixin, Base): -def _migrate_seed_email_account(): - """If email_accounts is empty and settings.json has legacy flat imap_host/smtp_host - keys, create a single default account from them so nothing breaks for users who - upgraded. Safe to run repeatedly — it short-circuits once any row exists.""" +def _migrate_email_account_default_invariant(): + """Normalize legacy duplicates and install durable at-most-one enforcement. + + Older databases only had a non-unique ``(owner, is_default)`` lookup index. + Keep the oldest default deterministically in each normalized owner scope, + then add the same partial functional unique index used for fresh schemas. + """ + dialect_name = engine.dialect.name + index_ddl = _EMAIL_ACCOUNT_DEFAULT_INDEX_DDL.get(dialect_name) + if index_ddl is None: + logger.warning( + "Email-account default uniqueness is not available for database " + "dialect %s; mutations remain serialized but are not protected by " + "a database constraint", + dialect_name, + ) + return + try: - with engine.connect() as conn: - tables = [r[0] for r in conn.execute(text( - "SELECT name FROM sqlite_master WHERE type='table' AND name='email_accounts'" - ))] - if "email_accounts" not in tables: - return - existing = conn.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0 - if existing > 0: + with engine.begin() as conn: + if not inspect(conn).has_table(EmailAccount.__tablename__): return + default_rows = conn.execute(text(""" + SELECT id, owner + FROM email_accounts + WHERE is_default IS TRUE + ORDER BY + COALESCE(owner, ''), + CASE WHEN created_at IS NULL THEN 1 ELSE 0 END, + created_at, + id + """)).mappings() + seen_owner_keys = set() + duplicate_ids = [] + for row in default_rows: + owner_key = row["owner"] or "" + if owner_key in seen_owner_keys: + duplicate_ids.append(row["id"]) + else: + seen_owner_keys.add(owner_key) - import json as _json - import uuid as _uuid - from pathlib import Path - settings_file = Path(SETTINGS_FILE) - if not settings_file.exists(): - return - try: - s = _json.loads(settings_file.read_text(encoding="utf-8")) - except Exception: - return + for account_id in duplicate_ids: + conn.execute( + text("UPDATE email_accounts SET is_default = :value WHERE id = :id"), + {"value": False, "id": account_id}, + ) + conn.execute(text(index_ddl)) - imap_host = (s.get("imap_host") or "").strip() - smtp_host = (s.get("smtp_host") or "").strip() - if not imap_host and not smtp_host: - return # nothing to migrate + if duplicate_ids: + logger.warning( + "Normalized %d duplicate default email account(s) before " + "installing %s", + len(duplicate_ids), + _EMAIL_ACCOUNT_DEFAULT_INDEX, + ) + except Exception: + # Starting without the constraint would silently retain the race this + # migration is intended to close. Fail startup so an operator sees and + # can repair an incompatible schema instead of accepting unsafe writes. + logger.exception("Failed to enforce the email-account default invariant") + raise + + +def _migrate_seed_email_account(): + """Atomically seed one legacy default account when no account exists. + + Reading settings is intentionally done before taking the owner mutex. The + decisive emptiness check and insert share one locked transaction, so two + application workers starting together cannot both seed a default row. + """ + import json as _json + import uuid as _uuid + + settings_file = Path(SETTINGS_FILE) + if not settings_file.exists(): + return + try: + s = _json.loads(settings_file.read_text(encoding="utf-8")) + except Exception: + return + + imap_host = (s.get("imap_host") or "").strip() + smtp_host = (s.get("smtp_host") or "").strip() + if not imap_host and not smtp_host: + return + + db = None + try: + if not inspect(engine).has_table(EmailAccount.__tablename__): + return + db = SessionLocal() + lock_email_account_owner_mutations(db, "") + existing = db.execute(text("SELECT COUNT(*) FROM email_accounts")).scalar() or 0 + if existing > 0: + return now = utcnow_naive() - with engine.begin() as conn: - conn.execute(text(""" - INSERT INTO email_accounts - (id, owner, name, is_default, enabled, - imap_host, imap_port, imap_user, imap_password, imap_starttls, - smtp_host, smtp_port, smtp_user, smtp_password, - from_address, created_at, updated_at) - VALUES - (:id, :owner, :name, :is_default, :enabled, - :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls, - :smtp_host, :smtp_port, :smtp_user, :smtp_password, - :from_address, :created_at, :updated_at) - """), { - "id": _uuid.uuid4().hex, - "owner": None, - "name": "Default", - "is_default": True, - "enabled": True, - "imap_host": imap_host, - "imap_port": int(s.get("imap_port") or 993), - "imap_user": s.get("imap_user") or "", - "imap_password": s.get("imap_password") or "", - "imap_starttls": bool(s.get("imap_starttls", True)), - "smtp_host": smtp_host, - "smtp_port": int(s.get("smtp_port") or 465), - "smtp_user": s.get("smtp_user") or "", - "smtp_password": s.get("smtp_password") or "", - "from_address": s.get("email_from") or "", - "created_at": now, - "updated_at": now, - }) - logging.getLogger(__name__).info("Seeded email_accounts 'Default' from settings.json") + db.execute(text(""" + INSERT INTO email_accounts + (id, owner, name, is_default, enabled, + imap_host, imap_port, imap_user, imap_password, imap_starttls, + smtp_host, smtp_port, smtp_user, smtp_password, + from_address, created_at, updated_at) + VALUES + (:id, :owner, :name, :is_default, :enabled, + :imap_host, :imap_port, :imap_user, :imap_password, :imap_starttls, + :smtp_host, :smtp_port, :smtp_user, :smtp_password, + :from_address, :created_at, :updated_at) + """), { + "id": _uuid.uuid4().hex, + "owner": None, + "name": "Default", + "is_default": True, + "enabled": True, + "imap_host": imap_host, + "imap_port": int(s.get("imap_port") or 993), + "imap_user": s.get("imap_user") or "", + "imap_password": s.get("imap_password") or "", + "imap_starttls": bool(s.get("imap_starttls", True)), + "smtp_host": smtp_host, + "smtp_port": int(s.get("smtp_port") or 465), + "smtp_user": s.get("smtp_user") or "", + "smtp_password": s.get("smtp_password") or "", + "from_address": s.get("email_from") or "", + "created_at": now, + "updated_at": now, + }) + db.commit() + logger.info("Seeded email_accounts 'Default' from settings.json") except Exception as e: - logging.getLogger(__name__).warning(f"seed email account migration: {e}") + if db is not None: + db.rollback() + logger.warning("seed email account migration: %s", e) + finally: + if db is not None: + db.close() # WARNING: Foreign-key enforcement is enabled globally for all SQLite connections. @@ -1975,6 +2117,7 @@ def init_db(): _migrate_add_crew_member_id() _migrate_add_assistant_columns() _migrate_add_email_smtp_security() + _migrate_email_account_default_invariant() _migrate_seed_email_account() _migrate_add_calendar_metadata() _migrate_add_calendar_is_utc() diff --git a/routes/auth_routes.py b/routes/auth_routes.py index 5c7a4e04a..c0c370561 100644 --- a/routes/auth_routes.py +++ b/routes/auth_routes.py @@ -345,9 +345,61 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter: # docs, email accounts, tasks, etc. try: from sqlalchemy import func - from core.database import Base, SessionLocal + from core.database import ( + Base, + EmailAccount, + SessionLocal, + lock_email_account_owner_mutations, + ) db = SessionLocal() try: + # Email-account defaults are protected by per-owner mutex rows. + # A rename crosses two owner partitions, so lock both in the + # shared helper's canonical order before inspecting either. + lock_email_account_owner_mutations( + db, old_username, new_username + ) + + source_default_ids = [ + row[0] + for row in ( + db.query(EmailAccount.id) + .filter( + func.lower(EmailAccount.owner) == old_username, + EmailAccount.is_default == True, # noqa: E712 + ) + .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc()) + .all() + ) + ] + destination_default_ids = [ + row[0] + for row in ( + db.query(EmailAccount.id) + .filter( + func.lower(EmailAccount.owner) == new_username, + EmailAccount.is_default == True, # noqa: E712 + ) + .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc()) + .all() + ) + ] + if destination_default_ids: + clear_default_ids = ( + destination_default_ids[1:] + source_default_ids + ) + else: + clear_default_ids = source_default_ids[1:] + if clear_default_ids: + ( + db.query(EmailAccount) + .filter(EmailAccount.id.in_(clear_default_ids)) + .update( + {EmailAccount.is_default: False}, + synchronize_session=False, + ) + ) + for mapper in Base.registry.mappers: model = mapper.class_ if not hasattr(model, "owner"): diff --git a/routes/email_routes.py b/routes/email_routes.py index 163b0d367..b8c2adf57 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -45,6 +45,7 @@ from src.upload_limits import read_upload_limited, EMAIL_COMPOSE_UPLOAD_MAX_BYTE from routes.email_helpers import ( _strip_think, _extract_reply, _apply_email_style_mechanics, require_owner, require_user, _assert_owns_account, + _account_visible_to_owner, _q, _attach_compose_uploads, _cleanup_compose_uploads, _load_settings, _save_settings, _get_email_config, _send_smtp_message, _smtp_security_mode, @@ -194,43 +195,62 @@ 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. +def _lock_email_account_owner_mutation(db, *owners: str) -> None: + """Delegate account/default serialization to the shared DB primitive.""" + from core.database import lock_email_account_owner_mutations - 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 + lock_email_account_owner_mutations(db, *owners) - 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 +def _email_account_owner_scope(query, owner: str): + """Restrict a query to one normalized EmailAccount owner partition.""" + from core.database import EmailAccount + from sqlalchemy import or_ - inserted = False + if owner: + return query.filter(EmailAccount.owner == owner) + return query.filter(or_(EmailAccount.owner == None, EmailAccount.owner == "")) # noqa: E711 + + +def _discover_email_account_mutation_scope(account_id: str, owner: str) -> str: + """Read the initial lock key and fail closed before a mutation session.""" + from core.database import EmailAccount, SessionLocal + + db = SessionLocal() 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 + row = db.get(EmailAccount, account_id) + if row is None or (owner and not _account_visible_to_owner(row, owner)): + raise HTTPException(404, "Account not found") + return row.owner or "" + except HTTPException: + raise + except Exception as exc: + logger.error("Account-owner mutation check failed: %s", exc) + raise HTTPException(503, "Account check failed") + finally: + db.close() - if not inserted: - db.query(EmailAccountOwnerLock).filter( - EmailAccountOwnerLock.owner_key == owner_key - ).with_for_update().one() + +def _lock_and_reload_email_account(db, account_id: str, owner: str, scope: str): + """Lock, reload, and revalidate an account, retrying if its owner moved.""" + from core.database import EmailAccount + + owner_scopes = {scope or ""} + while True: + _lock_email_account_owner_mutation(db, *owner_scopes) + row = db.get(EmailAccount, account_id, populate_existing=True) + if row is None or (owner and not _account_visible_to_owner(row, owner)): + raise HTTPException(404, "Account not found") + + current_scope = row.owner or "" + if current_scope in owner_scopes or db.get_bind().dialect.name == "sqlite": + return row + + # The account changed owner after discovery but before lock acquisition. + # Release the partial lock set and reacquire all observed scopes in the + # shared helper's canonical order, then validate from the database again. + db.rollback() + owner_scopes.add(current_scope) def _email_tag_owner_aliases(account_id: str | None, owner: str = "") -> list[str]: @@ -5469,8 +5489,7 @@ def setup_email_routes(): 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) + q = _email_account_owner_scope(q, owner) row = q.first() if row is None: row = EmailAccount(id=_uuid.uuid4().hex, owner=owner, name="Default", is_default=True, enabled=True) @@ -5496,8 +5515,7 @@ def setup_email_routes(): if data.get("smtp_password"): row.smtp_password = _enc(data["smtp_password"]) clear_q = db.query(EmailAccount).filter(EmailAccount.id != row.id) - if owner: - clear_q = clear_q.filter(EmailAccount.owner == owner) + clear_q = _email_account_owner_scope(clear_q, owner) clear_q.update({EmailAccount.is_default: False}) db.commit() finally: @@ -5619,9 +5637,7 @@ def setup_email_routes(): # the one-default invariant — but scope it to THIS user's accounts, # otherwise creating a default would clear every other user's # default flag too. - scope_q = db.query(EmailAccount) - if owner: - scope_q = scope_q.filter(EmailAccount.owner == owner) + scope_q = _email_account_owner_scope(db.query(EmailAccount), owner) existing_count = scope_q.count() if row.is_default or existing_count == 0: scope_q.update({EmailAccount.is_default: False}) @@ -5672,16 +5688,20 @@ def setup_email_routes(): @router.delete("/accounts/{account_id}") async def delete_email_account(account_id: str, owner: str = Depends(require_user)): - _assert_owns_account(account_id, owner) + initial_scope = _discover_email_account_mutation_scope(account_id, owner) 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"} + row = _lock_and_reload_email_account( + db, account_id, owner, initial_scope + ) + row_scope = row.owner or "" was_default = bool(row.is_default) db.delete(row) + # Flush the removal before staging a replacement default. The + # partial unique index is checked statement-by-statement, and the + # ORM is otherwise free to UPDATE the promoted row before DELETE. + db.flush() # 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 @@ -5691,8 +5711,7 @@ def setup_email_routes(): EmailAccount.id != account_id, EmailAccount.enabled == True, # noqa: E712 ) - if owner: - promote_q = promote_q.filter(EmailAccount.owner == owner) + promote_q = _email_account_owner_scope(promote_q, row_scope) promote = promote_q.order_by( EmailAccount.created_at.asc(), EmailAccount.id.asc() ).first() @@ -5914,19 +5933,18 @@ def setup_email_routes(): @router.post("/accounts/{account_id}/set-default") async def set_default_account(account_id: str, owner: str = Depends(require_user)): - _assert_owns_account(account_id, owner) + initial_scope = _discover_email_account_mutation_scope(account_id, owner) 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"} - # SECURITY: scope the "clear other defaults" sweep to this user's - # accounts so we don't unset another user's default flag. - clear_q = db.query(EmailAccount) - if owner: - clear_q = clear_q.filter(EmailAccount.owner == owner) + row = _lock_and_reload_email_account( + db, account_id, owner, initial_scope + ) + # Scope the sweep to the target row's normalized owner partition; + # this also handles visible legacy NULL/empty-owner accounts. + clear_q = _email_account_owner_scope( + db.query(EmailAccount), row.owner or "" + ) clear_q.update({EmailAccount.is_default: False}) row.is_default = True db.commit() diff --git a/scripts/demo_email/demo_account.py b/scripts/demo_email/demo_account.py index 9555b6791..8a0f1190a 100755 --- a/scripts/demo_email/demo_account.py +++ b/scripts/demo_email/demo_account.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Create/remove the switchable, non-default 'Demo' EmailAccount in Odysseus. +"""Create/remove the switchable 'Demo' EmailAccount in Odysseus. Mirrors the existing local-Dovecot account (localhost:31143, STARTTLS) but points at the throwaway demo@odysseus.local mailbox. Password is stored Fernet-encrypted @@ -20,7 +20,14 @@ from pathlib import Path ROOT = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(ROOT)) -from core.database import SessionLocal, EmailAccount, Base, engine # noqa: E402 +from core.database import ( # noqa: E402 + Base, + EmailAccount, + SessionLocal, + engine, + lock_email_account_owner_mutations, +) +from sqlalchemy import or_ # noqa: E402 from src.secret_storage import encrypt # noqa: E402 NAME = "Demo" @@ -31,18 +38,98 @@ IMAP_PASSWORD = "demodemo" OWNER = "" -def setup() -> int: - Base.metadata.create_all(bind=engine) +def _owner_scope(query, owner: str): + if owner: + return query.filter(EmailAccount.owner == owner) + return query.filter(or_(EmailAccount.owner == None, EmailAccount.owner == "")) # noqa: E711 + + +def _discover_demo_scopes() -> set[str]: db = SessionLocal() try: - acct = db.query(EmailAccount).filter( - EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER - ).first() + return { + row.owner or "" + for row in db.query(EmailAccount).filter( + EmailAccount.name == NAME, + EmailAccount.imap_user == IMAP_USER, + ).all() + } + finally: + db.close() + + +def _lock_and_load_demo_rows(db, scopes: set[str]): + """Reload Demo rows under every observed owner lock.""" + scopes = set(scopes) or {OWNER} + while True: + lock_email_account_owner_mutations(db, *scopes) + rows = ( + db.query(EmailAccount) + .filter( + EmailAccount.name == NAME, + EmailAccount.imap_user == IMAP_USER, + ) + .order_by(EmailAccount.created_at.asc(), EmailAccount.id.asc()) + .all() + ) + current_scopes = {row.owner or "" for row in rows} + if current_scopes.issubset(scopes) or db.get_bind().dialect.name == "sqlite": + return rows + db.rollback() + scopes.update(current_scopes) + + +def _promote_oldest_enabled(db, owner: str, excluded_ids: list[str]) -> None: + remaining = _owner_scope( + db.query(EmailAccount).filter( + EmailAccount.enabled == True, # noqa: E712 + ~EmailAccount.id.in_(excluded_ids), + ), + owner, + ) + if remaining.filter(EmailAccount.is_default == True).first() is not None: # noqa: E712 + return + promote = remaining.order_by( + EmailAccount.created_at.asc(), EmailAccount.id.asc() + ).first() + if promote is not None: + promote.is_default = True + + +def setup() -> int: + Base.metadata.create_all(bind=engine) + scopes = _discover_demo_scopes() | {OWNER} + db = SessionLocal() + try: + rows = _lock_and_load_demo_rows(db, scopes) + acct = rows[0] if rows else None if acct is None: acct = EmailAccount(id=uuid.uuid4().hex, name=NAME) db.add(acct) + old_scope = acct.owner or "" + was_default = bool(acct.is_default) + if old_scope != OWNER: + # Move a non-default row first so the unique index cannot see two + # defaults transiently while SQLAlchemy flushes the owner move and + # old-scope promotion in separate UPDATE statements. + acct.is_default = False + acct.owner = OWNER + db.flush() + if was_default: + _promote_oldest_enabled(db, old_scope, [acct.id]) + + target_default = _owner_scope( + db.query(EmailAccount).filter( + EmailAccount.id != acct.id, + EmailAccount.is_default == True, # noqa: E712 + ), + OWNER, + ).first() acct.owner = OWNER - acct.is_default = False # never default — user switches to it + # Keep Demo non-default when a real default exists. If it is the only + # enabled account, it must be default to preserve normal create + # semantics and avoid leaving the owner partition without one. + acct.is_default = target_default is None acct.enabled = True acct.imap_host = "localhost" acct.imap_port = 31143 @@ -57,20 +144,27 @@ def setup() -> int: acct.smtp_password = encrypt(IMAP_PASSWORD) acct.from_address = IMAP_USER db.commit() - print(f"'{NAME}' account ready (id={acct.id}, non-default, switchable).") + state = "default" if acct.is_default else "non-default" + print(f"'{NAME}' account ready (id={acct.id}, {state}, switchable).") return 0 finally: db.close() def teardown() -> int: + scopes = _discover_demo_scopes() db = SessionLocal() try: - rows = db.query(EmailAccount).filter( - EmailAccount.name == NAME, EmailAccount.imap_user == IMAP_USER - ).all() + rows = _lock_and_load_demo_rows(db, scopes) + deleted_ids = [row.id for row in rows] + default_scopes = {row.owner or "" for row in rows if row.is_default} for r in rows: db.delete(r) + # Ensure the old default DELETE reaches the database before a + # replacement UPDATE; the unique index is enforced per statement. + db.flush() + for owner in default_scopes: + _promote_oldest_enabled(db, owner, deleted_ids) db.commit() print(f"removed {len(rows)} '{NAME}' account row(s).") return 0 diff --git a/tests/test_email_account_default_serialization.py b/tests/test_email_account_default_serialization.py index 04e1d1f30..6d2394378 100644 --- a/tests/test_email_account_default_serialization.py +++ b/tests/test_email_account_default_serialization.py @@ -6,11 +6,18 @@ rather than relying on an in-process Python lock. """ import asyncio +import json +import sys import threading +import types +from pathlib import Path +from types import SimpleNamespace from unittest import mock import pytest -from sqlalchemy import create_engine +from fastapi import HTTPException +from sqlalchemy import create_engine, create_mock_engine, text +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import NullPool @@ -46,6 +53,13 @@ def _endpoint(method, path): raise AssertionError(f"email route not found: {method} {path}") +def _named_endpoint(router, name): + for route in router.routes: + if getattr(getattr(route, "endpoint", None), "__name__", "") == name: + return route.endpoint + raise AssertionError(f"route not found: {name}") + + def _seed_account(factory, account_id, owner, *, is_default=False, enabled=True): from core.database import EmailAccount @@ -210,3 +224,299 @@ def test_delete_promotion_and_set_default_are_one_serial_transition( ("alice-c", "alice", True), ("bob-a", "bob", True), ] + + +def test_upgrade_normalizes_legacy_defaults_and_installs_unique_index( + tmp_path, monkeypatch +): + """A pre-index schema upgrades without requiring newer account columns.""" + from core import database as core_db + + engine = create_engine( + f"sqlite:///{tmp_path / 'legacy-accounts.db'}", + connect_args={"check_same_thread": False}, + poolclass=NullPool, + ) + try: + with engine.begin() as conn: + conn.execute(text(""" + CREATE TABLE email_accounts ( + id VARCHAR PRIMARY KEY, + owner VARCHAR, + name VARCHAR NOT NULL, + is_default BOOLEAN NOT NULL, + enabled BOOLEAN NOT NULL, + created_at DATETIME, + updated_at DATETIME + ) + """)) + conn.execute(text(""" + INSERT INTO email_accounts + (id, owner, name, is_default, enabled, created_at, updated_at) + VALUES + ('legacy-old', NULL, 'Old', 1, 1, '2024-01-01', '2024-01-01'), + ('legacy-new', '', 'New', 1, 1, '2025-01-01', '2025-01-01') + """)) + + monkeypatch.setattr(core_db, "engine", engine) + core_db._migrate_email_account_default_invariant() + core_db._migrate_email_account_default_invariant() # idempotent replay + + with engine.connect() as conn: + defaults = conn.execute(text(""" + SELECT id FROM email_accounts + WHERE is_default IS TRUE + ORDER BY id + """)).scalars().all() + index_names = { + row[1] for row in conn.execute(text("PRAGMA index_list(email_accounts)")) + } + assert defaults == ["legacy-old"] + assert core_db._EMAIL_ACCOUNT_DEFAULT_INDEX in index_names + + with pytest.raises(IntegrityError): + with engine.begin() as conn: + conn.execute(text(""" + INSERT INTO email_accounts + (id, owner, name, is_default, enabled, created_at, updated_at) + VALUES + ('legacy-third', NULL, 'Third', 1, 1, '2026-01-01', '2026-01-01') + """)) + finally: + engine.dispose() + + +def test_concurrent_legacy_seed_is_one_locked_transaction( + tmp_path, monkeypatch, caplog +): + from core import database as core_db + + engine = create_engine( + f"sqlite:///{tmp_path / 'seed-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) + settings_file = tmp_path / "settings.json" + settings_file.write_text( + json.dumps({"imap_host": "imap.example.test", "imap_user": "alice"}), + encoding="utf-8", + ) + monkeypatch.setattr(core_db, "engine", engine) + monkeypatch.setattr(core_db, "SessionLocal", factory) + monkeypatch.setattr(core_db, "SETTINGS_FILE", str(settings_file)) + + read_barrier = threading.Barrier(2) + real_read_text = Path.read_text + + def synchronized_read(path, *args, **kwargs): + value = real_read_text(path, *args, **kwargs) + if path == settings_file: + read_barrier.wait(5) + return value + + monkeypatch.setattr(Path, "read_text", synchronized_read) + threads = [ + threading.Thread(target=core_db._migrate_seed_email_account) + for _ in range(2) + ] + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join(5) + assert all(not thread.is_alive() for thread in threads) + + with engine.connect() as conn: + rows = conn.execute(text(""" + SELECT owner, is_default FROM email_accounts + ORDER BY id + """)).all() + assert rows == [(None, 1)] + assert "seed email account migration:" not in caplog.text + finally: + engine.dispose() + + +def test_multi_owner_row_locks_are_acquired_in_canonical_order(): + from core.database import lock_email_account_owner_mutations + + class FakeSession: + def __init__(self): + self.locked = [] + + def get_bind(self): + return SimpleNamespace(dialect=SimpleNamespace(name="postgresql")) + + def get(self, _model, owner_key, **kwargs): + assert kwargs == {"with_for_update": True} + self.locked.append(owner_key) + return object() + + db = FakeSession() + lock_email_account_owner_mutations(db, "zeta", "", "alpha", "zeta") + assert db.locked == ["", "alpha", "zeta"] + + +def test_postgresql_fresh_schema_emits_default_unique_index(): + from core import database as core_db + + statements = [] + engine_holder = {} + + def capture(statement, *_args, **_kwargs): + statements.append( + str(statement.compile(dialect=engine_holder["engine"].dialect)) + ) + + mock_engine = create_mock_engine("postgresql://", capture) + engine_holder["engine"] = mock_engine + core_db.EmailAccount.__table__.create(mock_engine) + + assert any( + core_db._EMAIL_ACCOUNT_DEFAULT_INDEX in statement + and "COALESCE(owner, '')" in statement + and "WHERE is_default IS TRUE" in statement + for statement in statements + ) + + +def test_rename_serializes_old_and_new_owner_and_stale_set_default_fails_closed( + account_db, monkeypatch, tmp_path +): + from core import database as core_db + from routes import auth_routes + + _seed_account(account_db, "alice-a", "alice", is_default=True) + _seed_account(account_db, "alice-b", "alice") + _seed_account(account_db, "bob-a", "bob", is_default=True) + + prefs_module = types.ModuleType("routes.prefs_routes") + prefs_module._load = lambda: {} + prefs_module._save = lambda _data: None + monkeypatch.setitem(sys.modules, "routes.prefs_routes", prefs_module) + monkeypatch.setattr( + auth_routes, "DEEP_RESEARCH_DIR", str(tmp_path / "deep_research") + ) + monkeypatch.setattr(auth_routes, "MEMORY_FILE", str(tmp_path / "memory.json")) + monkeypatch.setattr(auth_routes, "SKILLS_DIR", str(tmp_path / "skills")) + + auth_manager = mock.MagicMock() + auth_manager.get_username_for_token.return_value = "admin" + auth_manager.is_admin.return_value = True + auth_manager.users = {"admin": {}, "alice": {}} + auth_manager.rename_user.return_value = True + rename_user = _named_endpoint( + auth_routes.setup_auth_routes(auth_manager), "rename_user" + ) + set_default = _endpoint("POST", "/api/email/accounts/{account_id}/set-default") + + rename_acquired = threading.Event() + release_rename = threading.Event() + set_attempted = threading.Event() + set_acquired = threading.Event() + real_lock = core_db.lock_email_account_owner_mutations + + def controlled_lock(db, *owners): + thread_name = threading.current_thread().name + if thread_name == "rename-owner": + real_lock(db, *owners) + rename_acquired.set() + assert release_rename.wait(5) + return + if thread_name == "stale-set-default": + set_attempted.set() + real_lock(db, *owners) + set_acquired.set() + return + real_lock(db, *owners) + + monkeypatch.setattr(core_db, "lock_email_account_owner_mutations", controlled_lock) + request = SimpleNamespace( + cookies={"odysseus_session": "admin-token"}, + app=SimpleNamespace( + state=SimpleNamespace( + invalidate_token_cache=lambda: None, + session_manager=None, + research_handler=None, + upload_handler=None, + personal_docs_manager=None, + ) + ), + ) + results = {} + + def rename_owner(): + results["rename"] = asyncio.run( + rename_user("alice", SimpleNamespace(username="bob"), request) + ) + + def select_stale_default(): + try: + results["set"] = asyncio.run( + set_default("alice-b", owner="alice") + ) + except Exception as exc: # asserted below with its HTTP status + results["set_error"] = exc + + rename_thread = threading.Thread(target=rename_owner, name="rename-owner") + set_thread = threading.Thread( + target=select_stale_default, name="stale-set-default" + ) + rename_thread.start() + assert rename_acquired.wait(5) + set_thread.start() + assert set_attempted.wait(5) + assert not set_acquired.wait(0.1), "set-default bypassed the rename lock" + + release_rename.set() + rename_thread.join(5) + set_thread.join(5) + + assert not rename_thread.is_alive() + assert not set_thread.is_alive() + assert results["rename"]["ok"] is True + assert isinstance(results["set_error"], HTTPException) + assert results["set_error"].status_code == 404 + assert _rows(account_db) == [ + ("alice-a", "bob", False), + ("alice-b", "bob", False), + ("bob-a", "bob", True), + ] + + +def test_demo_teardown_promotes_replacement_in_same_transaction( + account_db, monkeypatch +): + from core.database import EmailAccount + from scripts.demo_email import demo_account + + db = account_db() + try: + db.add_all([ + EmailAccount( + id="real", + owner="", + name="Real", + is_default=False, + enabled=True, + ), + EmailAccount( + id="demo", + owner="", + name=demo_account.NAME, + imap_user=demo_account.IMAP_USER, + is_default=True, + enabled=True, + ), + ]) + db.commit() + finally: + db.close() + + monkeypatch.setattr(demo_account, "SessionLocal", account_db) + monkeypatch.setattr(demo_account, "engine", account_db.kw["bind"]) + + assert demo_account.teardown() == 0 + assert _rows(account_db) == [("real", "", True)] diff --git a/tests/test_rename_user_owner_sync.py b/tests/test_rename_user_owner_sync.py index 7e9e5d911..b7ac7e200 100644 --- a/tests/test_rename_user_owner_sync.py +++ b/tests/test_rename_user_owner_sync.py @@ -114,6 +114,12 @@ def _force_sql_owner_migration_failure(monkeypatch): def filter(self, *_args, **_kwargs): return self + def order_by(self, *_args, **_kwargs): + return self + + def all(self): + return [] + def update(self, *_args, **_kwargs): raise RuntimeError("forced owner migration failure") @@ -125,6 +131,12 @@ def _force_sql_owner_migration_failure(monkeypatch): def query(self, _model): return FailingQuery() + def get_bind(self): + return SimpleNamespace(dialect=SimpleNamespace(name="postgresql")) + + def get(self, _model, _key, **_kwargs): + return object() + def rollback(self): self.rolled_back = True