mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-12 15:09:47 +00:00
bb75f8d736
* feat(sandbox): share sandbox identity derivation and acquire serialization (#4741) Remote providers (AIO, E2B, BoxLite, Tenki, OpenSandbox) each inlined the same sha256(user:thread)[:16] sandbox-id expression and kept per-scope lock dicts that grew unboundedly until shutdown. This extracts both mechanisms into shared components without changing provider lifecycle, ids, capacity semantics, or public tool behavior: - sandbox/identity.py: keyword-only derive_sandbox_scope_token (byte-pinned compatibility contract) + is_sandbox_scope_token; per-provider golden vectors pin current behavior including BoxLite's raw-None quirk and each provider's private user_id resolution. - sandbox/acquire_serialization.py: AcquireSerializer — per-key lock table with holder/waiter refcount reclamation, bounded dedicated executor (async waits off both the event loop and the default executor), worker-owned cancellation cleanup (no event-loop callback dependency), idempotent close(). - Each provider adopts both components; AIO/E2B key by (user_id, thread_id) with acquire and (E2B) release serialized; BoxLite/Tenki/OpenSandbox key by derived sandbox id and offload the whole sync acquire to the serializer's executor so a cancelled awaiter cannot overlap a retried same-scope body (leaked-remote-VM regression caught in review). - thread_id=None acquires stay unserialized; provider shutdown()/reset() close the serializer; E2B capacity/ledger/reconciliation and AIO ownership/flock machinery untouched. - blocking-IO anchor proves contended OpenSandbox acquire_async stays off the event loop (teeth verified red/green); AGENTS.md documents the shared components. * refactor(sandbox): address review on acquire serialization (#5089) - Replace unreachable checkin branch with an assertion: run() returns False only after abandon(), which the except handler always re-raises; the old _checkin would have double-decremented the refcount. - Document the task.cancelling() == 0 assumption in hold_async. - Drop unused thread_id/user_id kwargs from BoxLite and Tenki _acquire_scope_locked (OpenSandbox still forwards them). * fix(sandbox): preserve request ContextVars in acquire executor bridge (#5089) loop.run_in_executor() does not copy contextvars, unlike the inherited SandboxProvider.acquire_async() which used asyncio.to_thread(). The BoxLite/OpenSandbox/Tenki acquire_async bridges introduced in this PR therefore dropped the request trace id (logged as trace_id=-). Add AcquireSerializer.run_on_executor(), which copies the calling context and runs the callable through ctx.run, and route all three providers through it. Add regression tests binding request_trace_context and verifying the worker thread observes it.
57 lines
2.4 KiB
Python
57 lines
2.4 KiB
Python
"""Anchor: contended OpenSandbox acquire_async must not block the event loop.
|
|
|
|
Two concurrent acquire_async calls for the same scope serialize on the
|
|
AcquireSerializer; the loser's wait and both callers' creation path must stay
|
|
off the loop. The fake SDK boundary below performs REAL file IO inside
|
|
create(), so any regression that moves creation back onto the loop trips the
|
|
Blockbuster gate (FILE_IO rules give this anchor teeth).
|
|
|
|
Blockbuster's default rule set is blind to ``threading.Lock.acquire`` (verified
|
|
empirically under ``detect_blocking_io_strict``), so the serializer lock wait
|
|
itself cannot be pinned this way — the lock-wait placement is covered by Task
|
|
7's ExplodingExecutor contract test. What this anchor pins instead is the
|
|
provider creation path: under contention, no blocking IO may run on the loop.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from blockbuster import BlockingError
|
|
from test_opensandbox_provider import _FakeRemote, _FakeSandboxClass, _install
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
class _FileIOProbingSandboxClass(_FakeSandboxClass):
|
|
"""Fake SDK whose create() performs real blocking file IO at the boundary."""
|
|
|
|
def __init__(self, probe_dir: Path) -> None:
|
|
super().__init__()
|
|
self._probe_dir = probe_dir
|
|
|
|
def create(self, image: str, **kwargs) -> _FakeRemote:
|
|
probe = self._probe_dir / f"create-{len(self.create_calls) + 1}.probe"
|
|
probe.write_text("x" * 4096) # real blocking file IO at the SDK boundary
|
|
return super().create(image, **kwargs)
|
|
|
|
|
|
async def test_concurrent_acquire_async_stays_off_event_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
provider, _fake_sdk = _install(monkeypatch, sdk=_FileIOProbingSandboxClass(tmp_path))
|
|
first, second = await asyncio.gather(
|
|
provider.acquire_async("thread-anchor", user_id="u-anchor"),
|
|
provider.acquire_async("thread-anchor", user_id="u-anchor"),
|
|
)
|
|
assert first == second # same scope serialized, second caller reuses
|
|
provider.shutdown()
|
|
|
|
|
|
async def test_sync_acquire_on_loop_trips_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""Meta-check (teeth): a synchronous acquire on the loop MUST be caught."""
|
|
provider, _fake_sdk = _install(monkeypatch, sdk=_FileIOProbingSandboxClass(tmp_path))
|
|
with pytest.raises(BlockingError):
|
|
provider.acquire("thread-anchor", user_id="u-anchor")
|
|
provider.shutdown()
|