mirror of
https://github.com/simonw/datasette.git
synced 2026-10-18 23:36:57 +02:00
Fix concurrency, lifecycle and contract problems from the second review
Schema watcher:
- Scan coalescing is loop-agnostic: in-flight scans are concurrent.futures
futures any event loop can wait for, taken over only when their loop has
closed or stopped; each refresh() resolves only its own scans. A write on
one loop no longer hangs forever when another loop's scan is in flight.
- At most one polling task across event loops; replaced pollers exit.
- A scan or catalog write that fails transiently (internal database locked,
shared-cache lock, cancellation) re-marks the database and retries it from
later requests and sweeps, with back-off; requests are never failed by it.
Write threads mark schema changes directly (under a lock) instead of
relying on a notification reaching a loop that may have closed.
- add_database() file databases and named in-memory databases use the
default_schema_watch mode (external unless configured). Named in-memory
databases can be external: polled with PRAGMA schema_version on their
write connection. They are not scanned at registration time, which raced
plugins filling them (datasette-copy-to-memory).
- close() no longer persists a fingerprint for an owned database whose file
header shows a schema the catalog has not seen.
- The derived-table permission rule (external-content FTS) trusts the
catalog only when a stat and header read prove it is current; otherwise
it asks the database, and schedules a rescan if the catalog is behind.
Connections and write threads:
- A BaseException from a write or isolated callback is delivered to the
caller as a RuntimeError instead of killing the thread silently.
- conn.close() on a write lease raises ConnectionLeaseError, and the write
thread reopens a connection something else closed.
- Write connections pin PRAGMA recursive_triggers=on (sqlite-utils FTS
triggers rely on it; idle reopen lost the old drift).
- execute_fn() accepts generators that no longer hold the connection or a
cursor.
- --crossdb _memory connections count against max_open_connections, are
reaped, and are discarded when databases are added or removed.
Lifecycle:
- The temp internal database file uses weakref.finalize, not atexit, so
closed or abandoned instances can be garbage collected.
- The read-pool reaper no longer keeps the last pool alive in its frame.
- The scratch directory is loaded (and locked) at the end of __init__, so a
failed construction leaves nothing locked.
- config_dir/scratch is used only if empty or created by Datasette; a second
instance on the same config dir starts without it; sidecar files are never
deleted the first time a directory is opened.
Regression tests: tests/test_connection_review_fixes.py (24 of 27 fail on
a509b81; the other 3 guard behaviour that must not change).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv
This commit is contained in:
Claude
parent
a509b81f2c
commit
d7a2ce91cc
10 changed files with 1413 additions and 153 deletions
|
|
@ -256,7 +256,7 @@ SETTINGS = (
|
|||
Setting(
|
||||
"default_schema_watch",
|
||||
"external",
|
||||
"Schema watch mode for databases opened from files: external (poll for changes made by other processes) or owned (only Datasette changes them)",
|
||||
"Schema watch mode for database files and named in-memory databases without a mode of their own: external (poll for changes made outside Datasette) or owned (only Datasette changes them)",
|
||||
),
|
||||
Setting("sql_time_limit_ms", 1000, "Time limit for a SQL query in milliseconds"),
|
||||
Setting(
|
||||
|
|
@ -540,12 +540,16 @@ class Datasette:
|
|||
self._internal_database.name = INTERNAL_DB_NAME
|
||||
|
||||
# Scratch databases: an explicit directory, else config_dir/scratch
|
||||
# if it exists, else a temporary directory created on first use
|
||||
if scratch_dir is None and config_dir and (config_dir / "scratch").is_dir():
|
||||
scratch_dir = config_dir / "scratch"
|
||||
self._scratch = ScratchDatabases(self, scratch_dir)
|
||||
# Attaches existing scratch databases without opening any of them
|
||||
self._scratch.load()
|
||||
# if Datasette created it (or it is empty), else a temporary
|
||||
# directory created on first use. Loaded - and the directory locked
|
||||
# - at the end of __init__, so a failed construction leaves nothing
|
||||
# locked.
|
||||
self._scratch = ScratchDatabases(
|
||||
self,
|
||||
scratch_dir,
|
||||
inferred=scratch_dir is None and config_dir is not None,
|
||||
fallback_dir=(config_dir / "scratch") if config_dir else None,
|
||||
)
|
||||
|
||||
self.cache_headers = cache_headers
|
||||
self._static_asset_hashes = {}
|
||||
|
|
@ -702,6 +706,13 @@ class Datasette:
|
|||
self.client = DatasetteClient(self)
|
||||
# ds.config is available now: resolve per-database schema_watch modes
|
||||
self._schema_watcher.configure()
|
||||
# Attaches existing scratch databases without opening any of them.
|
||||
# Takes the scratch directory's lock, so nothing after this may fail
|
||||
try:
|
||||
self._scratch.load()
|
||||
except BaseException:
|
||||
self._scratch.close()
|
||||
raise
|
||||
# Last, so metric callbacks never see a partially initialized instance
|
||||
register_datasette(self)
|
||||
|
||||
|
|
@ -1002,8 +1013,19 @@ class Datasette:
|
|||
self._schema_watcher.register(
|
||||
db, schema_watch or getattr(db, "schema_watch", None)
|
||||
)
|
||||
self._crossdb_attachments_changed(db)
|
||||
return db
|
||||
|
||||
def _crossdb_attachments_changed(self, db):
|
||||
# --crossdb: each pooled _memory connection ATTACHed the databases
|
||||
# that existed when it was opened. Discard them so the next query
|
||||
# sees databases added since, and none removed since
|
||||
if not self.crossdb or db.is_memory or db.is_scratch:
|
||||
return
|
||||
memory = self.databases.get("_memory")
|
||||
if memory is not None and memory is not db:
|
||||
memory._invalidate_connections()
|
||||
|
||||
def add_memory_database(self, memory_name, name=None, route=None):
|
||||
return self.add_database(
|
||||
Database(self, memory_name=memory_name), name=name, route=route
|
||||
|
|
@ -1022,6 +1044,7 @@ class Datasette:
|
|||
# Deletes this database's catalog rows (explicitly - there is no
|
||||
# periodic stale-catalog scan any more)
|
||||
self._schema_watcher.unregister(name)
|
||||
self._crossdb_attachments_changed(db)
|
||||
return db
|
||||
|
||||
@property
|
||||
|
|
|
|||
|
|
@ -35,7 +35,10 @@ callback and returns it afterwards.
|
|||
Write callbacks get the same proxy (see ``Database._execute_writes``).
|
||||
* In-memory databases are leased the same way but do not count against the
|
||||
cap and are never reaped: they use no file descriptors, and closing every
|
||||
connection to a named in-memory database would destroy its contents.
|
||||
connection to a named in-memory database would destroy its contents. The
|
||||
exception is ``_memory`` under ``--crossdb``: each of its connections
|
||||
ATTACHes up to ten database files, so those are counted and reaped like
|
||||
file connections.
|
||||
"""
|
||||
|
||||
import collections
|
||||
|
|
@ -132,15 +135,39 @@ class LeasedConnection:
|
|||
return self._live().cursor(*args, **kwargs)
|
||||
|
||||
def close(self):
|
||||
conn = self._live()
|
||||
if self._kind == "write":
|
||||
# The write connection belongs to the database's write thread,
|
||||
# which keeps using it for every later write
|
||||
raise ConnectionLeaseError(
|
||||
f"Cannot close the write connection to database "
|
||||
f"{self._db_name!r} from inside an execute_write_fn() "
|
||||
"callback: Datasette owns it and closes it when it is idle"
|
||||
)
|
||||
# Closing a pooled read connection is allowed; the pool notices on
|
||||
# release and discards it instead of returning it to the idle stack
|
||||
self._live().close()
|
||||
# release and discards it instead of returning it to the idle stack.
|
||||
# An isolated connection is closed afterwards anyway.
|
||||
conn.close()
|
||||
|
||||
def __repr__(self):
|
||||
state = "expired" if self._conn is None else "active"
|
||||
return f"<LeasedConnection {self._kind} database={self._db_name!r} {state}>"
|
||||
|
||||
|
||||
def _generator_uses_connection(result):
|
||||
"""True for a generator that would step the leased connection after
|
||||
the callback returned: one holding the connection (a closure over
|
||||
``conn``) or a live cursor (``(r for r in conn.execute(...))``). A
|
||||
generator over rows already fetched inside the callback is fine."""
|
||||
if not inspect.isgenerator(result):
|
||||
return False
|
||||
try:
|
||||
values = inspect.getgeneratorlocals(result).values()
|
||||
except Exception: # noqa: BLE001
|
||||
return False
|
||||
return any(isinstance(v, (sqlite3.Connection, sqlite3.Cursor)) for v in values)
|
||||
|
||||
|
||||
class _Entry:
|
||||
__slots__ = ("conn", "generation", "last_used", "state")
|
||||
|
||||
|
|
@ -163,7 +190,15 @@ class _DatabaseState:
|
|||
self.leased = 0
|
||||
self.closed = False
|
||||
# In-memory databases use no file descriptors and are never evicted
|
||||
self.counted = not db.is_memory
|
||||
# (closing every connection to a named one would destroy it) -
|
||||
# except _memory under --crossdb, whose connections each ATTACH up
|
||||
# to ten database files and are private, so closing one loses
|
||||
# nothing
|
||||
self.counted = not db.is_memory or (
|
||||
db.name == "_memory"
|
||||
and not db.memory_name
|
||||
and bool(getattr(db.ds, "crossdb", False))
|
||||
)
|
||||
|
||||
|
||||
class ReadConnectionPool:
|
||||
|
|
@ -189,7 +224,7 @@ class ReadConnectionPool:
|
|||
finally:
|
||||
lease._expire()
|
||||
self.release(entry)
|
||||
if isinstance(result, sqlite3.Cursor) or inspect.isgenerator(result):
|
||||
if isinstance(result, sqlite3.Cursor) or _generator_uses_connection(result):
|
||||
raise ConnectionLeaseError(
|
||||
f"An execute_fn() callback for database {db.name!r} returned a {type(result).__name__}, which "
|
||||
"would keep using the pooled connection after the callback "
|
||||
|
|
@ -474,14 +509,10 @@ class _Reaper:
|
|||
self.wake_at = float("inf")
|
||||
self._dirty = False
|
||||
pools = list(self._pools)
|
||||
next_wake = None
|
||||
for pool in pools:
|
||||
try:
|
||||
expiry = pool.reap_expired()
|
||||
except Exception: # noqa: BLE001
|
||||
expiry = None
|
||||
if expiry is not None and (next_wake is None or expiry < next_wake):
|
||||
next_wake = expiry
|
||||
# In a helper so that no reference to a pool (and through its
|
||||
# _prepare_connection, the whole Datasette) survives in this
|
||||
# long-lived frame while it waits
|
||||
next_wake = self._reap(pools)
|
||||
del pools
|
||||
with self._cond:
|
||||
if self._dirty:
|
||||
|
|
@ -495,5 +526,17 @@ class _Reaper:
|
|||
# together is handled in one pass
|
||||
self._cond.wait(max(0.0, next_wake - time.monotonic()) + 0.05)
|
||||
|
||||
@staticmethod
|
||||
def _reap(pools):
|
||||
next_wake = None
|
||||
for pool in pools:
|
||||
try:
|
||||
expiry = pool.reap_expired()
|
||||
except Exception: # noqa: BLE001
|
||||
expiry = None
|
||||
if expiry is not None and (next_wake is None or expiry < next_wake):
|
||||
next_wake = expiry
|
||||
return next_wake
|
||||
|
||||
|
||||
_reaper = _Reaper()
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
import asyncio
|
||||
import atexit
|
||||
import contextvars
|
||||
import inspect
|
||||
import os
|
||||
|
|
@ -9,6 +8,7 @@ import tempfile
|
|||
import threading
|
||||
import time
|
||||
import uuid
|
||||
import weakref
|
||||
from collections import namedtuple
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -120,7 +120,13 @@ class Database:
|
|||
self.is_mutable = True
|
||||
self.mode = "rwc"
|
||||
self._wal_enabled = False
|
||||
atexit.register(self._cleanup_temp_file)
|
||||
# Unlinked by close(), when this Database is garbage collected,
|
||||
# or at exit. The finalizer holds only the path: an
|
||||
# atexit.register() of a bound method would keep this Database -
|
||||
# and through it the whole Datasette - alive until exit
|
||||
self._temp_file_finalizer = weakref.finalize(
|
||||
self, _unlink_temp_files, temp_path
|
||||
)
|
||||
else:
|
||||
self._wal_enabled = False
|
||||
self.cached_hash = None
|
||||
|
|
@ -266,6 +272,7 @@ class Database:
|
|||
generation = self._conn_generation
|
||||
conn = self.connect(write=True)
|
||||
try:
|
||||
_pin_write_connection_pragmas(conn)
|
||||
self.ds._prepare_connection(conn, self.name)
|
||||
except BaseException:
|
||||
# Never cache a half-prepared connection
|
||||
|
|
@ -429,11 +436,7 @@ class Database:
|
|||
|
||||
def _cleanup_temp_file(self):
|
||||
if self.is_temp_disk and self.path:
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
try:
|
||||
os.unlink(self.path + suffix)
|
||||
except OSError:
|
||||
pass
|
||||
self._temp_file_finalizer()
|
||||
|
||||
async def execute_write(
|
||||
self,
|
||||
|
|
@ -815,11 +818,13 @@ class Database:
|
|||
return
|
||||
if task is _SHUTDOWN:
|
||||
return
|
||||
if writer.generation != self._conn_generation:
|
||||
if writer.generation != self._conn_generation or not writer.usable():
|
||||
# The file was replaced or deleted since this connection was
|
||||
# opened (the SchemaWatcher bumps the generation): reopen by
|
||||
# path. connect() uses mode=rw for a file Datasette has seen,
|
||||
# so a deleted file is reported, not silently recreated.
|
||||
# Also reopened if something closed it (a callback reaching
|
||||
# the raw connection through cursor.connection).
|
||||
writer.reopen()
|
||||
idle_timeout = current_idle_timeout()
|
||||
# block=True: the caller awaits the result, so the write spans
|
||||
|
|
@ -870,11 +875,9 @@ class Database:
|
|||
self._write_thread_connections.discard(
|
||||
isolated_connection
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
except BaseException as e: # noqa: BLE001
|
||||
# Write thread must survive any task failure or the database wedges
|
||||
sys.stderr.write(f"{e}\n")
|
||||
sys.stderr.flush()
|
||||
exception = e
|
||||
exception = self._write_task_failed(e)
|
||||
else:
|
||||
conn = writer.conn
|
||||
try:
|
||||
|
|
@ -902,15 +905,33 @@ class Database:
|
|||
)
|
||||
finally:
|
||||
self._schema_check(conn, task.loop)
|
||||
except Exception as e: # noqa: BLE001
|
||||
sys.stderr.write(f"{e}\n")
|
||||
sys.stderr.flush()
|
||||
exception = e
|
||||
except BaseException as e: # noqa: BLE001
|
||||
exception = self._write_task_failed(e)
|
||||
_deliver_write_result(task, result, exception)
|
||||
finally:
|
||||
if token is not None:
|
||||
otel_context_api.detach(token)
|
||||
|
||||
def _write_task_failed(self, e):
|
||||
"""The exception to deliver to the caller of a write task that
|
||||
raised ``e``. An Exception is delivered as it is. Any other
|
||||
BaseException (SystemExit, or a plugin's own BaseException subclass)
|
||||
is delivered wrapped in a RuntimeError: re-raised by ``await`` on the
|
||||
caller's event loop, SystemExit or KeyboardInterrupt would stop that
|
||||
loop. The write thread keeps running (``with conn`` has rolled the
|
||||
transaction back). Before this the thread died and the caller
|
||||
waited forever."""
|
||||
if isinstance(e, Exception):
|
||||
sys.stderr.write(f"{e}\n")
|
||||
sys.stderr.flush()
|
||||
return e
|
||||
wrapped = RuntimeError(
|
||||
f"A write callback for database {self.name!r} raised "
|
||||
f"{type(e).__name__}: {e}"
|
||||
)
|
||||
wrapped.__cause__ = e
|
||||
return wrapped
|
||||
|
||||
async def execute_fn(self, fn):
|
||||
"""Run `fn(conn)` on a read connection, traced as a `db.query` span.
|
||||
|
||||
|
|
@ -1276,23 +1297,26 @@ class Database:
|
|||
async def derived_table_dependencies(self):
|
||||
"""Return implementation tables and the tables they derive from."""
|
||||
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
|
||||
):
|
||||
if state is not None and self.ds._schema_watcher.catalog_is_current(state):
|
||||
# From the catalog rows (computed by the SchemaWatcher when it
|
||||
# scanned this database, or read back from _internal), so a
|
||||
# permission check on one of this database's tables does not
|
||||
# open it and agrees with what allowed_resources() lists. As
|
||||
# fresh as the catalog itself.
|
||||
# open it and agrees with what allowed_resources() lists. Only
|
||||
# when the catalog provably matches the file (a stat and a
|
||||
# header read): this decides whether a table that derives from
|
||||
# another - an external-content FTS table - may be shown, and a
|
||||
# catalog that lags an out-of-band change would allow a table
|
||||
# it has not seen yet.
|
||||
from .utils.catalog import catalog_derived_table_dependencies
|
||||
|
||||
return (await catalog_derived_table_dependencies(self.ds, [self]))[
|
||||
self.name
|
||||
]
|
||||
schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
|
||||
if state is not None:
|
||||
# Found the catalog behind: rebuild it now rather than at the
|
||||
# next write (owned) or poll (external)
|
||||
self.ds._schema_watcher.note_live_version(state, schema_version)
|
||||
if (
|
||||
self._cached_derived_table_dependencies is None
|
||||
or self._cached_derived_table_dependencies[0] != schema_version
|
||||
|
|
@ -1396,6 +1420,21 @@ def _apply_write_wrapper(fn, wrapper_factory, track_event):
|
|||
return wrapped
|
||||
|
||||
|
||||
def _pin_write_connection_pragmas(conn):
|
||||
"""Connection state Datasette's writes rely on, set explicitly on every
|
||||
new write connection (before prepare_connection hooks, which may change
|
||||
it). Write connections are reopened after an idle period, so anything
|
||||
that used to "drift" into place on a long-lived connection must be set
|
||||
here instead.
|
||||
|
||||
recursive_triggers: sqlite-utils turns it on whenever it wraps a
|
||||
connection (core's JSON write API does that), and the triggers that keep
|
||||
sqlite-utils FTS indexes in sync need it for INSERT OR REPLACE to remove
|
||||
the replaced row's index entry.
|
||||
"""
|
||||
conn.execute("PRAGMA recursive_triggers = on")
|
||||
|
||||
|
||||
def _call_with_lease(fn, conn, db_name, kind):
|
||||
"""Call fn with a LeasedConnection for conn that expires when it returns.
|
||||
|
||||
|
|
@ -1438,6 +1477,7 @@ class _WriteConnection:
|
|||
try:
|
||||
self.conn = db.connect(write=True)
|
||||
db._write_thread_connections.add(self.conn)
|
||||
_pin_write_connection_pragmas(self.conn)
|
||||
# Threads do not inherit the caller's context, so any spans
|
||||
# created by prepare_connection hooks here are root spans
|
||||
db.ds._prepare_connection(self.conn, db.name)
|
||||
|
|
@ -1449,6 +1489,18 @@ class _WriteConnection:
|
|||
self.close()
|
||||
self.open()
|
||||
|
||||
def usable(self):
|
||||
"""False if the connection has been closed behind the write thread's
|
||||
back. A failed open (``exception`` set) counts as usable: that error
|
||||
is reported to each write until the thread exits and retries."""
|
||||
conn = self.conn
|
||||
if conn is None:
|
||||
return self.exception is not None
|
||||
try:
|
||||
return conn.total_changes >= 0
|
||||
except sqlite3.ProgrammingError:
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
conn, self.conn = self.conn, None
|
||||
if conn is not None:
|
||||
|
|
@ -1513,6 +1565,14 @@ def _deliver_write_result(task, result, exception):
|
|||
pass
|
||||
|
||||
|
||||
def _unlink_temp_files(path):
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
try:
|
||||
os.unlink(path + suffix)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
class QueryInterrupted(Exception):
|
||||
def __init__(self, e, sql, params):
|
||||
self.e = e
|
||||
|
|
|
|||
|
|
@ -26,12 +26,26 @@ Every attached database has a *schema watch mode*:
|
|||
``immutable``
|
||||
Scanned once at startup and never again.
|
||||
|
||||
Defaults: files passed to ``Datasette(files=...)`` / the CLI use the
|
||||
Defaults: database files and named in-memory databases use the
|
||||
``default_schema_watch`` setting (``external`` unless set to ``owned``),
|
||||
immutable files are ``immutable``, anything added later with
|
||||
``datasette.add_database()`` is ``owned``. Override per database with
|
||||
whether they were passed to ``Datasette(files=...)`` / the CLI or added
|
||||
later with ``datasette.add_database()``. Immutable files are ``immutable``
|
||||
and private ``:memory:`` databases ``owned``. Override per database with
|
||||
``databases: {name: {schema_watch: owned|external|immutable}}`` in
|
||||
``datasette.yaml`` or ``add_database(..., schema_watch=...)``.
|
||||
``datasette.yaml`` or ``add_database(..., schema_watch=...)``. A named
|
||||
in-memory database in ``external`` mode is polled with ``PRAGMA
|
||||
schema_version`` on its write connection (shared-cache table locks do not
|
||||
wait, so reading it from another connection could fail Datasette's writes).
|
||||
|
||||
Several event loops
|
||||
-------------------
|
||||
One Datasette can be driven by several event loops at once, each in its own
|
||||
thread. Scans in flight are tracked with ``concurrent.futures`` futures that
|
||||
any loop can wait for; a scan is only taken over when the loop running it
|
||||
has closed or stopped. There is at most one live polling task. State that a
|
||||
write thread marks (``needs_scan``, ``_pending``) is set under a lock, before
|
||||
the write's result is delivered, so nothing depends on a notification
|
||||
reaching a loop that may have gone away.
|
||||
|
||||
Replaced and deleted files
|
||||
--------------------------
|
||||
|
|
@ -60,6 +74,7 @@ open.
|
|||
"""
|
||||
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
|
@ -92,6 +107,11 @@ 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
|
||||
# Seconds before a database whose catalog scan or catalog write failed for a
|
||||
# transient reason (internal database locked, cancelled) is scanned again by
|
||||
# a request or sweep; doubled after each consecutive failure up to the max
|
||||
RETRY_DELAY_S = 1.0
|
||||
RETRY_DELAY_MAX_S = 30.0
|
||||
|
||||
|
||||
class WatchState:
|
||||
|
|
@ -108,6 +128,8 @@ class WatchState:
|
|||
"notified_version",
|
||||
"removed",
|
||||
"requested_mode",
|
||||
"retry_at",
|
||||
"retry_delay",
|
||||
"scan_future",
|
||||
"stats",
|
||||
)
|
||||
|
|
@ -122,7 +144,12 @@ class WatchState:
|
|||
self.catalog_version = None
|
||||
self.notified_version = None
|
||||
self.needs_scan = True
|
||||
# An _InflightScan while a catalog scan of this database runs
|
||||
self.scan_future = None
|
||||
# time.monotonic() before which a failed scan is not retried, and
|
||||
# the delay used for the next failure (doubles, see RETRY_DELAY_S)
|
||||
self.retry_at = 0.0
|
||||
self.retry_delay = RETRY_DELAY_S
|
||||
self.missing = False
|
||||
self.removed = False
|
||||
self.error = None
|
||||
|
|
@ -221,6 +248,68 @@ def header_schema_version(path):
|
|||
return int.from_bytes(header[40:44], "big")
|
||||
|
||||
|
||||
class _InflightScan:
|
||||
"""One catalog scan in progress for one database.
|
||||
|
||||
Not bound to an event loop: one Datasette is driven by several loops
|
||||
(``--get``, TestClient, pytest-asyncio), each in its own thread, and a
|
||||
write on any of them may need to wait for a scan started on another.
|
||||
Waiters use :func:`_wait_for` on ``future``, a ``concurrent.futures``
|
||||
future that the scan's owner resolves in a ``finally`` block.
|
||||
"""
|
||||
|
||||
__slots__ = ("future", "loop")
|
||||
|
||||
def __init__(self, loop):
|
||||
self.future = concurrent.futures.Future()
|
||||
self.loop = loop
|
||||
|
||||
def done(self):
|
||||
return self.future.done()
|
||||
|
||||
def abandoned(self):
|
||||
"""True if nothing will ever finish this scan: the loop running it
|
||||
has closed, or has stopped (``run_until_complete()`` returned while
|
||||
the scan task was still pending)."""
|
||||
loop = self.loop
|
||||
return loop.is_closed() or not loop.is_running()
|
||||
|
||||
def finish(self):
|
||||
try:
|
||||
self.future.set_result(None)
|
||||
except concurrent.futures.InvalidStateError:
|
||||
pass
|
||||
|
||||
|
||||
def _wait_for(cf_future):
|
||||
"""An asyncio future on the running loop that completes when the
|
||||
concurrent future does. Cancelling it does not cancel cf_future, which
|
||||
other waiters (maybe on other loops) share."""
|
||||
loop = asyncio.get_running_loop()
|
||||
waiter = loop.create_future()
|
||||
|
||||
def _set():
|
||||
if not waiter.done():
|
||||
waiter.set_result(None)
|
||||
|
||||
def _done(_):
|
||||
try:
|
||||
loop.call_soon_threadsafe(_set)
|
||||
except RuntimeError:
|
||||
# That loop has closed; nobody is waiting any more
|
||||
pass
|
||||
|
||||
cf_future.add_done_callback(_done)
|
||||
return waiter
|
||||
|
||||
|
||||
def _is_transient(error):
|
||||
if isinstance(error, sqlite3.OperationalError):
|
||||
message = str(error)
|
||||
return "locked" in message or "busy" in message
|
||||
return False
|
||||
|
||||
|
||||
def _fp_to_json(fp, t_ns, closed=False):
|
||||
data = {"fp": fp, "t": t_ns}
|
||||
if closed:
|
||||
|
|
@ -263,9 +352,21 @@ class SchemaWatcher:
|
|||
def __init__(self, ds):
|
||||
self.ds = ds
|
||||
self.states = {}
|
||||
self._pending = set() # states registered after startup, not yet scanned
|
||||
# States whose catalog must be (re)built: registered after startup,
|
||||
# changed by a write whose notification may never run, or whose last
|
||||
# scan failed for a transient reason. Flushed by the next request or
|
||||
# sweep.
|
||||
self._pending = set()
|
||||
self._pending_removals = set() # names whose catalog rows must go
|
||||
# Guards claiming scans (WatchState.needs_scan / scan_future) and
|
||||
# _pending: refresh() runs on several event loops in different
|
||||
# threads, and write threads mark states from their own thread
|
||||
self._lock = threading.Lock()
|
||||
# The current polling task. At most one is live: a poller whose loop
|
||||
# has closed or stopped is replaced, and a replaced poller exits at
|
||||
# its next wake-up (it is no longer self._task)
|
||||
self._task = None
|
||||
self._pollers = set()
|
||||
self._spawned = set()
|
||||
self._last_sweep = 0.0
|
||||
# Bumped after every write to the catalog tables, so readers can
|
||||
|
|
@ -320,7 +421,13 @@ class SchemaWatcher:
|
|||
db = state.db
|
||||
mode = self._config_mode(state.name) or state.requested_mode
|
||||
if mode is None:
|
||||
mode = "owned"
|
||||
if state.is_file or db.memory_name:
|
||||
# Other code may change it: a plugin's own connection, a
|
||||
# backup() into a named in-memory database, another process
|
||||
mode = self._default_mode_for_files()
|
||||
else:
|
||||
# A private ":memory:" database: nothing else can see it
|
||||
mode = "owned"
|
||||
elif mode == FILES_DEFAULT:
|
||||
mode = self._default_mode_for_files()
|
||||
if mode not in MODES:
|
||||
|
|
@ -333,8 +440,10 @@ class SchemaWatcher:
|
|||
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
|
||||
elif mode == "external" and not state.is_file and not db.memory_name:
|
||||
# A private ":memory:" database: there is nothing to poll.
|
||||
# Named in-memory databases are polled with PRAGMA
|
||||
# schema_version on their write connection.
|
||||
mode = "owned"
|
||||
return mode
|
||||
|
||||
|
|
@ -363,8 +472,16 @@ class SchemaWatcher:
|
|||
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]))
|
||||
with self._lock:
|
||||
self._pending.add(state)
|
||||
if not db.memory_name:
|
||||
self._spawn(self.refresh([state]))
|
||||
# A shared-cache in-memory database is scanned by the next
|
||||
# request or sweep instead: plugins typically fill one right
|
||||
# after adding it (VACUUM INTO, backup()) from their own
|
||||
# connection, and shared-cache table locks do not wait - a scan
|
||||
# reading sqlite_master at that moment makes their writes fail
|
||||
# with "database table is locked"
|
||||
return state
|
||||
|
||||
def unregister(self, name):
|
||||
|
|
@ -373,7 +490,8 @@ class SchemaWatcher:
|
|||
return
|
||||
state.removed = True
|
||||
state.db._watch_state = None
|
||||
self._pending.discard(state)
|
||||
with self._lock:
|
||||
self._pending.discard(state)
|
||||
if self.ds.internal_db_created:
|
||||
self._pending_removals.add(name)
|
||||
self._spawn(self._delete_catalog([name]))
|
||||
|
|
@ -408,34 +526,51 @@ class SchemaWatcher:
|
|||
task.add_done_callback(_done)
|
||||
return task
|
||||
|
||||
def loop_running(self):
|
||||
def _poller_alive(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
|
||||
loop = task.get_loop()
|
||||
return not loop.is_closed() and loop.is_running()
|
||||
|
||||
def loop_running(self):
|
||||
"""True while a polling task is alive, on any event loop."""
|
||||
return self._poller_alive()
|
||||
|
||||
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():
|
||||
"""Start the polling loop unless one is already alive on any event
|
||||
loop. Runs one sweep first so the catalog is current when the first
|
||||
request is served."""
|
||||
if self.interval_s <= 0 or self.ds._closed:
|
||||
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")
|
||||
with self._lock:
|
||||
if self._poller_alive():
|
||||
return
|
||||
# Claim the slot before awaiting so concurrent start() calls,
|
||||
# on this loop or another, no-op
|
||||
task = loop.create_task(self._run(), name="datasette-schema-watcher")
|
||||
self._task = task
|
||||
self._pollers.add(task)
|
||||
task.add_done_callback(self._poller_done)
|
||||
if self.ds.internal_db_created:
|
||||
try:
|
||||
await self.ds._refresh_schemas(background=True)
|
||||
except Exception:
|
||||
logger.exception("schema watcher initial sweep failed")
|
||||
|
||||
def _poller_done(self, task):
|
||||
with self._lock:
|
||||
self._pollers.discard(task)
|
||||
|
||||
async def _run(self):
|
||||
# Wait one interval first: start() has just swept
|
||||
while not self.ds._closed:
|
||||
me = asyncio.current_task()
|
||||
# Wait one interval first: start() has just swept. A poller that
|
||||
# has been replaced (its loop stopped for a while and another loop
|
||||
# started a new one) exits instead of sweeping alongside it.
|
||||
while not self.ds._closed and self._task is me:
|
||||
await asyncio.sleep(self.interval_s)
|
||||
if self.ds._closed:
|
||||
if self.ds._closed or self._task is not me:
|
||||
return
|
||||
try:
|
||||
await self.ds._refresh_schemas(background=True)
|
||||
|
|
@ -444,8 +579,12 @@ class SchemaWatcher:
|
|||
except Exception:
|
||||
logger.exception("schema watcher sweep failed")
|
||||
|
||||
def _all_tasks(self):
|
||||
with self._lock:
|
||||
return [t for t in [self._task, *self._pollers, *self._spawned] if t]
|
||||
|
||||
def stop(self):
|
||||
tasks = [t for t in [self._task, *self._spawned] if t is not None]
|
||||
tasks = self._all_tasks()
|
||||
self._task = None
|
||||
for task in tasks:
|
||||
if task.done():
|
||||
|
|
@ -464,7 +603,7 @@ class SchemaWatcher:
|
|||
pass
|
||||
|
||||
async def astop(self):
|
||||
tasks = [t for t in [self._task, *self._spawned] if t is not None]
|
||||
tasks = self._all_tasks()
|
||||
self.stop()
|
||||
loop = asyncio.get_running_loop()
|
||||
mine = [t for t in tasks if t.get_loop() is loop and not t.done()]
|
||||
|
|
@ -472,23 +611,35 @@ class SchemaWatcher:
|
|||
await asyncio.wait(mine, timeout=5)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# request path: O(1) unless add/remove_database left work pending
|
||||
# request path: O(1) unless add/remove_database or a write 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()
|
||||
try:
|
||||
await self.flush_pending()
|
||||
except Exception:
|
||||
# Never fail the request: what is still pending is retried
|
||||
# (with back-off) by later requests and sweeps
|
||||
logger.warning("Could not update the catalog", exc_info=True)
|
||||
if (
|
||||
not self.loop_running()
|
||||
and self.interval_s > 0
|
||||
self.interval_s > 0
|
||||
and self.ds.internal_db_created
|
||||
and not self._poller_alive()
|
||||
):
|
||||
await self.start()
|
||||
|
||||
def _ready_pending(self):
|
||||
now = time.monotonic()
|
||||
with self._lock:
|
||||
return [s for s in self._pending if not s.removed and s.retry_at <= now]
|
||||
|
||||
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))
|
||||
ready = self._ready_pending()
|
||||
if ready:
|
||||
await self.refresh(ready)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# write path
|
||||
|
|
@ -501,9 +652,25 @@ class SchemaWatcher:
|
|||
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
|
||||
with self._lock:
|
||||
if (
|
||||
state.removed
|
||||
or version == state.catalog_version
|
||||
or version == state.notified_version
|
||||
):
|
||||
return
|
||||
state.notified_version = version
|
||||
# Marked here, on the write thread, before the write's result is
|
||||
# delivered: a blocking writer always finds needs_scan set in
|
||||
# Database._after_write(). The notification below only starts a
|
||||
# scan sooner; if it never runs (a block=False write from an
|
||||
# event loop that has since closed or stopped) the next request
|
||||
# or sweep picks the state up from _pending.
|
||||
state.needs_scan = True
|
||||
state.retry_at = 0.0
|
||||
self._pending.add(state)
|
||||
state.stats["write_detections"] += 1
|
||||
self.counters["write_detections"] += 1
|
||||
if loop is None:
|
||||
self._on_write_schema_change(state)
|
||||
else:
|
||||
|
|
@ -515,9 +682,6 @@ class SchemaWatcher:
|
|||
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]))
|
||||
|
||||
|
|
@ -562,7 +726,7 @@ class SchemaWatcher:
|
|||
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_scan = self._ready_pending()
|
||||
to_clear = []
|
||||
for state, kind, fp, t_ns, version in outcomes:
|
||||
if state.removed:
|
||||
|
|
@ -797,40 +961,69 @@ class SchemaWatcher:
|
|||
|
||||
async def refresh(self, states):
|
||||
"""Rebuild the catalog for every state that needs it, coalescing with
|
||||
scans already in flight for the same database."""
|
||||
scans already in flight for the same database - on this event loop
|
||||
or any other."""
|
||||
loop = asyncio.get_running_loop()
|
||||
pending = list(states)
|
||||
# States whose scan this call ran and that failed transiently: left
|
||||
# in _pending for a later request or sweep, not retried here
|
||||
failed = set()
|
||||
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)
|
||||
with self._lock:
|
||||
for state in pending:
|
||||
inflight = state.scan_future
|
||||
if inflight is not None and not inflight.done():
|
||||
if not inflight.abandoned():
|
||||
waits.append(inflight.future)
|
||||
continue
|
||||
# Its event loop has closed or stopped, so nothing
|
||||
# will finish that scan. Take over: its results are
|
||||
# discarded (_store() checks scan_future) and its
|
||||
# waiters are woken to wait for this one instead
|
||||
state.scan_future = None
|
||||
state.needs_scan = True
|
||||
inflight.finish()
|
||||
if state.needs_scan and not state.removed and state not in failed:
|
||||
state.needs_scan = False
|
||||
state.scan_future = _InflightScan(loop)
|
||||
mine.append(state)
|
||||
if mine:
|
||||
records = {state: state.scan_future for state in mine}
|
||||
handled = set()
|
||||
try:
|
||||
await self._scan_and_store(mine, records, handled)
|
||||
finally:
|
||||
now = time.monotonic()
|
||||
with self._lock:
|
||||
for state, record in records.items():
|
||||
if state.scan_future is not record:
|
||||
# Taken over by another loop (see above)
|
||||
continue
|
||||
state.scan_future = None
|
||||
if state not in handled and not state.removed:
|
||||
# The scan or the catalog write failed
|
||||
# transiently (internal database locked, a
|
||||
# shared-cache lock) or was cancelled: the
|
||||
# catalog is not current, so make sure a
|
||||
# later request or sweep tries again
|
||||
state.needs_scan = True
|
||||
state.retry_at = now + state.retry_delay
|
||||
state.retry_delay = min(
|
||||
state.retry_delay * 2, RETRY_DELAY_MAX_S
|
||||
)
|
||||
self._pending.add(state)
|
||||
failed.add(state)
|
||||
for record in records.values():
|
||||
record.finish()
|
||||
if waits:
|
||||
await asyncio.wait(waits)
|
||||
await asyncio.wait([_wait_for(f) for f in waits])
|
||||
pending = [
|
||||
s
|
||||
for s in pending
|
||||
if not s.removed
|
||||
and s not in failed
|
||||
and (
|
||||
s.needs_scan
|
||||
or (s.scan_future is not None and not s.scan_future.done())
|
||||
|
|
@ -839,18 +1032,25 @@ class SchemaWatcher:
|
|||
]
|
||||
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)
|
||||
with self._lock:
|
||||
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):
|
||||
async def wait_for_scan(self, state):
|
||||
"""Wait for a catalog scan of state that is in flight, if any."""
|
||||
inflight = state.scan_future
|
||||
if inflight is not None and not inflight.done() and not inflight.abandoned():
|
||||
await _wait_for(inflight.future)
|
||||
|
||||
async def _scan_and_store(self, states, records, handled):
|
||||
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)
|
||||
await self._store(results, records, handled)
|
||||
states = [s for s in states if s not in memory]
|
||||
if not states:
|
||||
return
|
||||
|
|
@ -860,24 +1060,49 @@ class SchemaWatcher:
|
|||
async def one(chunk):
|
||||
async with semaphore:
|
||||
results = await self._off_loop(self._scan_chunk_sync, chunk)
|
||||
await self._store(results)
|
||||
await self._store(results, records, handled)
|
||||
|
||||
if len(chunks) == 1:
|
||||
await one(chunks[0])
|
||||
else:
|
||||
await asyncio.gather(*(one(c) for c in chunks))
|
||||
|
||||
async def _store(self, results):
|
||||
async def _store(self, results, records=None, handled=None):
|
||||
"""Write scan results to the catalog. ``records`` maps each state to
|
||||
the _InflightScan this scan was started as: results for a state whose
|
||||
scan has since been taken over by another loop are dropped. States
|
||||
whose outcome was recorded are added to ``handled``; the others
|
||||
(transient failures) are retried later by refresh()."""
|
||||
records = records or {}
|
||||
if handled is None:
|
||||
handled = set()
|
||||
|
||||
def current(state):
|
||||
record = records.get(state)
|
||||
return record is None or state.scan_future is record
|
||||
|
||||
entries = []
|
||||
missing = []
|
||||
stored = []
|
||||
for r in results:
|
||||
state = r["state"]
|
||||
if state.removed:
|
||||
if state.removed or not current(state):
|
||||
handled.add(state)
|
||||
continue
|
||||
if "error" in r:
|
||||
if state.db._closed:
|
||||
# Closed (or deleted) while the scan was running
|
||||
handled.add(state)
|
||||
continue
|
||||
if _is_transient(r["error"]):
|
||||
# "database is locked" / "database table is locked" (a
|
||||
# shared-cache memory database being filled from another
|
||||
# connection): not a property of the database, retry
|
||||
logger.info(
|
||||
"Schema of database %r is locked, will retry: %s",
|
||||
state.name,
|
||||
r["error"],
|
||||
)
|
||||
continue
|
||||
state.error = r["error"]
|
||||
state.needs_scan = True
|
||||
|
|
@ -887,23 +1112,28 @@ class SchemaWatcher:
|
|||
logger.warning(
|
||||
"Could not read schema of database %r: %s", state.name, r["error"]
|
||||
)
|
||||
handled.add(state)
|
||||
continue
|
||||
state.error = None
|
||||
if r.get("missing"):
|
||||
missing.append(state)
|
||||
state.fp = r["fp"]
|
||||
state.fp_racy = True
|
||||
handled.add(state)
|
||||
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"],
|
||||
state,
|
||||
(
|
||||
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)
|
||||
|
|
@ -911,15 +1141,27 @@ class SchemaWatcher:
|
|||
watcher = self
|
||||
|
||||
def _write(conn):
|
||||
# Skip databases removed while we were scanning
|
||||
live = [e for e in entries if e[0] in watcher.states]
|
||||
# Runs on the internal database's write thread, which
|
||||
# serializes catalog writes: skip databases removed (or
|
||||
# re-added) while we were scanning, and scans another loop
|
||||
# has taken over - its newer rows must not be overwritten
|
||||
live = [
|
||||
entry
|
||||
for state, entry in entries
|
||||
if watcher.states.get(entry[0]) is state and current(state)
|
||||
]
|
||||
write_catalog_entries(conn, live)
|
||||
|
||||
await self.ds.get_internal_database().execute_write_fn(_write)
|
||||
for r in stored:
|
||||
state = r["state"]
|
||||
handled.add(state)
|
||||
if state.removed or not current(state):
|
||||
continue
|
||||
state.catalog_version = r["version"]
|
||||
state.missing = False
|
||||
state.retry_at = 0.0
|
||||
state.retry_delay = RETRY_DELAY_S
|
||||
if r["fp"] is not None:
|
||||
state.fp = r["fp"]
|
||||
state.fp_racy = is_racy(r["fp"], r["t"])
|
||||
|
|
@ -1028,7 +1270,10 @@ class SchemaWatcher:
|
|||
state.needs_scan = True
|
||||
if to_scan:
|
||||
await self.refresh(to_scan)
|
||||
self._pending.clear()
|
||||
with self._lock:
|
||||
# Keep states that a write marked during the scan, or whose scan
|
||||
# failed transiently
|
||||
self._pending = {s for s in self._pending if s.needs_scan and not s.removed}
|
||||
|
||||
def _reusable_sync(self, states, persisted):
|
||||
"""Databases whose stored fingerprint matches the file right now and
|
||||
|
|
@ -1080,13 +1325,77 @@ class SchemaWatcher:
|
|||
fp = fingerprint(state.db.path)
|
||||
except OSError:
|
||||
continue
|
||||
if fp[0] is None:
|
||||
if fp[0] is None or fp[1] is not None or fp[2] is not None:
|
||||
# Missing, or a -wal/-journal with content: the header
|
||||
# check below would not describe the whole database
|
||||
continue
|
||||
if state.fp is None or state.fp[0] is None or state.fp[0][:2] != fp[0][:2]:
|
||||
# Replaced since the catalog was built
|
||||
continue
|
||||
if header_schema_version(os.fspath(state.db.path)) != state.catalog_version:
|
||||
# Changed by something other than Datasette while it ran:
|
||||
# the catalog is stale and must not be restored next time
|
||||
continue
|
||||
out.append(
|
||||
(state.name, state.catalog_version, _fp_to_json(fp, t_ns, closed=True))
|
||||
)
|
||||
return out
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# single-database freshness checks (permission decisions)
|
||||
# ------------------------------------------------------------------
|
||||
def catalog_is_current(self, state):
|
||||
"""True if this database's catalog rows provably describe it as it
|
||||
is right now - checked without opening a SQLite connection.
|
||||
|
||||
Used before trusting the catalog for a decision about one database
|
||||
(the derived-table permission rule), where a catalog that lags an
|
||||
out-of-band change could expose a table. Immutable databases cannot
|
||||
change. A file database is current if it is the same file the
|
||||
catalog was built from, has no -wal/-journal content, and the schema
|
||||
cookie in its header equals the catalog's version. Anything else
|
||||
(in-memory databases, live WAL content) returns False and the caller
|
||||
asks the database itself.
|
||||
"""
|
||||
if state.mode == "immutable":
|
||||
return True
|
||||
if (
|
||||
not state.is_file
|
||||
or state.catalog_version is None
|
||||
or state.needs_scan
|
||||
or state.missing
|
||||
):
|
||||
return False
|
||||
path = os.fspath(state.db.path)
|
||||
try:
|
||||
fp = fingerprint(path)
|
||||
except OSError:
|
||||
return False
|
||||
if fp[0] is None or fp[1] is not None or fp[2] is not None:
|
||||
return False
|
||||
prev = state.fp
|
||||
if prev is None or prev[0] is None or prev[0][:2] != fp[0][:2]:
|
||||
return False
|
||||
return header_schema_version(path) == state.catalog_version
|
||||
|
||||
def note_live_version(self, state, version):
|
||||
"""A caller read ``PRAGMA schema_version`` from the database itself.
|
||||
If the catalog is behind, schedule a rescan instead of waiting for
|
||||
the next write (owned) or poll (external)."""
|
||||
with self._lock:
|
||||
if (
|
||||
state.removed
|
||||
or state.error is not None
|
||||
or state.needs_scan
|
||||
or state.scan_future is not None
|
||||
or version == state.catalog_version
|
||||
):
|
||||
return
|
||||
state.needs_scan = True
|
||||
self._pending.add(state)
|
||||
if self.ds.internal_db_created:
|
||||
self._spawn(self.refresh([state]))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# introspection
|
||||
# ------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -108,6 +108,10 @@ class ScratchDatabaseError(Exception):
|
|||
"""A scratch database operation could not be carried out."""
|
||||
|
||||
|
||||
class ScratchDirectoryInUse(StartupError):
|
||||
"""Another Datasette instance holds the scratch directory's lock."""
|
||||
|
||||
|
||||
class ScratchDatabaseExists(ScratchDatabaseError):
|
||||
"""A database with that name already exists."""
|
||||
|
||||
|
|
@ -291,11 +295,19 @@ class ScratchDatabases:
|
|||
available as ``datasette._scratch``; plugins use the Datasette methods
|
||||
(``create_scratch_database()`` etc.)."""
|
||||
|
||||
def __init__(self, ds, directory=None):
|
||||
def __init__(self, ds, directory=None, *, inferred=False, fallback_dir=None):
|
||||
self.ds = ds
|
||||
self.configured_dir = (
|
||||
Path(directory).expanduser().resolve() if directory else None
|
||||
)
|
||||
# Configuration directory mode: config_dir/scratch, used only if
|
||||
# Datasette created it (it holds the registry) or it is empty - a
|
||||
# user's own "scratch" folder is never taken over
|
||||
self._inferred_dir = (
|
||||
Path(fallback_dir) if inferred and fallback_dir is not None else None
|
||||
)
|
||||
# Whether the registry already existed when the directory was opened
|
||||
self._registry_existed = False
|
||||
# The directory in use: configured_dir, or a temporary directory
|
||||
# created by the first create_scratch_database() call
|
||||
self.directory = None
|
||||
|
|
@ -331,12 +343,46 @@ class ScratchDatabases:
|
|||
def load(self):
|
||||
"""Called from Datasette.__init__. Attaches every scratch database
|
||||
in the configured directory without opening any of them."""
|
||||
if self.configured_dir is None:
|
||||
if self.configured_dir is None and self._inferred_dir is not None:
|
||||
directory = self._accept_inferred_dir(self._inferred_dir)
|
||||
if directory is None:
|
||||
return
|
||||
try:
|
||||
self._open_directory(directory)
|
||||
except ScratchDirectoryInUse as e:
|
||||
# Another Datasette (a server) is using the configuration
|
||||
# directory's scratch databases: this one (datasette ...
|
||||
# --get, a second process) runs without them rather than
|
||||
# failing to start
|
||||
logger.warning("%s; not attaching its scratch databases", e)
|
||||
return
|
||||
self.configured_dir = directory
|
||||
elif self.configured_dir is None:
|
||||
return
|
||||
self._open_directory(self.configured_dir)
|
||||
else:
|
||||
self._open_directory(self.configured_dir)
|
||||
for entry in self._reconcile():
|
||||
self._attach(entry)
|
||||
|
||||
@staticmethod
|
||||
def _accept_inferred_dir(path):
|
||||
if not path.is_dir():
|
||||
return None
|
||||
if (path / REGISTRY_FILENAME).exists():
|
||||
return path.resolve()
|
||||
try:
|
||||
empty = not os.listdir(path)
|
||||
except OSError:
|
||||
return None
|
||||
if empty:
|
||||
return path.resolve()
|
||||
logger.warning(
|
||||
"Not using %s for scratch databases: Datasette did not create it "
|
||||
"and it is not empty. Use --scratch-dir to use it anyway",
|
||||
path,
|
||||
)
|
||||
return None
|
||||
|
||||
def _open_directory(self, directory):
|
||||
directory = Path(directory)
|
||||
if any(c in str(directory) for c in UNSAFE_PATH_CHARS):
|
||||
|
|
@ -349,6 +395,7 @@ class ScratchDatabases:
|
|||
except OSError as e:
|
||||
raise StartupError(f"Cannot create scratch directory {directory}: {e}")
|
||||
self._acquire_directory_lock(directory)
|
||||
self._registry_existed = (directory / REGISTRY_FILENAME).exists()
|
||||
self.directory = str(directory)
|
||||
|
||||
def _acquire_directory_lock(self, directory):
|
||||
|
|
@ -362,7 +409,7 @@ class ScratchDatabases:
|
|||
except OSError as e:
|
||||
os.close(fd)
|
||||
if e.errno in (errno.EWOULDBLOCK, errno.EAGAIN, errno.EACCES):
|
||||
raise StartupError(
|
||||
raise ScratchDirectoryInUse(
|
||||
f"Scratch directory {directory} is in use by another Datasette instance"
|
||||
)
|
||||
# flock() not supported here (some network file systems,
|
||||
|
|
@ -467,7 +514,9 @@ class ScratchDatabases:
|
|||
entries[name] = _Entry(name, created)
|
||||
self.counters["adopted"] += 1
|
||||
# -wal/-shm/-journal files whose database is gone would be replayed
|
||||
# into the next database created under that name
|
||||
# into the next database created under that name. Only in a
|
||||
# directory Datasette has used before: never delete files the first
|
||||
# time a directory is opened
|
||||
files = set(os.listdir(directory))
|
||||
for filename in files:
|
||||
for suffix in SIDECAR_SUFFIXES:
|
||||
|
|
@ -476,6 +525,15 @@ class ScratchDatabases:
|
|||
and not filename.startswith(".")
|
||||
and filename[: -len(suffix)] not in files
|
||||
):
|
||||
if not self._registry_existed:
|
||||
logger.warning(
|
||||
"Scratch directory %s contains %s but not its "
|
||||
"database file; remove it before creating a "
|
||||
"scratch database with that name",
|
||||
directory,
|
||||
filename,
|
||||
)
|
||||
continue
|
||||
try:
|
||||
os.unlink(os.path.join(directory, filename))
|
||||
self.counters["stray_files_removed"] += 1
|
||||
|
|
@ -796,16 +854,9 @@ class ScratchDatabases:
|
|||
db._scratch_gone = reason
|
||||
state = db._watch_state
|
||||
if state is not None:
|
||||
fut = state.scan_future
|
||||
if fut is not None and not fut.done():
|
||||
try:
|
||||
same_loop = fut.get_loop() is asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
same_loop = False
|
||||
if same_loop:
|
||||
# A catalog scan in flight: let it finish, its rows are
|
||||
# deleted below
|
||||
await asyncio.wait([fut])
|
||||
# A catalog scan in flight (on any event loop): let it finish,
|
||||
# its rows are deleted below
|
||||
await self.ds._schema_watcher.wait_for_scan(state)
|
||||
if self.ds.databases.get(entry.name) is db:
|
||||
self.ds._detach_database(entry.name)
|
||||
await self._run_blocking(self._close_and_wait, db)
|
||||
|
|
|
|||
|
|
@ -262,9 +262,10 @@ These can be passed to ``datasette serve`` using ``datasette serve --setting nam
|
|||
schema_watch_interval_ms How often to check external database files for
|
||||
schema changes - 0 to disable polling
|
||||
(default=1000)
|
||||
default_schema_watch Schema watch mode for databases opened from
|
||||
files: external (poll for changes made by other
|
||||
processes) or owned (only Datasette changes them)
|
||||
default_schema_watch Schema watch mode for database files and named
|
||||
in-memory databases without a mode of their own:
|
||||
external (poll for changes made outside
|
||||
Datasette) or owned (only Datasette changes them)
|
||||
(default=external)
|
||||
sql_time_limit_ms Time limit for a SQL query in milliseconds
|
||||
(default=1000)
|
||||
|
|
|
|||
|
|
@ -1328,7 +1328,7 @@ Removes the column type assignment for the specified column.
|
|||
This will be used in the URL path. If not specified, it will default to the same thing as the ``name``.
|
||||
|
||||
``schema_watch`` - string, optional
|
||||
How Datasette keeps its catalog of this database's tables current: ``"owned"`` (the default - only Datasette's own writes change the schema), ``"external"`` (other processes may change the file too, so it is polled) or ``"immutable"``. A ``schema_watch`` setting for this database in ``datasette.yaml`` takes precedence. See :ref:`setting_schema_watch_interval_ms`.
|
||||
How Datasette keeps its catalog of this database's tables current: ``"owned"`` (only Datasette's own writes change the schema), ``"external"`` (other code may change it too - another process, or a plugin's own connection - so it is polled) or ``"immutable"``. The default for a database file or a named in-memory database is the :ref:`setting_default_schema_watch` setting, ``"external"`` unless configured. Pass ``"owned"`` if every change to the database goes through Datasette's write methods: it is then never polled. A ``schema_watch`` setting for this database in ``datasette.yaml`` takes precedence. See :ref:`setting_schema_watch_interval_ms`.
|
||||
|
||||
The ``datasette.add_database(db)`` method lets you add a new database to the current Datasette instance.
|
||||
|
||||
|
|
@ -1405,10 +1405,10 @@ A *scratch database* is a SQLite file that a plugin (or a user, through a plugin
|
|||
They live in the **scratch directory**:
|
||||
|
||||
- ``datasette --scratch-dir path/to/scratch`` (or the ``DATASETTE_SCRATCH_DIR`` environment variable, or ``Datasette(scratch_dir=...)``). The directory is created if it does not exist.
|
||||
- In :ref:`config_dir`, a ``scratch/`` directory inside the configuration directory is used if it exists.
|
||||
- In :ref:`config_dir`, a ``scratch/`` directory inside the configuration directory is used if it exists and either is empty or was created by Datasette (it contains ``.datasette-scratch.db``). A ``scratch/`` folder of your own that holds other files is left alone, with a warning in the log. If another Datasette instance is already using it - a server, while you run ``datasette mydir --get ...`` - the second instance starts without its scratch databases, with a warning.
|
||||
- Otherwise scratch databases are **temporary**: they go in a temporary directory that is created the first time a plugin creates one and deleted when Datasette shuts down, the same way the internal database is temporary unless ``--internal`` is used. ``datasette.scratch_dir`` is ``None`` until then. Plugins can use the same API either way.
|
||||
|
||||
Use a directory just for scratch databases. Only one Datasette instance can use a scratch directory at a time: a second one fails to start with an error.
|
||||
Use a directory just for scratch databases. Only one Datasette instance can use a scratch directory at a time: a second one given the same ``--scratch-dir`` fails to start with an error. The first time Datasette opens a directory it adopts any SQLite files already in it, but never deletes anything: leftover ``-wal``, ``-shm`` or ``-journal`` files without their database are only removed once the directory holds Datasette's registry.
|
||||
|
||||
The directory contains one ``<name>.db`` file per scratch database, plus a registry (``.datasette-scratch.db``) that records each database's owner, metadata, creation time and when it was last used. Scratch databases are created in WAL mode (rollback journal mode with ``--nolock``).
|
||||
|
||||
|
|
@ -2233,7 +2233,9 @@ Connections are only valid inside your function
|
|||
|
||||
The ``conn`` passed to an ``execute_fn()``, ``execute_write_fn()`` or ``execute_isolated_fn()`` function - and to :ref:`plugin_hook_write_wrapper` hooks - is lent to that function for the duration of the call. It behaves like a ``sqlite3.Connection`` (``isinstance(conn, sqlite3.Connection)`` is ``True`` and it works with ``sqlite_utils.Database(conn)``), but once the function returns it stops working: any further use raises ``datasette.database.ConnectionLeaseError``. Do not store the connection on an object, in a global or in a closure for later use. Do everything you need inside the function and return plain Python values.
|
||||
|
||||
``execute_fn()`` also raises ``ConnectionLeaseError`` if your function returns a cursor or a generator, since those would keep reading from the connection after it had been returned to the pool.
|
||||
``execute_fn()`` also raises ``ConnectionLeaseError`` if your function returns a cursor, or a generator that holds the connection or a cursor (for example ``(row for row in conn.execute(...))``), since those would keep reading from the connection after it had been returned to the pool. A generator over rows already fetched inside the function is fine.
|
||||
|
||||
The write connection belongs to Datasette: calling ``conn.close()`` inside an ``execute_write_fn()`` function raises ``ConnectionLeaseError``. Closing the connection passed to an ``execute_fn()`` or ``execute_isolated_fn()`` function is allowed.
|
||||
|
||||
Datasette cannot detect every misuse. These are unsupported and may fail unpredictably, or read from a connection that is now being used by someone else:
|
||||
|
||||
|
|
@ -2366,7 +2368,7 @@ Example usage:
|
|||
|
||||
version = await db.execute_fn(get_version)
|
||||
|
||||
The connection is borrowed from a pool for the duration of the call (see :ref:`database_connections`) and is only valid inside the function. Using it after the function has returned - for example by storing it on an object - raises a ``datasette.database.ConnectionLeaseError``. Returning a cursor or a generator from the function raises the same error: fetch the rows you need before returning. See :ref:`database_connection_leases`.
|
||||
The connection is borrowed from a pool for the duration of the call (see :ref:`database_connections`) and is only valid inside the function. Using it after the function has returned - for example by storing it on an object - raises a ``datasette.database.ConnectionLeaseError``. Returning a cursor, or a generator that is still reading from the connection, raises the same error: fetch the rows you need before returning. See :ref:`database_connection_leases`.
|
||||
|
||||
Calls to ``execute_fn()`` are not guaranteed to use the same connection, so temporary tables, ``ATTACH`` statements and functions created inside one call may not be visible in the next.
|
||||
|
||||
|
|
@ -2455,7 +2457,9 @@ This method works like ``.execute_write()``, but instead of a SQL statement you
|
|||
|
||||
The function can then perform multiple actions, safe in the knowledge that it has exclusive access to the single writable connection for as long as it is executing.
|
||||
|
||||
The connection is only valid while your function is running: using it after the function has returned raises ``datasette.database.ConnectionLeaseError`` - see :ref:`database_connection_leases`. The write connection is closed after :ref:`setting_connection_idle_timeout_ms` without writes, so temporary tables, ``ATTACH`` statements and functions created by one write function may be gone by the next.
|
||||
The connection is only valid while your function is running: using it after the function has returned raises ``datasette.database.ConnectionLeaseError`` - see :ref:`database_connection_leases`. The write connection is closed after :ref:`setting_connection_idle_timeout_ms` without writes, so temporary tables, ``ATTACH`` statements and functions created by one write function may be gone by the next. Every new write connection has ``PRAGMA recursive_triggers`` turned on (before any :ref:`plugin_hook_prepare_connection` hook runs), which is what ``sqlite-utils`` sets and what the triggers it creates to keep full-text search indexes current rely on for ``INSERT OR REPLACE``.
|
||||
|
||||
If your function raises an exception, ``execute_write_fn()`` raises it too and the transaction is rolled back. An exception that is not a subclass of ``Exception``, such as ``SystemExit``, is raised as a ``RuntimeError`` whose ``__cause__`` is the original exception.
|
||||
|
||||
Like ``execute_fn()``, the call is traced as a ``db.query`` OpenTelemetry span carrying ``datasette.callback`` rather than ``db.query.text``, above the write-queue spans - see :ref:`internals_telemetry`. A named function gives the span a readable identity; a lambda reports ``<lambda>``.
|
||||
|
||||
|
|
|
|||
|
|
@ -240,7 +240,7 @@ Set this to 0 for no limit, in which case idle connections are only closed by :r
|
|||
|
||||
datasette *.db --setting max_open_connections 64
|
||||
|
||||
The cap does not cover write connections (one per database that is being written to, see :ref:`setting_connection_idle_timeout_ms`), the short-lived connections used by :ref:`database_execute_isolated_fn` and by schema checks, or in-memory databases, which use no file descriptors.
|
||||
The cap does not cover write connections (one per database that is being written to, see :ref:`setting_connection_idle_timeout_ms`), the short-lived connections used by :ref:`database_execute_isolated_fn` and by schema checks, or in-memory databases, which use no file descriptors. The exception is the ``_memory`` database under ``--crossdb``: each of its connections ``ATTACH``\ es up to ten database files, so those connections count against the cap and are closed when idle like file connections. Adding or removing a database closes them too, so the next cross-database query sees the current set of databases.
|
||||
|
||||
Each pooled connection to a file in WAL mode uses three file descriptors (the database, its ``-wal`` and its ``-shm`` file); other databases use one. Under a small file descriptor limit such as ``ulimit -n 256``, a cap of 64 keeps pooled read connections below about 200 descriptors even if every database uses WAL, leaving room for write connections and Datasette itself.
|
||||
|
||||
|
|
@ -298,7 +298,7 @@ Set this to ``0`` to turn polling off. External databases are then only checked
|
|||
|
||||
datasette mydatabase.db --setting schema_watch_interval_ms 0
|
||||
|
||||
Databases opened from files use the :ref:`setting_default_schema_watch` mode. Databases added by plugins with :ref:`datasette.add_database() <datasette_add_database>` are ``owned`` unless the plugin passes ``schema_watch="external"``. You can set the mode for an individual database in ``datasette.yaml``, which overrides both:
|
||||
Database files, and named in-memory databases, use the :ref:`setting_default_schema_watch` mode - including those added by plugins with :ref:`datasette.add_database() <datasette_add_database>`, unless the plugin passes ``schema_watch="owned"`` (or ``"external"``). A named in-memory database in ``external`` mode is checked with ``PRAGMA schema_version`` on its write connection every ``schema_watch_interval_ms``, which notices tables created through a plugin's own connection to it. You can set the mode for an individual database in ``datasette.yaml``, which overrides both:
|
||||
|
||||
.. code-block:: yaml
|
||||
|
||||
|
|
@ -319,7 +319,7 @@ Default: ``external``
|
|||
|
||||
.. [[[end]]]
|
||||
|
||||
The schema watch mode, ``external`` or ``owned``, for mutable database files passed on the command line or with ``Datasette(files=...)`` - see :ref:`setting_schema_watch_interval_ms`.
|
||||
The schema watch mode, ``external`` or ``owned``, for mutable database files and named in-memory databases that do not have a mode of their own: files passed on the command line or with ``Datasette(files=...)``, and databases added with :ref:`datasette.add_database() <datasette_add_database>` without a ``schema_watch`` argument - see :ref:`setting_schema_watch_interval_ms`.
|
||||
|
||||
The default, ``external``, suits the common workflow of changing a database with another tool such as ``sqlite-utils`` while Datasette is serving it: new tables and columns show up within about a second. If Datasette is the only thing that ever changes your database files, set this to ``owned`` and they will not be polled. This is worth doing for instances with thousands of databases, where polling costs a few milliseconds of CPU per second::
|
||||
|
||||
|
|
|
|||
746
tests/test_connection_review_fixes.py
Normal file
746
tests/test_connection_review_fixes.py
Normal file
|
|
@ -0,0 +1,746 @@
|
|||
"""
|
||||
Regression tests for problems found by the second review of the connection
|
||||
management redesign (concurrency, lifecycle and public-contract findings).
|
||||
Each test names the problem it pins down.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import gc
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
import weakref
|
||||
|
||||
import pytest
|
||||
|
||||
from datasette import schema_watcher as schema_watcher_module
|
||||
from datasette.app import Datasette
|
||||
from datasette.connection_pool import ConnectionLeaseError
|
||||
from datasette.database import Database
|
||||
from datasette.scratch import LOCK_FILENAME, REGISTRY_FILENAME
|
||||
|
||||
|
||||
def make_db(path, sql="create table t (id integer primary key, v text)"):
|
||||
conn = sqlite3.connect(path)
|
||||
conn.executescript(sql)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
async def catalog_tables(ds, name):
|
||||
rows = await ds.get_internal_database().execute(
|
||||
"select table_name from catalog_tables where database_name = ? "
|
||||
"order by table_name",
|
||||
[name],
|
||||
)
|
||||
return [r[0] for r in rows.rows]
|
||||
|
||||
|
||||
def live_tables(path):
|
||||
conn = sqlite3.connect(path)
|
||||
try:
|
||||
return sorted(
|
||||
r[0]
|
||||
for r in conn.execute(
|
||||
"select name from sqlite_master where type = 'table' "
|
||||
"and name not like 'sqlite_%'"
|
||||
)
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def run_in_threads(*coroutine_functions, timeout=20):
|
||||
"""Run each coroutine function with asyncio.run() in its own thread:
|
||||
several event loops driving one Datasette at the same time."""
|
||||
results = [None] * len(coroutine_functions)
|
||||
|
||||
def runner(i, fn):
|
||||
try:
|
||||
results[i] = ("ok", asyncio.run(fn()))
|
||||
except BaseException as e: # noqa: BLE001
|
||||
results[i] = ("error", e)
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=runner, args=(i, fn), daemon=True)
|
||||
for i, fn in enumerate(coroutine_functions)
|
||||
]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
deadline = time.monotonic() + timeout
|
||||
for thread in threads:
|
||||
thread.join(max(0.0, deadline - time.monotonic()))
|
||||
hung = [i for i, thread in enumerate(threads) if thread.is_alive()]
|
||||
assert not hung, f"event loop threads {hung} did not finish within {timeout}s"
|
||||
for status, value in results:
|
||||
if status == "error":
|
||||
raise value
|
||||
return [value for _, value in results]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def owned_db(tmp_path):
|
||||
path = str(tmp_path / "data.db")
|
||||
make_db(path)
|
||||
ds = Datasette(settings={"schema_watch_interval_ms": 0})
|
||||
db = ds.add_database(Database(ds, path=path), name="data", schema_watch="owned")
|
||||
asyncio.run(ds.invoke_startup())
|
||||
yield ds, db, path
|
||||
ds.close()
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Concurrency and lifecycle
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cross_loop_ddl_writes_do_not_hang(owned_db, monkeypatch):
|
||||
# C1: refresh() treated a scan started on another *live* event loop as
|
||||
# abandoned, stole it, and each loop then resolved or cleared the
|
||||
# other's future: blocking writes on one loop never returned
|
||||
ds, db, path = owned_db
|
||||
original = schema_watcher_module.SchemaWatcher._scan_chunk_sync
|
||||
|
||||
def slow_scan(self, states):
|
||||
time.sleep(0.2)
|
||||
return original(self, states)
|
||||
|
||||
monkeypatch.setattr(
|
||||
schema_watcher_module.SchemaWatcher, "_scan_chunk_sync", slow_scan
|
||||
)
|
||||
|
||||
def writer(prefix, count):
|
||||
async def run():
|
||||
for i in range(count):
|
||||
await asyncio.wait_for(
|
||||
db.execute_write(f"create table {prefix}{i} (id integer)"), 10
|
||||
)
|
||||
|
||||
return run
|
||||
|
||||
def inserter():
|
||||
async def run():
|
||||
async def worker():
|
||||
for _ in range(20):
|
||||
await asyncio.wait_for(
|
||||
db.execute_write("insert into t (v) values ('x')"), 10
|
||||
)
|
||||
|
||||
await asyncio.gather(*(worker() for _ in range(3)))
|
||||
|
||||
return run
|
||||
|
||||
run_in_threads(writer("a", 3), writer("b", 3), inserter())
|
||||
|
||||
async def check():
|
||||
return await catalog_tables(ds, "data")
|
||||
|
||||
assert asyncio.run(check()) == live_tables(path)
|
||||
state = db._watch_state
|
||||
assert state.scan_future is None
|
||||
assert not state.needs_scan
|
||||
|
||||
|
||||
def test_scan_from_stopped_loop_is_taken_over(owned_db):
|
||||
# The rule for taking over a scan: its loop has closed or stopped, so
|
||||
# nothing will finish it. A loop stopped by run_until_complete()
|
||||
# returning while a scan task is pending must not hang other loops.
|
||||
ds, db, _path = owned_db
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
gate = threading.Event()
|
||||
original = schema_watcher_module.SchemaWatcher._scan_chunk_sync
|
||||
|
||||
def gated_scan(self, states):
|
||||
gate.wait(5)
|
||||
return original(self, states)
|
||||
|
||||
schema_watcher_module.SchemaWatcher._scan_chunk_sync = gated_scan
|
||||
try:
|
||||
|
||||
async def start_scan():
|
||||
# block=False: the scan task is spawned on this loop and
|
||||
# left pending when run_until_complete() returns
|
||||
await db.execute_write(
|
||||
"create table stopped1 (id integer)", block=False
|
||||
)
|
||||
await asyncio.sleep(0.2)
|
||||
|
||||
loop.run_until_complete(start_scan())
|
||||
finally:
|
||||
schema_watcher_module.SchemaWatcher._scan_chunk_sync = original
|
||||
gate.set()
|
||||
|
||||
async def other_loop():
|
||||
await asyncio.wait_for(
|
||||
db.execute_write("create table other1 (id integer)"), 10
|
||||
)
|
||||
return await catalog_tables(ds, "data")
|
||||
|
||||
tables = run_in_threads(other_loop)[0]
|
||||
assert {"stopped1", "other1"} <= set(tables)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_one_poller_across_live_event_loops(tmp_path):
|
||||
# C2: every switch between two live event loops started another
|
||||
# poller task, each sweeping every interval
|
||||
path = str(tmp_path / "ext.db")
|
||||
make_db(path)
|
||||
ds = Datasette([path], settings={"schema_watch_interval_ms": 100})
|
||||
asyncio.run(ds.invoke_startup())
|
||||
watcher = ds._schema_watcher
|
||||
turns = [threading.Event(), threading.Event()]
|
||||
done = threading.Event()
|
||||
counts = {}
|
||||
|
||||
def client(me, other, rounds):
|
||||
async def run():
|
||||
for _ in range(rounds):
|
||||
turns[me].wait(5)
|
||||
turns[me].clear()
|
||||
assert (await ds.client.get("/ext.json")).status_code == 200
|
||||
turns[other].set()
|
||||
if me == 0:
|
||||
sweeps = watcher.counters["sweeps"]
|
||||
await asyncio.sleep(1.0)
|
||||
counts["sweeps_per_s"] = watcher.counters["sweeps"] - sweeps
|
||||
counts["pollers"] = len([t for t in watcher._pollers if not t.done()])
|
||||
done.set()
|
||||
else:
|
||||
done.wait(10)
|
||||
|
||||
return run
|
||||
|
||||
turns[0].set()
|
||||
run_in_threads(client(0, 1, 10), client(1, 0, 10))
|
||||
assert counts["pollers"] == 1
|
||||
# One poller at a 100ms interval: about 10 sweeps a second, not 10 per
|
||||
# loop switch
|
||||
assert counts["sweeps_per_s"] <= 15
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_catalog_write_is_retried(owned_db, monkeypatch):
|
||||
# C3: refresh() cleared needs_scan before scanning and never restored
|
||||
# it, so an owned database lost a catalog update for good if the
|
||||
# catalog write failed (internal database locked)
|
||||
ds, db, _path = owned_db
|
||||
original = schema_watcher_module.write_catalog_entries
|
||||
locked = threading.Event()
|
||||
locked.set()
|
||||
|
||||
def flaky(conn, entries):
|
||||
if locked.is_set():
|
||||
raise sqlite3.OperationalError("database is locked")
|
||||
return original(conn, entries)
|
||||
|
||||
monkeypatch.setattr(schema_watcher_module, "write_catalog_entries", flaky)
|
||||
await db.execute_write("create table added_while_locked (id integer)")
|
||||
# Let the refresh task spawned by the write's notification fail too
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(0.01)
|
||||
state = db._watch_state
|
||||
assert state.needs_scan
|
||||
assert state in ds._schema_watcher._pending
|
||||
assert await catalog_tables(ds, "data") == ["t"]
|
||||
# While still locked, a request is not failed by the catalog update
|
||||
state.retry_at = 0.0
|
||||
assert (await ds.client.get("/data.json")).status_code == 200
|
||||
assert state.retry_at > time.monotonic()
|
||||
# Once unlocked, the next request retries (after the back-off)
|
||||
locked.clear()
|
||||
state.retry_at = 0.0
|
||||
await ds.client.get("/data.json")
|
||||
assert await catalog_tables(ds, "data") == ["added_while_locked", "t"]
|
||||
response = await ds.client.get("/data.json")
|
||||
assert {t["name"] for t in response.json()["tables"]} == {
|
||||
"added_while_locked",
|
||||
"t",
|
||||
}
|
||||
|
||||
|
||||
def test_block_false_write_from_closed_loop_reaches_catalog(owned_db):
|
||||
# C3: the schema-change notification went to the caller's loop, which
|
||||
# had closed: notified_version was recorded anyway and the change was
|
||||
# never picked up
|
||||
ds, db, path = owned_db
|
||||
|
||||
async def write():
|
||||
await db.execute_write(
|
||||
"create table added_nonblocking (id integer)", block=False
|
||||
)
|
||||
|
||||
asyncio.run(write())
|
||||
# Let the write thread finish (it delivers to a closed loop)
|
||||
deadline = time.monotonic() + 5
|
||||
while "added_nonblocking" not in live_tables(path):
|
||||
assert time.monotonic() < deadline
|
||||
time.sleep(0.01)
|
||||
time.sleep(0.05)
|
||||
|
||||
async def later():
|
||||
await ds.client.get("/data.json")
|
||||
return await catalog_tables(ds, "data")
|
||||
|
||||
assert asyncio.run(later()) == ["added_nonblocking", "t"]
|
||||
|
||||
|
||||
def test_unclosed_datasette_releases_scratch_directory(tmp_path):
|
||||
# C4: the temp internal database's atexit.register(bound method) pinned
|
||||
# every Datasette, so the scratch directory lock was never released
|
||||
scratch = tmp_path / "scratch"
|
||||
ds = Datasette(scratch_dir=str(scratch))
|
||||
ref = weakref.ref(ds)
|
||||
del ds
|
||||
gc.collect()
|
||||
assert ref() is None
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
|
||||
|
||||
def test_failed_construction_releases_scratch_directory(tmp_path):
|
||||
# C4: a Datasette() whose __init__ raised after loading the scratch
|
||||
# directory kept it locked, with no object to close()
|
||||
from datasette.utils import StartupError
|
||||
|
||||
scratch = tmp_path / "scratch"
|
||||
with pytest.raises(StartupError):
|
||||
Datasette(scratch_dir=str(scratch), settings={"default_schema_watch": "bogus"})
|
||||
# No gc.collect(): the lock is taken last, after validation
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
|
||||
|
||||
def test_closed_datasette_is_collected(tmp_path):
|
||||
# C4 + C5: atexit and the read-pool reaper's loop variable both kept a
|
||||
# closed Datasette (and every Database, cache and plugin state) alive
|
||||
path = str(tmp_path / "data.db")
|
||||
make_db(path)
|
||||
ds = Datasette([path], settings={"connection_idle_timeout_ms": 20})
|
||||
|
||||
async def use(ds):
|
||||
await ds.invoke_startup()
|
||||
await ds.get_database("data").execute("select 1")
|
||||
await ds.get_database("data").execute_write("insert into t (v) values (1)")
|
||||
|
||||
asyncio.run(use(ds))
|
||||
# Let the reaper scan (and close) the idle connection
|
||||
time.sleep(0.3)
|
||||
ds.close()
|
||||
ref = weakref.ref(ds)
|
||||
del ds
|
||||
gc.collect()
|
||||
assert ref() is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_callback_cannot_close_write_connection(owned_db):
|
||||
# C6: closing the write lease closed the write thread's connection;
|
||||
# every later write failed until the thread idled out
|
||||
_ds, db, _path = owned_db
|
||||
|
||||
def close_it(conn):
|
||||
conn.close()
|
||||
|
||||
with pytest.raises(ConnectionLeaseError):
|
||||
await db.execute_write_fn(close_it)
|
||||
await db.execute_write("insert into t (v) values ('after')")
|
||||
|
||||
# Reaching the raw connection through a cursor still closes it: the
|
||||
# write thread notices and reopens before the next write
|
||||
def close_raw(conn):
|
||||
conn.execute("select 1").connection.close()
|
||||
|
||||
with pytest.raises(sqlite3.ProgrammingError):
|
||||
await db.execute_write_fn(close_raw)
|
||||
for _ in range(3):
|
||||
await db.execute_write("insert into t (v) values ('again')")
|
||||
rows = await db.execute("select count(*) from t")
|
||||
assert rows.first()[0] == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("num_sql_threads", (3, 0))
|
||||
async def test_isolated_lease_close_is_allowed(tmp_path, num_sql_threads):
|
||||
path = str(tmp_path / "data.db")
|
||||
make_db(path)
|
||||
ds = Datasette([path], settings={"num_sql_threads": num_sql_threads})
|
||||
db = ds.get_database("data")
|
||||
await db.execute_isolated_fn(lambda conn: conn.close())
|
||||
await db.execute_write("insert into t (v) values (1)")
|
||||
ds.close()
|
||||
|
||||
|
||||
class Abort(BaseException):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("exception_class", (Abort, SystemExit))
|
||||
async def test_write_callback_base_exception_is_delivered(owned_db, exception_class):
|
||||
# C7: a BaseException from a write callback killed the write thread
|
||||
# without answering the caller, who waited forever
|
||||
_ds, db, _path = owned_db
|
||||
|
||||
def boom(conn):
|
||||
raise exception_class("stop")
|
||||
|
||||
with pytest.raises(RuntimeError) as excinfo:
|
||||
await asyncio.wait_for(db.execute_write_fn(boom), 5)
|
||||
assert isinstance(excinfo.value.__cause__, exception_class)
|
||||
# Same for an isolated function
|
||||
with pytest.raises(RuntimeError):
|
||||
await asyncio.wait_for(db.execute_isolated_fn(boom), 5)
|
||||
await asyncio.wait_for(db.execute_write("insert into t (v) values (1)"), 5)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Public contract
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def _fts_setup(path):
|
||||
make_db(path, "create table secret (id integer primary key, body text)")
|
||||
conn = sqlite3.connect(path)
|
||||
conn.execute("insert into secret values (1, 'TOP SECRET PAYLOAD')")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def _add_fts(path):
|
||||
conn = sqlite3.connect(path)
|
||||
conn.executescript("""
|
||||
create virtual table secret_fts using fts5(body, content='secret', content_rowid='id');
|
||||
insert into secret_fts(secret_fts) values ('rebuild');
|
||||
""")
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("mode", ("external", "owned"))
|
||||
async def test_derived_table_rule_uses_current_schema(tmp_path, mode):
|
||||
# Contract C1: the "external-content FTS tables require access to their
|
||||
# content table" rule was decided from a catalog that had not seen an
|
||||
# FTS table created outside Datasette: its rows were served to
|
||||
# anonymous users (until the next poll, or for good for owned DBs)
|
||||
path = str(tmp_path / "data.db")
|
||||
_fts_setup(path)
|
||||
ds = Datasette(
|
||||
config={
|
||||
"databases": {"data": {"tables": {"secret": {"allow": {"id": "root"}}}}}
|
||||
},
|
||||
settings={"default_allow_sql": False, "schema_watch_interval_ms": 0},
|
||||
)
|
||||
ds.add_database(Database(ds, path=path), name="data", schema_watch=mode)
|
||||
await ds.invoke_startup()
|
||||
assert (await ds.client.get("/data/secret.json")).status_code == 403
|
||||
_add_fts(path)
|
||||
response = await ds.client.get("/data/secret_fts.json")
|
||||
assert response.status_code == 403
|
||||
assert "TOP SECRET" not in response.text
|
||||
# Finding the catalog behind schedules a rescan
|
||||
await ds.client.get("/data.json")
|
||||
assert "secret_fts" in await catalog_tables(ds, "data")
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_database_file_changes_are_picked_up(tmp_path):
|
||||
# Contract C2: add_database() defaulted to "owned" (never polled), so a
|
||||
# table created by another process never appeared in listings
|
||||
path = str(tmp_path / "added.db")
|
||||
make_db(path)
|
||||
ds = Datasette(settings={"schema_watch_interval_ms": 50})
|
||||
await ds.invoke_startup()
|
||||
ds.add_database(Database(ds, path=path), name="added")
|
||||
await ds.client.get("/added.json")
|
||||
make_db(path, "create table t_ext (id integer)")
|
||||
|
||||
async def listed():
|
||||
response = await ds.client.get("/added.json")
|
||||
return {t["name"] for t in response.json()["tables"]}
|
||||
|
||||
deadline = time.monotonic() + 5
|
||||
while "t_ext" not in await listed():
|
||||
assert time.monotonic() < deadline, "t_ext never appeared"
|
||||
await asyncio.sleep(0.05)
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_named_memory_database_out_of_band_changes(tmp_path):
|
||||
# Contract C2: named in-memory databases could not be watched at all;
|
||||
# datasette-app-support restores one with backup() into
|
||||
# db.connect(write=True) and then shows it
|
||||
ds = Datasette(settings={"schema_watch_interval_ms": 50})
|
||||
await ds.invoke_startup()
|
||||
memory = ds.add_memory_database("review_fixes_temporary", name="temporary")
|
||||
await memory.execute_write("create table m0 (id integer)")
|
||||
source_path = str(tmp_path / "backup.db")
|
||||
make_db(source_path, "create table restored_table (id integer)")
|
||||
source = sqlite3.connect(source_path)
|
||||
target = memory.connect(write=True)
|
||||
source.backup(target)
|
||||
source.close()
|
||||
target.close()
|
||||
|
||||
async def listed():
|
||||
response = await ds.client.get("/temporary.json")
|
||||
return {t["name"] for t in response.json()["tables"]}
|
||||
|
||||
deadline = time.monotonic() + 5
|
||||
while "restored_table" not in await listed():
|
||||
assert time.monotonic() < deadline, "restored_table never appeared"
|
||||
await asyncio.sleep(0.05)
|
||||
ds.close()
|
||||
|
||||
|
||||
def _config_dir(tmp_path):
|
||||
config_dir = tmp_path / "cfg"
|
||||
(config_dir / "scratch").mkdir(parents=True)
|
||||
make_db(str(config_dir / "main.db"))
|
||||
return config_dir
|
||||
|
||||
|
||||
def test_config_dir_user_scratch_folder_is_not_taken_over(tmp_path):
|
||||
# Contract C3: an existing config_dir/scratch folder was adopted:
|
||||
# its files were served, orphan sidecars deleted and the folder locked
|
||||
config_dir = _config_dir(tmp_path)
|
||||
make_db(str(config_dir / "scratch" / "private_notes.db"))
|
||||
(config_dir / "scratch" / "other.db-wal").write_bytes(b"junk")
|
||||
ds = Datasette(config_dir=config_dir)
|
||||
assert "private_notes" not in ds.databases
|
||||
assert ds.scratch_dir is None
|
||||
ds.close()
|
||||
assert sorted(os.listdir(config_dir / "scratch")) == [
|
||||
"other.db-wal",
|
||||
"private_notes.db",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_config_dir_scratch_shared_by_two_instances(tmp_path):
|
||||
# Contract C3: a Datasette-created config_dir/scratch locked out a
|
||||
# second process on the same config dir (datasette cfg --get ...)
|
||||
config_dir = _config_dir(tmp_path)
|
||||
first = Datasette(config_dir=config_dir)
|
||||
assert first.scratch_dir == str((config_dir / "scratch").resolve())
|
||||
await first.create_scratch_database("notes")
|
||||
assert (config_dir / "scratch" / REGISTRY_FILENAME).exists()
|
||||
assert (config_dir / "scratch" / LOCK_FILENAME).exists()
|
||||
second = Datasette(config_dir=config_dir)
|
||||
assert "notes" not in second.databases
|
||||
second.close()
|
||||
first.close()
|
||||
# Once the first has gone, the directory is used again
|
||||
third = Datasette(config_dir=config_dir)
|
||||
assert "notes" in third.databases
|
||||
third.close()
|
||||
|
||||
|
||||
def test_first_use_of_explicit_scratch_dir_keeps_sidecars(tmp_path, caplog):
|
||||
scratch = tmp_path / "scratch"
|
||||
scratch.mkdir()
|
||||
(scratch / "gone.db-wal").write_bytes(b"junk")
|
||||
with caplog.at_level(logging.WARNING, logger="datasette.scratch"):
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
assert (scratch / "gone.db-wal").exists()
|
||||
assert "gone.db-wal" in caplog.text
|
||||
# The next start (the registry exists now) removes it
|
||||
Datasette(scratch_dir=str(scratch)).close()
|
||||
assert not (scratch / "gone.db-wal").exists()
|
||||
|
||||
|
||||
FTS_SETUP = """
|
||||
create table docs (id integer primary key, body text);
|
||||
create virtual table docs_fts using fts5(body, content='docs', content_rowid='id');
|
||||
create trigger docs_ai after insert on docs begin
|
||||
insert into docs_fts (rowid, body) values (new.id, new.body);
|
||||
end;
|
||||
create trigger docs_ad after delete on docs begin
|
||||
insert into docs_fts (docs_fts, rowid, body) values ('delete', old.id, old.body);
|
||||
end;
|
||||
insert into docs (id, body) values (1, 'oldword');
|
||||
"""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("num_sql_threads", (3, 0))
|
||||
async def test_write_connection_pins_recursive_triggers(tmp_path, num_sql_threads):
|
||||
# Contract C4: INSERT OR REPLACE on a sqlite-utils style FTS table only
|
||||
# cleaned up the old index entry once some code had wrapped the write
|
||||
# connection in sqlite_utils.Database (recursive_triggers drift) - and
|
||||
# the idle write thread's fresh connections lost the drift again
|
||||
path = str(tmp_path / "fts.db")
|
||||
make_db(path, FTS_SETUP)
|
||||
ds = Datasette(
|
||||
[path],
|
||||
settings={
|
||||
"num_sql_threads": num_sql_threads,
|
||||
"connection_idle_timeout_ms": 50,
|
||||
},
|
||||
)
|
||||
db = ds.get_database("fts")
|
||||
for word in ("newword", "newerword"):
|
||||
flag = await db.execute_write_fn(
|
||||
lambda conn: conn.execute("PRAGMA recursive_triggers").fetchone()[0]
|
||||
)
|
||||
assert flag == 1
|
||||
await db.execute_write(
|
||||
"insert or replace into docs (id, body) values (1, ?)", [word]
|
||||
)
|
||||
# Let the idle write thread exit: the next write opens a new
|
||||
# connection
|
||||
await asyncio.sleep(0.3)
|
||||
stale = await db.execute(
|
||||
"select rowid from docs_fts where docs_fts match 'oldword OR newword'"
|
||||
)
|
||||
assert stale.rows == []
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_new_memory_database_is_not_scanned_at_registration():
|
||||
# Contract C5: add_memory_database() after startup scanned the
|
||||
# shared-cache database straight away, racing plugins that fill it from
|
||||
# their own connection (datasette-copy-to-memory's VACUUM INTO failed
|
||||
# with "database table is locked")
|
||||
ds = Datasette(settings={"schema_watch_interval_ms": 0})
|
||||
await ds.invoke_startup()
|
||||
watcher = ds._schema_watcher
|
||||
scans = watcher.counters["scans"]
|
||||
memory = ds.add_memory_database("review_fixes_copy", name="copy")
|
||||
await memory.execute("select 1 + 1")
|
||||
await asyncio.sleep(0.05)
|
||||
assert watcher.counters["scans"] == scans
|
||||
state = memory._watch_state
|
||||
assert state.scan_future is None
|
||||
assert state in watcher._pending
|
||||
# Fill it the way the plugin does
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.execute("create table copied (id integer)")
|
||||
conn.execute("VACUUM INTO 'file:review_fixes_copy?mode=memory&cache=shared'")
|
||||
conn.close()
|
||||
# The next request scans it
|
||||
response = await ds.client.get("/copy.json")
|
||||
assert {t["name"] for t in response.json()["tables"]} == {"copied"}
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_locked_memory_scan_is_retried(monkeypatch):
|
||||
ds = Datasette(settings={"schema_watch_interval_ms": 0})
|
||||
await ds.invoke_startup()
|
||||
memory = ds.add_memory_database("review_fixes_locked", name="locked")
|
||||
original = schema_watcher_module.collect_schema
|
||||
locked = threading.Event()
|
||||
locked.set()
|
||||
|
||||
def locked_scan(conn, name):
|
||||
if locked.is_set():
|
||||
raise sqlite3.OperationalError("database table is locked: sqlite_master")
|
||||
return original(conn, name)
|
||||
|
||||
monkeypatch.setattr(schema_watcher_module, "collect_schema", locked_scan)
|
||||
await memory.execute_write("create table m (id integer)")
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(0.01)
|
||||
state = memory._watch_state
|
||||
# Transient: not recorded as a broken database
|
||||
assert state.error is None
|
||||
assert state.needs_scan
|
||||
locked.clear()
|
||||
state.retry_at = 0.0
|
||||
await ds.client.get("/locked.json")
|
||||
assert await catalog_tables(ds, "locked") == ["m"]
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stale_owned_catalog_is_not_persisted_at_close(tmp_path):
|
||||
# Contract C6: close() stored a fresh fingerprint for an owned database
|
||||
# whose catalog was stale, so a restart restored the stale catalog
|
||||
path = str(tmp_path / "data.db")
|
||||
internal = str(tmp_path / "internal.db")
|
||||
make_db(path)
|
||||
config = {"databases": {"data": {"schema_watch": "owned"}}}
|
||||
|
||||
async def tables_after_start(change=None):
|
||||
ds = Datasette([path], internal=internal, config=config)
|
||||
await ds.invoke_startup()
|
||||
if change:
|
||||
change()
|
||||
response = await ds.client.get("/data.json")
|
||||
ds.close()
|
||||
return [t["name"] for t in response.json()["tables"]]
|
||||
|
||||
assert await tables_after_start() == ["t"]
|
||||
# Owned: a change made behind Datasette's back is not noticed while
|
||||
# it runs (documented) ...
|
||||
assert await tables_after_start(
|
||||
lambda: make_db(path, "create table t2 (id integer)")
|
||||
) == ["t"]
|
||||
# ... but it is not persisted as current either
|
||||
assert await tables_after_start() == ["t", "t2"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_crossdb_memory_connections_are_reaped_and_refreshed(tmp_path):
|
||||
# Contract C7: --crossdb _memory connections ATTACH up to ten files but
|
||||
# were never counted, reaped or refreshed: the fds stayed open and a
|
||||
# removed database stayed queryable
|
||||
paths = []
|
||||
for i in range(3):
|
||||
path = str(tmp_path / f"db{i}.db")
|
||||
make_db(path)
|
||||
paths.append(path)
|
||||
ds = Datasette(paths[:2], crossdb=True, settings={"connection_idle_timeout_ms": 50})
|
||||
await ds.invoke_startup()
|
||||
memory = ds.get_database("_memory")
|
||||
assert (await memory.execute("select count(*) from db0.t")).first()[0] == 0
|
||||
pool = ds._read_pool
|
||||
assert pool.snapshot()["open"] >= 1
|
||||
deadline = time.monotonic() + 5
|
||||
while pool.snapshot()["open"]:
|
||||
assert time.monotonic() < deadline, "crossdb connections were not reaped"
|
||||
await asyncio.sleep(0.05)
|
||||
# Added and removed databases are reflected immediately
|
||||
await memory.execute("select 1")
|
||||
ds.add_database(Database(ds, path=paths[2]), name="db2")
|
||||
assert (await memory.execute("select count(*) from db2.t")).first()[0] == 0
|
||||
ds.remove_database("db0")
|
||||
with pytest.raises(sqlite3.OperationalError):
|
||||
await memory.execute("select count(*) from db0.t")
|
||||
ds.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_fn_may_return_generator_over_fetched_rows():
|
||||
# Contract C8: every generator was rejected, including harmless ones
|
||||
ds = Datasette(memory=True)
|
||||
db = ds.get_database("_memory")
|
||||
|
||||
def fetched(conn):
|
||||
rows = conn.execute("select 1 union all select 2").fetchall()
|
||||
return (row[0] for row in rows)
|
||||
|
||||
assert list(await db.execute_fn(fetched)) == [1, 2]
|
||||
|
||||
def live_cursor(conn):
|
||||
return (row[0] for row in conn.execute("select 1"))
|
||||
|
||||
with pytest.raises(ConnectionLeaseError):
|
||||
await db.execute_fn(live_cursor)
|
||||
|
||||
def closure(conn):
|
||||
def rows():
|
||||
yield from conn.execute("select 1")
|
||||
|
||||
return rows()
|
||||
|
||||
with pytest.raises(ConnectionLeaseError):
|
||||
await db.execute_fn(closure)
|
||||
ds.close()
|
||||
|
|
@ -126,9 +126,30 @@ async def test_modes_and_defaults(tmp_path, db_path):
|
|||
watcher = ds._schema_watcher
|
||||
assert watcher.states["ext"].mode == "external"
|
||||
assert watcher.states["imm"].mode == "immutable"
|
||||
# add_database(): files and named in-memory databases can be changed
|
||||
# by code other than Datasette's write methods (a plugin's own
|
||||
# connection, a backup() into a named in-memory database), so they get
|
||||
# the default_schema_watch mode too unless the caller says otherwise
|
||||
added = ds.add_database(Database(ds, memory_name="sw_owned_mem"), name="added")
|
||||
assert watcher.states["added"].mode == "owned"
|
||||
assert watcher.states["added"].mode == "external"
|
||||
assert added._watch_state is watcher.states["added"]
|
||||
added_file = str(tmp_path / "added_file.db")
|
||||
make_db(added_file)
|
||||
ds.add_database(Database(ds, path=added_file), name="added_file")
|
||||
assert watcher.states["added_file"].mode == "external"
|
||||
ds.add_database(
|
||||
Database(ds, memory_name="sw_owned_mem2"), name="owned", schema_watch="owned"
|
||||
)
|
||||
assert watcher.states["owned"].mode == "owned"
|
||||
# A private ":memory:" database: nothing else can change it
|
||||
ds.add_database(Database(ds, is_memory=True), name="private")
|
||||
assert watcher.states["private"].mode == "owned"
|
||||
ds.close()
|
||||
# default_schema_watch=owned applies to add_database() too
|
||||
ds = Datasette(files=[db_path], settings={"default_schema_watch": "owned"})
|
||||
ds.add_database(Database(ds, memory_name="sw_owned_mem3"), name="added")
|
||||
assert ds._schema_watcher.states["added"].mode == "owned"
|
||||
assert ds._schema_watcher.states["ext"].mode == "owned"
|
||||
ds.close()
|
||||
|
||||
|
||||
|
|
@ -595,15 +616,17 @@ async def test_default_schema_watch_setting(tmp_path, db_path, setting, expected
|
|||
assert watcher.states["ext"].mode == expected
|
||||
# Immutable files are always immutable
|
||||
assert watcher.states["imm"].mode == "immutable"
|
||||
# add_database() still defaults to owned, and can ask for external
|
||||
# add_database() follows the setting too, and can ask for either mode
|
||||
other = str(tmp_path / "other.db")
|
||||
make_db(other)
|
||||
added = ds.add_database(Database(ds, path=other), name="added")
|
||||
assert added._watch_state.mode == "owned"
|
||||
assert added._watch_state.mode == expected
|
||||
polled = ds.add_database(
|
||||
Database(ds, path=other), name="polled", schema_watch="external"
|
||||
)
|
||||
assert polled._watch_state.mode == "external"
|
||||
owned = ds.add_database(Database(ds, path=other), name="owned", schema_watch="owned")
|
||||
assert owned._watch_state.mode == "owned"
|
||||
ds.close()
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue