From 7a3c534cedf12e61df93ed6d3e59bfb5fbf8eb4a Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 03:18:16 +0000 Subject: [PATCH 1/5] Spike: SchemaWatcher keeps the _internal catalog current off the request path Per-database schema_watch modes (owned / external / immutable): - write path reads PRAGMA schema_version after every write task and rebuilds that one database's catalog before the write returns - external files are polled with a stat() fingerprint prefilter (racily-clean fingerprints re-verified), PRAGMA schema_version only on change, rebuild only when the version moved or the inode changed - immutable databases are scanned once - remove_database deletes its catalog rows explicitly; startup deletes rows for databases that are not attached (no per-refresh full scans) - catalog reads use one short-lived connection per database inside one read transaction; shared-cache memory databases go through their write thread - replaced/deleted files invalidate cached connections (generation) - fingerprints persisted in catalog_databases.fingerprint (migration 0002) so restarts skip unchanged files - new setting schema_watch_interval_ms (default 1000, 0 = no polling) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/app.py | 104 ++-- datasette/database.py | 130 ++++- datasette/schema_watcher.py | 889 +++++++++++++++++++++++++++++++++ datasette/utils/internal_db.py | 200 ++++---- datasette/views/base.py | 2 +- datasette/views/database.py | 4 +- datasette/views/special.py | 4 +- datasette/views/table.py | 2 +- docs/settings.rst | 27 + tests/test_api.py | 1 + tests/test_internal_db.py | 6 +- tests/test_schema_watcher.py | 516 +++++++++++++++++++ 12 files changed, 1727 insertions(+), 158 deletions(-) create mode 100644 datasette/schema_watcher.py create mode 100644 tests/test_schema_watcher.py diff --git a/datasette/app.py b/datasette/app.py index 3e4c5acf..6ce72b10 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -49,6 +49,7 @@ from .events import Event from .plugins import DEFAULT_PLUGINS, get_plugins, pm from .renderer import json_renderer from .resources import DatabaseResource, TableResource +from .schema_watcher import SchemaWatcher from .telemetry import ( TelemetryMiddleware, _in_datasette_client, @@ -108,7 +109,7 @@ from .utils.asgi import ( asgi_send_redirect, asgi_static, ) -from .utils.internal_db import init_internal_db, populate_schema_tables +from .utils.internal_db import init_internal_db from .utils.sqlite import ( sqlite3, using_pysqlite3, @@ -303,6 +304,11 @@ SETTINGS = ( "Allow display of SQL trace debug information with ?_trace=1", ), Setting("base_url", "/", "Datasette URLs should use this base path"), + Setting( + "schema_watch_interval_ms", + 1000, + "How often to check external database files for schema changes - set 0 to disable polling", + ), ) _HASH_URLS_REMOVED = "The hash_urls setting has been removed, try the datasette-hashed-urls plugin instead" OBSOLETE_SETTINGS = { @@ -482,16 +488,21 @@ class Datasette: self._background_tasks = BackgroundTaskSupervisor(self) self.crossdb = crossdb self.nolock = nolock + self.internal_db_created = False + self._schema_watcher = SchemaWatcher(self) if memory or crossdb or not self.files: self.add_database( Database(self, is_mutable=False, is_memory=True), name="_memory" ) for file in self.files: + is_mutable = file not in self.immutables self.add_database( - Database(self, file, is_mutable=file not in self.immutables) + Database(self, file, is_mutable=is_mutable), + # Files named on the command line may be changed by other + # processes, so they are polled by default + schema_watch="external" if is_mutable else "immutable", ) - self.internal_db_created = False if internal is None: self._internal_database = Database(self, is_temp_disk=True) else: @@ -651,6 +662,8 @@ class Datasette: self.root_enabled = False self.default_deny = default_deny self.client = DatasetteClient(self) + # ds.config is available now: resolve per-database schema_watch modes + self._schema_watcher.configure() # Last, so metric callbacks never see a partially initialized instance register_datasette(self) @@ -713,66 +726,38 @@ class Datasette: return None async def refresh_schemas(self, *, force=False): - # Throttle schema refreshes to at most once per second + """Bring the _internal catalog up to date with every attached database. + + Uses the SchemaWatcher stat() prefilter, so it only opens connections + to databases whose files changed. Not called on the request path any + more - the watcher polls external databases in the background and + the write path catches schema changes made through Datasette. + """ + # Throttle non-forced refreshes to at most once per second if ( not force and time.monotonic() - getattr(self, "_last_schema_refresh", 0) < 1.0 ): return self._last_schema_refresh = time.monotonic() - if self._refresh_schemas_lock.locked(): + if self._refresh_schemas_lock.locked() and not force: return async with self._refresh_schemas_lock: await self._refresh_schemas() - async def _refresh_schemas(self): + async def _refresh_schemas(self, *, background=False): internal_db = self.get_internal_database() if not self.internal_db_created: + if background: + return await init_internal_db(internal_db) await self.apply_metadata_json() self.internal_db_created = True - current_schema_versions = { - row["database_name"]: row["schema_version"] - for row in await internal_db.execute( - "select database_name, schema_version from catalog_databases" - ) - } - catalog_table_names = ( - "catalog_columns", - "catalog_foreign_keys", - "catalog_indexes", - "catalog_views", - "catalog_tables", - "catalog_databases", - ) - # Delete stale entries for databases that are no longer attached - catalog_database_names = set(current_schema_versions.keys()) - for table in catalog_table_names[:-1]: - catalog_database_names.update( - row["database_name"] - for row in await internal_db.execute( - f"select distinct database_name from {table}" - ) - if row["database_name"] is not None - ) - stale_databases = catalog_database_names - set(self.databases.keys()) - if stale_databases: - - def delete_stale_database_catalog(conn): - for stale_db_name in stale_databases: - for table in catalog_table_names: - conn.execute( - f"DELETE FROM {table} WHERE database_name = ?", - [stale_db_name], - ) - - await internal_db.execute_write_fn(delete_stale_database_catalog) - for database_name, db in self.databases.items(): - schema_version = (await db.execute("PRAGMA schema_version")).first()[0] - # Compare schema versions to see if we should skip it - if schema_version == current_schema_versions.get(database_name): - continue - await populate_schema_tables(internal_db, db, schema_version) + # Full catalog build (skips files whose persisted fingerprint + # shows they are unchanged when --internal is a real file) + await self._schema_watcher.initial_scan() + return + await self._schema_watcher.sweep(background=background) @property def urls(self): @@ -940,7 +925,15 @@ class Datasette: name = next(iter(self.databases.keys())) return self.databases[name] - def add_database(self, db, name=None, route=None): + def add_database(self, db, name=None, route=None, schema_watch=None): + """Attach a database. + + ``schema_watch`` is how the _internal catalog is kept current for it: + ``"owned"`` (default - schema changes come through Datasette's write + methods), ``"external"`` (other processes may change the file, poll + it) or ``"immutable"`` (scan once). ``databases..schema_watch`` + in datasette.yaml overrides it. + """ new_databases = self.databases.copy() if name is None: # Pick a unique name for this database @@ -957,6 +950,9 @@ class Datasette: new_databases[name] = db # don't mutate! that causes race conditions with live import self.databases = new_databases + self._schema_watcher.register( + db, schema_watch or getattr(db, "schema_watch", None) + ) return db def add_memory_database(self, memory_name, name=None, route=None): @@ -965,10 +961,14 @@ class Datasette: ) def remove_database(self, name): - self.get_database(name).close() + db = self.get_database(name) new_databases = self.databases.copy() new_databases.pop(name) self.databases = new_databases + # Deletes this database's catalog rows (explicitly - there is no + # periodic stale-catalog scan any more) + self._schema_watcher.unregister(name) + db.close() def close(self): """Release all resources held by this Datasette instance. @@ -980,6 +980,7 @@ class Datasette: if self._closed: return self._closed = True + self._schema_watcher.stop() # Stop reporting metrics before closing databases unregister_datasette(self) first_exception = None @@ -3110,6 +3111,7 @@ ORDER BY allowed.parent, allowed.child """ await self.invoke_startup() await self._background_tasks.launch_all() + await self._schema_watcher.start() async def _launch_background_tasks(self): """Idempotently launch every registered background task. Private: @@ -3132,6 +3134,7 @@ ORDER BY allowed.parent, allowed.child if self._suppress_background_tasks: return await self._background_tasks.launch_all() + await self._schema_watcher.start() async def invoke_shutdown(self): """Run the graceful teardown sequence: plugin ``shutdown`` hooks, @@ -3147,6 +3150,7 @@ ORDER BY allowed.parent, allowed.child except Exception: logging.getLogger("datasette").exception("shutdown hook failed") await self._background_tasks.cancel_all(grace=5.0) + await self._schema_watcher.astop() self.close() def app(self): diff --git a/datasette/database.py b/datasette/database.py index 542b3012..056e8109 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -130,9 +130,16 @@ class Database: # These are used when in non-threaded mode: self._read_connection = None self._write_connection = None + self._read_connection_generation = 0 + self._write_connection_generation = 0 # Track file and memory connections, including reads on worker threads, # so close() can release all of them from the calling thread. self._all_connections = [] + # Set by SchemaWatcher.register(); None for the internal database + self._watch_state = None + # Bumped when the file behind this database is replaced or deleted, + # so cached connections still pointing at the old inode get reopened + self._conn_generation = 0 if not is_temp_disk: self.mode = mode @@ -144,6 +151,50 @@ class Database: with self._pending_execute_futures_lock: self._pending_execute_futures.discard(future) + def _invalidate_connections(self): + """Cached read connections and the write connection are reopened the + next time their thread uses them (they may point at a replaced or + deleted inode).""" + self._conn_generation += 1 + + def _forget_connection(self, conn): + try: + conn.close() + except Exception: # noqa: BLE001, S110 + pass + try: + self._all_connections.remove(conn) + except ValueError: + pass + + def _schema_check(self, conn, loop=None): + # Runs on the thread that owns conn, right after a write task + state = self._watch_state + if state is not None: + self.ds._schema_watcher.check_after_write(state, conn, loop) + + async def _execute_on_write_connection(self, fn): + """Run fn(conn) on the write connection, serialized with writes, + without write_wrapper hooks, events or a transaction. Used by the + SchemaWatcher for shared-cache memory databases, where a second + connection reading sqlite_master would make concurrent DDL fail + with "database table is locked" (shared-cache locks do not wait).""" + self._check_not_closed() + if self.ds.executor is None: + if self._write_connection is None: + self._write_connection = self.connect(write=True) + self._write_connection_generation = self._conn_generation + self.ds._prepare_connection(self._write_connection, self.name) + return fn(self._write_connection) + return await self._send_to_write_thread(fn, transaction=False) + + async def _after_write(self): + # Read-your-writes for the catalog: if that write changed the schema, + # wait for the catalog rows for this database to be rebuilt + state = self._watch_state + if state is not None and (state.needs_scan or state.scan_future is not None): + await self.ds._schema_watcher.after_write(state) + @property def cached_table_counts(self): if self._cached_table_counts is not None: @@ -393,6 +444,8 @@ class Database: try: return fn(isolated_connection) finally: + if write: + self._schema_check(isolated_connection) isolated_connection.close() try: self._all_connections.remove(isolated_connection) @@ -408,7 +461,9 @@ class Database: with record_operation_duration(self.name, "write" if write else "read"): if self.ds.executor is None: # non-threaded mode - return _run() + result = _run() + await self._after_write() + return result if not write: # Immutable database - no writes can ever occur, so there # is no write queue to block; run against a fresh @@ -418,7 +473,9 @@ class Database: self.ds.executor, ctx.run, _run ) # Threaded mode - send to write thread - return await self._send_to_write_thread(fn, isolated_connection=True) + result = await self._send_to_write_thread(fn, isolated_connection=True) + await self._after_write() + return result async def analyze_sql(self, sql, params=None) -> SQLAnalysis: self._check_not_closed() @@ -456,15 +513,25 @@ class Database: fn = self._wrap_fn_with_hooks(fn, request, transaction, track_event) if self.ds.executor is None: # non-threaded mode + if ( + self._write_connection is not None + and self._write_connection_generation != self._conn_generation + ): + self._forget_connection(self._write_connection) + self._write_connection = None if self._write_connection is None: self._write_connection = self.connect(write=True) + self._write_connection_generation = self._conn_generation self.ds._prepare_connection(self._write_connection, self.name) - if transaction: - with self._write_connection: - self._write_connection.execute("BEGIN IMMEDIATE") + try: + if transaction: + with self._write_connection: + self._write_connection.execute("BEGIN IMMEDIATE") + result = fn(self._write_connection) + else: result = fn(self._write_connection) - else: - result = fn(self._write_connection) + finally: + self._schema_check(self._write_connection) if not block: # There is no write thread here, so the write has already # finished. Hand back the same (task_id, reply_future) shape @@ -479,6 +546,7 @@ class Database: fn, block=block, transaction=transaction ) if block: + await self._after_write() for event in pending_events: await self.ds.track_event(event) else: @@ -572,6 +640,7 @@ class Database: # to this database conn_exception = None conn = None + conn_generation = self._conn_generation try: conn = self.connect(write=True) # Threads do not inherit the caller's context, so any spans @@ -582,6 +651,18 @@ class Database: conn_exception = e while True: task = self._write_queue.get() + if task is not _SHUTDOWN and conn_generation != self._conn_generation: + # The file was replaced or deleted: reopen by path + if conn is not None: + self._forget_connection(conn) + conn = None + conn_exception = None + conn_generation = self._conn_generation + try: + conn = self.connect(write=True) + self.ds._prepare_connection(conn, self.name) + except Exception as e: # noqa: BLE001 + conn_exception = e if task is _SHUTDOWN: if conn is not None: try: @@ -628,6 +709,7 @@ class Database: try: result = task.fn(isolated_connection) finally: + self._schema_check(isolated_connection, task.loop) isolated_connection.close() try: self._all_connections.remove(isolated_connection) @@ -649,12 +731,15 @@ class Database: task.isolated_connection, ) span.set_attribute(TRANSACTION, task.transaction) - if task.transaction: - with conn: - conn.execute("BEGIN IMMEDIATE") + try: + if task.transaction: + with conn: + conn.execute("BEGIN IMMEDIATE") + result = task.fn(conn) + else: result = task.fn(conn) - else: - result = task.fn(conn) + finally: + self._schema_check(conn, task.loop) except Exception as e: # noqa: BLE001 sys.stderr.write(f"{e}\n") sys.stderr.flush() @@ -688,18 +773,33 @@ class Database: self._check_not_closed() if self.ds.executor is None: # non-threaded mode + if ( + self._read_connection is not None + and self._read_connection_generation != self._conn_generation + ): + self._forget_connection(self._read_connection) + self._read_connection = None if self._read_connection is None: self._read_connection = self.connect() + self._read_connection_generation = self._conn_generation self.ds._prepare_connection(self._read_connection, self.name) return fn(self._read_connection) # threaded mode def in_thread(): - conn = getattr(connections, self._thread_local_id, None) - if not conn: + cached = getattr(connections, self._thread_local_id, None) + conn = None + if cached: + conn, generation = cached + if generation != self._conn_generation: + # File replaced/deleted since this connection was opened + self._forget_connection(conn) + conn = None + if conn is None: + generation = self._conn_generation conn = self.connect() self.ds._prepare_connection(conn, self.name) - setattr(connections, self._thread_local_id, conn) + setattr(connections, self._thread_local_id, (conn, generation)) return fn(conn) with self._pending_execute_futures_lock: diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py new file mode 100644 index 00000000..860a6ad7 --- /dev/null +++ b/datasette/schema_watcher.py @@ -0,0 +1,889 @@ +""" +Keeps the ``_internal`` catalog (``catalog_databases``, ``catalog_tables``, +``catalog_columns`` ...) current without checking every attached database on +the request path. + +Every attached database has a *schema watch mode*: + +``owned`` + Every schema change goes through Datasette. After each write task the + write thread reads ``PRAGMA schema_version`` on the connection it just + used (microseconds, same thread). If the value moved, the catalog for + that one database is rebuilt before the write call returns, so callers + get read-your-writes on the catalog. Owned databases are never polled. + +``external`` + The file can also be changed by other processes (sqlite-utils, cron + jobs...). On top of the write-path check, a background task stats the + database file, its ``-wal`` and its ``-journal`` every + ``schema_watch_interval_ms``. Only a database whose fingerprint + ``(st_dev, st_ino, st_size, st_mtime_ns, st_ctime_ns)`` changed - or + whose previous fingerprint was *racily clean* - gets a + ``PRAGMA schema_version`` check on a short-lived connection, and only a + changed ``schema_version`` (or a replaced or deleted file) triggers a + catalog rebuild for that database. + +``immutable`` + Scanned once at startup and never again. + +Defaults: files passed to ``Datasette(files=...)`` / the CLI are +``external``, immutable files are ``immutable``, anything added later with +``datasette.add_database()`` is ``owned``. Override per database with +``databases: {name: {schema_watch: owned|external|immutable}}`` in +``datasette.yaml`` or ``add_database(..., schema_watch=...)``. + +Racily clean fingerprints +------------------------- +Like git's index: a fingerprint whose newest timestamp is within +``RACY_WINDOW_NS`` of the moment it was taken could be followed by another +write inside the same timestamp tick that leaves size, inode and mtime +identical. Such a fingerprint is flagged ``racy`` and the next sweep runs +the ``PRAGMA schema_version`` check even though the fingerprint looks +unchanged. The stat is always taken *before* the pragma, so a write landing +between the two either shows up in the pragma or changes the next stat. + +All catalog reads use one short-lived connection per database inside a +single read transaction, so ``schema_version`` and the rows written to the +catalog describe the same snapshot. The watcher never keeps a connection +open. +""" + +import asyncio +import json +import logging +import os +import time + +from .utils import sqlite3 +from .utils.internal_db import CATALOG_TABLES, collect_schema, write_catalog_entries + +logger = logging.getLogger("datasette.schema_watcher") + +MODES = ("owned", "external", "immutable") +# Timestamps closer than this to the time of the stat are treated as racy. +# 2s covers coarse filesystems (FAT: 2s, HFS+: 1s) and clock skew on +# network filesystems; ext4/xfs/tmpfs on Linux are far finer than this. +RACY_WINDOW_NS = 2_000_000_000 +# Databases per thread hop / per internal-DB transaction for bulk scans +SCAN_CHUNK = 64 +SCAN_CONCURRENCY = 3 +# Directories with at least this many watched files are listed once per +# sweep with os.scandir() instead of probing -wal/-journal with stat() +SCANDIR_THRESHOLD = 8 +PRAGMA_BUSY_TIMEOUT_MS = 200 + + +class WatchState: + __slots__ = ( + "catalog_version", + "db", + "error", + "fp", + "fp_racy", + "missing", + "mode", + "name", + "needs_scan", + "notified_version", + "removed", + "requested_mode", + "scan_future", + "stats", + ) + + def __init__(self, db, name, requested_mode): + self.db = db + self.name = name + self.requested_mode = requested_mode + self.mode = None + self.fp = None + self.fp_racy = True + self.catalog_version = None + self.notified_version = None + self.needs_scan = True + self.scan_future = None + self.missing = False + self.removed = False + self.error = None + self.stats = {"checks": 0, "scans": 0, "write_detections": 0} + + @property + def is_file(self): + db = self.db + return bool(db.path) and not db.is_memory + + def __repr__(self): + return f"" + + +def _fp_main(st): + return (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns) + + +def _fp_side(st): + # No ctime for -wal/-journal: when running as root SQLite fchown()s + # these files to match the database every time a connection opens + # them, so their ctime moves on every open - including our own checks + return (st.st_ino, st.st_size, st.st_mtime_ns) + + +def fingerprint(path, present=None): + """(main, wal, journal) stat tuples; None for a file that does not exist. + + ``present`` is an optional set of file names in the directory (from + scandir) used to skip stat() calls for -wal/-journal files that are not + there. + """ + path = os.fspath(path) + try: + main = _fp_main(os.stat(path)) + except FileNotFoundError: + main = None + parts = [main] + base = os.path.basename(path) if present is not None else None + for suffix in ("-wal", "-journal"): + if present is not None and (base + suffix) not in present: + parts.append(None) + continue + try: + parts.append(_fp_side(os.stat(path + suffix))) + except FileNotFoundError: + parts.append(None) + return tuple(parts) + + +def newest_timestamp(fp): + main = fp[0] + newest = max(main[3], main[4]) if main is not None else 0 + for part in fp[1:]: + if part is not None: + newest = max(newest, part[2]) + return newest + + +def is_racy(fp, t_ns): + return newest_timestamp(fp) >= t_ns - RACY_WINDOW_NS + + +def _fp_to_json(fp, t_ns): + return json.dumps({"fp": fp, "t": t_ns}) + + +def _fp_from_json(value): + try: + data = json.loads(value) + fp = tuple(tuple(p) if p is not None else None for p in data["fp"]) + return fp, int(data["t"]) + except Exception: # noqa: BLE001 + return None, None + + +class SchemaWatcher: + def __init__(self, ds): + self.ds = ds + self.states = {} + self._pending = set() # states registered after startup, not yet scanned + self._pending_removals = set() # names whose catalog rows must go + self._task = None + self._spawned = set() + self._last_sweep = 0.0 + self.counters = { + "sweeps": 0, + "sweep_seconds": 0.0, + "stats": 0, + "pragma_checks": 0, + "scans": 0, + "write_detections": 0, + "replaced": 0, + "missing": 0, + "busy": 0, + "restored_from_persisted": 0, + } + + # ------------------------------------------------------------------ + # configuration + # ------------------------------------------------------------------ + @property + def interval_s(self): + ms = self.ds.setting("schema_watch_interval_ms") + return (ms or 0) / 1000.0 + + def _config_mode(self, name): + config = getattr(self.ds, "config", None) or {} + db_config = (config.get("databases") or {}).get(name) or {} + return db_config.get("schema_watch") + + def _resolve_mode(self, state): + db = state.db + mode = self._config_mode(state.name) or state.requested_mode + if mode is None: + mode = "owned" + if mode not in MODES: + from .utils import StartupError + + raise StartupError( + f"Invalid schema_watch mode {mode!r} for database {state.name!r}, " + f"expected one of {', '.join(MODES)}" + ) + if not db.is_mutable: + # Nothing can change an immutable database + mode = "immutable" + elif mode == "external" and not state.is_file: + # There is no file to stat: rely on the write path + mode = "owned" + return mode + + def configure(self): + """Resolve modes once ds.config is available (end of __init__).""" + for state in self.states.values(): + state.mode = self._resolve_mode(state) + + # ------------------------------------------------------------------ + # registration + # ------------------------------------------------------------------ + def register(self, db, requested_mode=None): + if requested_mode is not None and requested_mode not in MODES: + raise ValueError( + f"schema_watch must be one of {', '.join(MODES)}, not {requested_mode!r}" + ) + state = WatchState(db, db.name, requested_mode) + if hasattr(self.ds, "config"): + state.mode = self._resolve_mode(state) + old = self.states.get(db.name) + if old is not None: + old.removed = True + self.states[db.name] = state + db._watch_state = state + self._pending_removals.discard(db.name) + if self.ds.internal_db_created: + self._pending.add(state) + self._spawn(self.refresh([state])) + return state + + def unregister(self, name): + state = self.states.pop(name, None) + if state is None: + return + state.removed = True + state.db._watch_state = None + self._pending.discard(state) + if self.ds.internal_db_created: + self._pending_removals.add(name) + self._spawn(self._delete_catalog([name])) + + # ------------------------------------------------------------------ + # task helpers + # ------------------------------------------------------------------ + def _spawn(self, coro): + try: + loop = asyncio.get_running_loop() + except RuntimeError: + # No loop (sync caller): the work stays pending and is picked + # up by the next flush/sweep + coro.close() + return None + task = loop.create_task(coro) + self._spawned.add(task) + + def _done(t): + self._spawned.discard(t) + if not t.cancelled() and t.exception() is not None: + logger.error("schema watcher task failed", exc_info=t.exception()) + + task.add_done_callback(_done) + return task + + def loop_running(self): + task = self._task + if task is None or task.done(): + return False + try: + return task.get_loop() is asyncio.get_running_loop() + except RuntimeError: + return False + + async def start(self): + """Start the polling loop (idempotent). Runs one sweep first so the + catalog is current when the first request is served.""" + if self.interval_s <= 0 or self.ds._closed or self.loop_running(): + return + loop = asyncio.get_running_loop() + # Claim the slot before awaiting so concurrent start() calls no-op + self._task = loop.create_task(self._run(), name="datasette-schema-watcher") + if self.ds.internal_db_created: + try: + await self.ds._refresh_schemas(background=True) + except Exception: + logger.exception("schema watcher initial sweep failed") + + async def _run(self): + # Wait one interval first: start() has just swept + while not self.ds._closed: + await asyncio.sleep(self.interval_s) + if self.ds._closed: + return + try: + await self.ds._refresh_schemas(background=True) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("schema watcher sweep failed") + + def stop(self): + tasks = [t for t in [self._task, *self._spawned] if t is not None] + self._task = None + for task in tasks: + if task.done(): + continue + loop = task.get_loop() + try: + running = asyncio.get_running_loop() + except RuntimeError: + running = None + if running is loop: + task.cancel() + elif not loop.is_closed(): + try: + loop.call_soon_threadsafe(task.cancel) + except RuntimeError: + pass + + async def astop(self): + tasks = [t for t in [self._task, *self._spawned] if t is not None] + self.stop() + loop = asyncio.get_running_loop() + mine = [t for t in tasks if t.get_loop() is loop and not t.done()] + if mine: + await asyncio.wait(mine, timeout=5) + + # ------------------------------------------------------------------ + # request path: O(1) unless add/remove_database left work pending + # ------------------------------------------------------------------ + async def on_request(self): + if (self._pending or self._pending_removals) and self.ds.internal_db_created: + await self.flush_pending() + if ( + not self.loop_running() + and self.interval_s > 0 + and self.ds.internal_db_created + ): + await self.start() + + async def flush_pending(self): + if self._pending_removals: + await self._delete_catalog(list(self._pending_removals)) + if self._pending: + await self.refresh(list(self._pending)) + + # ------------------------------------------------------------------ + # write path + # ------------------------------------------------------------------ + def check_after_write(self, state, conn, loop=None): + """Called on the thread that owns ``conn`` right after a write task. + + ``loop`` is the event loop to notify (None when already on it).""" + try: + version = conn.execute("PRAGMA schema_version").fetchone()[0] + except Exception: # noqa: BLE001 + return + if version == state.catalog_version or version == state.notified_version: + return + state.notified_version = version + if loop is None: + self._on_write_schema_change(state) + else: + try: + loop.call_soon_threadsafe(self._on_write_schema_change, state) + except RuntimeError: + pass + + def _on_write_schema_change(self, state): + if state.removed: + return + state.needs_scan = True + state.stats["write_detections"] += 1 + self.counters["write_detections"] += 1 + if self.ds.internal_db_created: + self._spawn(self.refresh([state])) + + async def after_write(self, state): + """Awaited by blocking writes: gives read-your-writes on the catalog.""" + if not self.ds.internal_db_created or state.removed: + return + try: + await self.refresh([state]) + except Exception: + logger.exception("catalog refresh after write to %r failed", state.name) + + # ------------------------------------------------------------------ + # sweeps (stat prefilter) + # ------------------------------------------------------------------ + async def sweep(self, *, background): + """Check databases for changes made outside Datasette. + + background=True (the polling loop): external-mode files only. + background=False (explicit refresh_schemas()): every mutable + database - files via the stat prefilter, named in-memory databases + via a PRAGMA schema_version check. + """ + t0 = time.perf_counter() + if self._pending_removals: + await self._delete_catalog(list(self._pending_removals)) + candidates = [] + for state in list(self.states.values()): + if state.mode == "immutable" or state.mode is None: + if state.needs_scan: + candidates.append(state) + continue + if background and state.mode != "external": + continue + candidates.append(state) + outcomes = [] + file_candidates = [s for s in candidates if s.is_file] + if file_candidates: + outcomes = await asyncio.to_thread(self._sweep_sync, file_candidates) + for state in candidates: + if not state.is_file and state.db.memory_name and not state.removed: + outcomes.append(await self._memory_pragma_check(state)) + to_scan = [s for s in self._pending if not s.removed] + to_clear = [] + for state, kind, fp, t_ns, version in outcomes: + if state.removed: + continue + self.counters["stats"] += 1 + if fp is not None: + state.fp = fp + state.fp_racy = is_racy(fp, t_ns) + if kind == "clean": + pass + elif kind == "missing": + self.counters["missing"] += 1 + if not state.missing: + state.missing = True + state.db._invalidate_connections() + to_clear.append(state) + elif kind == "replaced": + self.counters["replaced"] += 1 + state.missing = False + state.db._invalidate_connections() + state.needs_scan = True + elif kind == "busy": + self.counters["busy"] += 1 + state.fp_racy = True # check again next sweep + elif kind == "checked": + self.counters["pragma_checks"] += 1 + state.stats["checks"] += 1 + if version != state.catalog_version: + state.needs_scan = True + elif kind == "error": + state.fp_racy = True + if state.needs_scan: + to_scan.append(state) + if to_clear: + await self._clear_missing(to_clear) + if to_scan: + await self.refresh(to_scan) + self.counters["sweeps"] += 1 + self.counters["sweep_seconds"] += time.perf_counter() - t0 + self._last_sweep = time.monotonic() + + def _sweep_sync(self, states): + """Runs in a worker thread. Reads state, never writes it.""" + by_dir = {} + for state in states: + if state.is_file: + by_dir.setdefault( + os.path.dirname(os.path.abspath(state.db.path)), [] + ).append(state) + out = [] + for directory, dir_states in by_dir.items(): + present = None + if len(dir_states) >= SCANDIR_THRESHOLD: + try: + with os.scandir(directory) as it: + present = {entry.name for entry in it} + except OSError: + present = None + for state in dir_states: + t_ns = time.time_ns() + fp = fingerprint(state.db.path, present) + prev = state.fp + if fp == prev and not state.fp_racy and not state.needs_scan: + out.append((state, "clean", fp, t_ns, None)) + continue + if fp[0] is None: + out.append((state, "missing", fp, t_ns, None)) + continue + if prev is None or prev[0] is None or prev[0][:2] != fp[0][:2]: + # New inode/device (atomic replace, delete+recreate) or + # first sight: the schema_version alone cannot be trusted + out.append((state, "replaced", fp, t_ns, None)) + continue + out.append(self._pragma_check(state, fp, t_ns)) + return out + + async def _memory_pragma_check(self, state): + def _version(conn): + return conn.execute("PRAGMA schema_version").fetchone()[0] + + try: + version = await state.db._execute_on_write_connection(_version) + except Exception: # noqa: BLE001 + return (state, "error", None, None, None) + return (state, "checked", None, None, version) + + def _pragma_check(self, state, fp, t_ns): + db = state.db + try: + conn = self._connect(db, prepare=False) + except Exception: # noqa: BLE001 + return (state, "error", fp, t_ns, None) + try: + conn.execute(f"PRAGMA busy_timeout={PRAGMA_BUSY_TIMEOUT_MS}") + version = conn.execute("PRAGMA schema_version").fetchone()[0] + return (state, "checked", fp, t_ns, version) + except sqlite3.OperationalError as e: + msg = str(e) + if "locked" in msg or "busy" in msg: + return (state, "busy", fp, t_ns, None) + return (state, "error", fp, t_ns, None) + except Exception: # noqa: BLE001 + return (state, "error", fp, t_ns, None) + finally: + self._close(db, conn) + + # ------------------------------------------------------------------ + # scanning + storing + # ------------------------------------------------------------------ + def _connect(self, db, prepare): + conn = db.connect() + try: + if prepare: + self.ds._prepare_connection(conn, db.name) + else: + conn.row_factory = None + except Exception: + self._close(db, conn) + raise + return conn + + @staticmethod + def _close(db, conn): + try: + conn.close() + finally: + try: + db._all_connections.remove(conn) + except ValueError: + pass + + def _scan_sync(self, state): + """Read schema_version + full schema in one read transaction.""" + db = state.db + fp = t_ns = None + if state.is_file: + t_ns = time.time_ns() + fp = fingerprint(db.path) + if fp[0] is None: + return {"state": state, "missing": True, "fp": fp, "t": t_ns} + conn = self._connect(db, prepare=True) + try: + conn.row_factory = sqlite3.Row + conn.execute("BEGIN") + try: + version = conn.execute("PRAGMA schema_version").fetchone()[0] + schema = collect_schema(conn, db.name) + finally: + conn.rollback() + finally: + self._close(db, conn) + return { + "state": state, + "fp": fp, + "t": t_ns, + "version": version, + "schema": schema, + } + + async def _scan_on_write_connection(self, state): + name = state.db.name + + def _scan(conn): + version = conn.execute("PRAGMA schema_version").fetchone()[0] + # The write thread's post-task schema check must not re-notify + # for the version this scan is about to store + state.notified_version = version + return version, collect_schema(conn, name) + + try: + version, schema = await state.db._execute_on_write_connection(_scan) + except Exception as e: # noqa: BLE001 + return {"state": state, "error": e} + return { + "state": state, + "fp": None, + "t": None, + "version": version, + "schema": schema, + } + + def _scan_chunk_sync(self, states): + results = [] + for state in states: + try: + results.append(self._scan_sync(state)) + except Exception as e: # noqa: BLE001 + results.append({"state": state, "error": e}) + return results + + async def refresh(self, states): + """Rebuild the catalog for every state that needs it, coalescing with + scans already in flight for the same database.""" + loop = asyncio.get_running_loop() + pending = list(states) + for _ in range(10): + mine = [] + waits = [] + for state in pending: + fut = state.scan_future + if fut is not None and not fut.done(): + if fut.get_loop() is loop: + waits.append(fut) + continue + # Scan started on an event loop that has gone away + state.scan_future = None + state.needs_scan = True + if state.needs_scan and not state.removed: + state.needs_scan = False + state.scan_future = loop.create_future() + mine.append(state) + if mine: + try: + await self._scan_and_store(mine) + finally: + for state in mine: + fut = state.scan_future + if fut is not None and not fut.done(): + fut.set_result(None) + state.scan_future = None + if waits: + await asyncio.wait(waits) + pending = [ + s + for s in pending + if not s.removed + and ( + s.needs_scan + or (s.scan_future is not None and not s.scan_future.done()) + ) + and s.error is None + ] + if not pending: + break + for state in states: + # Failed scans are retried by sweeps, not by every request + if not state.needs_scan or state.error is not None: + self._pending.discard(state) + + async def _scan_and_store(self, states): + memory = [s for s in states if not s.is_file and s.db.memory_name] + if memory: + results = [] + for state in memory: + results.append(await self._scan_on_write_connection(state)) + await self._store(results) + states = [s for s in states if s not in memory] + if not states: + return + chunks = [states[i : i + SCAN_CHUNK] for i in range(0, len(states), SCAN_CHUNK)] + semaphore = asyncio.Semaphore(SCAN_CONCURRENCY) + + async def one(chunk): + async with semaphore: + results = await asyncio.to_thread(self._scan_chunk_sync, chunk) + await self._store(results) + + if len(chunks) == 1: + await one(chunks[0]) + else: + await asyncio.gather(*(one(c) for c in chunks)) + + async def _store(self, results): + entries = [] + missing = [] + stored = [] + for r in results: + state = r["state"] + if state.removed: + continue + if "error" in r: + state.error = r["error"] + state.needs_scan = True + logger.warning( + "Could not read schema of database %r: %s", state.name, r["error"] + ) + continue + state.error = None + if r.get("missing"): + missing.append(state) + state.fp = r["fp"] + state.fp_racy = True + continue + db = state.db + fp_json = _fp_to_json(r["fp"], r["t"]) if r["fp"] is not None else None + entries.append( + ( + state.name, + str(db.path) if db.path is not None else None, + db.is_memory, + r["version"], + fp_json, + r["schema"], + ) + ) + stored.append(r) + if entries: + watcher = self + + def _write(conn): + # Skip databases removed while we were scanning + live = [e for e in entries if e[0] in watcher.states] + write_catalog_entries(conn, live) + + await self.ds.get_internal_database().execute_write_fn(_write) + for r in stored: + state = r["state"] + state.catalog_version = r["version"] + state.missing = False + if r["fp"] is not None: + state.fp = r["fp"] + state.fp_racy = is_racy(r["fp"], r["t"]) + state.stats["scans"] += 1 + self.counters["scans"] += 1 + if missing: + for state in missing: + if not state.missing: + state.missing = True + state.db._invalidate_connections() + await self._clear_missing(missing) + + async def _clear_missing(self, states): + names = [s.name for s in states] + + def _clear(conn): + for name in names: + for table in CATALOG_TABLES[:-1]: + conn.execute(f"DELETE FROM {table} WHERE database_name = ?", [name]) + conn.execute( + "UPDATE catalog_databases SET schema_version = NULL, fingerprint = NULL " + "WHERE database_name = ?", + [name], + ) + + await self.ds.get_internal_database().execute_write_fn(_clear) + for state in states: + state.catalog_version = None + + async def _delete_catalog(self, names): + watcher = self + + def _delete(conn): + for name in names: + if name in watcher.states: + # Re-added under the same name; its own scan replaces rows + continue + for table in CATALOG_TABLES: + conn.execute(f"DELETE FROM {table} WHERE database_name = ?", [name]) + + await self.ds.get_internal_database().execute_write_fn(_delete) + for name in names: + self._pending_removals.discard(name) + + # ------------------------------------------------------------------ + # startup + # ------------------------------------------------------------------ + async def initial_scan(self): + """Full catalog build at startup. + + Deletes catalog rows for databases that are not attached (a + persistent --internal database can carry rows from a previous run), + then scans every database - except those whose persisted fingerprint + proves the file is unchanged since the catalog was written. + """ + internal = self.ds.get_internal_database() + names = list(self.states.keys()) + names_json = json.dumps(names) + + def _delete_stale(conn): + for table in CATALOG_TABLES: + conn.execute( + f"DELETE FROM {table} WHERE database_name IS NULL OR database_name NOT IN " + "(SELECT value FROM json_each(?))", + [names_json], + ) + + await internal.execute_write_fn(_delete_stale) + self._pending_removals.clear() + states = [s for s in self.states.values() if not s.removed] + persisted = {} + if not internal.is_temp_disk: + rows = await internal.execute( + "select database_name, path, schema_version, fingerprint from catalog_databases" + ) + persisted = {row["database_name"]: row for row in rows.rows} + to_scan = [] + if persisted: + reuse = await asyncio.to_thread(self._reusable_sync, states, persisted) + for state in states: + hit = reuse.get(state.name) + if hit is None: + to_scan.append(state) + continue + fp, t_ns, version = hit + state.fp = fp + state.fp_racy = is_racy(fp, t_ns) + state.catalog_version = version + state.needs_scan = False + self.counters["restored_from_persisted"] += 1 + else: + to_scan = states + for state in to_scan: + state.needs_scan = True + if to_scan: + await self.refresh(to_scan) + self._pending.clear() + + def _reusable_sync(self, states, persisted): + """Databases whose stored fingerprint matches the file right now and + was not racy when it was stored. Stat calls only - no connections.""" + out = {} + for state in states: + row = persisted.get(state.name) + if row is None or not state.is_file or row["fingerprint"] is None: + continue + if row["path"] != str(state.db.path) or row["schema_version"] is None: + continue + stored_fp, stored_t = _fp_from_json(row["fingerprint"]) + if stored_fp is None or is_racy(stored_fp, stored_t): + continue + t_ns = time.time_ns() + fp = fingerprint(state.db.path) + if fp == stored_fp: + out[state.name] = (fp, t_ns, row["schema_version"]) + return out + + # ------------------------------------------------------------------ + # introspection + # ------------------------------------------------------------------ + def status(self): + modes = {} + for state in self.states.values(): + modes[state.mode] = modes.get(state.mode, 0) + 1 + return { + "interval_s": self.interval_s, + "loop_running": self.loop_running(), + "modes": modes, + "pending": len(self._pending), + "pending_removals": len(self._pending_removals), + **self.counters, + } diff --git a/datasette/utils/internal_db.py b/datasette/utils/internal_db.py index 0ddeb847..3f6406ad 100644 --- a/datasette/utils/internal_db.py +++ b/datasette/utils/internal_db.py @@ -180,87 +180,100 @@ async def init_internal_db(db): await db.execute_write_fn(apply_migrations, transaction=False) -async def populate_schema_tables(internal_db, db, schema_version): - database_name = db.name +# Children first, catalog_databases last, so deletes work with FK enforcement +CATALOG_TABLES = ( + "catalog_columns", + "catalog_foreign_keys", + "catalog_indexes", + "catalog_views", + "catalog_tables", + "catalog_databases", +) - tables = (await db.execute("select * from sqlite_master WHERE type = 'table'")).rows - views = (await db.execute("select * from sqlite_master WHERE type = 'view'")).rows - def collect_info(conn): - tables_to_insert = [] - views_to_insert = [] - columns_to_insert = [] - foreign_keys_to_insert = [] - indexes_to_insert = [] +@internal_migrations(name="0002_catalog_fingerprint") +def catalog_fingerprint_column(db): + # stat() fingerprint of the database file when the catalog rows were + # written, so a restart with a persistent internal database can skip + # unchanged files without opening them + if "fingerprint" not in db["catalog_databases"].columns_dict: + db.execute("ALTER TABLE catalog_databases ADD COLUMN fingerprint TEXT") - for view in views: - view_name = view["name"] - views_to_insert.append( - (database_name, view_name, view["rootpage"], view["sql"]) - ) - for table in tables: - table_name = table["name"] - tables_to_insert.append( - (database_name, table_name, table["rootpage"], table["sql"]) - ) - columns = table_column_details(conn, table_name) - columns_to_insert.extend( - { - "database_name": database_name, - "table_name": table_name, - **column._asdict(), - } - for column in columns - ) - foreign_keys = conn.execute( - f"PRAGMA foreign_key_list({escape_sqlite(table_name)})" - ).fetchall() - foreign_keys_to_insert.extend( - { - "database_name": database_name, - "table_name": table_name, - **dict(foreign_key), - } - for foreign_key in foreign_keys - ) - indexes = conn.execute( - f"PRAGMA index_list({escape_sqlite(table_name)})" - ).fetchall() - indexes_to_insert.extend( - { - "database_name": database_name, - "table_name": table_name, - **dict(index), - } - for index in indexes - ) - return ( - tables_to_insert, - views_to_insert, - columns_to_insert, - foreign_keys_to_insert, - indexes_to_insert, +def collect_schema(conn, database_name): + """Read everything the catalog needs from one connection. + + Run it inside a read transaction so it sees a single snapshot. + """ + tables = conn.execute("select * from sqlite_master WHERE type = 'table'").fetchall() + views = conn.execute("select * from sqlite_master WHERE type = 'view'").fetchall() + tables_to_insert = [] + views_to_insert = [] + columns_to_insert = [] + foreign_keys_to_insert = [] + indexes_to_insert = [] + + for view in views: + views_to_insert.append( + (database_name, view["name"], view["rootpage"], view["sql"]) ) - ( - tables_to_insert, - views_to_insert, - columns_to_insert, - foreign_keys_to_insert, - indexes_to_insert, - ) = await db.execute_fn(collect_info) + for table in tables: + table_name = table["name"] + tables_to_insert.append( + (database_name, table_name, table["rootpage"], table["sql"]) + ) + columns = table_column_details(conn, table_name) + columns_to_insert.extend( + { + "database_name": database_name, + "table_name": table_name, + **column._asdict(), + } + for column in columns + ) + foreign_keys = conn.execute( + f"PRAGMA foreign_key_list({escape_sqlite(table_name)})" + ).fetchall() + foreign_keys_to_insert.extend( + { + "database_name": database_name, + "table_name": table_name, + **dict(foreign_key), + } + for foreign_key in foreign_keys + ) + indexes = conn.execute( + f"PRAGMA index_list({escape_sqlite(table_name)})" + ).fetchall() + indexes_to_insert.extend( + { + "database_name": database_name, + "table_name": table_name, + **dict(index), + } + for index in indexes + ) + return { + "tables": tables_to_insert, + "views": views_to_insert, + "columns": columns_to_insert, + "foreign_keys": foreign_keys_to_insert, + "indexes": indexes_to_insert, + } - def replace_catalog(conn): + +def write_catalog_entries(conn, entries): + """Replace the catalog rows for each database in ``entries``. + + Each entry is (database_name, path, is_memory, schema_version, + fingerprint_json, schema) where schema comes from collect_schema(). + Runs on the internal database write connection, in one transaction. + """ + for database_name, path, is_memory, schema_version, fingerprint, schema in entries: # Delete child rows before their catalog_tables parents so this also # works if a prepare_connection plugin enables foreign key enforcement. - for table in ( - "catalog_columns", - "catalog_foreign_keys", - "catalog_indexes", - "catalog_views", - "catalog_tables", - ): + for table in CATALOG_TABLES[:-1]: conn.execute( f"DELETE FROM {table} WHERE database_name = ?", [database_name], @@ -268,29 +281,24 @@ async def populate_schema_tables(internal_db, db, schema_version): conn.execute( """ INSERT OR REPLACE INTO catalog_databases ( - database_name, path, is_memory, schema_version - ) VALUES (?, ?, ?, ?) + database_name, path, is_memory, schema_version, fingerprint + ) VALUES (?, ?, ?, ?, ?) """, - [ - database_name, - str(db.path) if db.path is not None else None, - db.is_memory, - schema_version, - ], + [database_name, path, is_memory, schema_version, fingerprint], ) conn.executemany( """ INSERT INTO catalog_tables (database_name, table_name, rootpage, sql) values (?, ?, ?, ?) """, - tables_to_insert, + schema["tables"], ) conn.executemany( """ INSERT INTO catalog_views (database_name, view_name, rootpage, sql) values (?, ?, ?, ?) """, - views_to_insert, + schema["views"], ) conn.executemany( """ @@ -300,7 +308,7 @@ async def populate_schema_tables(internal_db, db, schema_version): :database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden ) """, - columns_to_insert, + schema["columns"], ) conn.executemany( """ @@ -310,7 +318,7 @@ async def populate_schema_tables(internal_db, db, schema_version): :database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match ) """, - foreign_keys_to_insert, + schema["foreign_keys"], ) conn.executemany( """ @@ -320,7 +328,29 @@ async def populate_schema_tables(internal_db, db, schema_version): :database_name, :table_name, :seq, :name, :unique, :origin, :partial ) """, - indexes_to_insert, + schema["indexes"], ) + +async def populate_schema_tables(internal_db, db, schema_version): + """Rebuild the catalog rows for one database (kept for compatibility - + the SchemaWatcher reads the schema on its own short-lived connection).""" + database_name = db.name + + def _collect(conn): + return collect_schema(conn, database_name) + + schema = await db.execute_fn(_collect) + entry = ( + database_name, + str(db.path) if db.path is not None else None, + db.is_memory, + schema_version, + None, + schema, + ) + + def replace_catalog(conn): + write_catalog_entries(conn, [entry]) + await internal_db.execute_write_fn(replace_catalog) diff --git a/datasette/views/base.py b/datasette/views/base.py index 7262edb4..75a152d9 100644 --- a/datasette/views/base.py +++ b/datasette/views/base.py @@ -116,7 +116,7 @@ class BaseView: async def dispatch_request(self, request): if self.ds: - await self.ds.refresh_schemas() + await self.ds._schema_watcher.on_request() handler = getattr(self, request.method.lower(), None) response = await handler(request) if self.ds.cors: diff --git a/datasette/views/database.py b/datasette/views/database.py index da207601..14674c93 100644 --- a/datasette/views/database.py +++ b/datasette/views/database.py @@ -80,7 +80,7 @@ class DatabaseView(View): async def get(self, request, datasette): format_ = request.url_vars.get("format") or "html" - await datasette.refresh_schemas() + await datasette._schema_watcher.on_request() db = await datasette.resolve_database(request) database = db.name @@ -711,7 +711,7 @@ class QueryView(View): async def get(self, request, datasette): from datasette.app import TableNotFound - await datasette.refresh_schemas() + await datasette._schema_watcher.on_request() db = await datasette.resolve_database(request) database = db.name diff --git a/datasette/views/special.py b/datasette/views/special.py index 72e80316..421dc523 100644 --- a/datasette/views/special.py +++ b/datasette/views/special.py @@ -312,7 +312,7 @@ class AllowedResourcesView(BaseView): async def get(self, request): await self.ds.ensure_permission(action="view-instance", actor=request.actor) - await self.ds.refresh_schemas() + await self.ds._schema_watcher.on_request() # Check if user has permissions-debug (to show sensitive fields) has_debug_permission = await self.ds.allowed( @@ -530,7 +530,7 @@ class PermissionRulesView(BaseView): union_sql, union_params, _restriction_sqls = await build_permission_rules_sql( self.ds, actor, action ) - await self.ds.refresh_schemas() + await self.ds._schema_watcher.on_request() db = self.ds.get_internal_database() count_query = f""" diff --git a/datasette/views/table.py b/datasette/views/table.py index 0a908c69..2f412e9a 100644 --- a/datasette/views/table.py +++ b/datasette/views/table.py @@ -1738,7 +1738,7 @@ async def _sort_order(table_metadata, sortable_columns, request, order_by): async def table_view(datasette, request): - await datasette.refresh_schemas() + await datasette._schema_watcher.on_request() with tracer.trace_child_tasks(): response = await table_view_traced(datasette, request) diff --git a/docs/settings.rst b/docs/settings.rst index f3e6636d..0141475b 100644 --- a/docs/settings.rst +++ b/docs/settings.rst @@ -563,6 +563,33 @@ You can do that like so:: datasette mydatabase.db --setting base_url /tools/datasette/ +.. _setting_schema_watch_interval_ms: + +schema_watch_interval_ms +~~~~~~~~~~~~~~~~~~~~~~~~ + +.. [[[cog + setting_default(cog, "schema_watch_interval_ms") +.. ]]] + +Default: ``1000`` + +.. [[[end]]] + +How often, in milliseconds, Datasette checks database files that other processes may change for schema changes, so that the ``_internal`` catalog tables stay current. Each check is a ``stat()`` of the database file and its ``-wal`` and ``-journal`` files; a database is only opened if those changed. + +Databases passed on the command line are checked this way by default. Databases added by plugins using ``datasette.add_database()`` are ``owned``: Datasette detects their schema changes when it writes to them and never polls them. Immutable databases are scanned once at startup. You can override the mode for a database in ``datasette.yaml``: + +.. code-block:: yaml + + databases: + mydatabase: + schema_watch: owned # or external, or immutable + +Set this to ``0`` to turn polling off:: + + datasette mydatabase.db --setting schema_watch_interval_ms 0 + .. _setting_secret: Configuring the secret diff --git a/tests/test_api.py b/tests/test_api.py index 690cb080..8b9e82aa 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -767,6 +767,7 @@ async def test_settings_json(ds_client): "template_debug": False, "trace_debug": False, "base_url": "/", + "schema_watch_interval_ms": 1000, } diff --git a/tests/test_internal_db.py b/tests/test_internal_db.py index b4bb964d..76ccd148 100644 --- a/tests/test_internal_db.py +++ b/tests/test_internal_db.py @@ -29,7 +29,8 @@ async def test_internal_migrations_recorded(ds_client): order by id """) assert [tuple(row) for row in migrations.rows] == [ - ("datasette_internal", "0001_initial") + ("datasette_internal", "0001_initial"), + ("datasette_internal", "0002_catalog_fingerprint"), ] @@ -61,7 +62,8 @@ async def test_internal_migrations_adopt_existing_internal_db(tmp_path): order by id """) assert [tuple(row) for row in migrations.rows] == [ - ("datasette_internal", "0001_initial") + ("datasette_internal", "0001_initial"), + ("datasette_internal", "0002_catalog_fingerprint"), ] ds.close() diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py new file mode 100644 index 00000000..17f57a87 --- /dev/null +++ b/tests/test_schema_watcher.py @@ -0,0 +1,516 @@ +""" +SchemaWatcher tests. Changes to "external" databases are made from a +separate OS process (a child Python interpreter using plain sqlite3, or the +sqlite-utils CLI), the way other tools change files Datasette is serving. +""" + +import asyncio +import os +import sqlite3 +import subprocess +import sys +import textwrap +import time + +import pytest + +from datasette import schema_watcher as schema_watcher_module +from datasette.app import Datasette +from datasette.database import Database + +INTERVAL_MS = 50 + + +def run_external(path, sql, journal_mode=None, keep_wal=False): + """Run sql against path in a separate Python process; returns the + child's time.monotonic() right after COMMIT.""" + script = textwrap.dedent(""" + import sqlite3, sys, time + path, sql, journal_mode, keep_wal = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4] + conn = sqlite3.connect(path, isolation_level=None, timeout=10) + if journal_mode: + conn.execute("PRAGMA journal_mode=" + journal_mode) + if keep_wal == "1": + # Never checkpoint: changes stay in the -wal file + conn.execute("PRAGMA wal_autocheckpoint=0") + conn.executescript(sql) + print(time.monotonic(), flush=True) + if keep_wal == "1": + # Keep the connection open so close() does not checkpoint + sys.stdin.read() + conn.close() + """) + if keep_wal: + proc = subprocess.Popen( + [sys.executable, "-c", script, path, sql, journal_mode or "", "1"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + ) + line = proc.stdout.readline() + return float(line), proc + out = subprocess.run( + [sys.executable, "-c", script, path, sql, journal_mode or "", "0"], + check=True, + capture_output=True, + text=True, + ) + return float(out.stdout.strip()) + + +def run_process(*args): + subprocess.run([sys.executable, *args], check=True) + + +def make_db(path, sql="create table t (id integer primary key, a text)", wal=False): + conn = sqlite3.connect(path) + if wal: + conn.execute("PRAGMA journal_mode=wal") + conn.executescript(sql) + conn.execute("insert into t (a) values ('one')") + conn.commit() + conn.close() + + +async def catalog_columns(ds, db_name, table): + rows = await ds.get_internal_database().execute( + "select name from catalog_columns where database_name = ? and table_name = ? order by cid", + [db_name, table], + ) + return [r[0] for r in rows.rows] + + +async def catalog_tables(ds, db_name): + rows = await ds.get_internal_database().execute( + "select table_name from catalog_tables where database_name = ? order by table_name", + [db_name], + ) + return [r[0] for r in rows.rows] + + +async def wait_for(check, timeout=5.0): + t0 = time.monotonic() + while True: + result = await check() + if result: + return time.monotonic() - t0 + if time.monotonic() - t0 > timeout: + raise AssertionError(f"timed out after {timeout}s") + await asyncio.sleep(0.01) + + +@pytest.fixture +def db_path(tmp_path): + path = str(tmp_path / "ext.db") + make_db(path) + return path + + +async def start_ds(paths, interval_ms=INTERVAL_MS, settings=None, **kwargs): + ds = Datasette( + files=paths, + settings={"schema_watch_interval_ms": interval_ms, **(settings or {})}, + **kwargs, + ) + await ds.invoke_startup() + await ds.start_background_tasks() + return ds + + +@pytest.mark.asyncio +async def test_modes_and_defaults(tmp_path, db_path): + immutable_path = str(tmp_path / "imm.db") + make_db(immutable_path) + ds = Datasette(files=[db_path], immutables=[immutable_path]) + await ds.invoke_startup() + watcher = ds._schema_watcher + assert watcher.states["ext"].mode == "external" + assert watcher.states["imm"].mode == "immutable" + added = ds.add_database(Database(ds, memory_name="sw_owned_mem"), name="added") + assert watcher.states["added"].mode == "owned" + assert added._watch_state is watcher.states["added"] + ds.close() + + +@pytest.mark.asyncio +async def test_mode_from_config(tmp_path, db_path): + ds = Datasette( + files=[db_path], config={"databases": {"ext": {"schema_watch": "owned"}}} + ) + assert ds._schema_watcher.states["ext"].mode == "owned" + ds.close() + + +@pytest.mark.asyncio +async def test_no_request_path_sweep(db_path): + ds = await start_ds([db_path], interval_ms=0) + watcher = ds._schema_watcher + sweeps = watcher.counters["sweeps"] + for _ in range(5): + response = await ds.client.get("/ext/t.json") + assert response.status_code == 200 + assert watcher.counters["sweeps"] == sweeps + ds.close() + + +@pytest.mark.asyncio +async def test_owned_write_path_detects_schema_change(tmp_path): + ds = Datasette(settings={"schema_watch_interval_ms": 0}) + await ds.invoke_startup() + db = ds.add_database(Database(ds, memory_name="sw_owned"), name="owned") + await ds.refresh_schemas(force=True) + watcher = ds._schema_watcher + scans = watcher.counters["scans"] + # Data-only writes: no catalog rebuild + await db.execute_write("create table if not exists t (id integer primary key)") + assert await catalog_tables(ds, "owned") == ["t"] + scans = watcher.counters["scans"] + for i in range(5): + await db.execute_write("insert into t default values") + assert watcher.counters["scans"] == scans + # Each write API: catalog is current as soon as the call returns + await db.execute_write("create table t2 (id integer primary key)") + assert "t2" in await catalog_tables(ds, "owned") + await db.execute_write_script("create table t3 (id integer); create table t4 (id);") + assert {"t3", "t4"} <= set(await catalog_tables(ds, "owned")) + scans = watcher.counters["scans"] + await db.execute_write_many("insert into t2 (id) values (?)", [[1], [2]]) + assert watcher.counters["scans"] == scans + + def plugin_write(conn): + conn.execute("alter table t add column extra text") + + await db.execute_write_fn(plugin_write) + assert await catalog_columns(ds, "owned", "t") == ["id", "extra"] + + def isolated(conn): + conn.execute("create table t6 (id)") + + await db.execute_isolated_fn(isolated) + assert "t6" in await catalog_tables(ds, "owned") + # block=False: refreshed in the background + await db.execute_write("create table t7 (id)", block=False) + await wait_for(lambda: _contains(ds, "owned", "t7")) + ds.close() + + +async def _contains(ds, db_name, table): + return table in await catalog_tables(ds, db_name) + + +@pytest.mark.asyncio +async def test_owned_write_path_non_threaded(tmp_path): + path = str(tmp_path / "nt.db") + make_db(path) + ds = Datasette(settings={"num_sql_threads": 0, "schema_watch_interval_ms": 0}) + await ds.invoke_startup() + db = ds.add_database(Database(ds, path=path), name="nt") + await ds._schema_watcher.flush_pending() + await db.execute_write("create table t2 (id)") + assert await catalog_tables(ds, "nt") == ["t", "t2"] + ds.close() + + +@pytest.mark.asyncio +async def test_external_data_only_insert_does_not_rescan(db_path): + ds = await start_ds([db_path]) + watcher = ds._schema_watcher + state = watcher.states["ext"] + scans = state.stats["scans"] + checks = state.stats["checks"] + for i in range(3): + run_external(db_path, f"insert into t (a) values ('x{i}')") + await asyncio.sleep(INTERVAL_MS * 3 / 1000) + # The fingerprint changed, so schema_version was checked... + assert state.stats["checks"] > checks + # ... but the catalog was not rebuilt + assert state.stats["scans"] == scans + ds.close() + + +@pytest.mark.asyncio +async def test_external_add_column(db_path): + ds = await start_ds([db_path]) + run_external(db_path, "alter table t add column b text") + + async def has_b(): + return await catalog_columns(ds, "ext", "t") == ["id", "a", "b"] + + await wait_for(has_b) + ds.close() + + +@pytest.mark.asyncio +async def test_external_sqlite_utils_transform(db_path): + ds = await start_ds([db_path]) + run_process( + "-m", "sqlite_utils", "transform", db_path, "t", "--rename", "a", "renamed" + ) + + async def renamed(): + return await catalog_columns(ds, "ext", "t") == ["id", "renamed"] + + await wait_for(renamed) + # And the table page uses the new column too + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "renamed": "one"}] + ds.close() + + +@pytest.mark.asyncio +async def test_external_create_table_in_wal_before_checkpoint(tmp_path): + path = str(tmp_path / "walled.db") + make_db(path, wal=True) + ds = await start_ds([path]) + _, proc = run_external( + path, "create table in_wal (id integer)", journal_mode="wal", keep_wal=True + ) + try: + assert os.path.getsize(path + "-wal") > 0 + + async def seen(): + return "in_wal" in await catalog_tables(ds, "walled") + + await wait_for(seen) + # Main file still does not contain it: it was only in the WAL + assert os.path.exists(path + "-wal") + finally: + proc.stdin.close() + proc.wait() + ds.close() + + +@pytest.mark.asyncio +async def test_external_atomic_replace_by_rename(tmp_path, db_path): + ds = await start_ds([db_path]) + # Warm a cached read connection on the old inode + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "a": "one"}] + replacement = str(tmp_path / "replacement.db") + conn = sqlite3.connect(replacement) + # Same schema_version as the original file (1 CREATE TABLE) but a + # different table: only the inode change gives it away + conn.execute("create table t (id integer primary key, different text)") + conn.execute("insert into t (different) values ('new file')") + conn.commit() + assert conn.execute("pragma schema_version").fetchone()[0] == 1 + conn.close() + old_inode = os.stat(db_path).st_ino + run_process( + "-c", + "import os, sys; os.replace(sys.argv[1], sys.argv[2])", + replacement, + db_path, + ) + assert os.stat(db_path).st_ino != old_inode + + async def replaced(): + return await catalog_columns(ds, "ext", "t") == ["id", "different"] + + await wait_for(replaced) + # Cached connections were invalidated, so queries see the new file + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "different": "new file"}] + ds.close() + + +@pytest.mark.asyncio +async def test_old_inode_connections_without_watcher(tmp_path, db_path): + # Documents the hazard the watcher fixes: with polling off, a cached + # read connection keeps reading the replaced (unlinked) file + # One SQL thread, so every request reuses the same cached connection + ds = await start_ds([db_path], interval_ms=0, settings={"num_sql_threads": 1}) + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "a": "one"}] + replacement = str(tmp_path / "replacement.db") + conn = sqlite3.connect(replacement) + conn.execute("create table t (id integer primary key, a text)") + conn.execute("insert into t (a) values ('new file')") + conn.commit() + conn.close() + os.replace(replacement, db_path) + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "a": "one"}] # stale! + # An explicit refresh notices the inode change and fixes it + await ds.refresh_schemas(force=True) + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "a": "new file"}] + ds.close() + + +@pytest.mark.asyncio +async def test_external_vacuum(db_path): + run_external( + db_path, + "create table big (x); insert into big select zeroblob(1000) from " + "(with recursive c(i) as (select 1 union all select i+1 from c where i < 200) select i from c);" + "drop table big;", + ) + ds = await start_ds([db_path]) + before = await ds.get_internal_database().execute( + "select schema_version from catalog_databases where database_name = 'ext'" + ) + run_external(db_path, "vacuum") + + async def bumped(): + after = await ds.get_internal_database().execute( + "select schema_version from catalog_databases where database_name = 'ext'" + ) + return after.first()[0] == before.first()[0] + 1 + + await wait_for(bumped) + ds.close() + + +@pytest.mark.asyncio +async def test_external_rapid_changes(db_path): + ds = await start_ds([db_path]) + for i in range(10): + run_external(db_path, f"create table rapid_{i} (id)") + + async def all_seen(): + tables = await catalog_tables(ds, "ext") + return all(f"rapid_{i}" in tables for i in range(10)) + + await wait_for(all_seen) + ds.close() + + +def _coarse_fingerprint(real): + # Simulate a filesystem with 1 second timestamp granularity + def fingerprint(path, present=None): + parts = real(path, present) + out = [] + for i, part in enumerate(parts): + if part is None: + out.append(None) + continue + part = list(part) + # main: (dev, ino, size, mtime, ctime); side files: (ino, size, mtime) + for idx in (3, 4) if i == 0 else (2,): + part[idx] = part[idx] // 10**9 * 10**9 + out.append(tuple(part)) + return tuple(out) + + return fingerprint + + +@pytest.mark.asyncio +@pytest.mark.parametrize("racy_rule", (True, False)) +async def test_racily_clean_same_tick_change(db_path, monkeypatch, racy_rule): + """Two schema changes inside one timestamp tick, same file size and + inode: only the racy-clean rule catches the second one.""" + monkeypatch.setattr( + schema_watcher_module, + "fingerprint", + _coarse_fingerprint(schema_watcher_module.fingerprint), + ) + if not racy_rule: + monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", -(10**18)) + ds = Datasette(files=[db_path], settings={"schema_watch_interval_ms": 0}) + await ds.invoke_startup() + # Line the first change up with the start of a fresh second + await asyncio.sleep(1 - (time.time() % 1) + 0.05) + size = os.path.getsize(db_path) + run_external(db_path, "alter table t add column b text") + await ds._refresh_schemas() # sweep: sees change 1 + assert await catalog_columns(ds, "ext", "t") == ["id", "a", "b"] + run_external(db_path, "alter table t add column c text") + assert os.path.getsize(db_path) == size + assert time.time() % 1 < 0.9, "test machine too slow to stay in one tick" + await ds._refresh_schemas() # sweep: fingerprint identical to last one + columns = await catalog_columns(ds, "ext", "t") + if racy_rule: + assert columns == ["id", "a", "b", "c"] + else: + # Without the rule the change is missed + assert columns == ["id", "a", "b"] + ds.close() + + +@pytest.mark.asyncio +async def test_external_file_deleted_and_recreated(tmp_path, db_path): + ds = await start_ds([db_path]) + assert await catalog_tables(ds, "ext") == ["t"] + run_process("-c", "import os, sys; os.unlink(sys.argv[1])", db_path) + + async def gone(): + return await catalog_tables(ds, "ext") == [] + + await wait_for(gone) + response = await ds.client.get("/ext/t.json") + assert response.status_code in (404, 500) + make_db(db_path, sql="create table t (id integer primary key, a text, back text)") + + async def back(): + return await catalog_columns(ds, "ext", "t") == ["id", "a", "back"] + + await wait_for(back) + response = await ds.client.get("/ext/t.json?_shape=array") + assert response.json() == [{"id": 1, "a": "one", "back": None}] + ds.close() + + +@pytest.mark.asyncio +async def test_remove_database_deletes_catalog_rows(tmp_path, db_path): + other = str(tmp_path / "other.db") + make_db(other) + ds = Datasette(files=[db_path, other]) + await ds.invoke_startup() + assert await catalog_tables(ds, "other") == ["t"] + ds.remove_database("other") + await ds._schema_watcher.on_request() + internal = ds.get_internal_database() + for table in ( + "catalog_databases", + "catalog_tables", + "catalog_columns", + "catalog_indexes", + ): + rows = await internal.execute( + f"select count(*) from {table} where database_name = 'other'" + ) + assert rows.first()[0] == 0 + assert await catalog_tables(ds, "ext") == ["t"] + ds.close() + + +@pytest.mark.asyncio +async def test_add_database_visible_on_next_request(tmp_path): + ds = Datasette(settings={"schema_watch_interval_ms": 0}) + await ds.invoke_startup() + await ds.client.get("/-/versions.json") + path = str(tmp_path / "late.db") + make_db(path) + ds.add_database(Database(ds, path=path), name="late") + response = await ds.client.get("/late.json") + assert response.status_code == 200 + assert await catalog_tables(ds, "late") == ["t"] + ds.close() + + +@pytest.mark.asyncio +async def test_persisted_fingerprints_skip_unchanged_files(tmp_path, monkeypatch): + # Shrink the racy window so freshly written files count as stable + monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", 1_000_000) + internal = str(tmp_path / "internal.db") + paths = [] + for i in range(3): + path = str(tmp_path / f"p{i}.db") + make_db(path) + paths.append(path) + await asyncio.sleep(0.05) + ds = Datasette(files=paths, internal=internal) + await ds.invoke_startup() + assert ds._schema_watcher.counters["scans"] == 3 + ds.close() + # Change one file while Datasette is "down" + run_external(paths[1], "create table added_offline (id)") + ds2 = Datasette(files=paths, internal=internal) + await ds2.invoke_startup() + counters = ds2._schema_watcher.counters + assert counters["restored_from_persisted"] == 2 + assert counters["scans"] == 1 + assert await catalog_tables(ds2, "p1") == ["added_offline", "t"] + assert await catalog_tables(ds2, "p0") == ["t"] + ds2.close() From 4492d3ee3d93c0d39303e5be7b61c3f604238ad0 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 03:25:15 +0000 Subject: [PATCH 2/5] Schema watcher: faster catalog scans, stable WAL fingerprints - collect_schema uses pragma_table_xinfo/foreign_key_list/index_list table-valued functions joined to sqlite_master: 3 queries per database instead of 3 per table (falls back to the per-table loop on error) - bulk scans run one chunk at a time (thread concurrency only added GIL contention); internal-DB writes of chunk k overlap the scan of k+1 - an empty -wal/-journal counts as absent (read-only opens of WAL databases create an empty -wal) and side files ignore ctime (SQLite fchown()s them on every open when running as root) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/schema_watcher.py | 10 +++- datasette/utils/internal_db.py | 84 +++++++++++++++++++++++++++++++++- 2 files changed, 90 insertions(+), 4 deletions(-) diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index 860a6ad7..ea7be2ea 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -66,7 +66,7 @@ MODES = ("owned", "external", "immutable") RACY_WINDOW_NS = 2_000_000_000 # Databases per thread hop / per internal-DB transaction for bulk scans SCAN_CHUNK = 64 -SCAN_CONCURRENCY = 3 +SCAN_CONCURRENCY = 1 # Directories with at least this many watched files are listed once per # sweep with os.scandir() instead of probing -wal/-journal with stat() SCANDIR_THRESHOLD = 8 @@ -146,9 +146,15 @@ def fingerprint(path, present=None): parts.append(None) continue try: - parts.append(_fp_side(os.stat(path + suffix))) + st = os.stat(path + suffix) except FileNotFoundError: parts.append(None) + continue + # An empty -wal holds no frames (and an empty -journal nothing to + # roll back): the database is exactly the main file, same as when + # the file is absent. Read-only opens of a WAL database create an + # empty -wal, which must not look like a change. + parts.append(_fp_side(st) if st.st_size else None) return tuple(parts) diff --git a/datasette/utils/internal_db.py b/datasette/utils/internal_db.py index 3f6406ad..57f27314 100644 --- a/datasette/utils/internal_db.py +++ b/datasette/utils/internal_db.py @@ -3,7 +3,8 @@ import textwrap from sqlite_utils import Database as SQLiteUtilsDatabase from sqlite_utils import Migrations -from datasette.utils import escape_sqlite, table_column_details +from datasette.utils import escape_sqlite, sqlite3, table_column_details +from datasette.utils.sqlite import supports_table_xinfo INTERNAL_DB_SCHEMA_TABLES = { "catalog_databases", @@ -200,11 +201,90 @@ def catalog_fingerprint_column(db): db.execute("ALTER TABLE catalog_databases ADD COLUMN fingerprint TEXT") +_COLUMNS_SQL = """ +SELECT m.name, p.cid, p.name, p.type, p."notnull", p.dflt_value, p.pk, p.hidden +FROM sqlite_master m, pragma_table_xinfo(m.name) p WHERE m.type = 'table' +""" +_FOREIGN_KEYS_SQL = """ +SELECT m.name, p.id, p.seq, p."table", p."from", p."to", p.on_update, p.on_delete, p."match" +FROM sqlite_master m, pragma_foreign_key_list(m.name) p WHERE m.type = 'table' +""" +_INDEXES_SQL = """ +SELECT m.name, p.seq, p.name, p."unique", p.origin, p.partial +FROM sqlite_master m, pragma_index_list(m.name) p WHERE m.type = 'table' +""" + + def collect_schema(conn, database_name): """Read everything the catalog needs from one connection. - Run it inside a read transaction so it sees a single snapshot. + Run it inside a read transaction so it sees a single snapshot. Uses + three joined pragma table-valued-function queries per database instead + of three PRAGMA calls per table, falling back to the per-table loop if + that fails (e.g. a virtual table whose module is not loaded). """ + if supports_table_xinfo(): + try: + return _collect_schema_joined(conn, database_name) + except sqlite3.DatabaseError: + pass + return _collect_schema_per_table(conn, database_name) + + +def _collect_schema_joined(conn, database_name): + tables = conn.execute("select * from sqlite_master WHERE type = 'table'").fetchall() + views = conn.execute("select * from sqlite_master WHERE type = 'view'").fetchall() + columns = [ + { + "database_name": database_name, + "table_name": r[0], + "cid": r[1], + "name": r[2], + "type": r[3], + "notnull": r[4], + "default_value": r[5], + "is_pk": r[6], + "hidden": r[7], + } + for r in conn.execute(_COLUMNS_SQL).fetchall() + ] + foreign_keys = [ + { + "database_name": database_name, + "table_name": r[0], + "id": r[1], + "seq": r[2], + "table": r[3], + "from": r[4], + "to": r[5], + "on_update": r[6], + "on_delete": r[7], + "match": r[8], + } + for r in conn.execute(_FOREIGN_KEYS_SQL).fetchall() + ] + indexes = [ + { + "database_name": database_name, + "table_name": r[0], + "seq": r[1], + "name": r[2], + "unique": r[3], + "origin": r[4], + "partial": r[5], + } + for r in conn.execute(_INDEXES_SQL).fetchall() + ] + return { + "tables": [(database_name, t["name"], t["rootpage"], t["sql"]) for t in tables], + "views": [(database_name, v["name"], v["rootpage"], v["sql"]) for v in views], + "columns": columns, + "foreign_keys": foreign_keys, + "indexes": indexes, + } + + +def _collect_schema_per_table(conn, database_name): tables = conn.execute("select * from sqlite_master WHERE type = 'table'").fetchall() views = conn.execute("select * from sqlite_master WHERE type = 'view'").fetchall() tables_to_insert = [] From 3ec3b490df6757539c7e29ea794600a30349f626 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 03:51:24 +0000 Subject: [PATCH 3/5] Schema watcher: never recreate deleted files; O(1) derived-table cache key - write connections reopened after the watcher saw the file replaced or deleted use mode=rw, so a write to a deleted database fails instead of silently creating a new empty file - derived_table_dependencies() keys its cache on the schema_version the watcher stored in the catalog (plus its scan count) instead of running PRAGMA schema_version on a pooled read connection - allowed_resources() calls it for every database on every page Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/database.py | 22 +++++++++++++++++++++- tests/test_schema_watcher.py | 4 ++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/datasette/database.py b/datasette/database.py index 056e8109..32e1d19d 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -251,6 +251,11 @@ class Database: assert not (write and not self.is_mutable) if write: qs = "" + state = self._watch_state + if self._conn_generation or (state is not None and state.missing): + # Reopening after the file was replaced or deleted: never + # silently create a new empty database in its place + qs = "?mode=rw" if self.mode is not None: qs = f"?mode={self.mode}" conn = sqlite3.connect( @@ -1129,7 +1134,22 @@ class Database: async def derived_table_dependencies(self): """Return implementation tables and the tables they derive from.""" - schema_version = (await self.execute("PRAGMA schema_version")).first()[0] + state = self._watch_state + if ( + state is not None + and state.catalog_version is not None + and not state.needs_scan + and not state.missing + ): + # O(1): the schema_version the SchemaWatcher last stored in the + # catalog. allowed_resources() calls this for every database on + # every page, so a PRAGMA here was a per-request sweep of all + # databases (one pooled connection each). The cache is now as + # fresh as the catalog itself. The scan count is part of the key + # because a replaced file can have the same schema_version. + schema_version = ("catalog", state.catalog_version, state.stats["scans"]) + else: + schema_version = (await self.execute("PRAGMA schema_version")).first()[0] if ( self._cached_derived_table_dependencies is None or self._cached_derived_table_dependencies[0] != schema_version diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py index 17f57a87..2f26f79d 100644 --- a/tests/test_schema_watcher.py +++ b/tests/test_schema_watcher.py @@ -440,6 +440,10 @@ async def test_external_file_deleted_and_recreated(tmp_path, db_path): await wait_for(gone) response = await ds.client.get("/ext/t.json") assert response.status_code in (404, 500) + # Writes fail instead of silently creating a new empty file + with pytest.raises(sqlite3.OperationalError): + await ds.get_database("ext").execute_write("insert into t (a) values ('x')") + assert not os.path.exists(db_path) make_db(db_path, sql="create table t (id integer primary key, a text, back text)") async def back(): From a7359a7065111d1af2617684bf4f315fd0615b65 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 04:06:13 +0000 Subject: [PATCH 4/5] Schema watcher: lock-free header check when no -wal/-journal has content When the fingerprint shows no non-empty -wal or -journal the main file is the whole database, so the sweep reads the schema cookie (offset 40 of the header) with pread instead of opening a connection. No SHARED lock means no busy back-off for a concurrent writer: polling was costing write-heavy phases ~10% throughput and up to 2x p99. The fingerprint is still taken first, so a commit in flight is covered by the racy rule. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/schema_watcher.py | 47 ++++++++++++++++++++++++++++++++++-- tests/test_schema_watcher.py | 36 +++++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 4 deletions(-) diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index ea7be2ea..6fa5ed7f 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -71,6 +71,9 @@ SCAN_CONCURRENCY = 1 # sweep with os.scandir() instead of probing -wal/-journal with stat() SCANDIR_THRESHOLD = 8 PRAGMA_BUSY_TIMEOUT_MS = 200 +# Read the schema cookie from the file header instead of opening a +# connection when no -wal/-journal has content +USE_HEADER_CHECK = True class WatchState: @@ -171,6 +174,35 @@ def is_racy(fp, t_ns): return newest_timestamp(fp) >= t_ns - RACY_WINDOW_NS +SQLITE_HEADER = b"SQLite format 3\x00" + + +def header_schema_version(path): + """The schema cookie (offset 40 of the database header) read straight + from the file - what PRAGMA schema_version returns - without opening a + SQLite connection or taking any lock. + + Only meaningful when there is no non-empty -wal or -journal: then the + main file is the whole database. The caller must have taken the + fingerprint *before* calling this; a commit in flight at that moment has + already bumped the mtime it saw, so the fingerprint is racy and the next + sweep looks again. + """ + try: + fd = os.open(path, os.O_RDONLY) + except OSError: + return None + try: + header = os.pread(fd, 100, 0) + except OSError: + return None + finally: + os.close(fd) + if len(header) < 100 or not header.startswith(SQLITE_HEADER): + return None + return int.from_bytes(header[40:44], "big") + + def _fp_to_json(fp, t_ns): return json.dumps({"fp": fp, "t": t_ns}) @@ -198,6 +230,7 @@ class SchemaWatcher: "sweep_seconds": 0.0, "stats": 0, "pragma_checks": 0, + "header_checks": 0, "scans": 0, "write_detections": 0, "replaced": 0, @@ -476,8 +509,10 @@ class SchemaWatcher: elif kind == "busy": self.counters["busy"] += 1 state.fp_racy = True # check again next sweep - elif kind == "checked": - self.counters["pragma_checks"] += 1 + elif kind in ("checked", "header"): + self.counters[ + "pragma_checks" if kind == "checked" else "header_checks" + ] += 1 state.stats["checks"] += 1 if version != state.catalog_version: state.needs_scan = True @@ -525,6 +560,14 @@ class SchemaWatcher: # first sight: the schema_version alone cannot be trusted out.append((state, "replaced", fp, t_ns, None)) continue + if USE_HEADER_CHECK and fp[1] is None and fp[2] is None: + # No -wal/-journal content: read the cookie from the + # header. No connection, no SHARED lock that could make + # a concurrent writer back off. + version = header_schema_version(os.fspath(state.db.path)) + if version is not None: + out.append((state, "header", fp, t_ns, version)) + continue out.append(self._pragma_check(state, fp, t_ns)) return out diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py index 2f26f79d..2b876809 100644 --- a/tests/test_schema_watcher.py +++ b/tests/test_schema_watcher.py @@ -229,7 +229,9 @@ async def test_external_data_only_insert_does_not_rescan(db_path): @pytest.mark.asyncio -async def test_external_add_column(db_path): +@pytest.mark.parametrize("use_header", (True, False)) +async def test_external_add_column(db_path, monkeypatch, use_header): + monkeypatch.setattr(schema_watcher_module, "USE_HEADER_CHECK", use_header) ds = await start_ds([db_path]) run_external(db_path, "alter table t add column b text") @@ -395,11 +397,41 @@ def _coarse_fingerprint(real): return fingerprint +@pytest.mark.parametrize("journal_mode", ("delete", "wal", "truncate")) +def test_header_schema_version_matches_pragma(tmp_path, journal_mode): + path = str(tmp_path / "h.db") + conn = sqlite3.connect(path, isolation_level=None) + conn.execute(f"PRAGMA journal_mode={journal_mode}") + for i in range(300): + conn.execute(f"create table t{i} (id integer)") + conn.execute("alter table t1 add column b") + version = conn.execute("PRAGMA schema_version").fetchone()[0] + if journal_mode == "wal": + # Pages still in the -wal: the main file header is stale, which is + # why the watcher only reads it when the -wal is empty + assert schema_watcher_module.fingerprint(path)[1] is not None + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + assert schema_watcher_module.fingerprint(path)[1:] == (None, None) + assert schema_watcher_module.header_schema_version(path) == version == 301 + conn.close() + assert ( + schema_watcher_module.header_schema_version(str(tmp_path / "nope.db")) is None + ) + (tmp_path / "junk.db").write_bytes(b"not a database" * 20) + assert ( + schema_watcher_module.header_schema_version(str(tmp_path / "junk.db")) is None + ) + + @pytest.mark.asyncio +@pytest.mark.parametrize("use_header", (True, False)) @pytest.mark.parametrize("racy_rule", (True, False)) -async def test_racily_clean_same_tick_change(db_path, monkeypatch, racy_rule): +async def test_racily_clean_same_tick_change( + db_path, monkeypatch, racy_rule, use_header +): """Two schema changes inside one timestamp tick, same file size and inode: only the racy-clean rule catches the second one.""" + monkeypatch.setattr(schema_watcher_module, "USE_HEADER_CHECK", use_header) monkeypatch.setattr( schema_watcher_module, "fingerprint", From c87eeceb48a47859f9a9810a57bee2b0a0a08419 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 04:27:33 +0000 Subject: [PATCH 5/5] Schema watcher: scan connections are untracked so close() cannot pull them from under a scan remove_database() closes the Database, which closed every connection in _all_connections - including a catalog scan connection still in use on a worker thread: segfault (crashed an xdist worker in the full suite, reproduced with an add/remove stress loop). Database.connect(track=False) leaves the watcher's short-lived connections to the watcher, which always closes them on the thread that opened them. Regression test added. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/database.py | 15 +++++++++++---- datasette/schema_watcher.py | 15 ++++++--------- tests/test_schema_watcher.py | 27 +++++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 13 deletions(-) diff --git a/datasette/database.py b/datasette/database.py index 32e1d19d..73687c51 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -223,7 +223,11 @@ class Database: else: return "db" - def connect(self, write=False): + def connect(self, write=False, track=True): + """Open a new connection. ``track=False`` leaves it out of + ``_all_connections``: the caller closes it itself and close() must + never close it from another thread while it is in use (the + SchemaWatcher's short-lived scan connections).""" extra_kwargs = {} if write: extra_kwargs["isolation_level"] = "IMMEDIATE" @@ -234,11 +238,13 @@ class Database: ) if not write: conn.execute("PRAGMA query_only=1") - self._all_connections.append(conn) + if track: + self._all_connections.append(conn) return conn if self.is_memory: conn = sqlite3.connect(":memory:", uri=True, check_same_thread=False) - self._all_connections.append(conn) + if track: + self._all_connections.append(conn) return conn # mode=ro or immutable=1? @@ -261,7 +267,8 @@ class Database: conn = sqlite3.connect( f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs ) - self._all_connections.append(conn) + if track: + self._all_connections.append(conn) if self.is_temp_disk and not self._wal_enabled: conn.execute("PRAGMA journal_mode=WAL") self._wal_enabled = True diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index 6fa5ed7f..d3f84e92 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -605,26 +605,23 @@ class SchemaWatcher: # scanning + storing # ------------------------------------------------------------------ def _connect(self, db, prepare): - conn = db.connect() + # Untracked: Database.close() on the event loop thread must not close + # this connection while a worker thread is using it (that segfaults). + # It is always closed by _close() in the same thread that opened it. + conn = db.connect(track=False) try: if prepare: self.ds._prepare_connection(conn, db.name) else: conn.row_factory = None except Exception: - self._close(db, conn) + conn.close() raise return conn @staticmethod def _close(db, conn): - try: - conn.close() - finally: - try: - db._all_connections.remove(conn) - except ValueError: - pass + conn.close() def _scan_sync(self, state): """Read schema_version + full schema in one read transaction.""" diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py index 2b876809..df9f1a73 100644 --- a/tests/test_schema_watcher.py +++ b/tests/test_schema_watcher.py @@ -550,3 +550,30 @@ async def test_persisted_fingerprints_skip_unchanged_files(tmp_path, monkeypatch assert await catalog_tables(ds2, "p1") == ["added_offline", "t"] assert await catalog_tables(ds2, "p0") == ["t"] ds2.close() + + +@pytest.mark.asyncio +async def test_remove_database_during_catalog_scan(tmp_path): + # remove_database() closes the Database while the watcher may still be + # scanning it in a worker thread. The scan connection is not tracked by + # the Database, so close() must not pull it out from under the scan + # (that used to segfault). + path = str(tmp_path / "big.db") + conn = sqlite3.connect(path) + for i in range(150): + conn.execute(f"create table t{i} (a, b, c integer references t0(a))") + conn.execute(f"create index i{i} on t{i}(a)") + conn.commit() + conn.close() + ds = Datasette(settings={"schema_watch_interval_ms": 0}) + await ds.invoke_startup() + for i in range(60): + ds.add_database(Database(ds, path=path, is_mutable=i % 2 == 0), name=f"x{i}") + await asyncio.sleep(0.0005 * (i % 5)) + ds.remove_database(f"x{i}") + await asyncio.sleep(0.2) + rows = await ds.get_internal_database().execute( + "select count(*) from catalog_tables where database_name like 'x%'" + ) + assert rows.first()[0] == 0 + ds.close()