Refresh table metadata when psycodict announces schema changes

.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"
     },

.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'

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,22 @@ 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.  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()
+
 ##############################
 #       Top-level pages      #
 ##############################

lmfdb/schema_refresh.py

@@ -0,0 +1,234 @@
+# -*- 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 ``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.  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,
+  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 remains a no-op, so this module is safe to
+deploy against current psycodict.
+
+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
+status quo (restart to pick up schema changes) unless a polling fallback is
+added later.
+"""
+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 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")
+
+
+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.  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:
+
+    - ``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._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.
+        # 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
+    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 self._disabled:
+            return
+        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: 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:
+                return
+            try:
+                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
+                )
+                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()