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 <noreply@anthropic.com>

Author: roed314

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      #
 ##############################

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()

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