mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(aio): keep renewal shutdown retryable (#6317)
* fix(aio): keep renewal shutdown retryable * test(aio): cover renewal shutdown retry * test(aio): stabilize renewal shutdown retry liveness * fix(aio): preserve signal forwarding on shutdown timeout * test(aio): match renewal shutdown timeout diagnostic
This commit is contained in:
parent
9c5e5253ea
commit
c5b9964c25
@ -1364,8 +1364,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
def _stop_lease_renewal(self) -> None:
|
||||
self._renewal_stop.set()
|
||||
thread = self._renewal_thread
|
||||
if thread is not None and thread.is_alive() and thread is not threading.current_thread():
|
||||
if thread is None or thread is threading.current_thread():
|
||||
return
|
||||
if thread.is_alive():
|
||||
thread.join(timeout=5)
|
||||
if thread.is_alive():
|
||||
raise RuntimeError("Sandbox lease-renewal thread is still running after stop timeout")
|
||||
|
||||
def _lease_renewal_loop(self) -> None:
|
||||
interval = self._ownership_config.renewal_interval_seconds
|
||||
@ -1557,7 +1561,10 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._original_sighup = signal.getsignal(signal.SIGHUP) if hasattr(signal, "SIGHUP") else None
|
||||
|
||||
def signal_handler(signum, frame):
|
||||
self.shutdown()
|
||||
try:
|
||||
self.shutdown()
|
||||
except Exception:
|
||||
logger.exception("Sandbox shutdown failed while handling signal %s; forwarding signal", signum)
|
||||
if signum == signal.SIGTERM:
|
||||
original = self._original_sigterm
|
||||
elif hasattr(signal, "SIGHUP") and signum == signal.SIGHUP:
|
||||
@ -2664,24 +2671,25 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
|
||||
try:
|
||||
self._stop_idle_checker()
|
||||
# Stop renewing before detaching tracked ownership: the destroy paths
|
||||
# claim ownership themselves, and a renewal racing teardown can
|
||||
# re-publish leases we are about to drop. A bounded join timeout must
|
||||
# fail closed so a retry still owns every active and warm entry.
|
||||
self._stop_lease_renewal()
|
||||
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.
|
||||
# Do not detach tracked sandboxes before both maintenance workers are
|
||||
# known stopped. If either bounded join fails, a retry must still own
|
||||
# every active and 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()
|
||||
|
||||
# Stop renewing before destroying: the destroy paths claim ownership
|
||||
# themselves, and a renewal racing them only re-publishes leases we are
|
||||
# about to drop.
|
||||
self._stop_lease_renewal()
|
||||
|
||||
logger.info(f"Shutting down {len(sandbox_ids)} active + {len(warm_items)} warm-pool sandbox(es)")
|
||||
|
||||
for sandbox_id in sandbox_ids:
|
||||
|
||||
@ -2189,6 +2189,77 @@ def test_shutdown_keeps_aio_warm_entries_owned_when_idle_checker_stop_times_out(
|
||||
assert provider._warm_pool_identity == {}
|
||||
|
||||
|
||||
def test_shutdown_keeps_aio_warm_entries_owned_when_lease_renewal_stop_times_out(tmp_path):
|
||||
"""A live renewal worker must keep AIO ownership attached for shutdown 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-renewal-retry", sandbox_url="http://warm-renewal-retry")
|
||||
provider._warm_pool = {"warm-renewal-retry": (warm_info, 1.0)}
|
||||
provider._warm_pool_identity = {"warm-renewal-retry": ("default", "thread-retry")}
|
||||
provider._stop_idle_checker = MagicMock()
|
||||
provider._destroy_warm_entry = MagicMock()
|
||||
provider._ownership.close = MagicMock()
|
||||
|
||||
renewal_thread = MagicMock()
|
||||
renewal_thread.is_alive.side_effect = [True, True, False, False]
|
||||
provider._renewal_thread = renewal_thread
|
||||
|
||||
with pytest.raises(RuntimeError, match="lease-renewal thread is still running after stop timeout"):
|
||||
provider.shutdown()
|
||||
|
||||
assert provider._shutdown_called is False
|
||||
assert provider._warm_pool == {"warm-renewal-retry": (warm_info, 1.0)}
|
||||
assert provider._warm_pool_identity == {"warm-renewal-retry": ("default", "thread-retry")}
|
||||
renewal_thread.join.assert_called_once_with(timeout=5)
|
||||
provider._destroy_warm_entry.assert_not_called()
|
||||
provider._ownership.close.assert_not_called()
|
||||
|
||||
provider.shutdown()
|
||||
|
||||
provider._destroy_warm_entry.assert_called_once_with(
|
||||
"warm-renewal-retry",
|
||||
warm_info,
|
||||
reason="shutdown",
|
||||
still_reapable=ANY,
|
||||
)
|
||||
assert provider._warm_pool == {}
|
||||
assert provider._warm_pool_identity == {}
|
||||
|
||||
|
||||
def test_signal_handler_forwards_original_signal_when_shutdown_fails(tmp_path, monkeypatch, caplog):
|
||||
"""A fail-closed shutdown must not swallow or replace the process signal."""
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
provider = _make_provider(tmp_path)
|
||||
provider.shutdown = MagicMock(side_effect=RuntimeError("renewal worker still alive"))
|
||||
|
||||
forwarded = MagicMock()
|
||||
installed = {}
|
||||
|
||||
def fake_getsignal(signum):
|
||||
return forwarded if signum == aio_mod.signal.SIGTERM else aio_mod.signal.SIG_IGN
|
||||
|
||||
def fake_signal(signum, handler):
|
||||
installed[signum] = handler
|
||||
|
||||
monkeypatch.setattr(aio_mod.signal, "getsignal", fake_getsignal)
|
||||
monkeypatch.setattr(aio_mod.signal, "signal", fake_signal)
|
||||
|
||||
provider._register_signal_handlers()
|
||||
|
||||
with caplog.at_level("ERROR"):
|
||||
installed[aio_mod.signal.SIGTERM](aio_mod.signal.SIGTERM, None)
|
||||
|
||||
provider.shutdown.assert_called_once_with()
|
||||
forwarded.assert_called_once_with(aio_mod.signal.SIGTERM, None)
|
||||
assert "Sandbox shutdown failed while handling signal" in caplog.text
|
||||
|
||||
|
||||
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")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user