From da1b2c8be14e054ea6f10e6d99d83c63448c2712 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Fri, 17 Jul 2026 05:54:55 -0700 Subject: [PATCH] Gate the subscription notifier to one worker (#161) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app.py | 10 +++++-- singleton.py | 50 ++++++++++++++++++++++++++++++++ tests/test_singleton.py | 64 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 2 deletions(-) create mode 100644 singleton.py create mode 100644 tests/test_singleton.py diff --git a/app.py b/app.py index 1714864..2f2315c 100644 --- a/app.py +++ b/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() diff --git a/singleton.py b/singleton.py new file mode 100644 index 0000000..3adb59b --- /dev/null +++ b/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 diff --git a/tests/test_singleton.py b/tests/test_singleton.py new file mode 100644 index 0000000..a2c142b --- /dev/null +++ b/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