fix(metrics): take the version advisory lock in the recompute

The recompute assumed its only caller was refresh_task_browse_summaries, which
already holds a sorted advisory lock per version id. The backfill calls it
directly and held no lock, so a backfill batch racing a live refresh could
overwrite a fresh row with the snapshot it aggregated moments earlier -- silent
staleness rather than a crash, and invisible to a single-threaded fixture test.

pg_advisory_xact_lock is re-entrant within a transaction, so taking an
already-held lock on the browse-summary path costs nothing, and the shared sort
order keeps both callers in one global ordering.

Proven by holding the lock from a second connection and asserting the recompute
blocks; the test fails with DID NOT RAISE when the lock is removed.

Author: stateofkate

oddish/src/oddish/core/task_version_model_metrics.py

@@ -3,7 +3,7 @@
 from collections.abc import Iterable
 from typing import Any
 
-from sqlalchemy import Integer, case, delete, func, select
+from sqlalchemy import Integer, case, delete, func, select, text
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -207,10 +207,22 @@ async def refresh_task_version_model_metrics(
         return
     await session.flush()
 
-    # No locking here: the sole caller is refresh_task_browse_summaries, which
-    # already holds a sorted transaction-scoped advisory lock per version id.
-    # Taking them again would be redundant; taking them in a different order
-    # would reintroduce the deadlock those locks exist to prevent.
+    # Lock here rather than relying on the caller. refresh_task_browse_summaries
+    # already holds these, but the backfill calls this function directly, and an
+    # unlocked backfill batch racing a live refresh can overwrite a fresh row
+    # with the snapshot it aggregated moments earlier -- silent staleness, not a
+    # crash. pg_advisory_xact_lock is re-entrant within a transaction, so taking
+    # an already-held lock is free; the same sorted order keeps both paths in a
+    # single global ordering and cannot deadlock against each other.
+    for version_id in version_ids:
+        await session.execute(
+            text(
+                "SELECT pg_advisory_xact_lock("
+                "hashtextextended(CAST(:version_id AS text), 0))"
+            ),
+            {"version_id": version_id},
+        )
+
     known = (
         await session.execute(
             select(TaskVersionModel.id)

oddish/tests/test_task_version_model_metrics.py

@@ -347,3 +347,47 @@ async def _count() -> int:
         assert await _count() == first
     finally:
         await session.rollback()
+
+
+@pytest.mark.asyncio
+async def test_recompute_takes_the_version_advisory_lock(session):
+    """The backfill calls this directly, so it cannot rely on a caller's lock.
+
+    Without a lock, a backfill batch racing a live refresh overwrites a fresh
+    row with the snapshot it aggregated moments earlier. Proven by holding the
+    lock from a second connection and asserting the recompute blocks on it.
+    """
+    import asyncio
+
+    import oddish.db.connection as conn
+    from sqlalchemy import text as sa_text
+
+    _, version_id = await _seed(session, [{"reward": 1.0, "total_steps": 12}])
+    await session.commit()
+
+    holder = conn.async_session_maker()
+    try:
+        # Same hash expression as refresh_task_browse_summaries, so the two
+        # paths contend on one lock rather than two.
+        await holder.execute(
+            sa_text(
+                "SELECT pg_advisory_xact_lock("
+                "hashtextextended(CAST(:v AS text), 0))"
+            ),
+            {"v": version_id},
+        )
+
+        blocked = conn.async_session_maker()
+        try:
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(
+                    refresh_task_version_model_metrics(blocked, [version_id]),
+                    timeout=2.0,
+                )
+        finally:
+            await blocked.rollback()
+            await blocked.close()
+    finally:
+        await holder.rollback()
+        await holder.close()
+        await session.rollback()