From 8acae8f473a1e3decd33fb1f71f3d977c22abae9 Mon Sep 17 00:00:00 2001 From: David Roe Date: Tue, 21 Jul 2026 23:18:10 -0400 Subject: [PATCH 1/5] Refresh table metadata when psycodict announces schema changes Companion to roed314/psycodict#111: each web worker keeps a NotificationListener subscribed to the psycodict_schema channel and, on a non-blocking poll from a before_request hook, calls db.refresh_tables() when a schema change is announced, so column and table changes become visible without restarting workers. Reconnects with a catch-up refresh after listener failures, and is a no-op (one log line) when psycodict does not provide the notification API. Co-Authored-By: Claude Fable 5 --- lmfdb/app.py | 16 +++ lmfdb/schema_refresh.py | 195 +++++++++++++++++++++++++++++ lmfdb/tests/test_schema_refresh.py | 150 ++++++++++++++++++++++ 3 files changed, 361 insertions(+) create mode 100644 lmfdb/schema_refresh.py create mode 100644 lmfdb/tests/test_schema_refresh.py diff --git a/lmfdb/app.py b/lmfdb/app.py index 423dc93303..b2b3ab9242 100644 --- a/lmfdb/app.py +++ b/lmfdb/app.py @@ -22,6 +22,7 @@ from .logger import critical from .homepage import load_boxes, contribs +from .schema_refresh import schema_refresher LMFDB_VERSION = "LMFDB Release 1.2.1" @@ -340,6 +341,21 @@ def get_menu_cookie(): """ g.show_menu = str(request.cookies.get('showmenu')) != "False" +############################## +# Schema refreshing # +############################## + + +@app.before_request +def refresh_schema_if_changed(): + """ + Pick up schema changes announced on psycodict's LISTEN/NOTIFY channel, so + that added or dropped columns and tables become visible to this worker + without a restart. A non-blocking poll (and a no-op when psycodict does + not provide the notification API). + """ + schema_refresher.check() + ############################## # Top-level pages # ############################## diff --git a/lmfdb/schema_refresh.py b/lmfdb/schema_refresh.py new file mode 100644 index 0000000000..2851c3d2f0 --- /dev/null +++ b/lmfdb/schema_refresh.py @@ -0,0 +1,195 @@ +# -*- coding: utf-8 -*- +""" +Keep a long-running website's table metadata in sync with the database. + +psycodict reads each table's columns, types and sort information from the +database once, when the connection is created. A website process therefore +does not see schema changes made elsewhere: after a column is dropped its +queries still mention the column and fail, and a newly added column stays +invisible, until every worker process is restarted. + +psycodict provides the two halves of the remedy: + +- ``db.refresh_tables()`` re-reads the schema and updates the table objects + in place, so references held by application code stay valid + (roed314/psycodict#99); +- schema-changing operations (``create_table``, ``drop_table``, + ``rename_table``, ``add_column``, ``drop_column`` and the reload swap) + announce themselves via PostgreSQL's LISTEN/NOTIFY on the channel + ``psycodict_schema``, with the affected table's name as payload, and + ``db.listener()`` subscribes to that channel (roed314/psycodict#111). + +This module ties the two together for the website. Each worker process owns +a :class:`SchemaRefresher`, driven by a non-blocking ``check()`` from a +``before_request`` hook: when a schema-change notification has arrived, the +worker refreshes its table metadata before handling the request. + +psycodict deliberately ships the notification API without threads, callbacks +or automatic reconnection, leaving those policies to the application. The +policies chosen here: + +- **No background thread.** Web workers spend their life handling requests, + so polling at request boundaries is both sufficient and free of the races a + refresh-from-another-thread would invite. An idle worker can lag behind + until its next request, which is harmless: with no requests there are no + queries to fail. A non-blocking poll on an idle connection is just a + socket read, so doing it every request costs nothing measurable. +- **Reconnect with catch-up.** If the listening connection is lost, any + notifications sent before a new ``LISTEN`` is issued are gone (PostgreSQL + delivers only what is sent after). So the refresher backs off briefly, + builds a fresh listener, and then does a full refresh to cover whatever it + may have missed -- including the window between process start and the + first subscription. +- **Refresh everything, not just the named table.** The payload names the + affected table, but ``refresh_tables()`` re-reads all metadata anyway; a + whole-catalog refresh is cheap relative to how rarely schemas change, and + it handles creates, drops and renames without special cases. The payload + is still used for logging and for collapsing a burst of notifications into + a single refresh. + +If psycodict does not provide the notification API (any release before 1.0), +the refresher logs once and disables itself, so this module is safe to +deploy against current psycodict. +""" +import os +import threading +import time +from logging import getLogger + +try: + from psycodict.notifications import SCHEMA_CHANNEL +except ImportError: + # psycodict without LISTEN/NOTIFY support; the refresher will disable + # itself, but the channel name is part of psycodict's contract either way. + SCHEMA_CHANNEL = "psycodict_schema" + +logger = getLogger("lmfdb.schema_refresh") + + +class SchemaRefresher: + """ + Refresh ``db``'s table metadata when a schema change is announced. + + Drive it by calling :meth:`check` regularly -- the LMFDB app does so in a + ``before_request`` hook. ``check`` never blocks and never raises, so it + cannot take a request down with it. + + INPUT: + + - ``db`` -- the database whose metadata to refresh; defaults to lmfdb's + ``db``, imported lazily so this module stays import-light + - ``retry_interval`` -- seconds to wait before rebuilding the listener + after a failure (default 30) + """ + + def __init__(self, db=None, retry_interval=30.0): + self._db = db + self.retry_interval = retry_interval + self._listener = None + self._pid = None + self._next_attempt = 0.0 + self._logged_unavailable = False + # before_request hooks may run concurrently under threaded or gevent + # servers; one poller at a time is plenty, so extra callers just skip. + self._lock = threading.Lock() + + @property + def db(self): + if self._db is None: + from lmfdb import db + self._db = db + return self._db + + def available(self): + """ + Whether psycodict provides both the notification API (``listener``) + and the refresh API (``refresh_tables``). + """ + return hasattr(self.db, "listener") and hasattr(self.db, "refresh_tables") + + def check(self): + """ + Poll for schema-change notifications, refreshing metadata if any arrived. + """ + if not self._lock.acquire(blocking=False): + # Another thread is polling; it will see anything we would have. + return + try: + self._check() + except Exception: + # A refresher bug must never take down the request that ran it. + logger.exception("Unexpected error while checking for schema changes") + finally: + self._lock.release() + + def _check(self): + if not self.available(): + if not self._logged_unavailable: + logger.info( + "psycodict does not provide schema-change notifications; " + "table metadata will refresh only on restart" + ) + self._logged_unavailable = True + return + if self._listener is not None and self._pid != os.getpid(): + # This process was forked (gunicorn --preload) after the listener + # was built, so the socket is shared with the parent. Abandon it + # without closing -- a close would corrupt the parent's copy -- + # and build our own below. + self._listener = None + if self._listener is None: + if time.monotonic() < self._next_attempt: + return + try: + self._listener = self.db.listener() + self._pid = os.getpid() + except Exception as err: + self._next_attempt = time.monotonic() + self.retry_interval + logger.warning( + "Could not subscribe to schema-change notifications (%s); will retry", err + ) + return + # Notifications sent while we were not subscribed are lost, so + # catch up with a full refresh on every (re)subscription. + self._refresh("subscribed to schema-change notifications") + return + try: + notifications = self._listener.poll() + except Exception as err: + self._drop_listener() + self._next_attempt = time.monotonic() + self.retry_interval + logger.warning("Lost the schema-change listener (%s); will resubscribe", err) + return + tables = sorted({payload for channel, payload in notifications if channel == SCHEMA_CHANNEL}) + if tables: + self._refresh("schema changed for %s" % (", ".join(tables))) + + def _refresh(self, reason): + try: + self.db.refresh_tables() + except Exception as err: + # Staying subscribed with stale metadata would silently swallow + # the failure; drop the listener so the next check resubscribes + # and the catch-up refresh retries this one. + self._drop_listener() + self._next_attempt = time.monotonic() + self.retry_interval + logger.warning("Failed to refresh table metadata (%s): %s; will retry", reason, err) + else: + logger.info("Refreshed table metadata: %s", reason) + + def _drop_listener(self): + if self._listener is not None: + try: + self._listener.close() + except Exception: + pass + self._listener = None + + def close(self): + """ + Close the listener (if any); a later ``check`` subscribes anew. + """ + self._drop_listener() + + +schema_refresher = SchemaRefresher() diff --git a/lmfdb/tests/test_schema_refresh.py b/lmfdb/tests/test_schema_refresh.py new file mode 100644 index 0000000000..06f2227d7c --- /dev/null +++ b/lmfdb/tests/test_schema_refresh.py @@ -0,0 +1,150 @@ +# -*- coding: utf-8 -*- +""" +Tests for lmfdb.schema_refresh. + +These use stub database/listener objects, so they exercise the refresher's +control flow (subscribe, poll, refresh, failure, fork handling) without +needing a psycodict that provides the LISTEN/NOTIFY API; they pass against +any psycodict version. +""" + +from lmfdb.schema_refresh import SCHEMA_CHANNEL, SchemaRefresher + + +class StubListener: + def __init__(self, db): + self._db = db + self.closed = False + + def poll(self, timeout=0.0): + if self._db.poll_error is not None: + raise self._db.poll_error + batch, self._db.pending = self._db.pending, [] + return batch + + def close(self): + self.closed = True + + +class StubDB: + """ + Duck-types the (tiny) psycodict surface the refresher touches. + """ + def __init__(self): + self.refreshes = 0 + self.pending = [] + self.listeners = [] + self.listen_error = None + self.poll_error = None + self.refresh_error = None + + def listener(self): + if self.listen_error is not None: + raise self.listen_error + listener = StubListener(self) + self.listeners.append(listener) + return listener + + def refresh_tables(self): + if self.refresh_error is not None: + raise self.refresh_error + self.refreshes += 1 + + +def test_unavailable_psycodict_is_a_noop(): + # An object with neither listener() nor refresh_tables(), like psycodict + # before 1.0: the refresher must disable itself, not crash the request. + refresher = SchemaRefresher(db=object()) + refresher.check() + refresher.check() + assert refresher._listener is None + + +def test_subscribe_then_notify(): + db = StubDB() + refresher = SchemaRefresher(db=db) + # The first check subscribes and does a catch-up refresh (notifications + # sent before LISTEN are lost, so a new subscriber cannot assume it has + # seen everything). + refresher.check() + assert len(db.listeners) == 1 + assert db.refreshes == 1 + # A quiet poll does not refresh. + refresher.check() + assert db.refreshes == 1 + # One batch of notifications = one refresh; other channels are ignored. + db.pending = [ + (SCHEMA_CHANNEL, "nf_fields"), + (SCHEMA_CHANNEL, "ec_curvedata"), + ("some_other_channel", "ignored"), + ] + refresher.check() + assert db.refreshes == 2 + refresher.check() + assert db.refreshes == 2 + + +def test_subscription_failure_backs_off_then_recovers(): + db = StubDB() + db.listen_error = RuntimeError("connection refused") + refresher = SchemaRefresher(db=db, retry_interval=1000) + refresher.check() + assert refresher._listener is None + assert db.refreshes == 0 + # Within the retry interval, no new attempt is made even though the + # database has recovered. + db.listen_error = None + refresher.check() + assert refresher._listener is None + # Once the interval has passed, it subscribes and catches up. + refresher._next_attempt = 0.0 + refresher.check() + assert len(db.listeners) == 1 + assert db.refreshes == 1 + + +def test_lost_listener_resubscribes_with_catchup(): + db = StubDB() + refresher = SchemaRefresher(db=db, retry_interval=0.0) + refresher.check() + assert db.refreshes == 1 + db.poll_error = RuntimeError("server closed the connection unexpectedly") + refresher.check() + assert refresher._listener is None + assert db.listeners[0].closed + db.poll_error = None + # The resubscription's catch-up refresh covers notifications that were + # lost while disconnected. + refresher.check() + assert len(db.listeners) == 2 + assert db.refreshes == 2 + + +def test_failed_refresh_drops_listener_for_retry(): + db = StubDB() + refresher = SchemaRefresher(db=db, retry_interval=0.0) + db.refresh_error = RuntimeError("could not read meta_tables") + refresher.check() + # The catch-up refresh failed: rather than staying subscribed with stale + # metadata, the listener is dropped so the next check retries in full. + assert refresher._listener is None + assert db.refreshes == 0 + db.refresh_error = None + refresher.check() + assert db.refreshes == 1 + assert refresher._listener is not None + + +def test_forked_worker_builds_its_own_listener(): + db = StubDB() + refresher = SchemaRefresher(db=db) + refresher.check() + # Simulate a fork: the recorded pid no longer matches this process. + refresher._pid = -1 + refresher.check() + # The inherited listener is abandoned *without* close (its socket is + # shared with the parent process) and a fresh one is built, followed by + # the usual catch-up refresh. + assert len(db.listeners) == 2 + assert not db.listeners[0].closed + assert db.refreshes == 2 From a1bd212aaa6c50a983cd5d7e364365bb6f554f98 Mon Sep 17 00:00:00 2001 From: David Roe Date: Tue, 21 Jul 2026 23:45:44 -0400 Subject: [PATCH 2/5] Disable schema refresher permanently on hot standbys A server in recovery refuses LISTEN outright (SQLSTATE 25006) and can never deliver notifications (NOTIFY is not WAL-logged), so retrying every 30s would warn forever. Verified against devmirror, which is a physical replica (PG 18.1, pg_is_in_recovery() = true) -- this is the situation for development copies of the website. Also document why abandoning an inherited listener without close() is safe (psycopg's pid-guarded GC). Co-Authored-By: Claude Fable 5 --- lmfdb/schema_refresh.py | 31 ++++++++++++++++++++++++++++-- lmfdb/tests/test_schema_refresh.py | 18 +++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/lmfdb/schema_refresh.py b/lmfdb/schema_refresh.py index 2851c3d2f0..65b0ccc6ab 100644 --- a/lmfdb/schema_refresh.py +++ b/lmfdb/schema_refresh.py @@ -50,6 +50,14 @@ If psycodict does not provide the notification API (any release before 1.0), the refresher logs once and disables itself, so this module is safe to deploy against current psycodict. + +The refresher likewise disables itself, for the life of the process, when +the database is a hot standby: a server in recovery refuses ``LISTEN`` +outright (SQLSTATE 25006), and notifications cannot traverse physical +replication anyway (``NOTIFY`` is not WAL-logged). This is the situation +for development copies of the website pointing at devmirror; they keep the +status quo (restart to pick up schema changes) unless a polling fallback is +added later. """ import os import threading @@ -88,6 +96,7 @@ def __init__(self, db=None, retry_interval=30.0): self._listener = None self._pid = None self._next_attempt = 0.0 + self._disabled = False self._logged_unavailable = False # before_request hooks may run concurrently under threaded or gevent # servers; one poller at a time is plenty, so extra callers just skip. @@ -123,6 +132,8 @@ def check(self): self._lock.release() def _check(self): + if self._disabled: + return if not self.available(): if not self._logged_unavailable: logger.info( @@ -134,8 +145,11 @@ def _check(self): if self._listener is not None and self._pid != os.getpid(): # This process was forked (gunicorn --preload) after the listener # was built, so the socket is shared with the parent. Abandon it - # without closing -- a close would corrupt the parent's copy -- - # and build our own below. + # without closing: an explicit close would send a protocol + # Terminate over the shared socket, killing the parent's copy, + # while just dropping the reference is safe (psycopg skips the + # protocol shutdown when collecting a connection in a process + # other than the one that created it). Then build our own below. self._listener = None if self._listener is None: if time.monotonic() < self._next_attempt: @@ -144,6 +158,19 @@ def _check(self): self._listener = self.db.listener() self._pid = os.getpid() except Exception as err: + code = getattr(err, "sqlstate", None) or getattr(err, "pgcode", None) + if code == "25006": + # "cannot execute LISTEN during recovery": the database is + # a hot standby, which can never deliver notifications + # (NOTIFY is not WAL-logged), so this is permanent for the + # life of the server -- disable rather than retry forever. + self._disabled = True + logger.info( + "Database is a hot standby (%s); schema-change " + "notifications are unavailable there, so table " + "metadata will refresh only on restart", err + ) + return self._next_attempt = time.monotonic() + self.retry_interval logger.warning( "Could not subscribe to schema-change notifications (%s); will retry", err diff --git a/lmfdb/tests/test_schema_refresh.py b/lmfdb/tests/test_schema_refresh.py index 06f2227d7c..ffd8c10f33 100644 --- a/lmfdb/tests/test_schema_refresh.py +++ b/lmfdb/tests/test_schema_refresh.py @@ -135,6 +135,24 @@ def test_failed_refresh_drops_listener_for_retry(): assert refresher._listener is not None +def test_hot_standby_disables_permanently(): + class RecoveryError(RuntimeError): + sqlstate = "25006" + + db = StubDB() + db.listen_error = RecoveryError("cannot execute LISTEN during recovery") + refresher = SchemaRefresher(db=db, retry_interval=0.0) + refresher.check() + assert refresher._listener is None + # Permanent: even after the retry interval (0s here) and with the error + # cleared, no new subscription is attempted -- a standby can never + # deliver notifications, so retrying would just warn forever. + db.listen_error = None + refresher.check() + assert db.listeners == [] + assert db.refreshes == 0 + + def test_forked_worker_builds_its_own_listener(): db = StubDB() refresher = SchemaRefresher(db=db) From 46bda2c99f0b3501d126209a699d958e22a73fd0 Mon Sep 17 00:00:00 2001 From: David Roe Date: Wed, 5 Aug 2026 02:25:27 -0400 Subject: [PATCH 3/5] Register test_schema_refresh.py with CI and sharpen its claims The new test file was never added to the explicit CI inventory, so every shard died at the "didn't miss any test files" guard before running any test or lint: schedule it in the paired proddb/devmirror matrix entries that already cover lmfdb/tests, and bump the expected file count to 45. Also make the payload handling actually observable in the tests -- the old batch mixed channels but only counted refreshes, so it would have passed with the channel filter removed, and it never repeated a payload, so it never exercised the deduplication the PR claims. A batch of purely foreign notifications now asserts no refresh, and a burst with repeats asserts the computed reason names each table once, sorted, with the other channel's payload absent. Documentation corrections, no behavior change: only the steady-state poll is non-blocking (subscribing and refreshing do database I/O); the lock stops two pollers colliding but does not serialize refresh_tables() against queries in other requests, so it does not by itself make threaded or gevent workers safe; and against pre-1.0 psycodict the refresher logs once and stays a no-op rather than setting _disabled. Co-Authored-By: Claude Opus 5 --- .github/workflows/matrix_includes.json | 4 +- .github/workflows/python-package.yml | 2 +- lmfdb/schema_refresh.py | 26 +++++++++---- lmfdb/tests/test_schema_refresh.py | 51 ++++++++++++++++++++++++-- 4 files changed, 70 insertions(+), 13 deletions(-) diff --git a/.github/workflows/matrix_includes.json b/.github/workflows/matrix_includes.json index 9d4f37cbc0..99dce391fd 100644 --- a/.github/workflows/matrix_includes.json +++ b/.github/workflows/matrix_includes.json @@ -60,12 +60,12 @@ "server": "devmirror" }, { - "files": "lmfdb/users/test_users.py lmfdb/lattice/test_lattice.py lmfdb/maass_forms/test_maass.py lmfdb/higher_genus_w_automorphisms/test_hgcwa.py lmfdb/belyi/test_belyi.py lmfdb/hypergm/test_hgm.py lmfdb/tests/test_utils.py lmfdb/tests/test_connection_reset.py", + "files": "lmfdb/users/test_users.py lmfdb/lattice/test_lattice.py lmfdb/maass_forms/test_maass.py lmfdb/higher_genus_w_automorphisms/test_hgcwa.py lmfdb/belyi/test_belyi.py lmfdb/hypergm/test_hgm.py lmfdb/tests/test_utils.py lmfdb/tests/test_connection_reset.py lmfdb/tests/test_schema_refresh.py", "folders": "belyi higher_genus_w_automorphisms hypergm lattice maass_forms tests users", "server": "proddb" }, { - "files": "lmfdb/users/test_users.py lmfdb/lattice/test_lattice.py lmfdb/maass_forms/test_maass.py lmfdb/higher_genus_w_automorphisms/test_hgcwa.py lmfdb/belyi/test_belyi.py lmfdb/hypergm/test_hgm.py lmfdb/tests/test_utils.py lmfdb/tests/test_connection_reset.py", + "files": "lmfdb/users/test_users.py lmfdb/lattice/test_lattice.py lmfdb/maass_forms/test_maass.py lmfdb/higher_genus_w_automorphisms/test_hgcwa.py lmfdb/belyi/test_belyi.py lmfdb/hypergm/test_hgm.py lmfdb/tests/test_utils.py lmfdb/tests/test_connection_reset.py lmfdb/tests/test_schema_refresh.py", "folders": "belyi higher_genus_w_automorphisms hypergm lattice maass_forms tests users", "server": "devmirror" }, diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index e170e4d958..064378b62a 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -90,7 +90,7 @@ jobs: - name: checking that we didn't miss any test files shell: bash -l {0} # If this fails you need to update the file list above and file count - run: test $(find lmfdb -name 'test_*.py' -or -name '*_test.py' | wc -l) -eq 44 + run: test $(find lmfdb -name 'test_*.py' -or -name '*_test.py' | wc -l) -eq 45 - name: Config LMFDB to run tests against proddb if: matrix.files != 'lint' && matrix.server == 'proddb' diff --git a/lmfdb/schema_refresh.py b/lmfdb/schema_refresh.py index 65b0ccc6ab..ecf757cae4 100644 --- a/lmfdb/schema_refresh.py +++ b/lmfdb/schema_refresh.py @@ -30,10 +30,15 @@ - **No background thread.** Web workers spend their life handling requests, so polling at request boundaries is both sufficient and free of the races a - refresh-from-another-thread would invite. An idle worker can lag behind - until its next request, which is harmless: with no requests there are no - queries to fail. A non-blocking poll on an idle connection is just a - socket read, so doing it every request costs nothing measurable. + refresh-from-another-thread would invite. This assumes LMFDB's + single-threaded sync workers, where a request boundary is a moment with no + query in flight; the lock in :meth:`SchemaRefresher.check` keeps concurrent + callers from polling or refreshing simultaneously, but it does not + serialize a refresh against queries running in other requests, so it is not + by itself enough for a threaded or gevent server. An idle worker can lag + behind until its next request, which is harmless: with no requests there + are no queries to fail. A non-blocking poll on an idle connection is just + a socket read, so doing it every request costs nothing measurable. - **Reconnect with catch-up.** If the listening connection is lost, any notifications sent before a new ``LISTEN`` is issued are gone (PostgreSQL delivers only what is sent after). So the refresher backs off briefly, @@ -48,7 +53,7 @@ a single refresh. If psycodict does not provide the notification API (any release before 1.0), -the refresher logs once and disables itself, so this module is safe to +the refresher logs once and remains a no-op, so this module is safe to deploy against current psycodict. The refresher likewise disables itself, for the life of the process, when @@ -79,8 +84,10 @@ class SchemaRefresher: Refresh ``db``'s table metadata when a schema change is announced. Drive it by calling :meth:`check` regularly -- the LMFDB app does so in a - ``before_request`` hook. ``check`` never blocks and never raises, so it - cannot take a request down with it. + ``before_request`` hook. The steady-state poll of an established listener + is non-blocking, and ``check`` never lets an exception reach the request, + so it cannot take a request down with it. Establishing a listener and + refreshing metadata do talk to the database, and may block. INPUT: @@ -100,6 +107,11 @@ def __init__(self, db=None, retry_interval=30.0): self._logged_unavailable = False # before_request hooks may run concurrently under threaded or gevent # servers; one poller at a time is plenty, so extra callers just skip. + # This keeps two refreshers off the same listener -- it does not + # serialize refresh_tables() against queries running in other + # requests, so it is not on its own enough to share the database + # object across concurrent requests. The design assumes LMFDB's + # single-threaded workers, where check() runs with nothing in flight. self._lock = threading.Lock() @property diff --git a/lmfdb/tests/test_schema_refresh.py b/lmfdb/tests/test_schema_refresh.py index ffd8c10f33..5bd1524535 100644 --- a/lmfdb/tests/test_schema_refresh.py +++ b/lmfdb/tests/test_schema_refresh.py @@ -51,9 +51,23 @@ def refresh_tables(self): self.refreshes += 1 +class RecordingRefresher(SchemaRefresher): + """ + Records the reason computed for each refresh, which is where the payload + handling (channel filtering, deduplication, ordering) is observable. + """ + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.reasons = [] + + def _refresh(self, reason): + self.reasons.append(reason) + super()._refresh(reason) + + def test_unavailable_psycodict_is_a_noop(): # An object with neither listener() nor refresh_tables(), like psycodict - # before 1.0: the refresher must disable itself, not crash the request. + # before 1.0: the refresher must stay a cheap no-op, not crash the request. refresher = SchemaRefresher(db=object()) refresher.check() refresher.check() @@ -72,11 +86,10 @@ def test_subscribe_then_notify(): # A quiet poll does not refresh. refresher.check() assert db.refreshes == 1 - # One batch of notifications = one refresh; other channels are ignored. + # One batch of notifications = one refresh. db.pending = [ (SCHEMA_CHANNEL, "nf_fields"), (SCHEMA_CHANNEL, "ec_curvedata"), - ("some_other_channel", "ignored"), ] refresher.check() assert db.refreshes == 2 @@ -84,6 +97,38 @@ def test_subscribe_then_notify(): assert db.refreshes == 2 +def test_other_channels_do_not_refresh(): + db = StubDB() + refresher = SchemaRefresher(db=db) + refresher.check() + assert db.refreshes == 1 + # Nothing on our channel, so nothing to do: a batch made up entirely of + # someone else's notifications must not trigger a refresh. + db.pending = [("some_other_channel", "ignored")] + refresher.check() + assert db.refreshes == 1 + + +def test_batch_collapses_duplicates_and_filters_channels(): + db = StubDB() + refresher = RecordingRefresher(db=db) + refresher.check() + assert refresher.reasons == ["subscribed to schema-change notifications"] + # A burst naming the same table repeatedly, mixed with another channel's + # traffic: one refresh, and the reason names each affected table once, in + # sorted order, with the other channel's payload nowhere in sight. + db.pending = [ + (SCHEMA_CHANNEL, "nf_fields"), + (SCHEMA_CHANNEL, "ec_curvedata"), + (SCHEMA_CHANNEL, "nf_fields"), + ("some_other_channel", "gps_groups"), + (SCHEMA_CHANNEL, "ec_curvedata"), + ] + refresher.check() + assert db.refreshes == 2 + assert refresher.reasons[-1] == "schema changed for ec_curvedata, nf_fields" + + def test_subscription_failure_backs_off_then_recovers(): db = StubDB() db.listen_error = RuntimeError("connection refused") From 6689041a458d72e8aca141faab022d5cd021ffe4 Mon Sep 17 00:00:00 2001 From: David Roe Date: Wed, 5 Aug 2026 02:28:08 -0400 Subject: [PATCH 4/5] Match the before_request docstring to the refresher's actual guarantee Co-Authored-By: Claude Opus 5 --- lmfdb/app.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lmfdb/app.py b/lmfdb/app.py index b2b3ab9242..27ca156af8 100644 --- a/lmfdb/app.py +++ b/lmfdb/app.py @@ -351,8 +351,9 @@ def refresh_schema_if_changed(): """ Pick up schema changes announced on psycodict's LISTEN/NOTIFY channel, so that added or dropped columns and tables become visible to this worker - without a restart. A non-blocking poll (and a no-op when psycodict does - not provide the notification API). + without a restart. Almost always just a non-blocking poll, and a no-op + when psycodict does not provide the notification API; subscribing and + refreshing do talk to the database, but happen only rarely. """ schema_refresher.check() From 6f05ec6cd3beb1252178856ce96289bf6e3ef067 Mon Sep 17 00:00:00 2001 From: David Roe Date: Wed, 5 Aug 2026 03:11:20 -0400 Subject: [PATCH 5/5] Drop the last two stale phrases from the schema_refresh docstrings The module docstring still introduced check() as "non-blocking" without qualification, and the ImportError fallback comment still said the refresher would "disable itself" against a pre-1.0 psycodict when it actually just stays a no-op. Both now match what the code does; the class docstring already drew the distinction correctly. While here, "likewise disables itself" for the hot-standby case had lost its antecedent when the pre-1.0 wording changed, so say plainly that this one case really does disable the refresher, in contrast to the no-op. Documentation only, no behavior change: _disabled is still set solely on SQLSTATE 25006, so capability re-detection is preserved. Co-Authored-By: Claude Opus 5 --- lmfdb/schema_refresh.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/lmfdb/schema_refresh.py b/lmfdb/schema_refresh.py index ecf757cae4..38403f93bb 100644 --- a/lmfdb/schema_refresh.py +++ b/lmfdb/schema_refresh.py @@ -20,9 +20,9 @@ ``db.listener()`` subscribes to that channel (roed314/psycodict#111). This module ties the two together for the website. Each worker process owns -a :class:`SchemaRefresher`, driven by a non-blocking ``check()`` from a -``before_request`` hook: when a schema-change notification has arrived, the -worker refreshes its table metadata before handling the request. +a :class:`SchemaRefresher`, driven by ``check()`` from a ``before_request`` +hook: when a schema-change notification has arrived, the worker refreshes its +table metadata before handling the request. psycodict deliberately ships the notification API without threads, callbacks or automatic reconnection, leaving those policies to the application. The @@ -56,8 +56,8 @@ the refresher logs once and remains a no-op, so this module is safe to deploy against current psycodict. -The refresher likewise disables itself, for the life of the process, when -the database is a hot standby: a server in recovery refuses ``LISTEN`` +The refresher does disable itself, for the life of the process, when the +database is a hot standby: a server in recovery refuses ``LISTEN`` outright (SQLSTATE 25006), and notifications cannot traverse physical replication anyway (``NOTIFY`` is not WAL-logged). This is the situation for development copies of the website pointing at devmirror; they keep the @@ -72,8 +72,8 @@ try: from psycodict.notifications import SCHEMA_CHANNEL except ImportError: - # psycodict without LISTEN/NOTIFY support; the refresher will disable - # itself, but the channel name is part of psycodict's contract either way. + # psycodict without LISTEN/NOTIFY support; the refresher will remain a + # no-op, but the channel name is part of psycodict's contract either way. SCHEMA_CHANNEL = "psycodict_schema" logger = getLogger("lmfdb.schema_refresh")