fix(persistence): retry feedback and conversation upserts that lose a race (#6852)

* 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.

CHANGELOG.md

@@ -5878,6 +5878,13 @@ This release closes that milestone with **621 merged pull requests**.
   was dropped when a Slack bot was saved on the other, and a disconnected bot
   came back. Every save now rereads the file under a lock shared across
   processes and changes only its own provider. ([#6844])
+- **persistence:** Two first ratings of one run by the same user at the same
+  moment (two tabs, two devices, a client retry) no longer fail one of them
+  with a 500. Both used to miss the existing-row lookup and insert, and the
+  second insert hit the `(thread, run, user)` unique key. On Postgres, 26 of
+  80 such requests failed. The same race in IM conversation binding, such as
+  a first message arriving together with `/new`, no longer fails the message.
+  The losing write now updates the row the other one created. ([#6852])
 
 ### Security
 
@@ -11659,3 +11666,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6785]: https://github.com/bytedance/deer-flow/pull/6785
 [#6798]: https://github.com/bytedance/deer-flow/pull/6798
 [#6844]: https://github.com/bytedance/deer-flow/pull/6844
+[#6852]: https://github.com/bytedance/deer-flow/pull/6852

CHANGELOG_zh.md

@@ -4474,6 +4474,11 @@
   副本重写整个文件，因此在两个 worker 下，一个 worker 上保存的 Telegram 机器人会在
   另一个 worker 保存 Slack 机器人时丢失，已断开的机器人也会重新出现。现在每次保存
   都会在跨进程共享的锁下重新读取文件，只修改自己的 provider。([#6844])
+- **持久化：** 同一用户同时对同一个运行首次评分（两个标签页、两台设备或客户端重试）时，
+  不再有一个请求返回 500。两个请求都会查不到已有记录并插入，第二个插入会触发
+  `(thread, run, user)` 唯一约束；在 Postgres 上，80 个这样的请求中有 26 个失败。
+  IM 会话绑定中的同类竞争（例如首条消息与 `/new` 同时到达）也不再导致消息处理失败。
+  落后的写入现在会更新另一方创建的记录。([#6852])
 
 ### 安全
 
@@ -9315,3 +9320,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6781]: https://github.com/bytedance/deer-flow/pull/6781
 [#6785]: https://github.com/bytedance/deer-flow/pull/6785
 [#6844]: https://github.com/bytedance/deer-flow/pull/6844
+[#6852]: https://github.com/bytedance/deer-flow/pull/6852

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -10,6 +10,10 @@ Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned b
 
 On SQLite, `BEGIN IMMEDIATE` takes the database-wide write lock, so every unrelated writer (run status, thread metadata, the scheduler) waits for the transaction and fails with `database is locked` after `busy_timeout` (30s). Do slow work such as document conversion before opening a locked transaction; lock only to revalidate and publish, as `ProjectDocumentRepository.publish_under_live_lock` does. `tests/test_project_document_tools.py::TestConversionSerialization` pins this.
 
+## Unique-key upserts
+
+A repository method that creates or updates a row by a unique key must survive two first writers: both can miss the SELECT, and the later INSERT raises `IntegrityError`, a 500 for the caller. Use a dialect `on_conflict_do_update` when every key column is non-null, or retry the select-then-write after rolling back the session, as `ChannelConnectionRepository.upsert_connection` / `set_thread_id` and `FeedbackRepository.upsert` do. Feedback keeps the retry because `user_id` is nullable and a NULL key never conflicts, so ON CONFLICT would stop updating the NULL-user row.
+
 ## JSON integer filters
 
 Stored JSON integers are not bounded by the signed-64-bit filter input contract.

backend/packages/harness/deerflow/persistence/channel_connections/sql.py

@@ -560,30 +560,46 @@ class ChannelConnectionRepository:
         thread_id: str,
         external_topic_id: str | None = None,
     ) -> None:
+        """Bind a conversation to ``thread_id``, creating or updating its row.
+
+        Two writers binding a new conversation (a first message racing ``/new``
+        on another task, or two Gateway instances) can both miss the SELECT;
+        the later INSERT then violates
+        ``uq_channel_conversation_connection_external``. Like
+        ``upsert_connection``, that writer rolls back and retries as an update.
+        """
         topic_id = external_topic_id or ""
+        stmt = select(ChannelConversationRow).where(
+            ChannelConversationRow.connection_id == connection_id,
+            ChannelConversationRow.external_conversation_id == external_conversation_id,
+            ChannelConversationRow.external_topic_id == topic_id,
+        )
         async with self.session_factory() as session:
-            stmt = select(ChannelConversationRow).where(
-                ChannelConversationRow.connection_id == connection_id,
-                ChannelConversationRow.external_conversation_id == external_conversation_id,
-                ChannelConversationRow.external_topic_id == topic_id,
-            )
-            row = (await session.execute(stmt)).scalar_one_or_none()
-            if row is None:
-                row = ChannelConversationRow(
-                    id=self._new_id(),
-                    connection_id=connection_id,
-                    owner_user_id=owner_user_id,
-                    provider=provider,
-                    external_conversation_id=external_conversation_id,
-                    external_topic_id=topic_id,
-                    thread_id=thread_id,
-                )
-                session.add(row)
-            else:
-                row.thread_id = thread_id
-                row.owner_user_id = owner_user_id
-                row.provider = provider
-            await session.commit()
+            last_error: IntegrityError | None = None
+            for _ in range(_UPSERT_MAX_ATTEMPTS):
+                try:
+                    row = (await session.execute(stmt)).scalar_one_or_none()
+                    if row is None:
+                        row = ChannelConversationRow(
+                            id=self._new_id(),
+                            connection_id=connection_id,
+                            owner_user_id=owner_user_id,
+                            provider=provider,
+                            external_conversation_id=external_conversation_id,
+                            external_topic_id=topic_id,
+                            thread_id=thread_id,
+                        )
+                        session.add(row)
+                    else:
+                        row.thread_id = thread_id
+                        row.owner_user_id = owner_user_id
+                        row.provider = provider
+                    await session.commit()
+                    return
+                except IntegrityError as exc:
+                    last_error = exc
+                    await session.rollback()
+            raise last_error  # type: ignore[misc]  # loop runs at least once
 
     async def get_thread_id(
         self,

backend/packages/harness/deerflow/persistence/feedback/sql.py

@@ -9,12 +9,15 @@ import uuid
 from datetime import UTC, datetime
 
 from sqlalchemy import case, delete, func, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
 from deerflow.utils.time import coerce_iso
 
+_UPSERT_MAX_ATTEMPTS = 3
+
 
 class FeedbackRepository:
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
@@ -133,36 +136,48 @@ class FeedbackRepository:
         user_id: str | None | _AutoSentinel = AUTO,
         comment: str | None = None,
     ) -> dict:
-        """Create or update feedback for (thread_id, run_id, user_id). rating must be +1 or -1."""
+        """Create or update feedback for (thread_id, run_id, user_id). rating must be +1 or -1.
+
+        Two first submissions for one key can both miss the SELECT, and the
+        later INSERT then violates ``uq_feedback_thread_run_user``. That writer
+        rolls back and retries, so it re-reads the committed row and updates it
+        instead of failing the request.
+        """
         if rating not in (1, -1):
             raise ValueError(f"rating must be +1 or -1, got {rating}")
         resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.upsert")
+        stmt = select(FeedbackRow).where(
+            FeedbackRow.thread_id == thread_id,
+            FeedbackRow.run_id == run_id,
+            FeedbackRow.user_id == resolved_user_id,
+        )
         async with self._sf() as session:
-            stmt = select(FeedbackRow).where(
-                FeedbackRow.thread_id == thread_id,
-                FeedbackRow.run_id == run_id,
-                FeedbackRow.user_id == resolved_user_id,
-            )
-            result = await session.execute(stmt)
-            row = result.scalar_one_or_none()
-            if row is not None:
-                row.rating = rating
-                row.comment = comment
-                row.created_at = datetime.now(UTC)
-            else:
-                row = FeedbackRow(
-                    feedback_id=str(uuid.uuid4()),
-                    run_id=run_id,
-                    thread_id=thread_id,
-                    user_id=resolved_user_id,
-                    rating=rating,
-                    comment=comment,
-                    created_at=datetime.now(UTC),
-                )
-                session.add(row)
-            await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
+            last_error: IntegrityError | None = None
+            for _ in range(_UPSERT_MAX_ATTEMPTS):
+                try:
+                    row = (await session.execute(stmt)).scalar_one_or_none()
+                    if row is not None:
+                        row.rating = rating
+                        row.comment = comment
+                        row.created_at = datetime.now(UTC)
+                    else:
+                        row = FeedbackRow(
+                            feedback_id=str(uuid.uuid4()),
+                            run_id=run_id,
+                            thread_id=thread_id,
+                            user_id=resolved_user_id,
+                            rating=rating,
+                            comment=comment,
+                            created_at=datetime.now(UTC),
+                        )
+                        session.add(row)
+                    await session.commit()
+                    await session.refresh(row)
+                    return self._row_to_dict(row)
+                except IntegrityError as exc:
+                    last_error = exc
+                    await session.rollback()
+            raise last_error  # type: ignore[misc]  # loop runs at least once
 
     async def delete_by_run(
         self,

backend/tests/_repository_race_helpers.py

@@ -0,0 +1,61 @@
+"""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

backend/tests/test_channel_connections_repository.py

@@ -6,15 +6,19 @@ import logging
 from datetime import UTC, datetime, timedelta
 
 import pytest
+from _repository_race_helpers import miss_select_factory
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 from deerflow.persistence.channel_connections import (
     ChannelConnectionRepository,
     ChannelConnectionRow,
+    ChannelConversationRow,
     ChannelCredentialCipher,
     ChannelCredentialRow,
     ChannelOAuthStateRow,
 )
+from deerflow.persistence.channel_connections.sql import _UPSERT_MAX_ATTEMPTS
 
 
 @pytest.fixture
@@ -455,6 +459,55 @@ class TestChannelConnectionRepository:
         assert len(consumed) == 1
         assert consumed[0]["owner_user_id"] == "alice"
 
+    @pytest.mark.anyio
+    async def test_set_thread_id_retries_as_update_when_concurrent_insert_wins(self, repo):
+        """A losing conversation-binding INSERT retries as an UPDATE instead of raising IntegrityError."""
+        connection = await repo.upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U1", workspace_id="T1")
+        binding = {
+            "connection_id": connection["id"],
+            "owner_user_id": "alice",
+            "provider": "slack",
+            "external_conversation_id": "C-race",
+            "external_topic_id": "1700000000.000100",
+        }
+        await repo.set_thread_id(**binding, thread_id="thread-from-first-message")
+
+        real_factory = repo.session_factory
+
+        repo.session_factory, _ = miss_select_factory(real_factory)
+        try:
+            await repo.set_thread_id(**binding, thread_id="thread-from-new-command")
+        finally:
+            repo.session_factory = real_factory
+
+        assert await repo.get_thread_id(connection["id"], "C-race", "1700000000.000100") == "thread-from-new-command"
+        async with real_factory() as session:
+            rows = (await session.execute(select(ChannelConversationRow).where(ChannelConversationRow.external_conversation_id == "C-race"))).scalars().all()
+        assert len(rows) == 1
+
+    @pytest.mark.anyio
+    async def test_set_thread_id_raises_the_integrity_error_once_retries_are_exhausted(self, repo):
+        """If every attempt still collides, the original IntegrityError propagates instead of being swallowed."""
+        connection = await repo.upsert_connection(owner_user_id="alice", provider="slack", external_account_id="U1", workspace_id="T1")
+        binding = {
+            "connection_id": connection["id"],
+            "owner_user_id": "alice",
+            "provider": "slack",
+            "external_conversation_id": "C-exhausted",
+        }
+        await repo.set_thread_id(**binding, thread_id="thread-original")
+
+        real_factory = repo.session_factory
+        repo.session_factory, sessions = miss_select_factory(real_factory, misses=None)
+        try:
+            with pytest.raises(IntegrityError, match="channel_conversations.connection_id"):
+                await repo.set_thread_id(**binding, thread_id="thread-never-written")
+        finally:
+            repo.session_factory = real_factory
+
+        assert [session.missed for session in sessions] == [_UPSERT_MAX_ATTEMPTS]
+        assert await repo.get_thread_id(connection["id"], "C-exhausted") == "thread-original"
+
     @pytest.mark.anyio
     async def test_upsert_connection_retries_as_update_when_concurrent_insert_wins(self, repo):
         """A losing concurrent INSERT retries as an UPDATE instead of raising IntegrityError."""
@@ -468,36 +521,7 @@ class TestChannelConnectionRepository:
 
         real_factory = repo.session_factory
 
-        class _EmptyResult:
-            @staticmethod
-            def scalar_one_or_none():
-                return None
-
-        class MissFirstSelectSession:
-            """Make the initial identity SELECT miss, as if a concurrent writer inserted after it."""
-
-            def __init__(self, session):
-                self._session = session
-                self._missed = False
-
-            def __getattr__(self, name):
-                return getattr(self._session, name)
-
-            async def execute(self, *args, **kwargs):
-                result = await self._session.execute(*args, **kwargs)
-                if not self._missed:
-                    self._missed = True
-                    return _EmptyResult()
-                return result
-
-            async def __aenter__(self):
-                await self._session.__aenter__()
-                return self
-
-            async def __aexit__(self, *args):
-                return await self._session.__aexit__(*args)
-
-        repo.session_factory = lambda: MissFirstSelectSession(real_factory())
+        repo.session_factory, _ = miss_select_factory(real_factory)
         try:
             second = await repo.upsert_connection(
                 owner_user_id="alice",

backend/tests/test_feedback.py

@@ -4,8 +4,11 @@ Uses temp SQLite DB for ORM tests.
 """
 
 import pytest
+from _repository_race_helpers import miss_select_factory
+from sqlalchemy.exc import IntegrityError
 
 from deerflow.persistence.feedback import FeedbackRepository
+from deerflow.persistence.feedback.sql import _UPSERT_MAX_ATTEMPTS
 
 
 async def _make_feedback_repo(tmp_path):
@@ -173,6 +176,47 @@ class TestFeedbackRepository:
         assert second["comment"] == "changed my mind"
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_upsert_retries_as_update_when_concurrent_insert_wins(self, tmp_path):
+        """A losing first-submission INSERT updates the winner's row instead of raising IntegrityError."""
+        repo = await _make_feedback_repo(tmp_path)
+        try:
+            first = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
+            real_factory = repo._sf
+            repo._sf, _ = miss_select_factory(real_factory)
+            try:
+                second = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u1", comment="second tab")
+            finally:
+                repo._sf = real_factory
+
+            assert second["feedback_id"] == first["feedback_id"]
+            assert second["rating"] == -1
+            assert second["comment"] == "second tab"
+            rows = await repo.list_by_run("t1", "r1", user_id="u1")
+            assert [row["feedback_id"] for row in rows] == [first["feedback_id"]]
+        finally:
+            await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_upsert_raises_the_integrity_error_once_retries_are_exhausted(self, tmp_path):
+        """If every attempt still collides, the original IntegrityError propagates instead of being swallowed."""
+        repo = await _make_feedback_repo(tmp_path)
+        try:
+            first = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
+            real_factory = repo._sf
+            repo._sf, sessions = miss_select_factory(real_factory, misses=None)
+            try:
+                with pytest.raises(IntegrityError, match="feedback.thread_id, feedback.run_id, feedback.user_id"):
+                    await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u1")
+            finally:
+                repo._sf = real_factory
+
+            assert [session.missed for session in sessions] == [_UPSERT_MAX_ATTEMPTS]
+            rows = await repo.list_by_run("t1", "r1", user_id="u1")
+            assert [(row["feedback_id"], row["rating"]) for row in rows] == [(first["feedback_id"], 1)]
+        finally:
+            await _cleanup()
+
     @pytest.mark.anyio
     async def test_upsert_different_users_separate(self, tmp_path):
         repo = await _make_feedback_repo(tmp_path)

backend/tests/test_feedback_router.py

@@ -0,0 +1,114 @@
+"""Route tests for ``PUT /api/threads/{thread_id}/runs/{run_id}/feedback``.
+
+The route is documented as idempotent: two first submissions for one run by
+one user (two tabs, two devices, a client retry) must both succeed and leave a
+single feedback row.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID
+
+import httpx
+import pytest
+from _router_auth_helpers import make_authed_test_app
+
+from app.gateway.auth.models import User
+from app.gateway.auth_disabled import AUTH_SOURCE_SESSION
+from app.gateway.routers import feedback
+from deerflow.persistence.feedback import FeedbackRepository
+
+THREAD_ID = "11111111-2222-3333-4444-555555555555"
+RUN_ID = "run-1"
+
+
+def _user() -> User:
+    return User(
+        id=UUID("99999999-8888-7777-6666-555555555555"),
+        email="alice@example.com",
+        password_hash="x",
+        system_role="user",
+    )
+
+
+class _SelectBarrierSession:
+    """Hold each session after its first SELECT until every writer has run its own.
+
+    Both requests then see no row and both INSERT, which is the interleaving
+    that two concurrent first submissions produce on a real database.
+    """
+
+    def __init__(self, session, barrier: asyncio.Barrier):
+        self._session = session
+        self._barrier = barrier
+        self._waited = False
+
+    def __getattr__(self, name):
+        return getattr(self._session, name)
+
+    async def execute(self, *args, **kwargs):
+        result = await self._session.execute(*args, **kwargs)
+        if not self._waited:
+            self._waited = True
+            async with asyncio.timeout(5):
+                await self._barrier.wait()
+        return result
+
+    async def __aenter__(self):
+        await self._session.__aenter__()
+        return self
+
+    async def __aexit__(self, *args):
+        return await self._session.__aexit__(*args)
+
+
+@pytest.fixture
+async def feedback_repo(tmp_path):
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'feedback.db'}", sqlite_dir=str(tmp_path))
+    try:
+        yield FeedbackRepository(get_session_factory())
+    finally:
+        await close_engine()
+
+
+@pytest.mark.anyio
+async def test_concurrent_first_feedback_puts_both_succeed_with_one_row(feedback_repo):
+    barrier = asyncio.Barrier(2)
+    real_factory = feedback_repo._sf
+    feedback_repo._sf = lambda: _SelectBarrierSession(real_factory(), barrier)
+
+    app = make_authed_test_app(user_factory=_user)
+
+    @app.middleware("http")
+    async def _session_auth_source(request, call_next):
+        # Production AuthMiddleware records how the user authenticated; the
+        # route only reads the user id from a recognized source.
+        request.state.auth_source = AUTH_SOURCE_SESSION
+        return await call_next(request)
+
+    run_store = MagicMock()
+    run_store.get = AsyncMock(return_value={"run_id": RUN_ID, "thread_id": THREAD_ID})
+    app.state.run_store = run_store
+    app.state.feedback_repo = feedback_repo
+    app.include_router(feedback.router)
+
+    url = f"/api/threads/{THREAD_ID}/runs/{RUN_ID}/feedback"
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") as client:
+        up, down = await asyncio.gather(
+            client.put(url, json={"rating": 1}),
+            client.put(url, json={"rating": -1}),
+        )
+
+    assert up.status_code == 200, up.text
+    assert down.status_code == 200, down.text
+    assert up.json()["user_id"] == str(_user().id)
+    assert up.json()["feedback_id"] == down.json()["feedback_id"]
+    feedback_repo._sf = real_factory
+    rows = await feedback_repo.list_by_run(THREAD_ID, RUN_ID, user_id=str(_user().id))
+    assert len(rows) == 1
+    # The later commit wins, as with sequential submissions.
+    assert rows[0]["rating"] in (1, -1)