fix(discovery): cache a successful but empty Tailscale lookup (#6228)

The host cache was gated on the list being non-empty, so "queried fine, no
eligible peers" looked exactly like a cold cache and every caller paid for
another `tailscale status --json` — a subprocess with a 5s timeout.

Gate on the timestamp instead. Failures still leave the timestamp unset, so a
missing binary, a non-zero exit or unparseable output stays retryable rather
than being cached for the full TTL.

Co-authored-by: Claude <noreply@anthropic.com>

requirements.txt

@@ -51,3 +51,8 @@ pytest-asyncio
 # TestClient import when only classic httpx is present. Runtime code keeps
 # using `httpx` above; this is test-client only.
 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

src/model_discovery.py

@@ -38,7 +38,10 @@ def discover_tailscale_hosts() -> List[str]:
     global _hosts_cache, _hosts_cache_time
 
     now = time.time()
-    if _hosts_cache and (now - _hosts_cache_time) < _HOSTS_CACHE_TTL:
+    # Gate on the timestamp, not the list: a successful query that found no
+    # 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)
 
     hosts = []

src/task_scheduler.py

@@ -84,19 +84,30 @@ async def _cached(key: Tuple, ttl: float, fetch: Callable[[], Awaitable[Any]]) -
             pending = fut
             owner = True
     if not owner:
-        return await pending
+        # A cancelled waiter must not cancel the shared Future for the owner
+        # and every other waiter.
+        return await asyncio.shield(pending)
     try:
         val = await fetch()
         async with _shared_cache_lock:
             _shared_cache[key] = (time.monotonic() + ttl, val)
-            _shared_cache_pending.pop(key, None)
         pending.set_result(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:
-        async with _shared_cache_lock:
-            _shared_cache_pending.pop(key, None)
         pending.set_exception(e)
         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,

tests/test_tailscale_discovery_cache.py

@@ -0,0 +1,69 @@
+"""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

tests/test_task_scheduler_cache.py

@@ -0,0 +1,86 @@
+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"