mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-29 22:50:37 +00:00
fix(sandbox): keep lease renewal independent of health probes
This commit is contained in:
parent
cd1ff39725
commit
96535c739b
@ -211,6 +211,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._idle_checker_thread: threading.Thread | None = None
|
||||
self._renewal_stop = threading.Event()
|
||||
self._renewal_thread: threading.Thread | None = None
|
||||
self._health_check_thread: threading.Thread | None = None
|
||||
# Per-instance id used for cross-instance sandbox ownership leases (#4206).
|
||||
self._owner_id = generate_owner_id()
|
||||
|
||||
@ -1305,6 +1306,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
thread = self._renewal_thread
|
||||
if thread is not None and thread.is_alive() and thread is not threading.current_thread():
|
||||
thread.join(timeout=5)
|
||||
health_thread = getattr(self, "_health_check_thread", None)
|
||||
if health_thread is not None and health_thread.is_alive() and health_thread is not threading.current_thread():
|
||||
health_thread.join(timeout=5)
|
||||
|
||||
def _lease_renewal_loop(self) -> None:
|
||||
interval = self._ownership_config.renewal_interval_seconds
|
||||
@ -1314,9 +1318,29 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
except Exception:
|
||||
logger.exception("Error in sandbox ownership renewal loop")
|
||||
try:
|
||||
self._health_check_owned_sandboxes()
|
||||
self._schedule_health_check()
|
||||
except Exception:
|
||||
logger.exception("Error in sandbox health check loop")
|
||||
logger.exception("Error scheduling sandbox health check")
|
||||
|
||||
def _schedule_health_check(self) -> None:
|
||||
"""Start at most one health scan without delaying lease renewal."""
|
||||
if not self._config.get("auto_restart", True) or self._renewal_stop.is_set():
|
||||
return
|
||||
thread = getattr(self, "_health_check_thread", None)
|
||||
if thread is not None and thread.is_alive():
|
||||
return
|
||||
self._health_check_thread = threading.Thread(
|
||||
target=self._run_health_check,
|
||||
name="sandbox-health-check",
|
||||
daemon=True,
|
||||
)
|
||||
self._health_check_thread.start()
|
||||
|
||||
def _run_health_check(self) -> None:
|
||||
try:
|
||||
self._health_check_owned_sandboxes()
|
||||
except Exception:
|
||||
logger.exception("Error in sandbox health check loop")
|
||||
|
||||
def _health_check_owned_sandboxes(self) -> None:
|
||||
"""Evict cached sandboxes whose containers died since last use.
|
||||
@ -1324,10 +1348,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
``get()``/``get_scoped()`` are intentionally pure in-memory lookups
|
||||
(async tool paths call them directly on the event loop), so a
|
||||
mid-run container crash is otherwise only caught the next time
|
||||
``acquire``/``reclaim`` runs. Riding the existing renewal thread lets a
|
||||
crash between two tool calls be detected and evicted — via the same
|
||||
ownership-fenced teardown as every other reap path — within one
|
||||
renewal interval instead of persisting for the rest of the run.
|
||||
``acquire``/``reclaim`` runs. A worker scheduled by the renewal loop
|
||||
detects a crash between tool calls without delaying the next lease
|
||||
refresh, then uses the usual ownership-fenced teardown.
|
||||
"""
|
||||
if not self._config.get("auto_restart", True):
|
||||
return
|
||||
@ -1336,7 +1359,11 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
tracked = list(self._sandbox_infos.items()) + [(sandbox_id, info) for sandbox_id, (info, _) in self._warm_pool.items()]
|
||||
|
||||
for sandbox_id, info in tracked:
|
||||
if getattr(self, "_shutdown_called", False):
|
||||
return
|
||||
if self._check_tracked_sandbox_alive(sandbox_id, info) is False:
|
||||
if getattr(self, "_shutdown_called", False):
|
||||
return
|
||||
self._drop_unhealthy_sandbox(
|
||||
sandbox_id,
|
||||
"periodic health check",
|
||||
|
||||
@ -98,6 +98,8 @@ to respond.
|
||||
|
||||
**Shared warm-pool lifecycle:** community sandbox providers that keep released sandboxes alive for fast reuse share `deerflow.community.warm_pool_lifecycle.WarmPoolLifecycleMixin`. The mixin owns the common `DEFAULT_IDLE_TIMEOUT=600`, `IDLE_CHECK_INTERVAL=60`, `DEFAULT_REPLICAS=3`, idle-checker loop, warm-pool expiry, oldest-warm eviction, replica counting, and soft-cap logging. Providers remain responsible for their own active registries, creation/discovery, health checks, and destroy hook (`_destroy_warm_entry`): AIO destroys `SandboxInfo` through its backend; Boxlite closes loop-affine `BoxliteBox` handles; Tenki closes the microVM session (`TenkiSandbox.close`, which terminates the remote sandbox). AIO keeps active-idle cleanup outside the mixin and delegates only warm-pool expiry to the shared helper.
|
||||
|
||||
**AIO health polling:** The lease thread schedules at most one separate health worker, so slow active/warm container probes never delay lease renewal.
|
||||
|
||||
**Virtual Path System**:
|
||||
- Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills`
|
||||
- Physical: `backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/...`; raw skills stay under `deer-flow/skills/` and managed integration storage. Unrestricted sandboxes read `backend/.deer-flow/skills_view/public/` and `backend/.deer-flow/users/{user_id}/skills_view/{custom,legacy,integrations}/`; explicit lead Agent policies read `backend/.deer-flow/users/{user_id}/threads/{thread_id}/skills_view/{public,custom,legacy,integrations}/`.
|
||||
|
||||
@ -58,7 +58,7 @@ def test_health_check_evicts_sandbox_that_crashed_between_tool_calls(tmp_path):
|
||||
|
||||
``get()``/``get_scoped()`` stay pure in-memory lookups (see
|
||||
``test_get_remains_in_memory_lookup`` / ``test_get_uses_in_memory_registry_only``),
|
||||
so the renewal-thread health check is what catches a mid-run crash: it must
|
||||
so the scheduled health worker is what catches a mid-run crash: it must
|
||||
evict the cached sandbox so the *next* cache lookup misses and the caller
|
||||
falls through to ``acquire()`` for a fresh container.
|
||||
"""
|
||||
@ -95,3 +95,44 @@ def test_health_check_respects_auto_restart_disabled(tmp_path):
|
||||
provider._backend.is_alive.assert_not_called()
|
||||
provider._backend.destroy.assert_not_called()
|
||||
assert provider.get("sandbox-crashed-disabled") is sandbox
|
||||
|
||||
|
||||
def test_slow_health_check_does_not_delay_lease_renewal():
|
||||
"""A blocked remote probe must not hold up the next ownership renewal."""
|
||||
module = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
provider = module.AioSandboxProvider.__new__(module.AioSandboxProvider)
|
||||
provider._config = {"auto_restart": True}
|
||||
provider._ownership_config = SimpleNamespace(renewal_interval_seconds=0.01, ttl_multiplier=4)
|
||||
provider._renewal_stop = threading.Event()
|
||||
provider._renewal_thread = None
|
||||
provider._health_check_thread = None
|
||||
|
||||
health_started = threading.Event()
|
||||
allow_health = threading.Event()
|
||||
second_renewal = threading.Event()
|
||||
renewals: list[int] = []
|
||||
health_checks: list[int] = []
|
||||
|
||||
def renew() -> None:
|
||||
renewals.append(1)
|
||||
if len(renewals) >= 2:
|
||||
second_renewal.set()
|
||||
|
||||
def check_health() -> None:
|
||||
health_checks.append(1)
|
||||
health_started.set()
|
||||
allow_health.wait(timeout=2)
|
||||
|
||||
provider._renew_owned_leases = renew
|
||||
provider._health_check_owned_sandboxes = check_health
|
||||
|
||||
provider._start_lease_renewal()
|
||||
try:
|
||||
assert health_started.wait(timeout=1)
|
||||
assert second_renewal.wait(timeout=0.5), "a slow health probe delayed lease renewal"
|
||||
assert len(health_checks) == 1, "health checks must not spawn unbounded workers"
|
||||
finally:
|
||||
allow_health.set()
|
||||
provider._stop_lease_renewal()
|
||||
assert provider._health_check_thread is not None
|
||||
assert not provider._health_check_thread.is_alive()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user