mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-13 15:40:00 +00:00
fix(runtime): resolve omitted run owners before idempotent reuse (#5401)
* fix(runtime): resolve omitted run owners before idempotent reuse HTTP run admissions do not pass a user_id. The SQL run store stamps the request user from the contextvar onto the row, but RunManager kept None on its process-local RunRecord, so the two disagreed about who owns the run. A keyed retry that reached a peer worker, or the owning worker after cleanup() released its local record, hydrated the stamped row, compared its owner with None, and raised "Run idempotency key resolved to a different thread or user", which start_run surfaced as a 500. MemoryRunStore stored None on both sides and never hit the check. The same mismatch hid HTTP runs from owner-scoped history reads, which filter local records by the current user, and skipped the worker's MCP background_tasks projection, which only runs for records with an owner. create() and _admit_thread_operation() now resolve an omitted owner from the current user before building the record, and keep None when no user is in context instead of falling back to the default bucket, so the local record and every store agree on the owner. HTTP runs now receive the background_tasks projection, so the replay golden's values frames gain that key. * docs(changelog): reference #5401 in the keyed retry owner fix entry * test(runtime): close the SQL engine when peer-reuse test setup fails Review follow-up on #5401: the sql case of the two-worker start_run test initialized the engine above the try whose finally calls close_engine(). init_engine() assigns the module-global engine and session factory before bootstrapping the schema, so a failure there skipped the teardown and left a stale engine for later tests in the same process. Store setup and the RunManager workers now live inside the try, so the teardown runs whether setup or the test body fails.
This commit is contained in:
Hyeonsang Cho
GitHub
parent
dfc8e72428
commit
6f81daefff
12
CHANGELOG.md
12
CHANGELOG.md
@ -582,6 +582,17 @@ This section accumulates work toward the **2.1.0** milestone
|
||||
|
||||
### Fixed
|
||||
|
||||
- **runtime:** Stop a keyed run retry from failing with 500 on the SQL run
|
||||
store. HTTP admissions do not pass a `user_id`; the SQL store stamps the
|
||||
request user on the row, but the process-local run record kept `None`. A
|
||||
retry with the same `Idempotency-Key` that reached another Gateway worker, or
|
||||
the same worker after the finished run was cleaned up, compared the two
|
||||
owners, took its own run for another user's, and raised. The same mismatch
|
||||
dropped HTTP runs from owner-scoped history reads and skipped the MCP
|
||||
`background_tasks` projection for them, so `values` events for these runs now
|
||||
include `background_tasks`. `RunManager` now resolves an omitted owner from
|
||||
the request user the way the SQL store does, so every store records the same
|
||||
owner. ([#5401])
|
||||
- **runtime:** Stop a cross-worker idempotent run reuse from permanently
|
||||
blocking the thread on the reusing worker. The reuse registered the hydrated
|
||||
store row as a local run record, but only the owning worker finalizes and
|
||||
@ -2773,3 +2784,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#5353]: https://github.com/bytedance/deer-flow/pull/5353
|
||||
[#5357]: https://github.com/bytedance/deer-flow/pull/5357
|
||||
[#5393]: https://github.com/bytedance/deer-flow/pull/5393
|
||||
[#5401]: https://github.com/bytedance/deer-flow/pull/5401
|
||||
|
||||
@ -397,6 +397,13 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **运行时：** 带 `Idempotency-Key` 的 run 重试在 SQL run 存储上不再返回 500。HTTP
|
||||
准入不会传入 `user_id`，SQL 存储会把请求用户写入该行，但进程内的 run 记录仍为
|
||||
`None`。同一 key 的重试若落到另一个 Gateway worker，或在已完成的 run 被清理后回到
|
||||
同一 worker，就会比较两边的拥有者，把自己的 run 误判为其他用户的并抛错。同一不一致
|
||||
还让 HTTP run 被按拥有者过滤的历史读取漏掉，并跳过了 MCP `background_tasks` 投影，
|
||||
因此这类 run 的 `values` 事件现在会包含 `background_tasks`。`RunManager` 现在按
|
||||
SQL 存储的方式用请求用户补全缺省的拥有者，各存储记录的拥有者保持一致。([#5401])
|
||||
- **运行时：** 跨 worker 的幂等 run 复用不再让复用方 worker 永久阻塞该线程。此前复用会
|
||||
把从存储中读取的行注册为本地 run 记录，但只有拥有该 run 的 worker 才会结束并清理自己的
|
||||
记录，因此这份副本会一直停留在准入时的 `pending`/`running` 状态：该 worker 上此线程后续
|
||||
@ -2127,3 +2134,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
|
||||
[#5353]: https://github.com/bytedance/deer-flow/pull/5353
|
||||
[#5357]: https://github.com/bytedance/deer-flow/pull/5357
|
||||
[#5393]: https://github.com/bytedance/deer-flow/pull/5393
|
||||
[#5401]: https://github.com/bytedance/deer-flow/pull/5401
|
||||
|
||||
@ -172,6 +172,8 @@ the number of required IDs, whichever is larger; missing exact runs use targeted
|
||||
|
||||
**`RunManager._runs` holds only records this worker admitted.** A cross-worker idempotent reuse returns the `store_only` row from `_record_from_store()` unregistered: the peer never finalizes or `cleanup()`s it, so a registered copy stays `pending`/`running`, 409s later same-thread admissions, hides the owner's orphan from reconciliation, and sends a peer `cancel()` down the local-owner path. Pinned by `test_peer_idempotent_reuse_*` and `test_peer_cancel_of_reused_run_*` (`tests/test_multi_worker_run_ownership.py`) plus `tests/test_gateway_services.py::test_start_run_peer_idempotent_reuse_*`.
|
||||
|
||||
**A `RunRecord` owner matches its durable row.** HTTP admissions omit `user_id` and the SQL store stamps the ambient user, so `create()` and `_admit_thread_operation()` resolve an omitted owner from the contextvar, keeping `None` without one — never `get_effective_user_id()`'s `default` bucket. A `None` record made SQL keyed retries on a peer or after `cleanup()` return 500, hid HTTP runs from owner-scoped history, and skipped their MCP `background_tasks` projection. Pinned by `test_run_record_owner_matches_row_stamped_from_context`, `test_keyed_retry_without_explicit_user_*`, the `sql` case of `test_start_run_peer_idempotent_reuse_*`, `test_run_manager_*_admitted_without_explicit_user`, and `test_run_manager_keeps_omitted_owner_unset_without_user_context`.
|
||||
|
||||
**Where things live**:
|
||||
- `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
|
||||
- `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`
|
||||
|
||||
@ -16,7 +16,7 @@ from typing import TYPE_CHECKING, Any
|
||||
|
||||
from sqlalchemy.exc import IntegrityError as SAIntegrityError
|
||||
|
||||
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
|
||||
from deerflow.runtime.user_context import AUTO, _AutoSentinel, get_current_user, resolve_user_id
|
||||
from deerflow.utils.time import is_lease_expired
|
||||
from deerflow.utils.time import now_iso as _now_iso
|
||||
|
||||
@ -63,6 +63,21 @@ def _generate_worker_id() -> str:
|
||||
return f"{socket.gethostname()}:{uuid.uuid4().hex}"
|
||||
|
||||
|
||||
def _resolve_record_user_id(user_id: str | None) -> str | None:
|
||||
"""Fill an omitted run owner from the ambient user, as the SQL store does.
|
||||
|
||||
The SQL store stamps ``user_id=None`` with the request user, so a local
|
||||
record left at ``None`` disagrees with its own durable row: owner-scoped
|
||||
reads skip it and idempotent reuse rejects it as another user's run.
|
||||
Resolving here gives every store the same owner. Without a user in context
|
||||
the owner stays ``None``.
|
||||
"""
|
||||
if user_id is not None:
|
||||
return user_id
|
||||
user = get_current_user()
|
||||
return str(user.id) if user is not None else None
|
||||
|
||||
|
||||
def _cursor_part(value: str | None) -> str | None:
|
||||
"""Treat missing/blank cursor fields as absent so a one-sided empty string fails."""
|
||||
if value is None:
|
||||
@ -609,6 +624,7 @@ class RunManager:
|
||||
"""
|
||||
run_id = str(uuid.uuid4())
|
||||
now = _now_iso()
|
||||
user_id = _resolve_record_user_id(user_id)
|
||||
lease_expires_at = self._compute_lease_expires_at()
|
||||
record = RunRecord(
|
||||
run_id=run_id,
|
||||
@ -1581,6 +1597,8 @@ class RunManager:
|
||||
"""
|
||||
run_id = str(uuid.uuid4())
|
||||
now = _now_iso()
|
||||
# Resolve before the idempotency checks below compare it with stored rows.
|
||||
user_id = _resolve_record_user_id(user_id)
|
||||
|
||||
_supported_strategies = ("reject", "interrupt", "rollback")
|
||||
if multitask_strategy not in _supported_strategies:
|
||||
|
||||
@ -13,6 +13,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -23,6 +24,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -34,6 +36,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -46,6 +49,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -58,6 +62,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -70,6 +75,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -83,6 +89,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"skill_context",
|
||||
@ -96,6 +103,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
@ -110,6 +118,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
@ -124,6 +133,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
@ -138,6 +148,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
@ -152,6 +163,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
@ -166,6 +178,7 @@
|
||||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"background_tasks",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
|
||||
@ -2579,14 +2579,21 @@ def test_start_run_session_caller_anti_forgery(_stub_app_config):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_owner_completes(_stub_app_config):
|
||||
"""Two Gateway workers share one run store; a retry landing on the peer must not strand the thread."""
|
||||
@pytest.mark.parametrize("run_store_backend", ["memory", "sql"])
|
||||
async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_owner_completes(_stub_app_config, run_store_backend, tmp_path):
|
||||
"""Two Gateway workers share one run store; a retry landing on the peer must not strand the thread.
|
||||
|
||||
HTTP admissions omit ``user_id``; the SQL store stamps the ambient user on
|
||||
the row, so the peer's reuse check must see the same owner there too.
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
|
||||
from app.gateway.services import start_run
|
||||
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
|
||||
from deerflow.persistence.run import RunRepository
|
||||
from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
|
||||
from deerflow.runtime import RunManager, RunStatus
|
||||
from deerflow.runtime.runs.store.memory import MemoryRunStore
|
||||
@ -2600,27 +2607,37 @@ async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_
|
||||
await run_manager.set_status(record.run_id, RunStatus.success)
|
||||
await run_manager.cleanup(record.run_id, delay=0)
|
||||
|
||||
run_store = MemoryRunStore()
|
||||
owner = RunManager(store=run_store, worker_id="worker-a")
|
||||
peer = RunManager(store=run_store, worker_id="worker-b")
|
||||
thread_store = MemoryThreadMetaStore(InMemoryStore())
|
||||
body = _run_create_request()
|
||||
with (
|
||||
patch("app.gateway.services.resolve_agent_factory", return_value=object()),
|
||||
patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
|
||||
):
|
||||
first = await start_run(body, "thread-peer-reuse", _make_start_run_request(owner, thread_store=thread_store), idempotency_key="http-run:retry")
|
||||
reused = await start_run(body, "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store), idempotency_key="http-run:retry")
|
||||
assert reused.run_id == first.run_id
|
||||
assert reused.status in (RunStatus.pending, RunStatus.running)
|
||||
try:
|
||||
# init_engine() assigns the module-global engine before bootstrapping
|
||||
# the schema, so a partial setup failure must still reach close_engine().
|
||||
if run_store_backend == "sql":
|
||||
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'runs.db'}", sqlite_dir=str(tmp_path))
|
||||
run_store = RunRepository(get_session_factory())
|
||||
else:
|
||||
run_store = MemoryRunStore()
|
||||
owner = RunManager(store=run_store, worker_id="worker-a")
|
||||
peer = RunManager(store=run_store, worker_id="worker-b")
|
||||
with (
|
||||
patch("app.gateway.services.resolve_agent_factory", return_value=object()),
|
||||
patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
|
||||
):
|
||||
first = await start_run(body, "thread-peer-reuse", _make_start_run_request(owner, thread_store=thread_store), idempotency_key="http-run:retry")
|
||||
reused = await start_run(body, "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store), idempotency_key="http-run:retry")
|
||||
assert reused.run_id == first.run_id
|
||||
assert reused.status in (RunStatus.pending, RunStatus.running)
|
||||
|
||||
release_owner_run.set()
|
||||
await asyncio.wait_for(first.task, timeout=1)
|
||||
try:
|
||||
follow_up = await start_run(_run_create_request("next turn"), "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store))
|
||||
except HTTPException as exc:
|
||||
pytest.fail(f"peer rejected a new run after the owner finished: {exc.status_code} {exc.detail}")
|
||||
await asyncio.wait_for(follow_up.task, timeout=1)
|
||||
release_owner_run.set()
|
||||
await asyncio.wait_for(first.task, timeout=1)
|
||||
try:
|
||||
follow_up = await start_run(_run_create_request("next turn"), "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store))
|
||||
except HTTPException as exc:
|
||||
pytest.fail(f"peer rejected a new run after the owner finished: {exc.status_code} {exc.detail}")
|
||||
await asyncio.wait_for(follow_up.task, timeout=1)
|
||||
finally:
|
||||
if run_store_backend == "sql":
|
||||
await close_engine()
|
||||
|
||||
assert follow_up.run_id != first.run_id
|
||||
|
||||
|
||||
@ -250,6 +250,38 @@ async def test_run_manager_batch_history_methods_default_to_current_user():
|
||||
assert set(records) == {"regen-alice"}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_run_manager_owner_scoped_history_includes_runs_admitted_without_explicit_user():
|
||||
"""A run admitted with the ambient user must count as that user's run."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
||||
|
||||
manager = RunManager(store=MemoryRunStore())
|
||||
token = set_current_user(SimpleNamespace(id="alice"))
|
||||
try:
|
||||
record = await manager.create_or_reject("t1", metadata={"regenerate_from_run_id": "source"})
|
||||
await manager.set_status(record.run_id, RunStatus.success)
|
||||
sources = await manager.list_successful_regenerate_sources("t1")
|
||||
records = await manager.get_many_by_thread("t1", {record.run_id})
|
||||
finally:
|
||||
reset_current_user(token)
|
||||
|
||||
assert sources == {"source"}
|
||||
assert set(records) == {record.run_id}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.no_auto_user
|
||||
async def test_run_manager_keeps_omitted_owner_unset_without_user_context():
|
||||
"""No ambient user means no owner, never a fallback bucket such as ``default``."""
|
||||
manager = RunManager(store=MemoryRunStore())
|
||||
|
||||
record = await manager.create_or_reject("t1")
|
||||
|
||||
assert record.user_id is None
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_run_manager_batch_history_methods_fail_closed_without_user_context():
|
||||
from deerflow.runtime import user_context
|
||||
|
||||
@ -856,6 +856,35 @@ class TestRunRepository:
|
||||
assert follow_up.run_id != first.run_id
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("admit", ["create", "create_or_reject"])
|
||||
async def test_run_record_owner_matches_row_stamped_from_context(self, tmp_path, admit):
|
||||
"""Both record constructors resolve an omitted owner the way the SQL store does."""
|
||||
repo = await _make_repo(tmp_path)
|
||||
manager = RunManager(store=repo)
|
||||
record = await getattr(manager, admit)("thread-T")
|
||||
|
||||
stored = await repo.get(record.run_id)
|
||||
assert stored is not None
|
||||
assert record.user_id == stored["user_id"] == "test-user-autouse"
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_keyed_retry_without_explicit_user_reuses_row_stamped_from_context(self, tmp_path):
|
||||
"""A retry that hydrates the stamped row must not read it as another user's run."""
|
||||
repo = await _make_repo(tmp_path)
|
||||
owner = RunManager(store=repo, worker_id="worker-a")
|
||||
peer = RunManager(store=repo, worker_id="worker-b")
|
||||
first = await owner.create_or_reject("thread-T", idempotency_key="http-run:retry")
|
||||
|
||||
await owner.set_status(first.run_id, RunStatus.success)
|
||||
await owner.cleanup(first.run_id, delay=0)
|
||||
for manager in (peer, owner):
|
||||
retried = await manager.create_or_reject("thread-T", idempotency_key="http-run:retry")
|
||||
assert retried.run_id == first.run_id
|
||||
assert retried.idempotency_reused is True
|
||||
await _cleanup()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_checkpoint_write_reservation_blocks_interrupt_run_on_sql_store(self, tmp_path):
|
||||
"""An interrupt-strategy run cannot displace a durable checkpoint writer."""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user