fix(ml): per-task async engine for recompute_centroid (#881)
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 28s
CI / integration (push) Successful in 3m14s

recompute_centroid + recompute_centroids were the only tasks still using
the process-wide singleton extensions.get_session() under asyncio.run().
The async engine's asyncpg pool is bound to the loop it was created on;
each Celery task runs a fresh asyncio.run() loop, so after the first
invocation the cached engine handed loop-A connections to loop B and raised
"Future attached to a different loop" — every recompute after the first in
a worker process failed (~35ms, fails on first DB await).

Convert both to the established per-task async_session_factory() pattern
(NullPool engine created + disposed inside the task's own loop), matching
scan/download/admin tasks. No get_session usages remain in tasks/.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-16 19:19:43 -04:00
parent e287802ecb
commit 51201b459e
+21 -9
View File
@@ -491,15 +491,22 @@ def _confidence_for_tag(session, tag, preds: dict) -> float | None:
def recompute_centroid(self, tag_id: int) -> bool:
import asyncio
from ..extensions import get_session
from ..services.ml.centroids import CentroidService
from ._async_session import async_session_factory
async def _run() -> bool:
async with get_session() as session:
svc = CentroidService(session)
result = await svc.recompute_for_tag(tag_id)
await session.commit()
return result
# Per-task NullPool engine bound to THIS asyncio.run loop — the shared
# process-wide engine reuses connections across loops and raises
# "Future attached to a different loop" on every call after the first.
async_factory, async_engine = async_session_factory()
try:
async with async_factory() as session:
svc = CentroidService(session)
result = await svc.recompute_for_tag(tag_id)
await session.commit()
return result
finally:
await async_engine.dispose()
return asyncio.run(_run())
@@ -515,12 +522,17 @@ def recompute_centroids(self) -> int:
"""Daily: find drifted centroids, enqueue recompute_centroid for each."""
import asyncio
from ..extensions import get_session
from ..services.ml.centroids import CentroidService
from ._async_session import async_session_factory
async def _list() -> list[int]:
async with get_session() as session:
return await CentroidService(session).list_drifted()
# Per-task NullPool engine bound to this loop (see recompute_centroid).
async_factory, async_engine = async_session_factory()
try:
async with async_factory() as session:
return await CentroidService(session).list_drifted()
finally:
await async_engine.dispose()
drifted = asyncio.run(_list())
for tid in drifted: