Czile 30841c3b9f
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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 08:06:17 +08:00

171 lines
6.6 KiB
Python

import asyncio
import threading
from abc import ABC, abstractmethod
from deerflow.config import get_app_config
from deerflow.reflection import resolve_class
from deerflow.sandbox.sandbox import Sandbox
class SandboxProvider(ABC):
"""Abstract base class for sandbox providers"""
uses_thread_data_mounts: bool = False
needs_upload_permission_adjustment: bool = True
@abstractmethod
def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
"""Acquire a sandbox environment and return its ID.
Returns:
The ID of the acquired sandbox environment.
"""
pass
async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str:
"""Acquire a sandbox without blocking the event loop.
Most sandbox providers expose a synchronous lifecycle API because local
Docker/provisioner operations are blocking. Async runtimes should call
this method so those blocking operations run in a worker thread instead
of stalling the event loop.
"""
return await asyncio.to_thread(self.acquire, thread_id, user_id=user_id)
@abstractmethod
def get(self, sandbox_id: str) -> Sandbox | None:
"""Get a sandbox environment by ID.
Args:
sandbox_id: The ID of the sandbox environment to retain.
"""
pass
@abstractmethod
def release(self, sandbox_id: str) -> None:
"""Release a sandbox environment.
Args:
sandbox_id: The ID of the sandbox environment to destroy.
"""
pass
def reset(self) -> None:
"""Clear cached state that survives provider instance replacement."""
pass
_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:
"""Get the sandbox provider singleton.
Returns a cached singleton instance. Use `reset_sandbox_provider()` to clear
the cache, or `shutdown_sandbox_provider()` to properly shutdown and clear.
Returns:
A sandbox provider instance.
"""
global _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:
"""Reset the sandbox provider singleton.
This clears the cached instance without calling shutdown.
The next call to `get_sandbox_provider()` will create a new instance.
Useful for testing or when switching configurations.
Providers can override `reset()` to clear any module-level state they keep
alive across instances (for example, `LocalSandboxProvider`'s cached
`LocalSandbox` singleton). Without it, config/mount changes would not take
effect on the next acquire().
Note: If the provider has active sandboxes, they will be orphaned.
Use `shutdown_sandbox_provider()` for proper cleanup.
"""
global _default_sandbox_provider
# 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:
"""Shutdown and reset the sandbox provider.
This properly shuts down the provider (releasing all sandboxes)
before clearing the singleton. Call this when the application
is shutting down or when you need to completely reset the sandbox system.
"""
global _default_sandbox_provider
# 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:
"""Set a custom sandbox provider instance.
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
with _provider_lock:
_default_sandbox_provider = provider