110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""A slow skill refresh must not keep all later tool workers waiting forever."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import concurrent.futures
|
|
import errno
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from deerflow.sandbox import skill_view
|
|
|
|
|
|
def test_process_lock_wait_times_out_and_recovers(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)
|
|
lock = threading.Lock()
|
|
monkeypatch.setattr(skill_view, "_PROCESS_LOCK", lock)
|
|
lock.acquire()
|
|
try:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
|
future = pool.submit(lambda: skill_view._view_lock(tmp_path).__enter__())
|
|
try:
|
|
with pytest.raises(TimeoutError, match="skill.*lock"):
|
|
future.result(timeout=1)
|
|
finally:
|
|
lock.release()
|
|
finally:
|
|
if lock.locked():
|
|
lock.release()
|
|
with skill_view._view_lock(tmp_path):
|
|
assert lock.locked()
|
|
assert not lock.locked()
|
|
|
|
|
|
def test_warm_follower_timeout_does_not_replace_running_leader(monkeypatch):
|
|
monkeypatch.setattr(skill_view, "_VIEW_WAIT_SECONDS", 0.03, raising=False)
|
|
entered, release = threading.Event(), threading.Event()
|
|
entry = object()
|
|
key = ("source", "target", "scope")
|
|
flights = {}
|
|
monkeypatch.setattr(skill_view, "_WARM_VALIDATION_FLIGHTS", flights)
|
|
|
|
def validate(*args):
|
|
entered.set()
|
|
assert release.wait(2)
|
|
return True
|
|
|
|
monkeypatch.setattr(skill_view, "_entry_matches", validate)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
|
|
leader = pool.submit(skill_view._entry_matches_singleflight, key, entry, Path("."), "config")
|
|
assert entered.wait(1)
|
|
follower = pool.submit(skill_view._entry_matches_singleflight, key, entry, Path("."), "config")
|
|
try:
|
|
with pytest.raises(TimeoutError, match="skill.*validation"):
|
|
follower.result(timeout=1)
|
|
assert len(flights) == 1
|
|
assert not leader.done()
|
|
finally:
|
|
release.set()
|
|
assert leader.result(timeout=1) is True
|
|
assert not flights
|
|
assert skill_view._entry_matches_singleflight(key, entry, Path("."), "config") is True
|
|
|
|
|
|
def test_file_lock_contention_is_bounded_and_process_lock_released(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)
|
|
|
|
def busy(*args):
|
|
raise OSError(errno.EAGAIN, "lock held by another worker")
|
|
|
|
if skill_view.os.name == "nt":
|
|
import msvcrt
|
|
|
|
monkeypatch.setattr(msvcrt, "locking", busy)
|
|
else:
|
|
import fcntl
|
|
|
|
monkeypatch.setattr(fcntl, "flock", busy)
|
|
with pytest.raises(TimeoutError, match="skill.*lock"):
|
|
with skill_view._view_lock(tmp_path):
|
|
pytest.fail("must not enter the protected view")
|
|
assert not skill_view._process_lock_held()
|
|
|
|
|
|
def test_same_thread_reenters_view_lock_without_waiting(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(skill_view, "_REBUILD_WAIT_SECONDS", 0.03, raising=False)
|
|
with skill_view._view_lock(tmp_path):
|
|
with skill_view._view_lock(tmp_path):
|
|
assert getattr(skill_view._VIEW_LOCK_DEPTH, "depth", 0) == 2
|
|
assert getattr(skill_view._VIEW_LOCK_DEPTH, "depth", 0) == 0
|
|
|
|
|
|
def test_file_lock_non_contention_error_is_not_retried(tmp_path, monkeypatch):
|
|
def broken(*args):
|
|
raise OSError(errno.EIO, "disk failure")
|
|
|
|
if skill_view.os.name == "nt":
|
|
import msvcrt
|
|
|
|
monkeypatch.setattr(msvcrt, "locking", broken)
|
|
else:
|
|
import fcntl
|
|
|
|
monkeypatch.setattr(fcntl, "flock", broken)
|
|
with pytest.raises(OSError, match="disk failure"):
|
|
with skill_view._view_lock(tmp_path):
|
|
pytest.fail("must not enter the protected view")
|
|
assert not skill_view._process_lock_held()
|