diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index a85799ac29..d1a73679d0 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -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: diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index 4bf1b0bf04..48bc3d7d37 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -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")