Gate the subscription notifier to one worker (#161)

Prod runs 3 uvicorn workers, but the in-process subscription notifier was
started in every worker's lifespan (it was written assuming a single-worker
deploy). Its sweep fetches recent-forecast/archive data from Open-Meteo on a
15-min timer independent of any request, so running it 3× tripled the
background upstream load against a shared quota — the source of the overnight
429/503 rate-limit errors in logs/errors, and 3× redundant subscription scans.

Elect one worker to own the notifier via a non-blocking exclusive flock on a
lockfile (THERMOGRAPH_SINGLETON_LOCK): the first worker wins and holds the fd
for its lifetime; others stand down; the OS releases the lock if the leader
dies, so a restart re-elects cleanly. Unset (single worker / dev / tests)
always wins, so behavior there is unchanged. Mirrors the shared-metrics store's
env-selected cross-worker coordination.

The neighbor warmer stays per-worker on purpose: each worker drains its own
request-fed queue, so gating it would leave non-leader queues undrained.

- backend/singleton.py: flock-based leader election (claim()).
- backend/app.py: gate notify.start() on singleton.claim().
- deploy/thermograph.service: default THERMOGRAPH_SINGLETON_LOCK to
  data/notifier.lock (needs the unit reinstalled on prod to take effect).
- deploy/thermograph.env.example: document it alongside WORKERS.
- Tests: leader election — single-worker default, first-wins, idempotent
  re-claim, second-holder stands down, re-election after release. 183 passed.

Co-authored-by: root <root@vmi3417050.contaboserver.net>

app.py

@@ -25,6 +25,7 @@ import grid
 import metrics
 import notify
 import places
+import singleton
 import store
 import users
 import views
@@ -128,8 +129,13 @@ async def _lifespan(app):
     await db.create_db_and_tables()
     places.start_loading()
     threading.Thread(target=_neighbor_warmer, name="neighbor-warmer", daemon=True).start()
-    # Background subscription evaluator (in-process; single-worker deploy assumed).
-    notify.start()
+    # Background subscription evaluator. Its periodic sweep fetches upstream on a timer
+    # (not per request), so with multiple workers it must run in exactly one — elect a
+    # leader via a lockfile (THERMOGRAPH_SINGLETON_LOCK). Unset (single worker / dev /
+    # tests) => always the leader. The neighbor warmer above stays per-worker: each
+    # drains its own request-fed queue.
+    if singleton.claim(os.environ.get("THERMOGRAPH_SINGLETON_LOCK")):
+        notify.start()
     yield
     notify.stop()
 

singleton.py

@@ -0,0 +1,50 @@
+"""Single-instance leader election for in-process background jobs.
+
+With multiple uvicorn workers each worker process runs its own app lifespan, so a
+background thread started there runs once *per worker*. That is correct for the
+neighbor warmer (each worker drains its own request-fed queue) but wrong for the
+subscription notifier: its periodic sweep scans every subscription and issues
+upstream forecast/archive fetches on a timer, independent of any request. Running
+it in N workers multiplies the Open-Meteo quota use N-fold (the source of the
+overnight 429/503 rate-limit errors after prod went to 3 workers) and does the
+same DB scan N times.
+
+``claim`` elects one worker to own such jobs via a non-blocking exclusive advisory
+lock (``flock``) on a lockfile:
+
+* the first worker to call it acquires the lock and holds the fd for the process
+  lifetime, so the lock is never dropped while it lives -> returns ``True``;
+* other workers fail the non-blocking lock and stand down -> return ``False``;
+* the OS releases the lock when the leader exits, so a restart (or a crash) lets a
+  surviving/new worker win on its next call -> clean re-election, no stale state.
+
+No lockfile configured (single-worker deploy, dev, tests) => always the leader.
+This mirrors the metrics store's env-selected coordination: unset keeps the
+zero-dependency single-worker behavior; a path opts into cross-worker arbitration.
+"""
+import fcntl
+import os
+
+# lock_path -> held fd. Kept open for the whole process lifetime: closing the fd (or
+# letting it be GC'd) would drop the flock and let a second worker also become leader.
+_held: dict[str, int] = {}
+
+
+def claim(lock_path: "str | None") -> bool:
+    """Return True if this process should own the single-instance job guarded by
+    ``lock_path``. ``None``/empty (no multi-worker coordination configured) is always
+    the leader. Idempotent: re-claiming a path this process already holds returns True
+    without touching the lock."""
+    if not lock_path:
+        return True
+    if lock_path in _held:
+        return True
+    fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o644)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        # Another process already holds it — we're not the leader.
+        os.close(fd)
+        return False
+    _held[lock_path] = fd
+    return True

tests/test_singleton.py

@@ -0,0 +1,64 @@
+"""Unit tests for the single-instance leader election (singleton.py)."""
+import fcntl
+import importlib
+import os
+
+import singleton
+
+
+def _fresh():
+    # _held is module-level state; reload for an isolated starting point per test.
+    return importlib.reload(singleton)
+
+
+def test_no_lock_path_is_always_leader():
+    s = _fresh()
+    assert s.claim(None) is True
+    assert s.claim("") is True
+
+
+def test_first_claimer_wins_and_holds_the_lock(tmp_path):
+    s = _fresh()
+    lock = str(tmp_path / "notifier.lock")
+    assert s.claim(lock) is True
+    # Held for the process lifetime: the fd stays open and flock'd.
+    assert lock in s._held
+
+
+def test_reclaiming_same_path_is_idempotent(tmp_path):
+    s = _fresh()
+    lock = str(tmp_path / "notifier.lock")
+    assert s.claim(lock) is True
+    fd = s._held[lock]
+    # Re-claiming must not re-open or drop the lock — same fd, still True.
+    assert s.claim(lock) is True
+    assert s._held[lock] == fd
+
+
+def test_second_holder_stands_down(tmp_path):
+    """When another holder already owns the lock, claim() returns False and leaks no
+    fd. Simulate the other worker by taking the exclusive flock on a separate open
+    file description (a fresh os.open, as a distinct process would)."""
+    s = _fresh()
+    lock = str(tmp_path / "notifier.lock")
+    other = os.open(lock, os.O_RDWR | os.O_CREAT, 0o644)
+    fcntl.flock(other, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    try:
+        assert s.claim(lock) is False
+        assert lock not in s._held  # did not record a role it doesn't own
+    finally:
+        fcntl.flock(other, fcntl.LOCK_UN)
+        os.close(other)
+
+
+def test_reelection_after_holder_releases(tmp_path):
+    """A restart drops the OS lock; the next claim() then wins."""
+    s = _fresh()
+    lock = str(tmp_path / "notifier.lock")
+    other = os.open(lock, os.O_RDWR | os.O_CREAT, 0o644)
+    fcntl.flock(other, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    assert s.claim(lock) is False
+    # Prior leader exits -> OS releases the lock.
+    fcntl.flock(other, fcntl.LOCK_UN)
+    os.close(other)
+    assert s.claim(lock) is True