Merge pull request 'Migration lock safety + remove the merge's full-table scan (the real 0040-hang fix)' (#85) from dev into main

alembic/env.py

@@ -1,5 +1,7 @@
 """Alembic environment — reads DATABASE_URL from app config."""
 
+import os
+import re
 from logging.config import fileConfig
 
 from sqlalchemy import engine_from_config, pool, text
@@ -8,14 +10,18 @@ from alembic import context
 from backend.app.config import get_config
 from backend.app.models import Base
 
-# Arbitrary fixed 64-bit key for the session/transaction advisory lock that
-# serializes concurrent `alembic upgrade head` runs. Every `web` replica runs
-# migrations in its entrypoint, so under `docker stack deploy` two replicas can
-# boot at once and race the same DDL — duplicate CREATE TABLE, then a crashed
-# replica (operator-flagged 2026-06-07: 0040 raced; one backend died with
-# AdminShutdown). The first replica to reach the lock migrates; the rest block,
-# then find the version table already at head and apply nothing.
-_MIGRATION_LOCK_KEY = 0xFCA1E35C
+# Fail a blocked migration FAST instead of hanging forever. Migrations run
+# against the live DB while workers hold locks; 0040's `ALTER series_page` queued
+# behind a tag-merge that held a series_page lock for minutes (the merge runs an
+# unindexed full scan over image_record while repointing series_page) and hung
+# with no timeout — silent, indefinite (operator-flagged 2026-06-07). With a
+# lock_timeout a blocked DDL errors ("canceling statement due to lock timeout")
+# and the entrypoint's `alembic upgrade head` exits non-zero, so the deploy
+# retries / surfaces loudly rather than wedging. Override via env when a known
+# slow-lock window is expected.
+_MIGRATION_LOCK_TIMEOUT = os.environ.get("MIGRATION_LOCK_TIMEOUT", "30s")
+if not re.fullmatch(r"\d+\s*(ms|s|min)?", _MIGRATION_LOCK_TIMEOUT.strip()):
+    _MIGRATION_LOCK_TIMEOUT = "30s"  # ignore a malformed override
 
 config = context.config
 
@@ -47,24 +53,21 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
+        # Session-level lock_timeout for every DDL statement in this run. Set
+        # (and commit) before alembic opens its own transaction so the GUC
+        # persists on this connection regardless of how alembic structures its
+        # transactions. Value is from our own env, so f-string interpolation is
+        # safe (and it's been pattern-validated above); SET takes no bind params.
+        connection.execute(
+            text(f"SET lock_timeout = '{_MIGRATION_LOCK_TIMEOUT}'")
+        )
+        connection.commit()
         context.configure(
             connection=connection,
             target_metadata=target_metadata,
             compare_type=True,
         )
         with context.begin_transaction():
-            # Serialize concurrent migrators (see _MIGRATION_LOCK_KEY). A
-            # transaction-scoped advisory lock: the first replica to get here
-            # holds it for the whole upgrade and is auto-released when this
-            # transaction ends. A sibling replica blocks on this line, and only
-            # once the leader commits does it proceed to read the version table
-            # — now at head — so it runs zero migrations instead of re-applying
-            # the same DDL. The lock is acquired BEFORE run_migrations() reads
-            # the current revision, which is what makes the no-op correct.
-            connection.execute(
-                text("SELECT pg_advisory_xact_lock(:k)"),
-                {"k": _MIGRATION_LOCK_KEY},
-            )
             context.run_migrations()
 
 

backend/app/services/tag_service.py

@@ -562,50 +562,33 @@ class TagService:
     async def _create_protective_aliases(
         self, src_name: str, src_kind: TagKind, tgt: int
     ) -> bool:
-        """One alias per category the tagger has actually emitted for
-        src_name (so future predictions resolve to target); fall back to
-        the tag's kind when the tagger never predicted this exact name.
-        Idempotent — never clobbers a pre-existing operator alias."""
+        """Alias (src_name, category) -> tgt so future tagger predictions of
+        src_name resolve to the merge survivor. Idempotent — never clobbers a
+        pre-existing operator alias.
+
+        The category is the tag's kind. The tagger's tag_to_category map is
+        one-to-one (a name has exactly ONE category), and a tag's kind is set
+        from that category when it's created — so kind already IS the tagger's
+        category for this name. This used to SELECT DISTINCT category by scanning
+        every image_record's tagger_predictions JSON (an unindexed full scan that,
+        on a large library, held a series_page lock for minutes inside the merge
+        and blocked migration 0040 — operator-flagged 2026-06-07). That scan only
+        ever rediscovered the kind, so it's gone."""
         from ..models.tag_alias import TagAlias
 
-        rows = (
-            await self.session.execute(
-                text(
-                    "SELECT DISTINCT "
-                    "  (tagger_predictions::jsonb -> :n ->> 'category') "
-                    "  AS cat "
-                    "FROM image_record "
-                    "WHERE tagger_predictions IS NOT NULL "
-                    "  AND (tagger_predictions::jsonb) ? :n"
-                ),
-                {"n": src_name},
+        category = src_kind.value if hasattr(src_kind, "value") else str(src_kind)
+        res = await self.session.execute(
+            pg_insert(TagAlias)
+            .values(
+                alias_string=src_name,
+                alias_category=category,
+                canonical_tag_id=tgt,
             )
-        ).all()
-        categories = {r.cat for r in rows if r.cat}
-        if not categories:
-            kind_val = (
-                src_kind.value
-                if hasattr(src_kind, "value")
-                else str(src_kind)
+            .on_conflict_do_nothing(
+                index_elements=["alias_string", "alias_category"]
             )
-            categories = {kind_val}
-
-        created = False
-        for cat in categories:
-            res = await self.session.execute(
-                pg_insert(TagAlias)
-                .values(
-                    alias_string=src_name,
-                    alias_category=cat,
-                    canonical_tag_id=tgt,
-                )
-                .on_conflict_do_nothing(
-                    index_elements=["alias_string", "alias_category"]
-                )
-            )
-            if res.rowcount:
-                created = True
-        return created
+        )
+        return bool(res.rowcount)
 
 
 # ---------------------------------------------------------------------------

tests/test_tag_merge.py

@@ -320,7 +320,12 @@ async def test_no_alias_when_purely_manual(db):
 
 
 @pytest.mark.asyncio
-async def test_alias_per_observed_prediction_category(db):
+async def test_protective_alias_uses_tag_kind(db):
+    # The protective alias category is the tag's KIND — the tagger maps each name
+    # to exactly one category and a tag's kind is set from it, so kind already IS
+    # the tagger's category. The merge no longer scans image_record's predictions
+    # to rediscover it. Even with a (contrived) differing prediction category
+    # present, the merge writes a single (name, kind) alias.
     from backend.app.models import ImageRecord
     from backend.app.models.tag_alias import TagAlias
 
@@ -332,11 +337,6 @@ async def test_alias_per_observed_prediction_category(db):
     await svc.add_to_image(img, a.id, source="ml_auto")
     r1 = await db.get(ImageRecord, img)
     r1.tagger_predictions = {
-        "predname": {"category": "general", "confidence": 0.9}
-    }
-    i2 = await _img(db)
-    r2 = await db.get(ImageRecord, i2)
-    r2.tagger_predictions = {
         "predname": {"category": "copyright", "confidence": 0.8}
     }
     await db.flush()
@@ -353,7 +353,6 @@ async def test_alias_per_observed_prediction_category(db):
     ).all()
     assert {(r.alias_string, r.alias_category) for r in rows} == {
         ("predname", "general"),
-        ("predname", "copyright"),
     }
     assert all(r.canonical_tag_id == b.id for r in rows)