mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
4e9f6c68ea
* fix(persistence): retry feedback and conversation upserts that lose a race FeedbackRepository.upsert and ChannelConnectionRepository.set_thread_id looked up the row and inserted it when missing, with no conflict handling. Two first writers for one key both missed the lookup, and the second INSERT violated uq_feedback_thread_run_user or uq_channel_conversation_connection_external: a 500 for the second of two concurrent first ratings, and a failed message when a conversation's first message raced /new. Both now catch IntegrityError, roll back and retry, so the losing writer re-reads the committed row and updates it, as upsert_connection already does. A retry rather than ON CONFLICT keeps the NULL-user feedback path updating its single row, since a NULL key never conflicts. * docs(changelog): reference #6852 for the upsert race fix * test(persistence): pin upsert retry exhaustion and share the miss-select double Both retry loops now have a test that makes every SELECT miss against a pre-seeded row, so all attempts hit the unique key: the original IntegrityError must propagate after exactly _UPSERT_MAX_ATTEMPTS attempts and the stored row must stay unchanged. The "first SELECT misses" session double lived in four copies (one per retry test, including upsert_connection's). It now lives once in tests/_repository_race_helpers.py, with a misses count so the same double drives both the single lost race and exhaustion.
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""Session test doubles for repository upserts that race a concurrent writer.
|
|
|
|
A select-then-write upsert loses a race when another writer inserts the same
|
|
key between its SELECT and its INSERT. ``MissSelectSession`` reproduces that
|
|
deterministically: its first ``misses`` SELECTs report no row even though one
|
|
exists, so the repository INSERTs and hits the unique key exactly as the
|
|
losing writer would.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
|
|
class _EmptyResult:
|
|
@staticmethod
|
|
def scalar_one_or_none() -> None:
|
|
return None
|
|
|
|
|
|
class MissSelectSession:
|
|
"""Wrap an ``AsyncSession`` so its first ``misses`` ``execute`` calls find no row.
|
|
|
|
``misses=None`` misses every call, which drives a retry loop to
|
|
exhaustion. ``missed`` counts the calls that were made to miss.
|
|
"""
|
|
|
|
def __init__(self, session: Any, *, misses: int | None = 1) -> None:
|
|
self._session = session
|
|
self._misses = misses
|
|
self.missed = 0
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
return getattr(self._session, name)
|
|
|
|
async def execute(self, *args: Any, **kwargs: Any) -> Any:
|
|
result = await self._session.execute(*args, **kwargs)
|
|
if self._misses is None or self.missed < self._misses:
|
|
self.missed += 1
|
|
return _EmptyResult()
|
|
return result
|
|
|
|
async def __aenter__(self) -> MissSelectSession:
|
|
await self._session.__aenter__()
|
|
return self
|
|
|
|
async def __aexit__(self, *args: Any) -> Any:
|
|
return await self._session.__aexit__(*args)
|
|
|
|
|
|
def miss_select_factory(real_factory: Callable[[], Any], *, misses: int | None = 1) -> tuple[Callable[[], MissSelectSession], list[MissSelectSession]]:
|
|
"""Return a session factory producing ``MissSelectSession`` wrappers, and the list of sessions it created."""
|
|
sessions: list[MissSelectSession] = []
|
|
|
|
def factory() -> MissSelectSession:
|
|
session = MissSelectSession(real_factory(), misses=misses)
|
|
sessions.append(session)
|
|
return session
|
|
|
|
return factory, sessions
|