mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(sandbox): keep shutdown retryable while idle reaper is live (#6231)
* fix(sandbox): fail shutdown while idle reaper is live * test(sandbox): pin live idle reaper timeout * fix(sandbox): keep shutdown retryable after reaper timeout * fix(sandbox): keep shutdown retryable after reaper timeout * fix(sandbox): keep shutdown retryable after reaper timeout * fix(sandbox): keep shutdown retryable after reaper timeout * docs(sandbox): pin idle-reaper shutdown ownership * fix(sandbox): keep AIO warm entries retryable * test(sandbox): fix AIO retry regression import * test(sandbox): cover retryable provider shutdown guards * test(sandbox): assert warm ownership survives failed shutdown
This commit is contained in:
parent
35023bb908
commit
68dcaa6025
@ -2661,12 +2661,22 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
if self._shutdown_called:
|
||||
return
|
||||
self._shutdown_called = True
|
||||
|
||||
try:
|
||||
self._stop_idle_checker()
|
||||
except Exception:
|
||||
with self._lock:
|
||||
self._shutdown_called = False
|
||||
raise
|
||||
|
||||
# Do not detach tracked sandboxes before the reaper is known stopped.
|
||||
# If the bounded join fails, a retry must still own every warm entry.
|
||||
with self._lock:
|
||||
sandbox_ids = list(self._sandboxes.keys())
|
||||
warm_items = list(self._warm_pool.items())
|
||||
self._warm_pool.clear()
|
||||
self._warm_pool_identity.clear()
|
||||
|
||||
self._stop_idle_checker()
|
||||
# Stop renewing before destroying: the destroy paths claim ownership
|
||||
# themselves, and a renewal racing them only re-publishes leases we are
|
||||
# about to drop.
|
||||
|
||||
@ -653,7 +653,12 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
return
|
||||
self._shutdown_called = True
|
||||
|
||||
self._stop_idle_checker()
|
||||
try:
|
||||
self._stop_idle_checker()
|
||||
except Exception:
|
||||
with self._lock:
|
||||
self._shutdown_called = False
|
||||
raise
|
||||
|
||||
with self._lock:
|
||||
active = list(self._boxes.values())
|
||||
|
||||
@ -391,7 +391,12 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro
|
||||
if self._shutdown_called:
|
||||
return
|
||||
self._shutdown_called = True
|
||||
self._stop_idle_checker()
|
||||
try:
|
||||
self._stop_idle_checker()
|
||||
except Exception:
|
||||
with self._lock:
|
||||
self._shutdown_called = False
|
||||
raise
|
||||
with self._lock:
|
||||
sandboxes = list(self._sandboxes.values()) + [entry for entry, _ in self._warm_pool.values()]
|
||||
self._sandboxes.clear()
|
||||
|
||||
@ -456,7 +456,12 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider
|
||||
return
|
||||
self._shutdown_called = True
|
||||
|
||||
self._stop_idle_checker()
|
||||
try:
|
||||
self._stop_idle_checker()
|
||||
except Exception:
|
||||
with self._lock:
|
||||
self._shutdown_called = False
|
||||
raise
|
||||
|
||||
with self._lock:
|
||||
active = list(self._sandboxes.values())
|
||||
|
||||
@ -99,11 +99,13 @@ class WarmPoolLifecycleMixin[WarmEntryT]:
|
||||
logger.info("Started warm-pool idle checker thread (timeout: %ss)", self._config.get("idle_timeout", DEFAULT_IDLE_TIMEOUT))
|
||||
|
||||
def _stop_idle_checker(self) -> None:
|
||||
"""Stop the idle checker thread and wait for it to exit when running."""
|
||||
"""Stop the idle checker thread and fail if bounded teardown cannot own it."""
|
||||
self._idle_checker_stop.set()
|
||||
thread = self._idle_checker_thread
|
||||
if thread is not None and thread.is_alive() and thread is not threading.current_thread():
|
||||
thread.join(timeout=5)
|
||||
if thread.is_alive():
|
||||
raise RuntimeError("warm-pool idle checker is still running after stop timeout")
|
||||
|
||||
def _idle_checker_loop(self) -> None:
|
||||
"""Run periodic idle cleanup until the stop event is set."""
|
||||
|
||||
@ -96,7 +96,7 @@ cancellation before propagating it.
|
||||
Health probes require an exact `ok` line and reject `Error:`/`Exit Code:` diagnostic lines; other login-shell stdout/stderr is tolerated. Failures log sandbox id and output before eviction.
|
||||
|
||||
|
||||
**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.
|
||||
**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. Shutdown must not tear down provider resources while the shared idle-checker thread is still alive: `_stop_idle_checker()` uses a bounded join and raises if that bound expires; providers reset their shutdown guard on that failure so a later shutdown can retry instead of silently becoming a no-op.
|
||||
|
||||
**Virtual Path System**:
|
||||
- Agent sees: `/mnt/user-data/{workspace,uploads,outputs}`, `/mnt/skills`
|
||||
|
||||
@ -8,7 +8,7 @@ import os
|
||||
import stat
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import ANY, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from _windows_acl_helpers import _windows_acl_owner_sid, _windows_acl_sids
|
||||
@ -2151,6 +2151,44 @@ def test_destroy_swallows_close_errors_and_still_destroys_backend(tmp_path, capl
|
||||
provider._backend.destroy.assert_called_once()
|
||||
|
||||
|
||||
def test_shutdown_keeps_aio_warm_entries_owned_when_idle_checker_stop_times_out(tmp_path):
|
||||
"""A failed reaper join must leave AIO warm entries available to a retry."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
provider = _make_provider(tmp_path)
|
||||
provider._lock = aio_mod.threading.Lock()
|
||||
provider._shutdown_called = False
|
||||
provider._sandboxes = {}
|
||||
provider._sandbox_infos = {}
|
||||
provider._thread_sandboxes = {}
|
||||
provider._last_activity = {}
|
||||
warm_info = aio_mod.SandboxInfo(sandbox_id="warm-retry", sandbox_url="http://warm-retry")
|
||||
provider._warm_pool = {"warm-retry": (warm_info, 1.0)}
|
||||
provider._warm_pool_identity = {"warm-retry": ("default", "thread-retry")}
|
||||
provider._stop_idle_checker = MagicMock(side_effect=[RuntimeError("reaper still alive"), None])
|
||||
provider._stop_lease_renewal = MagicMock()
|
||||
provider._destroy_warm_entry = MagicMock()
|
||||
provider._ownership.close = MagicMock()
|
||||
|
||||
with pytest.raises(RuntimeError, match="reaper still alive"):
|
||||
provider.shutdown()
|
||||
|
||||
assert provider._shutdown_called is False
|
||||
assert provider._warm_pool == {"warm-retry": (warm_info, 1.0)}
|
||||
assert provider._warm_pool_identity == {"warm-retry": ("default", "thread-retry")}
|
||||
provider._destroy_warm_entry.assert_not_called()
|
||||
|
||||
provider.shutdown()
|
||||
|
||||
provider._destroy_warm_entry.assert_called_once_with(
|
||||
"warm-retry",
|
||||
warm_info,
|
||||
reason="shutdown",
|
||||
still_reapable=ANY,
|
||||
)
|
||||
assert provider._warm_pool == {}
|
||||
assert provider._warm_pool_identity == {}
|
||||
|
||||
|
||||
def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(tmp_path):
|
||||
"""AIO active-idle cleanup must remain local while warm expiry uses the shared lifecycle."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
|
||||
@ -2,9 +2,13 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.community.warm_pool_lifecycle import DEFAULT_IDLE_TIMEOUT, DEFAULT_REPLICAS, WarmPoolLifecycleMixin
|
||||
|
||||
@ -78,6 +82,44 @@ def test_reap_expired_warm_noops_when_timeout_disabled() -> None:
|
||||
assert provider.destroyed == []
|
||||
|
||||
|
||||
def test_stop_idle_checker_rejects_live_thread_after_timeout() -> None:
|
||||
provider = _Provider()
|
||||
thread = mock.Mock()
|
||||
thread.is_alive.return_value = True
|
||||
provider._idle_checker_thread = thread
|
||||
|
||||
with pytest.raises(RuntimeError, match="still running after stop timeout"):
|
||||
provider._stop_idle_checker()
|
||||
|
||||
thread.join.assert_called_once_with(timeout=5)
|
||||
assert provider._idle_checker_thread is thread
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("module_name", "class_name"),
|
||||
[
|
||||
("deerflow.community.aio_sandbox.aio_sandbox_provider", "AioSandboxProvider"),
|
||||
("deerflow.community.boxlite.provider", "BoxliteProvider"),
|
||||
("deerflow.community.tenki.provider", "TenkiSandboxProvider"),
|
||||
("deerflow.community.opensandbox.provider", "OpenSandboxProvider"),
|
||||
],
|
||||
)
|
||||
def test_provider_shutdown_resets_guard_when_idle_checker_stop_fails(module_name: str, class_name: str) -> None:
|
||||
provider_cls = getattr(importlib.import_module(module_name), class_name)
|
||||
provider = provider_cls.__new__(provider_cls)
|
||||
provider._lock = threading.Lock()
|
||||
provider._shutdown_called = False
|
||||
warm_entry = object()
|
||||
provider._warm_pool = {"warm": (warm_entry, 1.0)}
|
||||
provider._stop_idle_checker = mock.Mock(side_effect=RuntimeError("reaper still alive"))
|
||||
|
||||
with pytest.raises(RuntimeError, match="reaper still alive"):
|
||||
provider.shutdown()
|
||||
|
||||
assert provider._shutdown_called is False
|
||||
assert provider._warm_pool == {"warm": (warm_entry, 1.0)}
|
||||
|
||||
|
||||
def test_start_idle_checker_uses_monkeypatchable_interval(monkeypatch) -> None:
|
||||
provider = _Provider(idle_timeout=0.01)
|
||||
monkeypatch.setattr(_Provider, "IDLE_CHECK_INTERVAL", 0.01)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user