mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-07 21:52:20 +01:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aaa9050988 |
@@ -51,8 +51,3 @@ pytest-asyncio
|
|||||||
# TestClient import when only classic httpx is present. Runtime code keeps
|
# TestClient import when only classic httpx is present. Runtime code keeps
|
||||||
# using `httpx` above; this is test-client only.
|
# using `httpx` above; this is test-client only.
|
||||||
httpx2
|
httpx2
|
||||||
# DATABASE_URL defaults to sqlite (core/database.py), but when pointed at an
|
|
||||||
# external Postgres, SQLAlchemy's postgresql dialect imports psycopg2 inside
|
|
||||||
# create_engine() and raises ModuleNotFoundError if missing. -binary avoids
|
|
||||||
# needing libpq-dev/pg_config on the host/image to compile it.
|
|
||||||
psycopg2-binary
|
|
||||||
|
|||||||
@@ -38,10 +38,7 @@ def discover_tailscale_hosts() -> List[str]:
|
|||||||
global _hosts_cache, _hosts_cache_time
|
global _hosts_cache, _hosts_cache_time
|
||||||
|
|
||||||
now = time.time()
|
now = time.time()
|
||||||
# Gate on the timestamp, not the list: a successful query that found no
|
if _hosts_cache and (now - _hosts_cache_time) < _HOSTS_CACHE_TTL:
|
||||||
# eligible peers is a real answer, and testing the list's truthiness made
|
|
||||||
# that case re-run `tailscale status` (up to a 5s timeout) on every call.
|
|
||||||
if _hosts_cache_time and (now - _hosts_cache_time) < _HOSTS_CACHE_TTL:
|
|
||||||
return list(_hosts_cache)
|
return list(_hosts_cache)
|
||||||
|
|
||||||
hosts = []
|
hosts = []
|
||||||
|
|||||||
+4
-15
@@ -84,30 +84,19 @@ async def _cached(key: Tuple, ttl: float, fetch: Callable[[], Awaitable[Any]]) -
|
|||||||
pending = fut
|
pending = fut
|
||||||
owner = True
|
owner = True
|
||||||
if not owner:
|
if not owner:
|
||||||
# A cancelled waiter must not cancel the shared Future for the owner
|
return await pending
|
||||||
# and every other waiter.
|
|
||||||
return await asyncio.shield(pending)
|
|
||||||
try:
|
try:
|
||||||
val = await fetch()
|
val = await fetch()
|
||||||
async with _shared_cache_lock:
|
async with _shared_cache_lock:
|
||||||
_shared_cache[key] = (time.monotonic() + ttl, val)
|
_shared_cache[key] = (time.monotonic() + ttl, val)
|
||||||
|
_shared_cache_pending.pop(key, None)
|
||||||
pending.set_result(val)
|
pending.set_result(val)
|
||||||
return val
|
return val
|
||||||
except asyncio.CancelledError:
|
|
||||||
# Cancellation is a BaseException on supported Python versions, so it
|
|
||||||
# bypasses the Exception handler below. Wake all current waiters while
|
|
||||||
# allowing a later caller to retry the fetch.
|
|
||||||
pending.cancel()
|
|
||||||
raise
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
async with _shared_cache_lock:
|
||||||
|
_shared_cache_pending.pop(key, None)
|
||||||
pending.set_exception(e)
|
pending.set_exception(e)
|
||||||
raise
|
raise
|
||||||
finally:
|
|
||||||
# Keep this cleanup synchronous so a second cancellation cannot
|
|
||||||
# interrupt it and leave a permanently pending Future behind. All
|
|
||||||
# access runs on the scheduler's event-loop thread.
|
|
||||||
if _shared_cache_pending.get(key) is pending:
|
|
||||||
_shared_cache_pending.pop(key, None)
|
|
||||||
|
|
||||||
|
|
||||||
def compute_next_run(schedule: str, scheduled_time: str,
|
def compute_next_run(schedule: str, scheduled_time: str,
|
||||||
|
|||||||
@@ -1,69 +0,0 @@
|
|||||||
"""A successful Tailscale query with no eligible hosts is still cached knowledge.
|
|
||||||
|
|
||||||
`discover_tailscale_hosts` gated its cache on the host list being non-empty, so a
|
|
||||||
valid "nothing to see here" answer looked identical to a cold cache and every
|
|
||||||
caller paid for another `tailscale status --json` (up to a 5s timeout). Failures
|
|
||||||
stay uncached so a peer coming online is still picked up promptly.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from src import model_discovery
|
|
||||||
|
|
||||||
|
|
||||||
class _Result:
|
|
||||||
def __init__(self, returncode, stdout):
|
|
||||||
self.returncode = returncode
|
|
||||||
self.stdout = stdout
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def tailscale(monkeypatch):
|
|
||||||
"""Count `tailscale status` invocations and start from a cold cache."""
|
|
||||||
calls = []
|
|
||||||
|
|
||||||
def _record(result):
|
|
||||||
def _run(*_args, **_kwargs):
|
|
||||||
calls.append(1)
|
|
||||||
if isinstance(result, Exception):
|
|
||||||
raise result
|
|
||||||
return result
|
|
||||||
monkeypatch.setattr(model_discovery.subprocess, "run", _run)
|
|
||||||
return calls
|
|
||||||
|
|
||||||
monkeypatch.setattr(model_discovery, "_hosts_cache", [])
|
|
||||||
monkeypatch.setattr(model_discovery, "_hosts_cache_time", 0)
|
|
||||||
return _record
|
|
||||||
|
|
||||||
|
|
||||||
def test_empty_but_successful_discovery_is_only_run_once(tailscale):
|
|
||||||
calls = tailscale(_Result(0, '{"Self":{},"Peer":{}}'))
|
|
||||||
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == []
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == []
|
|
||||||
assert len(calls) == 1
|
|
||||||
|
|
||||||
|
|
||||||
def test_nonempty_discovery_is_still_cached(tailscale):
|
|
||||||
calls = tailscale(_Result(0, '{"Self":{"TailscaleIPs":["100.1.1.1"]},"Peer":{}}'))
|
|
||||||
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == ["100.1.1.1"]
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == ["100.1.1.1"]
|
|
||||||
assert len(calls) == 1
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
"result",
|
|
||||||
[
|
|
||||||
_Result(1, ""), # tailscale installed but logged out
|
|
||||||
_Result(0, "not json"), # unparseable output
|
|
||||||
FileNotFoundError("tailscale"), # not installed
|
|
||||||
],
|
|
||||||
ids=["nonzero_exit", "bad_json", "not_installed"],
|
|
||||||
)
|
|
||||||
def test_failures_stay_retryable(tailscale, result):
|
|
||||||
calls = tailscale(result)
|
|
||||||
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == []
|
|
||||||
assert model_discovery.discover_tailscale_hosts() == []
|
|
||||||
assert len(calls) == 2
|
|
||||||
@@ -1,86 +0,0 @@
|
|||||||
import asyncio
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from src import task_scheduler
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
|
||||||
def clear_shared_cache():
|
|
||||||
task_scheduler._shared_cache.clear()
|
|
||||||
task_scheduler._shared_cache_pending.clear()
|
|
||||||
yield
|
|
||||||
task_scheduler._shared_cache.clear()
|
|
||||||
task_scheduler._shared_cache_pending.clear()
|
|
||||||
|
|
||||||
|
|
||||||
async def test_cached_owner_cancellation_wakes_waiters_and_allows_retry():
|
|
||||||
key = ("cancelled-owner",)
|
|
||||||
fetch_started = asyncio.Event()
|
|
||||||
|
|
||||||
async def blocked_fetch():
|
|
||||||
fetch_started.set()
|
|
||||||
await asyncio.Event().wait()
|
|
||||||
|
|
||||||
owner = asyncio.create_task(task_scheduler._cached(key, 60, blocked_fetch))
|
|
||||||
await fetch_started.wait()
|
|
||||||
|
|
||||||
async def unexpected_fetch():
|
|
||||||
pytest.fail("a waiter must share the owner's fetch")
|
|
||||||
|
|
||||||
waiter = asyncio.create_task(task_scheduler._cached(key, 60, unexpected_fetch))
|
|
||||||
await asyncio.sleep(0)
|
|
||||||
|
|
||||||
owner.cancel()
|
|
||||||
with pytest.raises(asyncio.CancelledError):
|
|
||||||
await owner
|
|
||||||
with pytest.raises(asyncio.CancelledError):
|
|
||||||
await asyncio.wait_for(waiter, timeout=1)
|
|
||||||
|
|
||||||
assert key not in task_scheduler._shared_cache_pending
|
|
||||||
|
|
||||||
async def retry_fetch():
|
|
||||||
return "fresh"
|
|
||||||
|
|
||||||
result = await asyncio.wait_for(
|
|
||||||
task_scheduler._cached(key, 60, retry_fetch),
|
|
||||||
timeout=1,
|
|
||||||
)
|
|
||||||
assert result == "fresh"
|
|
||||||
|
|
||||||
|
|
||||||
async def test_cached_waiter_cancellation_does_not_cancel_shared_fetch():
|
|
||||||
key = ("cancelled-waiter",)
|
|
||||||
fetch_started = asyncio.Event()
|
|
||||||
release_fetch = asyncio.Event()
|
|
||||||
|
|
||||||
async def blocked_fetch():
|
|
||||||
fetch_started.set()
|
|
||||||
await release_fetch.wait()
|
|
||||||
return "shared"
|
|
||||||
|
|
||||||
owner = asyncio.create_task(task_scheduler._cached(key, 60, blocked_fetch))
|
|
||||||
await fetch_started.wait()
|
|
||||||
|
|
||||||
async def unexpected_fetch():
|
|
||||||
pytest.fail("a waiter must share the owner's fetch")
|
|
||||||
|
|
||||||
waiter = asyncio.create_task(task_scheduler._cached(key, 60, unexpected_fetch))
|
|
||||||
await asyncio.sleep(0)
|
|
||||||
waiter.cancel()
|
|
||||||
|
|
||||||
with pytest.raises(asyncio.CancelledError):
|
|
||||||
await waiter
|
|
||||||
|
|
||||||
pending = task_scheduler._shared_cache_pending[key]
|
|
||||||
assert not pending.cancelled()
|
|
||||||
assert not owner.done()
|
|
||||||
|
|
||||||
release_fetch.set()
|
|
||||||
assert await asyncio.wait_for(owner, timeout=1) == "shared"
|
|
||||||
assert key not in task_scheduler._shared_cache_pending
|
|
||||||
|
|
||||||
async def cache_miss():
|
|
||||||
pytest.fail("the successful owner result should be cached")
|
|
||||||
|
|
||||||
assert await task_scheduler._cached(key, 60, cache_miss) == "shared"
|
|
||||||
Reference in New Issue
Block a user