From 30841c3b9fe785122787257c6c8046d39d19f175 Mon Sep 17 00:00:00 2001 From: Czile Date: Thu, 25 Jun 2026 08:06:17 +0800 Subject: [PATCH] fix(sandbox): synchronize sandbox provider singleton lifecycle (+ concurrency regression tests) (#3730) * fix(sandbox): synchronize sandbox provider singleton lifecycle get_sandbox_provider() used an unsynchronized check-then-create, so two OS threads (e.g. the main event loop and the Feishu channel thread, which runs its own loop) could double-initialize the provider. With AioSandboxProvider the overwritten instance leaks its idle-checker thread, since the only code that joins it (shutdown()) is reachable only through the reference that was overwritten. reset_sandbox_provider(), shutdown_sandbox_provider() and set_sandbox_provider() also touched the global without a lock, so a reset/shutdown racing an in-flight create could clear it mid-construction or tear down an instance another thread was about to return. Guard all four lifecycle sites with a single module-level threading.Lock and use double-checked locking in the getter, mirroring get_memory_storage(). * test(sandbox): add concurrent regression tests for provider singleton - 8 threads racing on cold start, synchronized with a threading.Barrier so the check-then-create race fires deterministically; asserts exactly one provider instance is created. - reset racing concurrent gets: asserts every returned value is a fully constructed provider (never None / half-built). * fix(sandbox): lock get read path, run provider callbacks outside the lock Addresses the three review findings on #3730: 1. get_sandbox_provider()'s read+return ran outside _provider_lock, so a concurrent reset/shutdown/set could null or tear the global between the check and the return, handing callers None / a torn instance. The hot read and the install reconciliation now both happen under the lock. 2. The non-reentrant _provider_lock was held across plugin-supplied callbacks (resolve_class import + provider __init__ in get; provider.reset()/shutdown() in reset/shutdown). A custom provider that re-entered these lifecycle functions would self-deadlock, and a slow teardown blocked every concurrent get(). resolve_class + construction now run outside the lock; reset/shutdown detach the reference under the lock and invoke the callback outside it. Tradeoff: racing cold-start callers may each construct a candidate. Exactly one is installed and returned to everyone; the losers (e.g. an AioSandbox instance that already started an idle-checker thread) are shut down so they do not leak the orphan thread #3721 is about. set_sandbox_provider() documents that it replaces but does not shut down the prior instance. 3. test_reset_racing_get reset the singleton to None *before* the barrier, so the racing reset was a no-op and never exercised reset-of-a-live-provider. It now populates the singleton up front so the reset tears down a live instance while getters read it. Tests: rename the cold-start test to assert "one installed singleton, observed by all" (construction is no longer single under the new design); add shutdown-vs-get, set-vs-get, and a losing-racer-shuts-down-its-orphan case. All five pass; the existing sandbox/middleware/mounts/uploads suites are green. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../deerflow/sandbox/sandbox_provider.py | 71 ++++- .../tests/test_sandbox_provider_lifecycle.py | 293 ++++++++++++++++++ 2 files changed, 353 insertions(+), 11 deletions(-) create mode 100644 backend/tests/test_sandbox_provider_lifecycle.py diff --git a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py index b58fbd87d..e8c7b163c 100644 --- a/backend/packages/harness/deerflow/sandbox/sandbox_provider.py +++ b/backend/packages/harness/deerflow/sandbox/sandbox_provider.py @@ -1,4 +1,5 @@ import asyncio +import threading from abc import ABC, abstractmethod from deerflow.config import get_app_config @@ -55,6 +56,21 @@ class SandboxProvider(ABC): _default_sandbox_provider: SandboxProvider | None = None +# Guards every read and write of `_default_sandbox_provider`. The singleton is +# reachable from more than one OS thread (e.g. the main event loop and the Feishu +# channel thread, which runs its own loop), so a bare check-then-create can double +# initialize the provider, and an unsynchronized reset/shutdown racing a get can +# hand a caller `None` or a torn instance. Every access to the global below takes +# this lock, including the read+return in `get_sandbox_provider()`. +# +# The lock guards only the reference swap. Provider callbacks (`__init__`, +# `reset()`, `shutdown()`) and the dynamic import in `resolve_class()` run +# *outside* the lock: they are plugin-supplied (`config.sandbox.use` resolves to +# an arbitrary class) and may be slow or, worse, re-enter these lifecycle +# functions. Holding a non-reentrant `threading.Lock` across them would +# self-deadlock such a provider and would block every concurrent `get()` during a +# slow teardown. Keeping callbacks off the lock avoids both. +_provider_lock = threading.Lock() def get_sandbox_provider(**kwargs) -> SandboxProvider: @@ -67,11 +83,33 @@ def get_sandbox_provider(**kwargs) -> SandboxProvider: A sandbox provider instance. """ global _default_sandbox_provider - if _default_sandbox_provider is None: - config = get_app_config() - cls = resolve_class(config.sandbox.use, SandboxProvider) - _default_sandbox_provider = cls(**kwargs) - return _default_sandbox_provider + # Fast path: a single locked read so a concurrent reset/shutdown can't null + # the global between the check and the return. + with _provider_lock: + if _default_sandbox_provider is not None: + return _default_sandbox_provider + + # Cold start. Resolve + construct outside the lock: the import and the + # provider constructor are plugin code and must not run under a non-reentrant + # lock. The construction may race another caller; we reconcile under the lock. + config = get_app_config() + cls = resolve_class(config.sandbox.use, SandboxProvider) + provider = cls(**kwargs) + + with _provider_lock: + if _default_sandbox_provider is None: + _default_sandbox_provider = provider + return provider + # We lost the install race: another thread got there first. `winner` is + # read under the same lock, so it is always a live instance, never None. + winner = _default_sandbox_provider + + # Discard the instance we just built (outside the lock). For providers with + # side-effectful constructors (e.g. AioSandboxProvider starts an idle-checker + # thread), this tears down the orphan so it does not leak — issue #3721. + if hasattr(provider, "shutdown"): + provider.shutdown() + return winner def reset_sandbox_provider() -> None: @@ -90,9 +128,13 @@ def reset_sandbox_provider() -> None: Use `shutdown_sandbox_provider()` for proper cleanup. """ global _default_sandbox_provider - if _default_sandbox_provider is not None: - _default_sandbox_provider.reset() + # Detach the reference under the lock, then run the provider's `reset()` + # callback outside it (see the `_provider_lock` note). + with _provider_lock: + provider = _default_sandbox_provider _default_sandbox_provider = None + if provider is not None: + provider.reset() def shutdown_sandbox_provider() -> None: @@ -103,10 +145,13 @@ def shutdown_sandbox_provider() -> None: is shutting down or when you need to completely reset the sandbox system. """ global _default_sandbox_provider - if _default_sandbox_provider is not None: - if hasattr(_default_sandbox_provider, "shutdown"): - _default_sandbox_provider.shutdown() + # Detach the reference under the lock, then run the (potentially slow) + # `shutdown()` callback outside it (see the `_provider_lock` note). + with _provider_lock: + provider = _default_sandbox_provider _default_sandbox_provider = None + if provider is not None and hasattr(provider, "shutdown"): + provider.shutdown() def set_sandbox_provider(provider: SandboxProvider) -> None: @@ -114,8 +159,12 @@ def set_sandbox_provider(provider: SandboxProvider) -> None: This allows injecting a custom or mock provider for testing purposes. + Note: any previously installed provider is replaced but not shut down; the + caller owns the lifecycle of the instance it is overwriting. + Args: provider: The SandboxProvider instance to use. """ global _default_sandbox_provider - _default_sandbox_provider = provider + with _provider_lock: + _default_sandbox_provider = provider diff --git a/backend/tests/test_sandbox_provider_lifecycle.py b/backend/tests/test_sandbox_provider_lifecycle.py new file mode 100644 index 000000000..2afb455c2 --- /dev/null +++ b/backend/tests/test_sandbox_provider_lifecycle.py @@ -0,0 +1,293 @@ +"""Concurrency regression tests for the sandbox provider singleton lifecycle. + +These guard the fix for the unsynchronized check-then-create in +``get_sandbox_provider`` and the unlocked ``reset``/``shutdown``/``set`` paths: +before the lock was added, concurrent cold-start callers could each construct a +separate provider and overwrite the global, and a ``reset``/``shutdown`` racing +a ``get`` could hand a caller ``None`` or a torn-down instance. + +Each test resets the process-global singleton on entry and in a ``finally`` on +exit, so tests never leak a provider into one another. +""" + +import threading +import time + +import deerflow.sandbox.sandbox_provider as sandbox_provider +from deerflow.sandbox.sandbox import Sandbox +from deerflow.sandbox.sandbox_provider import SandboxProvider + + +class SlowSandboxProvider(SandboxProvider): + """Provider whose constructor is slow, to widen the check-then-create gap.""" + + instances_created = 0 + instances_lock = threading.Lock() + + def __init__(self) -> None: + time.sleep(0.05) + with self.instances_lock: + type(self).instances_created += 1 + + def acquire(self, thread_id: str | None = None) -> str: + return "sandbox-id" + + def get(self, sandbox_id: str) -> Sandbox | None: + return None + + def release(self, sandbox_id: str) -> None: + pass + + +class ShutdownSandboxProvider(SlowSandboxProvider): + """Provider that also exposes ``shutdown``/``reset``, to exercise the paths + that run a provider callback outside ``_provider_lock``. + + Every constructed instance registers itself in ``registry`` so a test can + assert which instances were later torn down. + """ + + registry: list["ShutdownSandboxProvider"] = [] + registry_lock = threading.Lock() + + def __init__(self) -> None: + super().__init__() + self.shutdown_calls = 0 + self.reset_calls = 0 + with self.registry_lock: + type(self).registry.append(self) + + def shutdown(self) -> None: + # A non-trivial teardown: the fix runs this outside the lock, so a + # concurrent get() must not be blocked or torn by it. + time.sleep(0.02) + self.shutdown_calls += 1 + + def reset(self) -> None: + self.reset_calls += 1 + + +class _SandboxConfig: + use = "SlowSandboxProvider" + + +class _AppConfig: + sandbox = _SandboxConfig() + + +def _patch_provider_resolution(monkeypatch, cls=SlowSandboxProvider) -> None: + monkeypatch.setattr(sandbox_provider, "get_app_config", lambda: _AppConfig()) + monkeypatch.setattr(sandbox_provider, "resolve_class", lambda *args: cls) + + +def test_get_sandbox_provider_installs_one_singleton_under_concurrent_access(monkeypatch): + """Eight threads racing on a cold start must all observe the *same* installed + instance. + + Construction runs outside ``_provider_lock`` (so plugin ``__init__``/import + never runs under a non-reentrant lock), so racing callers may each build a + candidate; the contract is that exactly one is installed and every caller + sees it. The losers are torn down — see + ``test_losing_cold_start_racer_shuts_down_its_orphan``. + """ + sandbox_provider.reset_sandbox_provider() + SlowSandboxProvider.instances_created = 0 + _patch_provider_resolution(monkeypatch) + + n_threads = 8 + providers: list[SandboxProvider] = [] + providers_lock = threading.Lock() + # Barrier makes all threads enter get_sandbox_provider() at the same moment, + # so the race is triggered deterministically rather than probabilistically. + barrier = threading.Barrier(n_threads) + + def get_provider() -> None: + barrier.wait() + provider = sandbox_provider.get_sandbox_provider() + with providers_lock: + providers.append(provider) + + threads = [threading.Thread(target=get_provider) for _ in range(n_threads)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + # Every caller sees the one installed singleton, whichever candidate won. + assert len({id(provider) for provider in providers}) == 1 + installed = sandbox_provider.get_sandbox_provider() + assert all(p is installed for p in providers) + finally: + sandbox_provider.reset_sandbox_provider() + + +def test_reset_racing_get_of_live_singleton_never_returns_none_or_torn(monkeypatch): + """A reset racing concurrent gets of a *live* singleton must never hand back + ``None`` or a half-built instance: every returned value is a real provider. + + The singleton is populated *before* the barrier so the resetter tears down a + live instance while the getters read it — the interleaving that the unlocked + get-read path could turn into a ``None``/torn return. + """ + sandbox_provider.reset_sandbox_provider() + SlowSandboxProvider.instances_created = 0 + _patch_provider_resolution(monkeypatch) + + # Populate the singleton up front so the reset races a live instance. + sandbox_provider.get_sandbox_provider() + + results: list[object] = [] + results_lock = threading.Lock() + barrier = threading.Barrier(5) + + def getter() -> None: + barrier.wait() + provider = sandbox_provider.get_sandbox_provider() + with results_lock: + results.append(provider) + + def resetter() -> None: + barrier.wait() + sandbox_provider.reset_sandbox_provider() + + threads = [threading.Thread(target=getter) for _ in range(4)] + threads.append(threading.Thread(target=resetter)) + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + # Whatever each getter saw — the original singleton or a freshly rebuilt + # one after the reset — it must be a real provider, never None and never + # a partially constructed object. + assert results, "every getter recorded a result" + assert all(isinstance(p, SlowSandboxProvider) for p in results) + finally: + sandbox_provider.reset_sandbox_provider() + + +def test_shutdown_racing_get_of_live_singleton_never_returns_none_or_torn(monkeypatch): + """Same guarantee as the reset case, for ``shutdown_sandbox_provider()``. + + Uses a provider with a real (non-trivial) ``shutdown()`` so the teardown + runs outside the lock while getters read the global concurrently. + """ + sandbox_provider.reset_sandbox_provider() + SlowSandboxProvider.instances_created = 0 + _patch_provider_resolution(monkeypatch, cls=ShutdownSandboxProvider) + + sandbox_provider.get_sandbox_provider() # live singleton before the race + + results: list[object] = [] + results_lock = threading.Lock() + barrier = threading.Barrier(5) + + def getter() -> None: + barrier.wait() + provider = sandbox_provider.get_sandbox_provider() + with results_lock: + results.append(provider) + + def shutter() -> None: + barrier.wait() + sandbox_provider.shutdown_sandbox_provider() + + threads = [threading.Thread(target=getter) for _ in range(4)] + threads.append(threading.Thread(target=shutter)) + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + assert results + assert all(isinstance(p, ShutdownSandboxProvider) for p in results) + finally: + sandbox_provider.reset_sandbox_provider() + + +def test_set_racing_get_never_returns_none_or_torn(monkeypatch): + """``set_sandbox_provider()`` racing concurrent gets must never expose a + ``None`` global: every getter sees a fully constructed provider.""" + sandbox_provider.reset_sandbox_provider() + SlowSandboxProvider.instances_created = 0 + _patch_provider_resolution(monkeypatch) + + sandbox_provider.get_sandbox_provider() # live singleton before the race + injected = SlowSandboxProvider() + + results: list[object] = [] + results_lock = threading.Lock() + barrier = threading.Barrier(5) + + def getter() -> None: + barrier.wait() + provider = sandbox_provider.get_sandbox_provider() + with results_lock: + results.append(provider) + + def setter() -> None: + barrier.wait() + sandbox_provider.set_sandbox_provider(injected) + + threads = [threading.Thread(target=getter) for _ in range(4)] + threads.append(threading.Thread(target=setter)) + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + assert results + assert all(isinstance(p, SlowSandboxProvider) for p in results) + finally: + sandbox_provider.reset_sandbox_provider() + + +def test_losing_cold_start_racer_shuts_down_its_orphan(monkeypatch): + """When two cold-start callers race, the loser must shut down the instance it + built so a side-effectful constructor (idle-checker thread, etc.) does not + leak — the core consequence in issue #3721. + + With ``ShutdownSandboxProvider`` every constructed-but-discarded instance has + its ``shutdown()`` invoked, so exactly ``(constructed - 1)`` of them are torn + down (the single winner is kept). + """ + sandbox_provider.reset_sandbox_provider() + ShutdownSandboxProvider.instances_created = 0 + ShutdownSandboxProvider.registry = [] + _patch_provider_resolution(monkeypatch, cls=ShutdownSandboxProvider) + + n_threads = 8 + providers: list[ShutdownSandboxProvider] = [] + providers_lock = threading.Lock() + barrier = threading.Barrier(n_threads) + + def get_provider() -> None: + barrier.wait() + provider = sandbox_provider.get_sandbox_provider() + with providers_lock: + providers.append(provider) + + threads = [threading.Thread(target=get_provider) for _ in range(n_threads)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + try: + winner = sandbox_provider.get_sandbox_provider() + # Exactly one instance is installed and returned to every caller. + assert len({id(p) for p in providers}) == 1 + assert all(p is winner for p in providers) + # The winner is never torn down... + assert winner.shutdown_calls == 0 + # ...and every loser that was constructed had shutdown() called on it + # exactly once. + losers = [inst for inst in ShutdownSandboxProvider.registry if inst is not winner] + assert len(losers) == ShutdownSandboxProvider.instances_created - 1 + assert all(inst.shutdown_calls == 1 for inst in losers) + finally: + sandbox_provider.reset_sandbox_provider()