mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(sandbox): measure warm-pool idle age on a monotonic clock (#7035)
* 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()`.
This commit is contained in:
FanouZeng-TT
GitHub
parent
0d73798beb
commit
eb7c342fb1
@ -901,7 +901,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
deferred += 1
|
||||
logger.debug("Deferring container %s during reconciliation: this instance is tearing it down", info.sandbox_id)
|
||||
continue
|
||||
self._warm_pool[info.sandbox_id] = (info, current_time)
|
||||
self._warm_pool[info.sandbox_id] = (info, time.monotonic())
|
||||
self._warm_pool_identity[info.sandbox_id] = None
|
||||
self._unowned_since.pop(info.sandbox_id, None)
|
||||
adopted += 1
|
||||
@ -1509,7 +1509,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
logger.warning(f"Error closing sandbox {sandbox_id} after losing its lease: {e}")
|
||||
|
||||
def _cleanup_idle_sandboxes(self, idle_timeout: float) -> None:
|
||||
current_time = time.time()
|
||||
current_time = time.monotonic()
|
||||
active_to_destroy = []
|
||||
|
||||
with self._lock:
|
||||
@ -1535,7 +1535,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
# Already released or destroyed by another path — skip.
|
||||
logger.info(f"Sandbox {sandbox_id} already gone before idle destroy, skipping")
|
||||
return False
|
||||
if (time.time() - last_activity) < idle_timeout:
|
||||
if (time.monotonic() - last_activity) < idle_timeout:
|
||||
# Re-acquired (activity updated) since the snapshot — skip.
|
||||
logger.info(f"Sandbox {sandbox_id} was re-acquired before idle destroy, skipping")
|
||||
return False
|
||||
@ -1556,7 +1556,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
if timeout <= 0:
|
||||
return
|
||||
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
expired: list[tuple[str, SandboxInfo]] = []
|
||||
with self._lock:
|
||||
for sandbox_id, (entry, timestamp) in self._warm_pool.items():
|
||||
@ -1737,7 +1737,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
|
||||
suffix = " (post-lock check)" if post_lock else ""
|
||||
logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
|
||||
self._last_activity[existing_id] = time.time()
|
||||
self._last_activity[existing_id] = time.monotonic()
|
||||
|
||||
# Fail closed: an OwnershipBackendError propagates rather than handing out
|
||||
# a sandbox we could not publish ownership for.
|
||||
@ -1905,7 +1905,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
getattr(self, "_cleanup_retry_ids", set()).discard(sandbox_id)
|
||||
self._active_sandbox_identity[sandbox_id] = key
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
self._last_activity[sandbox_id] = time.monotonic()
|
||||
self._thread_sandboxes[key] = sandbox_id
|
||||
|
||||
suffix = " (post-lock check)" if post_lock else f" at {info.sandbox_url}"
|
||||
@ -1985,7 +1985,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._sandbox_infos[info.sandbox_id] = info
|
||||
getattr(self, "_cleanup_retry_ids", set()).discard(info.sandbox_id)
|
||||
self._active_sandbox_identity[info.sandbox_id] = key
|
||||
self._last_activity[info.sandbox_id] = time.time()
|
||||
self._last_activity[info.sandbox_id] = time.monotonic()
|
||||
self._thread_sandboxes[key] = info.sandbox_id
|
||||
except Exception:
|
||||
try:
|
||||
@ -2032,7 +2032,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
getattr(self, "_cleanup_retry_ids", set()).discard(sandbox_id)
|
||||
self._active_sandbox_identity[sandbox_id] = key
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
self._last_activity[sandbox_id] = time.monotonic()
|
||||
if key is not None:
|
||||
self._thread_sandboxes[key] = sandbox_id
|
||||
except Exception:
|
||||
@ -2707,7 +2707,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
return None
|
||||
sandbox = self._sandboxes.get(sandbox_id)
|
||||
if sandbox is not None:
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
self._last_activity[sandbox_id] = time.monotonic()
|
||||
return sandbox
|
||||
|
||||
def get_scoped(
|
||||
@ -2728,7 +2728,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
return None
|
||||
sandbox = self._sandboxes.get(sandbox_id)
|
||||
if sandbox is not None:
|
||||
self._last_activity[sandbox_id] = time.time()
|
||||
self._last_activity[sandbox_id] = time.monotonic()
|
||||
return sandbox
|
||||
|
||||
def release(self, sandbox_id: str) -> None:
|
||||
@ -2792,7 +2792,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
self._last_activity.pop(sandbox_id, None)
|
||||
# Park in warm pool — container keeps running
|
||||
if info and sandbox_id not in self._warm_pool:
|
||||
self._warm_pool[sandbox_id] = (info, time.time())
|
||||
self._warm_pool[sandbox_id] = (info, time.monotonic())
|
||||
self._warm_pool_identity[sandbox_id] = thread_keys_to_remove[0] if thread_keys_to_remove else active_identity
|
||||
|
||||
if sandbox is not None:
|
||||
|
||||
@ -338,7 +338,6 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
|
||||
def _adopt_existing_boxes(self) -> int:
|
||||
runtime_cls = _import_sync_boxlite_runtime()
|
||||
now = time.time()
|
||||
adopted = 0
|
||||
|
||||
list_runtime = runtime_cls.default().start()
|
||||
@ -372,7 +371,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
if sandbox_id in self._boxes or sandbox_id in self._warm_pool:
|
||||
box_runtime.stop()
|
||||
continue
|
||||
self._warm_pool[sandbox_id] = (wrapped, now)
|
||||
self._warm_pool[sandbox_id] = (wrapped, time.monotonic())
|
||||
self._warm_pool_identity[sandbox_id] = None
|
||||
adopted += 1
|
||||
logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name)
|
||||
@ -524,7 +523,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
self._skip_health_check_warm_ids.discard(sandbox_id)
|
||||
self._warm_pool_identity.pop(sandbox_id, None)
|
||||
else:
|
||||
self._warm_pool[sandbox_id] = (box, time.time())
|
||||
self._warm_pool[sandbox_id] = (box, time.monotonic())
|
||||
self._warm_pool_identity[sandbox_id] = released_keys[0] if released_keys else active_identity
|
||||
self._skip_health_check_warm_ids.add(sandbox_id)
|
||||
|
||||
@ -559,7 +558,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
skip_eligible = sandbox_id in self._skip_health_check_warm_ids
|
||||
|
||||
skip_seconds = self._config.get("health_check_skip_seconds", 0.0)
|
||||
if skip_eligible and skip_seconds > 0 and (time.time() - released_at) < skip_seconds:
|
||||
if skip_eligible and skip_seconds > 0 and (time.monotonic() - released_at) < skip_seconds:
|
||||
# Recently released by this provider — promote directly without a
|
||||
# health-check round trip, but never return an adapter that this
|
||||
# process already knows is closed.
|
||||
@ -595,7 +594,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
logger.debug(
|
||||
"Reclaimed warm-pool box %s (skipped health check, age=%.1fs)",
|
||||
sandbox_id,
|
||||
time.time() - released_at,
|
||||
time.monotonic() - released_at,
|
||||
)
|
||||
return sandbox_id
|
||||
|
||||
@ -675,7 +674,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider):
|
||||
reaper and atexit shutdown instead of orphaning them.
|
||||
"""
|
||||
with self._lock:
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
for sandbox_id, box in self._boxes.items():
|
||||
self._warm_pool.setdefault(sandbox_id, (box, now))
|
||||
self._warm_pool_identity.setdefault(sandbox_id, self._active_box_identity.get(sandbox_id))
|
||||
|
||||
@ -1662,9 +1662,9 @@ class E2BSandboxProvider(SandboxProvider):
|
||||
idle_timeout = float(self._config["idle_timeout"])
|
||||
if idle_timeout <= 0:
|
||||
return
|
||||
now_wall = time.time()
|
||||
now_monotonic = time.monotonic()
|
||||
with self._lock:
|
||||
expired = [(sandbox_id, entry) for sandbox_id, entry in self._warm_pool.items() if now_wall - entry[1] >= idle_timeout]
|
||||
expired = [(sandbox_id, entry) for sandbox_id, entry in self._warm_pool.items() if now_monotonic - entry[1] >= idle_timeout]
|
||||
for sandbox_id, entry in expired:
|
||||
with self._sandbox_lifecycle(sandbox_id):
|
||||
with self._lock:
|
||||
@ -2994,7 +2994,7 @@ class E2BSandboxProvider(SandboxProvider):
|
||||
with self._lock:
|
||||
should_kill = self._shutdown_called
|
||||
if not should_kill:
|
||||
self._warm_pool[sandbox_id] = (seed or "", time.time())
|
||||
self._warm_pool[sandbox_id] = (seed or "", time.monotonic())
|
||||
self._warm_pool.move_to_end(sandbox_id)
|
||||
self._end_transition_locked()
|
||||
transition_slot_held = False
|
||||
|
||||
@ -403,7 +403,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro
|
||||
self._thread_sandboxes.pop(key, None)
|
||||
destroy_after_unlock = self._shutdown_called
|
||||
if not destroy_after_unlock:
|
||||
self._warm_pool[sandbox_id] = (sandbox, time.time())
|
||||
self._warm_pool[sandbox_id] = (sandbox, time.monotonic())
|
||||
if destroy_after_unlock:
|
||||
self._destroy_quietly(sandbox, context="released during shutdown")
|
||||
|
||||
@ -552,7 +552,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro
|
||||
def reset(self) -> None:
|
||||
"""Park active clients so the detached provider still owns their cleanup."""
|
||||
with self._lock:
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
for sandbox_id, sandbox in self._sandboxes.items():
|
||||
self._warm_pool.setdefault(sandbox_id, (sandbox, now))
|
||||
self._sandboxes.clear()
|
||||
|
||||
@ -387,7 +387,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider
|
||||
if self._shutdown_called:
|
||||
close_sandbox = sandbox
|
||||
else:
|
||||
self._warm_pool[sandbox_id] = (sandbox, time.time())
|
||||
self._warm_pool[sandbox_id] = (sandbox, time.monotonic())
|
||||
|
||||
if close_sandbox is not None:
|
||||
self._close_quietly(close_sandbox, context="released during shutdown")
|
||||
@ -443,7 +443,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider
|
||||
instance's idle reaper and atexit shutdown instead of orphaning them.
|
||||
"""
|
||||
with self._lock:
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
for sandbox_id, sandbox in self._sandboxes.items():
|
||||
self._warm_pool.setdefault(sandbox_id, (sandbox, now))
|
||||
self._sandboxes.clear()
|
||||
|
||||
@ -72,7 +72,9 @@ class WarmPoolLifecycleMixin[WarmEntryT]:
|
||||
if timeout <= 0:
|
||||
return
|
||||
|
||||
now = time.time()
|
||||
# Idle age is a duration. Wall-clock steps (NTP, VM resume, manual
|
||||
# correction) must neither expire fresh entries nor preserve old ones.
|
||||
now = time.monotonic()
|
||||
expired: list[tuple[str, WarmEntryT]] = []
|
||||
with self._lock:
|
||||
for sandbox_id, (entry, timestamp) in self._warm_pool.items():
|
||||
|
||||
@ -1870,7 +1870,7 @@ def _make_provider_with_active_sandbox(tmp_path, sandbox_id: str):
|
||||
sandbox_id: aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox-host"),
|
||||
}
|
||||
provider._thread_sandboxes = {}
|
||||
provider._last_activity = {sandbox_id: 0.0}
|
||||
provider._last_activity = {sandbox_id: time.monotonic()}
|
||||
provider._local_teardown = set()
|
||||
provider._acquire_epoch = {}
|
||||
provider._acquire_epoch_counter = 0
|
||||
@ -3414,7 +3414,7 @@ def test_warm_reclaim_keeps_failed_stop_fence_for_unsafe_generation(failure_reco
|
||||
with pytest.raises(RuntimeError, match="stop failed"):
|
||||
provider.destroy(old.sandbox_id)
|
||||
candidate = type(old)(old.sandbox_id, old.sandbox_url, container_id=generation)
|
||||
provider._warm_pool[old.sandbox_id] = (candidate, time.time())
|
||||
provider._warm_pool[old.sandbox_id] = (candidate, time.monotonic())
|
||||
provider._warm_pool_identity[old.sandbox_id] = None
|
||||
provider_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
|
||||
@ -3444,7 +3444,7 @@ def test_warm_reclaim_does_not_clear_fence_if_entry_changes(failure_recovery_lif
|
||||
def swap_during_publish(sandbox_id):
|
||||
publish(sandbox_id)
|
||||
with provider._lock:
|
||||
provider._warm_pool[sandbox_id] = (unvalidated, time.time())
|
||||
provider._warm_pool[sandbox_id] = (unvalidated, time.monotonic())
|
||||
|
||||
monkeypatch.setattr(provider, "_publish_ownership", swap_during_publish)
|
||||
assert provider._reclaim_warm_pool_sandbox("thread-failure-recovery", old.sandbox_id, user_id="alice") is None
|
||||
@ -3779,7 +3779,7 @@ def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(t
|
||||
"active-old": aio_mod.SandboxInfo(sandbox_id="active-old", sandbox_url="http://active-old"),
|
||||
}
|
||||
provider._thread_sandboxes = {("default", "thread-old"): "active-old"}
|
||||
provider._last_activity = {"active-old": 0.0}
|
||||
provider._last_activity = {"active-old": time.monotonic() - 10_000}
|
||||
provider._warm_pool = {
|
||||
"warm-old": (
|
||||
aio_mod.SandboxInfo(sandbox_id="warm-old", sandbox_url="http://warm-old"),
|
||||
|
||||
@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from types import SimpleNamespace
|
||||
|
||||
@ -248,10 +249,12 @@ def test_image_override_and_failed_construction_restore_config(transport):
|
||||
|
||||
|
||||
def test_idle_reaping_forces_a_new_remote(transport, monkeypatch, tmp_path):
|
||||
from deerflow.community import warm_pool_lifecycle
|
||||
|
||||
clock = [100.0]
|
||||
monkeypatch.setattr(warm_pool_lifecycle.time, "time", lambda: clock[0])
|
||||
# Warm-pool idle age is written and read on the monotonic clock, so freeze
|
||||
# that one and let the benchmark's sleep advance it. Freezing `time.time`
|
||||
# instead leaves both ends of the comparison on the real clock: the entry
|
||||
# never expires, the reaper removes nothing, and the next turn is a warm hit.
|
||||
monkeypatch.setattr(time, "monotonic", lambda: clock[0])
|
||||
monkeypatch.setattr(bench.time, "sleep", lambda seconds: clock.__setitem__(0, clock[0] + seconds))
|
||||
output = tmp_path / "idle.jsonl"
|
||||
assert (
|
||||
|
||||
@ -746,7 +746,7 @@ def test_adopted_warm_pool_box_still_health_checks(monkeypatch):
|
||||
_fake_run,
|
||||
default_env={},
|
||||
)
|
||||
provider._warm_pool["adopted"] = (adopted, time.time())
|
||||
provider._warm_pool["adopted"] = (adopted, time.monotonic())
|
||||
calls = 0
|
||||
original_execute = adopted.execute_command
|
||||
|
||||
@ -1039,8 +1039,8 @@ def test_reset_parked_resources_are_reaped_after_idle_timeout(monkeypatch):
|
||||
|
||||
provider.reset()
|
||||
|
||||
provider._warm_pool[sid_active] = (active_box, time.time() - 9999)
|
||||
provider._warm_pool[sid_warm] = (warm_box, time.time() - 9999)
|
||||
provider._warm_pool[sid_active] = (active_box, time.monotonic() - 9999)
|
||||
provider._warm_pool[sid_warm] = (warm_box, time.monotonic() - 9999)
|
||||
provider._reap_expired_warm(idle_timeout=1)
|
||||
|
||||
assert provider._warm_pool == {}
|
||||
@ -1076,7 +1076,7 @@ def test_idle_reaper_destroys_expired_warm_boxes(monkeypatch):
|
||||
|
||||
# Backdate the warm-pool timestamp so it appears long-expired
|
||||
warm_box = provider._warm_pool[sid][0]
|
||||
provider._warm_pool[sid] = (warm_box, time.time() - 9999)
|
||||
provider._warm_pool[sid] = (warm_box, time.monotonic() - 9999)
|
||||
|
||||
# Wait long enough for the reaper to detect and destroy it
|
||||
time.sleep(0.3)
|
||||
@ -1118,10 +1118,10 @@ def test_replica_enforcement_evicts_oldest_warm(monkeypatch):
|
||||
|
||||
# Make sid_a definitely older by backdating its timestamp
|
||||
box_a = provider._warm_pool[sid_a][0]
|
||||
provider._warm_pool[sid_a] = (box_a, time.time() - 100)
|
||||
provider._warm_pool[sid_a] = (box_a, time.monotonic() - 100)
|
||||
# Refresh sid_b's timestamp so it's newer
|
||||
box_b = provider._warm_pool[sid_b][0]
|
||||
provider._warm_pool[sid_b] = (box_b, time.time())
|
||||
provider._warm_pool[sid_b] = (box_b, time.monotonic())
|
||||
|
||||
# Acquiring a third thread triggers replica enforcement:
|
||||
# warm pool count (2) >= replicas (2) → evict oldest (sid_a)
|
||||
@ -1507,12 +1507,12 @@ def test_failed_health_check_does_not_remove_swapped_warm_entry(monkeypatch):
|
||||
_fake_run,
|
||||
default_env={},
|
||||
)
|
||||
provider._warm_pool["shared-id"] = (stale, time.time())
|
||||
provider._warm_pool["shared-id"] = (stale, time.monotonic())
|
||||
provider._warm_pool_identity["shared-id"] = ("user-a", "thread-a")
|
||||
|
||||
def _swap_during_health_check(*args, **kwargs):
|
||||
with provider._lock:
|
||||
provider._warm_pool["shared-id"] = (replacement, time.time())
|
||||
provider._warm_pool["shared-id"] = (replacement, time.monotonic())
|
||||
provider._warm_pool_identity["shared-id"] = (
|
||||
"user-b",
|
||||
"thread-b",
|
||||
|
||||
@ -2457,7 +2457,7 @@ def test_reconcile_honors_wall_clock_budget(monkeypatch):
|
||||
fake_cls.list_return = [_info("sb-never-probed", "u1", "t1")]
|
||||
p._config["reconciliation_max_seconds"] = 0.5
|
||||
provider_mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider")
|
||||
ticks = iter([0.0, 0.0, 1.0, 1.0])
|
||||
ticks = iter([0.0, 0.0, 0.0, 1.0, 1.0])
|
||||
monkeypatch.setattr(provider_mod.time, "monotonic", lambda: next(ticks))
|
||||
|
||||
stats = p._reconcile_remote_sandboxes(now=100.0)
|
||||
@ -2485,7 +2485,7 @@ def test_reconcile_never_probes_or_adopts_a_warm_pool_sandbox(monkeypatch):
|
||||
p = _make_provider()
|
||||
fake_cls = _install_fake_sdk(monkeypatch, p)
|
||||
fake_cls.list_return = [_info("sb-warm", "u1", "t1")]
|
||||
p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.time())
|
||||
p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.monotonic())
|
||||
|
||||
stats = p._reconcile_remote_sandboxes(now=100.0)
|
||||
|
||||
@ -2649,7 +2649,7 @@ def test_reconcile_sweeps_expired_warm_pool_entry(monkeypatch):
|
||||
fake_cls = _install_fake_sdk(monkeypatch, p)
|
||||
store = _install_shared_deployment_capacity(p)
|
||||
fake_cls.list_return = []
|
||||
p._warm_pool["sb-old"] = (p._stable_seed("t1", "u1"), time.time() - 601)
|
||||
p._warm_pool["sb-old"] = (p._stable_seed("t1", "u1"), time.monotonic() - 601)
|
||||
p._owned_sandbox_ids.add("sb-old")
|
||||
p._mount_results["sb-old"] = MagicMock()
|
||||
p._ownership.take("sb-old")
|
||||
@ -2671,7 +2671,7 @@ def test_reconcile_keeps_unexpired_warm_pool_entry(monkeypatch):
|
||||
p = _make_provider(idle_timeout=600)
|
||||
fake_cls = _install_fake_sdk(monkeypatch, p)
|
||||
fake_cls.list_return = []
|
||||
p._warm_pool["sb-fresh"] = (p._stable_seed("t1", "u1"), time.time())
|
||||
p._warm_pool["sb-fresh"] = (p._stable_seed("t1", "u1"), time.monotonic())
|
||||
|
||||
p._reconcile_remote_sandboxes(now=100.0)
|
||||
|
||||
@ -2688,7 +2688,7 @@ def test_reconcile_kills_remote_duplicate_of_a_warm_pool_sandbox(monkeypatch):
|
||||
]
|
||||
duplicate = FakeClient(sandbox_id="sb-duplicate")
|
||||
fake_cls.connect_factory = lambda sid, **_kw: duplicate if sid == "sb-duplicate" else FakeClient(sandbox_id=sid)
|
||||
p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.time())
|
||||
p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.monotonic())
|
||||
p._config["reconciliation_grace_seconds"] = 0.0
|
||||
|
||||
stats = p._reconcile_remote_sandboxes(now=100.0)
|
||||
@ -2711,7 +2711,7 @@ def test_reconcile_adoption_recheck_treats_warm_pool_as_local(monkeypatch):
|
||||
client = FakeClient(sandbox_id="sb-race")
|
||||
|
||||
def connect_and_park(sid, **_kw):
|
||||
p._warm_pool[sid] = (p._stable_seed("t1", "u1"), time.time())
|
||||
p._warm_pool[sid] = (p._stable_seed("t1", "u1"), time.monotonic())
|
||||
return client
|
||||
|
||||
fake_cls.connect_factory = connect_and_park
|
||||
@ -3483,7 +3483,7 @@ def test_thread_restore_does_not_overwrite_live_sandbox(monkeypatch, tmp_path, r
|
||||
provider._sandboxes[client.sandbox_id] = _make_sandbox(client, sandbox_id=client.sandbox_id)
|
||||
provider._thread_sandboxes[provider._thread_key("t1", "u1")] = client.sandbox_id
|
||||
elif reuse == "warm":
|
||||
provider._warm_pool[client.sandbox_id] = (provider._stable_seed("t1", "u1"), time.time())
|
||||
provider._warm_pool[client.sandbox_id] = (provider._stable_seed("t1", "u1"), time.monotonic())
|
||||
else:
|
||||
sdk.list_return = [_info(client.sandbox_id, "u1", "t1")]
|
||||
sandbox_id = provider.acquire("t1", user_id="u1")
|
||||
@ -5077,7 +5077,7 @@ def test_shutdown_during_initial_eviction_reconnect_failure_tracks_vm(monkeypatc
|
||||
"""Shutdown must destroy a VM while its first eviction reconnects."""
|
||||
p = _make_provider()
|
||||
fake_cls = _install_fake_sdk(monkeypatch, p)
|
||||
p._warm_pool["warm-1"] = ("seed", time.time())
|
||||
p._warm_pool["warm-1"] = ("seed", time.monotonic())
|
||||
|
||||
reconnect_started = threading.Event()
|
||||
allow_failure = threading.Event()
|
||||
@ -5117,7 +5117,7 @@ def test_shutdown_during_gone_sandbox_eviction_releases_slot_once(monkeypatch):
|
||||
"""
|
||||
p = _make_provider()
|
||||
fake_cls = _install_fake_sdk(monkeypatch, p)
|
||||
p._warm_pool["warm-1"] = ("seed", time.time())
|
||||
p._warm_pool["warm-1"] = ("seed", time.monotonic())
|
||||
|
||||
reconnect_started = threading.Event()
|
||||
allow_failure = threading.Event()
|
||||
@ -6569,7 +6569,7 @@ def test_warm_sweep_cannot_delete_concurrent_discovery_lease(monkeypatch):
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
sdk = _install_fake_sdk(monkeypatch, provider)
|
||||
sdk.list_return = [_info("sb-race", "u1", "t1")]
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
provider._ownership.take("sb-race")
|
||||
provider._owned_sandbox_ids.add("sb-race")
|
||||
cleanup_started, finish_cleanup, acquire_reached = threading.Event(), threading.Event(), threading.Event()
|
||||
@ -6615,7 +6615,7 @@ def test_warm_sweep_cannot_delete_concurrent_discovery_lease(monkeypatch):
|
||||
|
||||
def test_warm_sweep_serializes_with_lapsed_lease_renewal(monkeypatch):
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
provider._owned_sandbox_ids.add("sb-race")
|
||||
renewal_started, finish_renewal, cleanup_reached = threading.Event(), threading.Event(), threading.Event()
|
||||
renew = provider._ownership.renew
|
||||
@ -6737,7 +6737,7 @@ def test_reconcile_does_not_readopt_a_release_in_progress(monkeypatch):
|
||||
def test_warm_sweep_leaves_shared_capacity_to_inventory(monkeypatch, inventory_complete):
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
store = _install_shared_deployment_capacity(provider)
|
||||
provider._warm_pool["sb-old"] = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
provider._warm_pool["sb-old"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
monkeypatch.setattr(provider, "_list_remote_entries", lambda _metadata: ([], False, inventory_complete))
|
||||
|
||||
provider._reconcile_remote_sandboxes()
|
||||
@ -6825,7 +6825,7 @@ def test_warm_sweep_cannot_release_reacquired_ownership(monkeypatch, acquire_pat
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
sdk = _install_fake_sdk(monkeypatch, provider)
|
||||
sdk.list_return = [_info("sb-race", "u1", "t1")]
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
provider._owned_sandbox_ids.add("sb-race")
|
||||
provider._ownership.take("sb-race")
|
||||
cleaning = threading.Event()
|
||||
@ -6876,7 +6876,7 @@ def test_warm_sweep_cannot_release_reacquired_ownership(monkeypatch, acquire_pat
|
||||
def test_warm_sweep_rechecks_snapshot_after_lifecycle_wait(monkeypatch, transition):
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
sid = "sb-stale-snapshot"
|
||||
old_entry = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
old_entry = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
provider._warm_pool[sid] = old_entry
|
||||
provider._owned_sandbox_ids.add(sid)
|
||||
provider._ownership.take(sid)
|
||||
@ -6893,7 +6893,7 @@ def test_warm_sweep_rechecks_snapshot_after_lifecycle_wait(monkeypatch, transiti
|
||||
# The sweep has already captured old_entry. A competing operation
|
||||
# completes before it acquires the lifecycle lock.
|
||||
if transition == "repark":
|
||||
provider._warm_pool[sid] = (old_entry[0], time.time())
|
||||
provider._warm_pool[sid] = (old_entry[0], time.monotonic())
|
||||
else:
|
||||
provider._publish_ownership(sid)
|
||||
with lifecycle(sandbox_id):
|
||||
@ -6980,7 +6980,7 @@ def test_warm_sweep_cannot_race_lapsed_lease_reclaim(monkeypatch):
|
||||
|
||||
provider = _make_provider(idle_timeout=30)
|
||||
sid = "sb-lapsed"
|
||||
provider._warm_pool[sid] = (provider._stable_seed("t1", "u1"), time.time() - 31)
|
||||
provider._warm_pool[sid] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31)
|
||||
provider._owned_sandbox_ids.add(sid)
|
||||
renewing = threading.Event()
|
||||
finish_renewal = threading.Event()
|
||||
|
||||
@ -762,7 +762,10 @@ def test_output_sync_is_bounded(output_sync_provider, monkeypatch, caplog, limit
|
||||
monkeypatch.setattr(provider, "_MAX_SYNC_LISTING_BYTES", 4)
|
||||
else:
|
||||
monkeypatch.setattr(provider, "_SYNC_DEADLINE_SECONDS", 1)
|
||||
monkeypatch.setattr("deerflow.community.opensandbox.provider.time", types.SimpleNamespace(monotonic=iter([0, 2]).__next__, time=time.time))
|
||||
monkeypatch.setattr(
|
||||
"deerflow.community.opensandbox.provider.time",
|
||||
types.SimpleNamespace(monotonic=iter([0, 2, 2]).__next__, time=time.time),
|
||||
)
|
||||
provider.release(sandbox_id)
|
||||
if limit in {"files", "bytes"}:
|
||||
assert (outputs / "first.txt").read_bytes() == b"123"
|
||||
@ -1071,6 +1074,27 @@ def test_reset_parks_active_and_shutdown_destroys_active_and_warm(monkeypatch: p
|
||||
assert provider._sandboxes == {} and provider._warm_pool == {}
|
||||
|
||||
|
||||
def test_reset_parks_with_monotonic_timestamp_and_reaper_expires_it(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
provider, sdk = _install(monkeypatch)
|
||||
sandbox_id = provider.acquire("active", user_id="user")
|
||||
remote = sdk.remotes[0]
|
||||
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
||||
monkeypatch.setattr(time, "time", lambda: 1_000_000.0)
|
||||
|
||||
provider.reset()
|
||||
assert provider._warm_pool[sandbox_id][1] == 100.0
|
||||
|
||||
# This reset path was omitted by #6996. A backwards wall-clock step must not
|
||||
# prevent the shared reaper from observing ten elapsed monotonic seconds.
|
||||
monkeypatch.setattr(time, "monotonic", lambda: 110.0)
|
||||
monkeypatch.setattr(time, "time", lambda: -1_000_000.0)
|
||||
provider._reap_expired_warm(idle_timeout=5)
|
||||
|
||||
assert sandbox_id not in provider._warm_pool
|
||||
assert remote.destroy_calls == 1
|
||||
provider.shutdown()
|
||||
|
||||
|
||||
def test_shutdown_stops_idle_reaper(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(OpenSandboxProvider, "IDLE_CHECK_INTERVAL", 0.01)
|
||||
provider, _ = _install(monkeypatch, config={"idle_timeout": 60})
|
||||
|
||||
@ -745,12 +745,12 @@ def test_idle_reap_does_not_destroy_peer_owned_warm_entry():
|
||||
worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared)
|
||||
worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
worker_b._config["idle_timeout"] = 60
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
info = SandboxInfo(
|
||||
sandbox_id="a99c8444",
|
||||
sandbox_url="http://localhost:8080",
|
||||
container_name="deer-flow-sandbox-a99c8444",
|
||||
created_at=now - 50,
|
||||
created_at=time.time() - 50,
|
||||
)
|
||||
# Simulate the bad old path: B already has it in warm (or adopted wrongly).
|
||||
worker_b._warm_pool["a99c8444"] = (info, now - 61)
|
||||
@ -791,7 +791,7 @@ def test_multi_worker_release_then_peer_reconcile_cannot_kill():
|
||||
worker_a._backend = backend
|
||||
worker_a._config["idle_timeout"] = 60
|
||||
# A released to warm and holds the lease.
|
||||
worker_a._warm_pool[sid] = (info, time.time())
|
||||
worker_a._warm_pool[sid] = (info, time.monotonic())
|
||||
worker_a._publish_ownership(sid)
|
||||
|
||||
worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared)
|
||||
@ -801,7 +801,7 @@ def test_multi_worker_release_then_peer_reconcile_cannot_kill():
|
||||
assert sid not in worker_b._warm_pool
|
||||
|
||||
# Even if B somehow had it warm, reap must refuse.
|
||||
worker_b._warm_pool[sid] = (info, time.time() - 61)
|
||||
worker_b._warm_pool[sid] = (info, time.monotonic() - 61)
|
||||
worker_b._reap_expired_warm(idle_timeout=60)
|
||||
assert sid not in destroyed
|
||||
assert sid in running
|
||||
@ -1013,7 +1013,7 @@ def test_renewal_keeps_the_sandbox_when_the_store_cannot_answer():
|
||||
worker._sandboxes["live02"] = sandbox
|
||||
worker._sandbox_infos["live02"] = info
|
||||
worker._thread_sandboxes[("u1", "t1")] = "live02"
|
||||
worker._warm_pool["warm02"] = (info, time.time())
|
||||
worker._warm_pool["warm02"] = (info, time.monotonic())
|
||||
|
||||
worker._renew_owned_leases()
|
||||
|
||||
@ -1161,7 +1161,7 @@ def test_renewal_covers_warm_entries_not_just_active():
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker._sandboxes["active01"] = MagicMock()
|
||||
worker._warm_pool["warm01"] = (info, time.time())
|
||||
worker._warm_pool["warm01"] = (info, time.monotonic())
|
||||
worker._publish_ownership("active01")
|
||||
worker._publish_ownership("warm01")
|
||||
|
||||
@ -1189,7 +1189,7 @@ def test_renewal_does_not_forget_a_warm_entry_mid_teardown():
|
||||
container_name="deer-flow-sandbox-warm-stop",
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker._warm_pool["warm-stop"] = (info, time.time())
|
||||
worker._warm_pool["warm-stop"] = (info, time.monotonic())
|
||||
worker._publish_ownership("warm-stop")
|
||||
|
||||
stop_started = threading.Event()
|
||||
@ -1250,7 +1250,7 @@ def test_lost_lease_drops_sandbox_without_destroying_container():
|
||||
worker_a._sandboxes["moved01"] = sandbox
|
||||
worker_a._sandbox_infos["moved01"] = info
|
||||
worker_a._thread_sandboxes[("u1", "t1")] = "moved01"
|
||||
worker_a._last_activity["moved01"] = time.time()
|
||||
worker_a._last_activity["moved01"] = time.monotonic()
|
||||
worker_a._publish_ownership("moved01")
|
||||
|
||||
# The thread's next turn routes to B, which takes over ownership.
|
||||
@ -1583,7 +1583,7 @@ def test_teardown_marker_is_held_for_a_stop_that_outlives_the_lease_ttl():
|
||||
release_stop.wait(timeout=5)
|
||||
|
||||
worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
|
||||
worker_a._warm_pool["doomed1"] = (info, time.time())
|
||||
worker_a._warm_pool["doomed1"] = (info, time.monotonic())
|
||||
|
||||
reaper = threading.Thread(
|
||||
target=lambda: worker_a._destroy_warm_entry("doomed1", info, reason="idle_timeout", still_reapable=lambda: True),
|
||||
@ -1857,7 +1857,7 @@ def test_teardown_release_waits_for_the_heartbeat_to_exit():
|
||||
assert blocking.heartbeat_blocked.wait(timeout=5), "the heartbeat never issued a refresh claim"
|
||||
|
||||
worker_a._backend.destroy = MagicMock(side_effect=slow_destroy)
|
||||
worker_a._warm_pool["defer1"] = (info, time.time())
|
||||
worker_a._warm_pool["defer1"] = (info, time.monotonic())
|
||||
|
||||
reaper = threading.Thread(
|
||||
target=lambda: worker_a._destroy_warm_entry("defer1", info, reason="idle_timeout", still_reapable=lambda: True),
|
||||
@ -1900,7 +1900,7 @@ def test_evict_keeps_the_warm_entry_when_the_claim_is_refused():
|
||||
container_name="deer-flow-sandbox-peer01",
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker_a._warm_pool["peer01"] = (info, time.time() - 5)
|
||||
worker_a._warm_pool["peer01"] = (info, time.monotonic() - 5)
|
||||
# A peer owns it, so A's eviction claim is refused.
|
||||
worker_b._publish_ownership("peer01")
|
||||
|
||||
@ -1929,7 +1929,7 @@ def test_reclaim_drops_a_container_a_peer_is_destroying():
|
||||
container_name="deer-flow-sandbox-dying02",
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker_a._warm_pool["dying02"] = (info, time.time())
|
||||
worker_a._warm_pool["dying02"] = (info, time.monotonic())
|
||||
worker_a._check_tracked_sandbox_alive = MagicMock(return_value=True)
|
||||
|
||||
# B's reaper marks the teardown; its stop is in flight.
|
||||
@ -1987,9 +1987,9 @@ def test_shutdown_does_not_stop_a_peers_warm_container():
|
||||
mine = SandboxInfo(sandbox_id="mine01", sandbox_url="http://localhost:8080", container_name="c-mine01", created_at=time.time())
|
||||
theirs = SandboxInfo(sandbox_id="peer02", sandbox_url="http://localhost:8081", container_name="c-peer02", created_at=time.time())
|
||||
|
||||
worker_a._warm_pool["mine01"] = (mine, time.time())
|
||||
worker_a._warm_pool["mine01"] = (mine, time.monotonic())
|
||||
worker_a._publish_ownership("mine01")
|
||||
worker_a._warm_pool["peer02"] = (theirs, time.time())
|
||||
worker_a._warm_pool["peer02"] = (theirs, time.monotonic())
|
||||
worker_b._publish_ownership("peer02") # a live peer owns this one
|
||||
|
||||
worker_a.shutdown()
|
||||
@ -2018,7 +2018,7 @@ def test_teardown_heartbeat_stops_when_the_stop_returns():
|
||||
container_name="deer-flow-sandbox-doomed2",
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker_a._warm_pool["doomed2"] = (info, time.time())
|
||||
worker_a._warm_pool["doomed2"] = (info, time.monotonic())
|
||||
|
||||
assert worker_a._destroy_warm_entry("doomed2", info, reason="idle_timeout", still_reapable=lambda: True) is True
|
||||
|
||||
@ -2097,7 +2097,7 @@ def test_refused_idle_destroy_keeps_the_warm_entry():
|
||||
container_name="deer-flow-sandbox-warmpeer",
|
||||
created_at=time.time(),
|
||||
)
|
||||
worker_a._warm_pool["warmpeer"] = (info, time.time() - 999)
|
||||
worker_a._warm_pool["warmpeer"] = (info, time.monotonic() - 999)
|
||||
worker_b._publish_ownership("warmpeer")
|
||||
|
||||
worker_a._reap_expired_warm(idle_timeout=60)
|
||||
@ -2247,7 +2247,7 @@ def _active_sandbox(provider, sandbox_id, info, *, thread_key=("u", "t1")):
|
||||
|
||||
provider._sandboxes[sandbox_id] = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
|
||||
provider._sandbox_infos[sandbox_id] = info
|
||||
provider._last_activity[sandbox_id] = time.time()
|
||||
provider._last_activity[sandbox_id] = time.monotonic()
|
||||
provider._thread_sandboxes[thread_key] = sandbox_id
|
||||
provider._backend.is_alive.return_value = True
|
||||
|
||||
@ -2327,7 +2327,7 @@ def test_warm_reaper_does_not_stop_a_container_this_instance_just_reclaimed(reas
|
||||
provider._ownership = gate
|
||||
info = _info("warm1")
|
||||
# Old enough for the idle reaper; the only entry, so also the eviction pick.
|
||||
provider._warm_pool["warm1"] = (info, time.time() - 10_000)
|
||||
provider._warm_pool["warm1"] = (info, time.monotonic() - 10_000)
|
||||
provider._ownership._inner.take("warm1")
|
||||
|
||||
reap = provider._evict_oldest_warm if reap_name == "_evict_oldest_warm" else lambda: provider._reap_expired_warm(600)
|
||||
@ -2359,7 +2359,7 @@ def test_idle_destroy_does_not_stop_a_container_this_instance_just_reused():
|
||||
provider._ownership = gate
|
||||
info = _info("idle1")
|
||||
_active_sandbox(provider, "idle1", info)
|
||||
provider._last_activity["idle1"] = 0.0 # long idle
|
||||
provider._last_activity["idle1"] = time.monotonic() - 10_000 # long idle
|
||||
provider._ownership._inner.take("idle1")
|
||||
|
||||
handed_out = _run_reap_vs_promote(
|
||||
@ -2386,7 +2386,7 @@ def test_reuse_does_not_return_a_sandbox_reserved_during_its_health_check():
|
||||
provider._ownership = gate
|
||||
info = _info("idlerace")
|
||||
_active_sandbox(provider, "idlerace", info)
|
||||
provider._last_activity["idlerace"] = 0.0
|
||||
provider._last_activity["idlerace"] = time.monotonic() - 10_000
|
||||
provider._ownership._inner.take("idlerace")
|
||||
|
||||
health_started, let_health_finish = threading.Event(), threading.Event()
|
||||
@ -2622,14 +2622,14 @@ def test_idle_destroy_skips_a_sandbox_re_acquired_after_the_idle_snapshot():
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("idle2")
|
||||
_active_sandbox(provider, "idle2", info)
|
||||
provider._last_activity["idle2"] = 0.0
|
||||
provider._last_activity["idle2"] = time.monotonic() - 10_000
|
||||
provider._ownership.take("idle2")
|
||||
|
||||
real_destroy_tracked = provider._destroy_tracked
|
||||
|
||||
def turn_lands_first(sandbox_id, **kwargs):
|
||||
# A turn arrives between the idle snapshot and the reservation.
|
||||
provider._last_activity[sandbox_id] = time.time()
|
||||
provider._last_activity[sandbox_id] = time.monotonic()
|
||||
return real_destroy_tracked(sandbox_id, **kwargs)
|
||||
|
||||
provider._destroy_tracked = turn_lands_first
|
||||
@ -2829,7 +2829,7 @@ def test_teardown_reservation_is_cleared_once_the_stop_returns():
|
||||
"""
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("cleared")
|
||||
provider._warm_pool["cleared"] = (info, time.time() - 10_000)
|
||||
provider._warm_pool["cleared"] = (info, time.monotonic() - 10_000)
|
||||
provider._ownership.take("cleared")
|
||||
|
||||
assert provider._destroy_warm_entry("cleared", info, reason="idle_timeout", still_reapable=lambda: True) is True
|
||||
@ -2849,7 +2849,7 @@ def test_reclaim_does_not_hand_out_a_container_a_reaper_is_stopping():
|
||||
"""
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("handout")
|
||||
provider._warm_pool["handout"] = (info, time.time() - 10_000)
|
||||
provider._warm_pool["handout"] = (info, time.monotonic() - 10_000)
|
||||
provider._ownership.take("handout")
|
||||
|
||||
# Park reclaim between its `take()` and its install.
|
||||
@ -2899,7 +2899,7 @@ def test_warm_entry_is_removed_before_its_teardown_reservation_is_released():
|
||||
"""
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("ordered")
|
||||
provider._warm_pool["ordered"] = (info, time.time() - 10_000)
|
||||
provider._warm_pool["ordered"] = (info, time.monotonic() - 10_000)
|
||||
provider._ownership.take("ordered")
|
||||
|
||||
observed = {}
|
||||
@ -2927,7 +2927,7 @@ def test_reclaim_short_circuits_a_reserved_entry_without_touching_the_backend():
|
||||
"""
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("shortcut")
|
||||
provider._warm_pool["shortcut"] = (info, time.time())
|
||||
provider._warm_pool["shortcut"] = (info, time.monotonic())
|
||||
provider._ownership = MagicMock(wraps=provider._ownership)
|
||||
with provider._lock:
|
||||
provider._local_teardown.add("shortcut")
|
||||
@ -2977,7 +2977,7 @@ def test_registering_a_sandbox_clears_any_stale_warm_entry(register):
|
||||
"""
|
||||
provider = _make_provider_for_reconciliation()
|
||||
info = _info("stale")
|
||||
provider._warm_pool["stale"] = (info, time.time())
|
||||
provider._warm_pool["stale"] = (info, time.monotonic())
|
||||
|
||||
if register == "discovered":
|
||||
provider._register_discovered_sandbox("t1", info, user_id="u")
|
||||
|
||||
@ -1112,6 +1112,29 @@ def test_reset_parks_running_for_later_cleanup(monkeypatch):
|
||||
assert active_box.is_closed
|
||||
|
||||
|
||||
def test_reset_parks_running_with_monotonic_timestamp_for_idle_cleanup(monkeypatch):
|
||||
client = _FakeClient()
|
||||
provider = _install(monkeypatch, client=client)
|
||||
sid_active = provider.acquire("thread-active", user_id="u1")
|
||||
active_box = provider._sandboxes[sid_active]
|
||||
monkeypatch.setattr(time, "monotonic", lambda: 100.0)
|
||||
monkeypatch.setattr(time, "time", lambda: 1_000_000.0)
|
||||
|
||||
provider.reset()
|
||||
assert provider._sandboxes == {}
|
||||
assert provider._warm_pool[sid_active] == (active_box, 100.0)
|
||||
|
||||
# Ten elapsed monotonic seconds exceed the configured timeout even though
|
||||
# the wall clock moves backwards. This is the reset writer missed by #6996.
|
||||
monkeypatch.setattr(time, "monotonic", lambda: 110.0)
|
||||
monkeypatch.setattr(time, "time", lambda: -1_000_000.0)
|
||||
provider._reap_expired_warm(idle_timeout=5)
|
||||
|
||||
assert sid_active not in provider._warm_pool
|
||||
assert active_box.is_closed
|
||||
provider.shutdown()
|
||||
|
||||
|
||||
def test_idle_reaper_destroys_expired_warm(monkeypatch):
|
||||
client = _FakeClient()
|
||||
monkeypatch.setattr(TenkiSandboxProvider, "IDLE_CHECK_INTERVAL", 0.1)
|
||||
@ -1119,7 +1142,7 @@ def test_idle_reaper_destroys_expired_warm(monkeypatch):
|
||||
sid = provider.acquire("thread-1", user_id="u1")
|
||||
provider.release(sid)
|
||||
warm_box = provider._warm_pool[sid][0]
|
||||
provider._warm_pool[sid] = (warm_box, time.time() - 9999)
|
||||
provider._warm_pool[sid] = (warm_box, time.monotonic() - 9999)
|
||||
time.sleep(0.3)
|
||||
assert sid not in provider._warm_pool
|
||||
assert warm_box.is_closed
|
||||
@ -1134,8 +1157,8 @@ def test_replica_enforcement_evicts_oldest_warm(monkeypatch):
|
||||
sid_b = provider.acquire("thread-b", user_id="u1")
|
||||
provider.release(sid_b)
|
||||
box_a = provider._warm_pool[sid_a][0]
|
||||
provider._warm_pool[sid_a] = (box_a, time.time() - 100)
|
||||
provider._warm_pool[sid_b] = (provider._warm_pool[sid_b][0], time.time())
|
||||
provider._warm_pool[sid_a] = (box_a, time.monotonic() - 100)
|
||||
provider._warm_pool[sid_b] = (provider._warm_pool[sid_b][0], time.monotonic())
|
||||
provider.acquire("thread-c", user_id="u1")
|
||||
assert sid_a not in provider._warm_pool
|
||||
assert box_a.is_closed
|
||||
|
||||
@ -34,7 +34,7 @@ class _Provider(WarmPoolLifecycleMixin[str]):
|
||||
|
||||
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.time())
|
||||
provider._warm_pool["warm-1"] = ("entry-1", time.monotonic())
|
||||
|
||||
assert provider._replica_count() == (2, 2)
|
||||
|
||||
@ -61,7 +61,7 @@ def test_evict_oldest_warm_returns_none_when_pool_empty() -> None:
|
||||
|
||||
def test_reap_expired_warm_destroys_only_expired_entries() -> None:
|
||||
provider = _Provider()
|
||||
now = time.time()
|
||||
now = time.monotonic()
|
||||
provider._warm_pool["expired"] = ("entry-expired", now - 100)
|
||||
provider._warm_pool["fresh"] = ("entry-fresh", now)
|
||||
|
||||
@ -74,7 +74,7 @@ def test_reap_expired_warm_destroys_only_expired_entries() -> None:
|
||||
|
||||
def test_reap_expired_warm_noops_when_timeout_disabled() -> None:
|
||||
provider = _Provider(idle_timeout=0)
|
||||
provider._warm_pool["expired"] = ("entry-expired", time.time() - 100)
|
||||
provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 100)
|
||||
|
||||
provider._reap_expired_warm(idle_timeout=0)
|
||||
|
||||
@ -123,7 +123,7 @@ def test_provider_shutdown_resets_guard_when_idle_checker_stop_fails(module_name
|
||||
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.time() - 10)
|
||||
provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 10)
|
||||
|
||||
provider._start_idle_checker()
|
||||
deadline = time.time() + 1
|
||||
@ -135,3 +135,31 @@ def test_start_idle_checker_uses_monkeypatchable_interval(monkeypatch) -> None:
|
||||
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")]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user