From 4e9f6c68ea107265d28017d332285b843b033e57 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Wed, 14 Oct 2026 13:19:26 +0900 Subject: [PATCH] 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 | 8 ++ CHANGELOG_zh.md | 6 + .../harness/deerflow/persistence/AGENTS.md | 4 + .../persistence/channel_connections/sql.py | 60 +++++---- .../deerflow/persistence/feedback/sql.py | 67 ++++++---- backend/tests/_repository_race_helpers.py | 61 ++++++++++ .../test_channel_connections_repository.py | 84 ++++++++----- backend/tests/test_feedback.py | 44 +++++++ backend/tests/test_feedback_router.py | 114 ++++++++++++++++++ 9 files changed, 370 insertions(+), 78 deletions(-) create mode 100644 backend/tests/_repository_race_helpers.py create mode 100644 backend/tests/test_feedback_router.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c9f2bd58b8..f63b86a9a9 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 3cbb57b292..d5b680514c 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 1908c69666..ec490197eb 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py b/backend/packages/harness/deerflow/persistence/channel_connections/sql.py index 984c77e67f..b99718e59f 100644 --- a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py +++ b/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, diff --git a/backend/packages/harness/deerflow/persistence/feedback/sql.py b/backend/packages/harness/deerflow/persistence/feedback/sql.py index 400263b142..dec5514c6a 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/sql.py +++ b/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, diff --git a/backend/tests/_repository_race_helpers.py b/backend/tests/_repository_race_helpers.py new file mode 100644 index 0000000000..38b9f708be --- /dev/null +++ b/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 diff --git a/backend/tests/test_channel_connections_repository.py b/backend/tests/test_channel_connections_repository.py index 04878763b7..6356c3199d 100644 --- a/backend/tests/test_channel_connections_repository.py +++ b/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", diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py index 0d9c939aff..9d4ef4f802 100644 --- a/backend/tests/test_feedback.py +++ b/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) diff --git a/backend/tests/test_feedback_router.py b/backend/tests/test_feedback_router.py new file mode 100644 index 0000000000..4288d9080d --- /dev/null +++ b/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)