From a0acdda103deb1c835f098d096aabe47eba34e73 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Wed, 15 Jul 2026 20:33:43 +0800 Subject: [PATCH] fix(gateway): keep create_thread idempotent when the insert loses a race (#3800) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(gateway): keep create_thread idempotent when the insert loses a race The /api/threads create endpoint checks for an existing row and then inserts, but the two steps are not atomic. When two requests create the same thread_id concurrently, one commits first and the other's INSERT fails on the duplicate primary key. The SQL-backed thread store raises, the handler catches it as a generic failure, and the loser gets a 500 — even though the docstring promises the call is idempotent. Re-read the row after a failed insert: if it is now present (the competing request won), return it instead of surfacing the conflict. A genuine write failure where the row is still absent keeps the 500. Covered by a regression test that simulates the lost race and asserts the endpoint returns the existing thread rather than erroring. Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com> * fix(gateway): scope create_thread race recovery and mirror owner reconciliation Address review on #3800: - Scope the insert-race recovery to sqlalchemy IntegrityError instead of a broad `except Exception`, so a non-race failure that coincides with an existing row is no longer silently returned as a 200. Any other error logs and surfaces as a 500. (The memory store overwrites on duplicate rather than raising, so it never reaches this path.) - Route both the fast path and the recovery path through a shared `_resolve_existing_thread` helper so the recovery performs the same trusted- owner reconciliation (claiming a legacy unscoped `user_id=None` row) the fast path does. Thread ownership no longer diverges based on which path resolved the record. Add regression tests for both: the recovery claims an unscoped row for a trusted owner, and a non-IntegrityError failure surfaces as a 500. --------- Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Co-authored-by: Willem Jiang --- backend/app/gateway/routers/threads.py | 65 +++++++++--- backend/tests/test_threads_router.py | 140 +++++++++++++++++++++++++ 2 files changed, 191 insertions(+), 14 deletions(-) diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index a9e251110..5e1f938cb 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -22,6 +22,7 @@ from typing import Any from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from langgraph.checkpoint.base import empty_checkpoint, uuid6 from pydantic import BaseModel, Field, field_validator +from sqlalchemy.exc import IntegrityError from app.gateway.authz import require_permission from app.gateway.deps import get_checkpointer, get_run_manager @@ -503,6 +504,39 @@ async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteRe return response +async def _resolve_existing_thread( + thread_store: Any, + thread_id: str, + thread_owner_user_id: str | None, + thread_owner_kwargs: dict[str, Any], +) -> dict | None: + """Return the existing thread_meta record for an idempotent create. + + When the caller carries a trusted internal owner but only a legacy unscoped + (``user_id=None``) row exists, claim it for that owner before returning. + Both the fast path and the insert-race recovery path resolve through here so + a thread's ownership does not diverge based on which path found the record. + """ + existing_record = await thread_store.get(thread_id, **thread_owner_kwargs) + if existing_record is None and thread_owner_user_id: + unscoped_record = await thread_store.get(thread_id, user_id=None) + if unscoped_record is not None: + if unscoped_record.get("user_id") != thread_owner_user_id: + await thread_store.update_owner(thread_id, thread_owner_user_id, user_id=None) + existing_record = await thread_store.get(thread_id, **thread_owner_kwargs) + return existing_record + + +def _existing_thread_response(thread_id: str, record: dict) -> ThreadResponse: + return ThreadResponse( + thread_id=thread_id, + status=record.get("status", "idle"), + created_at=coerce_iso(record.get("created_at", "")), + updated_at=coerce_iso(record.get("updated_at", "")), + metadata=record.get("metadata", {}), + ) + + @router.post("", response_model=ThreadResponse) async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadResponse: """Create a new thread. @@ -523,21 +557,9 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe # ``ThreadCreateRequest._strip_reserved`` — see the model definition. # Idempotency: return existing record when already present - existing_record = await thread_store.get(thread_id, **thread_owner_kwargs) - if existing_record is None and thread_owner_user_id: - unscoped_record = await thread_store.get(thread_id, user_id=None) - if unscoped_record is not None: - if unscoped_record.get("user_id") != thread_owner_user_id: - await thread_store.update_owner(thread_id, thread_owner_user_id, user_id=None) - existing_record = await thread_store.get(thread_id, **thread_owner_kwargs) + existing_record = await _resolve_existing_thread(thread_store, thread_id, thread_owner_user_id, thread_owner_kwargs) if existing_record is not None: - return ThreadResponse( - thread_id=thread_id, - status=existing_record.get("status", "idle"), - created_at=coerce_iso(existing_record.get("created_at", "")), - updated_at=coerce_iso(existing_record.get("updated_at", "")), - metadata=existing_record.get("metadata", {}), - ) + return _existing_thread_response(thread_id, existing_record) # Write thread_meta so the thread appears in /threads/search immediately try: @@ -547,7 +569,22 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe **thread_owner_kwargs, metadata=body.metadata, ) + except IntegrityError: + # The idempotency read above and this insert are not atomic: a + # concurrent request for the same thread_id can commit in between, so + # the SQL-backed store rejects ours on the duplicate primary key. + # Honour the documented idempotency contract by resolving the + # now-existing record — running the same owner reconciliation the fast + # path does — instead of surfacing the conflict as a 500. (The memory + # store overwrites rather than raising, so it never reaches here.) + existing_record = await _resolve_existing_thread(thread_store, thread_id, thread_owner_user_id, thread_owner_kwargs) + if existing_record is not None: + return _existing_thread_response(thread_id, existing_record) + # A duplicate-key error with no row we can read back is a real failure. + logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to create thread") except Exception: + # Any non-race failure must surface, not be silently swallowed as a 200. logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id)) raise HTTPException(status_code=500, detail="Failed to create thread") diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index 42e5abefd..3823db3f3 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -251,6 +251,146 @@ def test_create_thread_returns_iso_timestamps() -> None: assert body["created_at"] == body["updated_at"] +def test_create_thread_returns_existing_when_insert_loses_race() -> None: + """A concurrent create that loses the INSERT race stays idempotent. + + The idempotency ``get`` check and the ``create`` INSERT are not atomic: + a competing request for the same ``thread_id`` can commit in between, and + the SQL-backed store then rejects ours on the duplicate primary key. The + endpoint documents idempotency ("returns the existing record when + ``thread_id`` already exists"), so it must surface the now-present row + rather than turning the integrity error into an HTTP 500. + """ + from sqlalchemy.exc import IntegrityError + + app, store, _checkpointer = _build_thread_app() + + class _RacingThreadMetaStore(_PermissiveThreadMetaStore): + """First create loses the race: the row is committed by a competing + request, then our INSERT fails with an integrity violation.""" + + def __init__(self, backing): + super().__init__(backing) + self._raised = False + + async def create(self, thread_id, *, assistant_id=None, user_id=None, display_name=None, metadata=None): # type: ignore[override] + if not self._raised: + self._raised = True + await super().create( + thread_id, + assistant_id=assistant_id, + user_id=user_id, + display_name=display_name, + metadata=metadata, + ) + raise IntegrityError( + "INSERT INTO threads_meta", + {}, + Exception("UNIQUE constraint failed: threads_meta.thread_id"), + ) + return await super().create( + thread_id, + assistant_id=assistant_id, + user_id=user_id, + display_name=display_name, + metadata=metadata, + ) + + app.state.thread_store = _RacingThreadMetaStore(store) + + with TestClient(app) as client: + response = client.post( + "/api/threads", + json={"thread_id": "race-thread", "metadata": {"k": "v"}}, + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["thread_id"] == "race-thread" + assert body["metadata"] == {"k": "v"} + + +def test_insert_race_recovery_claims_unscoped_row_for_trusted_owner() -> None: + """The insert-race recovery mirrors the fast path's owner reconciliation. + + When a competing request commits a legacy unscoped (``user_id=None``) row + between our idempotency read and our insert, and our insert then loses the + duplicate-key race, a trusted internal owner must still claim the row rather + than return it unowned — otherwise ownership of the same thread would depend + on whether the fast path or the recovery path resolved it. + """ + import asyncio + + from sqlalchemy.exc import IntegrityError + + from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE + + store = InMemoryStore() + checkpointer = InMemorySaver() + + class _RacingOwnerStore(MemoryThreadMetaStore): + """Our insert loses to a competing create that already wrote an + unscoped row, exactly the interleaving the recovery path exists for.""" + + async def create(self, thread_id, *, assistant_id=None, user_id=None, display_name=None, metadata=None): # type: ignore[override] + # The competing request commits its (owner-less) row here, then our + # insert loses the primary-key race. + await super().create(thread_id, user_id=None, metadata=metadata) + raise IntegrityError( + "INSERT INTO threads_meta", + {}, + Exception("UNIQUE constraint failed: threads_meta.thread_id"), + ) + + thread_store = _RacingOwnerStore(store) + request = SimpleNamespace( + headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"}, + state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), + app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer, thread_store=thread_store)), + ) + + async def _scenario(): + response = await threads.create_thread( + threads.ThreadCreateRequest(thread_id="channel-thread", metadata={"k": "v"}), + request, + ) + owner_row = await thread_store.get("channel-thread", user_id="owner-1") + unscoped_lookup = await thread_store.get("channel-thread", user_id=None) + return response, owner_row, unscoped_lookup + + response, owner_row, unscoped_lookup = asyncio.run(_scenario()) + + assert response.thread_id == "channel-thread" + # Recovery claimed the legacy row for the trusted owner, same as the fast path. + assert owner_row is not None + assert owner_row["user_id"] == "owner-1" + assert unscoped_lookup["user_id"] == "owner-1" + + +def test_create_thread_does_not_swallow_non_integrity_errors() -> None: + """A non-race insert failure must surface as 500, even when a row now exists. + + The recovery path only rescues the duplicate-key ``IntegrityError`` race; an + arbitrary failure that happens to coincide with an existing row must not be + silently returned as a 200 (previously the broad ``except`` did exactly that). + """ + app, store, _checkpointer = _build_thread_app() + + class _BrokenAfterWriteStore(_PermissiveThreadMetaStore): + async def create(self, thread_id, *, assistant_id=None, user_id=None, display_name=None, metadata=None): # type: ignore[override] + # A row exists after this call, but the insert failed for a reason + # unrelated to the idempotency race. + await super().create(thread_id, metadata=metadata) + raise RuntimeError("unexpected store failure") + + app.state.thread_store = _BrokenAfterWriteStore(store) + + with TestClient(app) as client: + response = client.post("/api/threads", json={"thread_id": "broken-thread", "metadata": {}}) + + assert response.status_code == 500, response.text + + def test_put_goal_creates_missing_thread_checkpoint_and_returns_goal() -> None: app, _store, _checkpointer = _build_thread_app()