mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 13:40:37 +00:00
eb7c342fb1
* fix(sandbox): measure warm-pool idle age on a monotonic clock Warm-pool expiry measures a duration, but every provider stamped and reaped entries using the wall clock. A forward correction could therefore destroy fresh sandboxes, while a backward correction could retain expired ones until the clock caught up. Move every warm-pool writer and reader to time.monotonic(), including the Tenki and OpenSandbox reset paths missed by the earlier closed attempt. Keep absolute SandboxInfo.created_at and reconciliation grace timestamps on the wall clock. Convert AIO's sibling active-idle timestamps as one clock domain, and update finite clock mocks for the additional duration read. Fixes #6970 * test(sandbox): keep active-idle race fixtures in the monotonic domain The production AIO activity clock moved to monotonic time, but the shared orphan-reconciliation helper still stamped active fixtures with wall time. CI then computed a negative idle age, so two teardown race tests never reached the ownership gate. Use the same monotonic domain as production. * test(sandbox): read warm-pool idle fixtures on the monotonic clock The idle-age conversion left several fixtures seeding the fields the reaper now reads with values from the clock it no longer reads: - `test_idle_reaping_forces_a_new_remote` froze `warm_pool_lifecycle.time.time`, so the benchmark's reaper compared against real elapsed monotonic seconds — zero over the test's lifetime. Nothing expired, the entry survived, and the second turn came back as a warm hit. - Three orphan-reconciliation fixtures seeded `_last_activity` with `0.0` and two aio fixtures with `time.time()`, both of which are far below a running monotonic clock: the computed idle age went negative, so the reaper never reached the ownership claim these tests are built to exercise. Seed them from `time.monotonic()` instead, and keep `created_at` (a wall-clock field) on `time.time()`.
166 lines
6.1 KiB
Python
166 lines
6.1 KiB
Python
"""Unit tests for shared warm-pool lifecycle mechanics."""
|
|
|
|
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
|
|
|
|
|
|
class _Provider(WarmPoolLifecycleMixin[str]):
|
|
_idle_checker_thread_name = "test-warm-pool-reaper"
|
|
|
|
def __init__(self, *, replicas: int = DEFAULT_REPLICAS, idle_timeout: float = DEFAULT_IDLE_TIMEOUT, active_count: int = 0) -> None:
|
|
self._lock = threading.Lock()
|
|
self._warm_pool: dict[str, tuple[str, float]] = {}
|
|
self._config: dict[str, Any] = {"replicas": replicas, "idle_timeout": idle_timeout}
|
|
self._idle_checker_stop = threading.Event()
|
|
self._idle_checker_thread: threading.Thread | None = None
|
|
self.active_count = active_count
|
|
self.destroyed: list[tuple[str, str, str]] = []
|
|
|
|
def _active_count_locked(self) -> int:
|
|
return self.active_count
|
|
|
|
def _destroy_warm_entry(self, sandbox_id: str, entry: str, *, reason: str) -> None:
|
|
self.destroyed.append((sandbox_id, entry, reason))
|
|
|
|
|
|
def test_replica_count_includes_active_and_warm_entries() -> None:
|
|
provider = _Provider(replicas=2, active_count=1)
|
|
provider._warm_pool["warm-1"] = ("entry-1", time.monotonic())
|
|
|
|
assert provider._replica_count() == (2, 2)
|
|
|
|
|
|
def test_evict_oldest_warm_removes_and_destroys_oldest_entry() -> None:
|
|
provider = _Provider()
|
|
provider._warm_pool["new"] = ("entry-new", 200.0)
|
|
provider._warm_pool["old"] = ("entry-old", 100.0)
|
|
|
|
evicted = provider._evict_oldest_warm()
|
|
|
|
assert evicted == "old"
|
|
assert "old" not in provider._warm_pool
|
|
assert "new" in provider._warm_pool
|
|
assert provider.destroyed == [("old", "entry-old", "replica_enforcement")]
|
|
|
|
|
|
def test_evict_oldest_warm_returns_none_when_pool_empty() -> None:
|
|
provider = _Provider()
|
|
|
|
assert provider._evict_oldest_warm() is None
|
|
assert provider.destroyed == []
|
|
|
|
|
|
def test_reap_expired_warm_destroys_only_expired_entries() -> None:
|
|
provider = _Provider()
|
|
now = time.monotonic()
|
|
provider._warm_pool["expired"] = ("entry-expired", now - 100)
|
|
provider._warm_pool["fresh"] = ("entry-fresh", now)
|
|
|
|
provider._reap_expired_warm(idle_timeout=10)
|
|
|
|
assert "expired" not in provider._warm_pool
|
|
assert "fresh" in provider._warm_pool
|
|
assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]
|
|
|
|
|
|
def test_reap_expired_warm_noops_when_timeout_disabled() -> None:
|
|
provider = _Provider(idle_timeout=0)
|
|
provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 100)
|
|
|
|
provider._reap_expired_warm(idle_timeout=0)
|
|
|
|
assert "expired" in provider._warm_pool
|
|
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)
|
|
provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 10)
|
|
|
|
provider._start_idle_checker()
|
|
deadline = time.time() + 1
|
|
while "expired" in provider._warm_pool and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
provider._stop_idle_checker()
|
|
|
|
assert "expired" not in provider._warm_pool
|
|
assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]
|
|
assert provider._idle_checker_thread is not None
|
|
assert not provider._idle_checker_thread.is_alive()
|
|
|
|
|
|
@pytest.mark.parametrize("wall_step", [10_000.0, -10_000.0])
|
|
def test_reap_expired_warm_ignores_wall_clock_steps(monkeypatch, wall_step: float) -> None:
|
|
"""NTP/VM wall-clock jumps cannot change a warm entry's monotonic idle age."""
|
|
provider = _Provider()
|
|
parked_at = time.monotonic() - 1
|
|
provider._warm_pool["fresh"] = ("entry-fresh", parked_at)
|
|
wall_now = time.time()
|
|
monkeypatch.setattr(time, "time", lambda: wall_now + wall_step)
|
|
|
|
provider._reap_expired_warm(idle_timeout=600)
|
|
|
|
assert provider._warm_pool == {"fresh": ("entry-fresh", parked_at)}
|
|
assert provider.destroyed == []
|
|
|
|
|
|
def test_reap_expired_warm_uses_elapsed_monotonic_time_when_wall_clock_moves_backward(monkeypatch) -> None:
|
|
"""A backward wall step cannot preserve an entry whose real idle age expired."""
|
|
provider = _Provider()
|
|
provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 601)
|
|
wall_now = time.time()
|
|
monkeypatch.setattr(time, "time", lambda: wall_now - 10_000)
|
|
|
|
provider._reap_expired_warm(idle_timeout=600)
|
|
|
|
assert provider._warm_pool == {}
|
|
assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]
|