remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-10-19 07:47:00 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge spike/schema-watcher into spike/integrated, reconciling the three spikes

Browse source 
Read path: the leased pool replaces the thread-local cache. Pooled
connections record Database._conn_generation when opened; a stale one is
discarded on checkout or return, and _invalidate_connections() (called
by the SchemaWatcher on a replaced/deleted file) closes that database's
idle connections at once. Leased ones are only ever closed by their own
thread on release.

Write path: the idle-exit write thread keeps its lock protocol; its
connection lives in a _WriteConnection holder that is reopened when the
generation moves, before the next task. Post-write schema_version check
and _after_write catalog await are kept for every write API in both
modes. Non-threaded write connections are reopened on a generation change
and never cached half-prepared.

Deleted files: write connections use mode=rw once Datasette has seen the
file exist (_file_seen), so a deleted file is never silently recreated,
while a database added before its file exists is still created by its
first write.

Settings: max_open_connections (clamped to >= 4 x num_sql_threads),
connection_idle_timeout_ms (30000, read connections and write threads),
schema_watch_interval_ms, default_schema_watch (external|owned). The
spike-only knobs (connection_idle_timeout, connection_pool_wait_ms,
pool_read_connections, write_thread_idle_timeout_ms) are gone, along with
the pool's wait/disable code paths. The reaper falls back to checkout-time
reaping where threads cannot be started (Pyodide).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv
This commit is contained in:
Claude 2026-10-04 04:49:04 +00:00
commit 990c8782db
No known key found for this signature in database
17 changed files with 2508 additions and 440 deletions
Download patch file Download diff file Expand all files Collapse all files

154
datasette/app.py
View file

@ -49,6 +49,7 @@ from .events import Event
from .plugins import DEFAULT_PLUGINS, get_plugins, pm
from .renderer import json_renderer
from .resources import DatabaseResource, TableResource
from .schema_watcher import FILES_DEFAULT, SchemaWatcher
from .telemetry import (
TelemetryMiddleware,
_in_datasette_client,
@ -108,7 +109,7 @@ from .utils.asgi import (
asgi_send_redirect,
asgi_static,
)
from .utils.internal_db import init_internal_db, populate_schema_tables
from .utils.internal_db import init_internal_db
from .utils.sqlite import (
sqlite3,
using_pysqlite3,
@ -233,27 +234,22 @@ SETTINGS = (
Setting(
"max_open_connections",
128,
"Maximum number of pooled read connections open across all databases - least recently used idle connections are closed first, 0 for no limit",
"Maximum number of pooled read connections open across all databases (at least 4 x num_sql_threads) - 0 for no limit",
),
Setting(
"connection_idle_timeout",
60,
"Close pooled read connections that have been idle for this many seconds - 0 to keep them open",
),
Setting(
"connection_pool_wait_ms",
0,
"When max_open_connections is reached and every connection is in use, wait this long for one to be returned before opening one more",
),
Setting(
"pool_read_connections",
True,
"Reuse read connections between queries - turn off to open a new connection for every query",
),
Setting(
"write_thread_idle_timeout_ms",
"connection_idle_timeout_ms",
30000,
"Milliseconds a database's write thread waits for a write before closing its connection and exiting - set 0 to keep it running",
"Close read connections, and stop write threads, that have been idle for this many milliseconds - 0 to keep them open",
),
Setting(
"schema_watch_interval_ms",
1000,
"How often to check external database files for schema changes - 0 to disable polling",
),
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)",
),
Setting("sql_time_limit_ms", 1000, "Time limit for a SQL query in milliseconds"),
Setting(
@ -507,16 +503,21 @@ class Datasette:
self._background_tasks = BackgroundTaskSupervisor(self)
self.crossdb = crossdb
self.nolock = nolock
self.internal_db_created = False
self._schema_watcher = SchemaWatcher(self)
if memory or crossdb or not self.files:
self.add_database(
Database(self, is_mutable=False, is_memory=True), name="_memory"
)
for file in self.files:
is_mutable = file not in self.immutables
self.add_database(
Database(self, file, is_mutable=file not in self.immutables)
Database(self, file, is_mutable=is_mutable),
# Files named on the command line or in files= use the
# default_schema_watch setting (external unless configured)
schema_watch=FILES_DEFAULT if is_mutable else "immutable",
)
self.internal_db_created = False
if internal is None:
self._internal_database = Database(self, is_temp_disk=True)
else:
@ -676,6 +677,8 @@ class Datasette:
self.root_enabled = False
self.default_deny = default_deny
self.client = DatasetteClient(self)
# ds.config is available now: resolve per-database schema_watch modes
self._schema_watcher.configure()
# Last, so metric callbacks never see a partially initialized instance
register_datasette(self)
@ -738,66 +741,38 @@ class Datasette:
return None
async def refresh_schemas(self, *, force=False):
# Throttle schema refreshes to at most once per second
"""Bring the _internal catalog up to date with every attached database.
Uses the SchemaWatcher stat() prefilter, so it only opens connections
to databases whose files changed. Not called on the request path any
more - the watcher polls external databases in the background and
the write path catches schema changes made through Datasette.
"""
# Throttle non-forced refreshes to at most once per second
if (
not force
and time.monotonic() - getattr(self, "_last_schema_refresh", 0) < 1.0
):
return
self._last_schema_refresh = time.monotonic()
if self._refresh_schemas_lock.locked():
if self._refresh_schemas_lock.locked() and not force:
return
async with self._refresh_schemas_lock:
await self._refresh_schemas()
async def _refresh_schemas(self):
async def _refresh_schemas(self, *, background=False):
internal_db = self.get_internal_database()
if not self.internal_db_created:
if background:
return
await init_internal_db(internal_db)
await self.apply_metadata_json()
self.internal_db_created = True
current_schema_versions = {
row["database_name"]: row["schema_version"]
for row in await internal_db.execute(
"select database_name, schema_version from catalog_databases"
)
}
catalog_table_names = (
"catalog_columns",
"catalog_foreign_keys",
"catalog_indexes",
"catalog_views",
"catalog_tables",
"catalog_databases",
)
# Delete stale entries for databases that are no longer attached
catalog_database_names = set(current_schema_versions.keys())
for table in catalog_table_names[:-1]:
catalog_database_names.update(
row["database_name"]
for row in await internal_db.execute(
f"select distinct database_name from {table}"
)
if row["database_name"] is not None
)
stale_databases = catalog_database_names - set(self.databases.keys())
if stale_databases:
def delete_stale_database_catalog(conn):
for stale_db_name in stale_databases:
for table in catalog_table_names:
conn.execute(
f"DELETE FROM {table} WHERE database_name = ?",
[stale_db_name],
)
await internal_db.execute_write_fn(delete_stale_database_catalog)
for database_name, db in self.databases.items():
schema_version = (await db.execute("PRAGMA schema_version")).first()[0]
# Compare schema versions to see if we should skip it
if schema_version == current_schema_versions.get(database_name):
continue
await populate_schema_tables(internal_db, db, schema_version)
# Full catalog build (skips files whose persisted fingerprint
# shows they are unchanged when --internal is a real file)
await self._schema_watcher.initial_scan()
return
await self._schema_watcher.sweep(background=background)
@property
def urls(self):
@ -965,7 +940,16 @@ class Datasette:
name = next(iter(self.databases.keys()))
return self.databases[name]
def add_database(self, db, name=None, route=None):
def add_database(self, db, name=None, route=None, schema_watch=None):
"""Attach a database.
``schema_watch`` is how the _internal catalog is kept current for it:
``"owned"`` (default - schema changes come through Datasette's write
methods), ``"external"`` (other processes may change the file, poll
it) or ``"immutable"`` (scan once). ``databases.<name>.schema_watch``
in datasette.yaml overrides it. Databases that are not mutable are
always ``"immutable"``.
"""
new_databases = self.databases.copy()
if name is None:
# Pick a unique name for this database
@ -982,6 +966,9 @@ class Datasette:
new_databases[name] = db
# don't mutate! that causes race conditions with live import
self.databases = new_databases
self._schema_watcher.register(
db, schema_watch or getattr(db, "schema_watch", None)
)
return db
def add_memory_database(self, memory_name, name=None, route=None):
@ -990,10 +977,14 @@ class Datasette:
)
def remove_database(self, name):
self.get_database(name).close()
db = self.get_database(name)
new_databases = self.databases.copy()
new_databases.pop(name)
self.databases = new_databases
# Deletes this database's catalog rows (explicitly - there is no
# periodic stale-catalog scan any more)
self._schema_watcher.unregister(name)
db.close()
def close(self):
"""Release all resources held by this Datasette instance.
@ -1005,6 +996,7 @@ class Datasette:
if self._closed:
return
self._closed = True
self._schema_watcher.stop()
# Stop reporting metrics before closing databases
unregister_datasette(self)
first_exception = None
@ -1039,14 +1031,29 @@ class Datasette:
pool = ReadConnectionPool(
self._prepare_connection,
max_open=self.setting("max_open_connections") or 0,
idle_timeout=float(self.setting("connection_idle_timeout") or 0),
wait_ms=self.setting("connection_pool_wait_ms") or 0,
enabled=bool(self.setting("pool_read_connections")),
max_open=self._effective_max_open_connections(),
idle_timeout=self._connection_idle_timeout_s(),
)
pool = self.__dict__.setdefault("_read_pool_instance", pool)
return pool
def _effective_max_open_connections(self):
"""max_open_connections, raised to at least 4 x num_sql_threads.
Every executor thread can hold one read connection at a time, so a
cap below the thread count would close and reopen a connection for
nearly every query. 0 means no limit."""
configured = self.setting("max_open_connections") or 0
if configured <= 0:
return 0
threads = self.setting("num_sql_threads") or 0
return max(configured, 4 * threads)
def _connection_idle_timeout_s(self):
"connection_idle_timeout_ms in seconds; 0 means never close idle ones."
timeout_ms = self.setting("connection_idle_timeout_ms") or 0
return max(timeout_ms, 0) / 1000
@property
def _read_pool_or_none(self):
return self.__dict__.get("_read_pool_instance")
@ -3158,6 +3165,7 @@ ORDER BY allowed.parent, allowed.child
"""
await self.invoke_startup()
await self._background_tasks.launch_all()
await self._schema_watcher.start()
async def _launch_background_tasks(self):
"""Idempotently launch every registered background task. Private:
@ -3180,6 +3188,7 @@ ORDER BY allowed.parent, allowed.child
if self._suppress_background_tasks:
return
await self._background_tasks.launch_all()
await self._schema_watcher.start()
async def invoke_shutdown(self):
"""Run the graceful teardown sequence: plugin ``shutdown`` hooks,
@ -3195,6 +3204,7 @@ ORDER BY allowed.parent, allowed.child
except Exception:
logging.getLogger("datasette").exception("shutdown hook failed")
await self._background_tasks.cancel_all(grace=5.0)
await self._schema_watcher.astop()
self.close()
def app(self):

214
datasette/connection_pool.py
View file

@ -1,27 +1,38 @@
"""Leased read-connection pool.
"""Leased read-connection pool, and the lease proxy every callback receives.
Read connections are no longer cached per worker thread. Instead every
``execute_fn()`` / ``execute()`` call checks a connection out of a pool for
the duration of one callback and returns it afterwards.
Read connections are not cached per worker thread. Every ``execute_fn()`` /
``execute()`` call checks a connection out of a pool for the duration of one
callback and returns it afterwards.
* One ``ReadConnectionPool`` per ``Datasette`` instance. It enforces a global
cap (``max_open_connections``) on open pooled read connections across all
file-backed databases. When a new connection is needed and the cap has been
reached, the least recently used *idle* connection (from any database) is
closed first.
cap (the ``max_open_connections`` setting) on open pooled read connections
across all file-backed databases. When a new connection is needed and the
cap has been reached, the least recently used *idle* connection (from any
database) is closed first.
* Each database keeps a LIFO stack of its idle connections, so the most
recently used (warmest) connection is reused and surplus ones age out.
* Idle connections are closed after ``connection_idle_timeout`` seconds. This
is done opportunistically on every checkout and by one process-wide daemon
* Idle connections are closed after ``connection_idle_timeout_ms``. This is
done opportunistically on every checkout and by one process-wide daemon
reaper thread, so it works with or without a running event loop
(non-threaded mode, sync callers, closed loops in tests).
* When the cap has been reached and every connection is checked out the pool
waits up to ``connection_pool_wait_ms`` for one to come back and then opens
a connection anyway (the cap is soft). Each checked-out connection occupies
one executor thread, so the overflow can never exceed ``num_sql_threads``.
(non-threaded mode, sync callers, closed loops in tests). Where threads
cannot be started at all (Pyodide) only the opportunistic reaping runs.
* Every connection is opened with ``Database.connect()`` and prepared with
``Datasette._prepare_connection()`` exactly once, outside the pool lock.
* Each pooled connection remembers the database's ``_conn_generation`` when
it was opened. The SchemaWatcher bumps that generation when a file is
replaced or deleted and calls ``invalidate_database()``: idle connections
are closed at once, and leased ones are discarded by their own thread when
they are returned. A connection is never closed while another thread may
be stepping it (that segfaults).
* The cap is soft. Each lease occupies one executor thread (or the event
loop in non-threaded mode), and Datasette clamps the cap to at least
``4 x num_sql_threads``, so "cap reached and every connection leased"
cannot happen with the public settings. If it does (a cap lowered through
the private attribute) the pool opens one more connection and closes it
on release.
* Callbacks receive a ``LeasedConnection`` proxy. Once the callback returns
the proxy is invalidated and any further use raises
``ConnectionLeaseError``.
the proxy is expired and any further use raises ``ConnectionLeaseError``.
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.
@ -37,11 +48,18 @@ from .utils import sqlite3
class ConnectionLeaseError(RuntimeError):
"""A pooled read connection was used outside the callback it was lent to."""
"""A connection was used outside the callback it was lent to."""
_CALLBACK_NAMES = {
"read": "execute_fn()",
"write": "execute_write_fn()",
"isolated": "execute_isolated_fn()",
}
class LeasedConnection:
"""Forwarding proxy handed to read callbacks.
"""Forwarding proxy handed to read, write and isolated callbacks.
Every attribute access, method call and attribute assignment is forwarded
to the underlying ``sqlite3.Connection`` while the lease is active. After
@ -53,22 +71,27 @@ class LeasedConnection:
``isinstance(proxy, sqlite3.Connection)`` is True (via ``__class__``), but
C functions that require a real connection object (for example the
*target* argument of ``Connection.backup()``) reject the proxy.
Not enforceable: a cursor created inside the callback and used after it
returns, and the raw connection reached through ``cursor.connection``.
"""
__slots__ = ("_conn", "_db_name")
__slots__ = ("_conn", "_db_name", "_kind")
def __init__(self, conn, db_name):
def __init__(self, conn, db_name, kind="read"):
object.__setattr__(self, "_conn", conn)
object.__setattr__(self, "_db_name", db_name)
object.__setattr__(self, "_kind", kind)
def _live(self):
conn = self._conn
if conn is None:
raise ConnectionLeaseError(
f"A read connection to database {self._db_name!r} was used after the "
"execute_fn() callback it was passed to had returned. Read "
"connections are pooled and only valid inside the callback: do "
"not store the connection (or a cursor) for later use"
f"A {self._kind} connection to database {self._db_name!r} was used "
f"after the {_CALLBACK_NAMES.get(self._kind, 'callback')} callback "
"it was passed to had returned. Connections are only valid inside "
"the callback: do not store the connection (or a cursor) for "
"later use"
)
return conn
@ -91,7 +114,9 @@ class LeasedConnection:
# Dunder methods are looked up on the type, so forward them explicitly
def __enter__(self):
return self._live().__enter__()
self._live().__enter__()
# "with conn as c" must not hand out the raw connection
return self
def __exit__(self, *args):
return self._live().__exit__(*args)
@ -107,21 +132,22 @@ class LeasedConnection:
return self._live().cursor(*args, **kwargs)
def close(self):
# Closing a pooled connection is allowed; the pool notices on release
# and discards it instead of returning it to the idle stack.
# 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()
def __repr__(self):
state = "expired" if self._conn is None else "active"
return f"<LeasedConnection database={self._db_name!r} {state}>"
return f"<LeasedConnection {self._kind} database={self._db_name!r} {state}>"
class _Entry:
__slots__ = ("conn", "last_used", "state")
__slots__ = ("conn", "generation", "last_used", "state")
def __init__(self, conn, state):
def __init__(self, conn, state, generation):
self.conn = conn
self.state = state
self.generation = generation
self.last_used = 0.0
@ -141,25 +167,14 @@ class _DatabaseState:
class ReadConnectionPool:
def __init__(
self,
prepare_connection,
max_open=128,
idle_timeout=60.0,
wait_ms=0,
enabled=True,
):
def __init__(self, prepare_connection, max_open=128, idle_timeout=30.0):
self._prepare_connection = prepare_connection
self.max_open = max_open # 0 = no limit
self.idle_timeout = idle_timeout # seconds, 0 = never
self.wait_s = wait_ms / 1000.0
self.enabled = enabled
self._lock = threading.Lock()
self._cond = threading.Condition(self._lock)
# Idle, counted entries, least recently used first
self._lru = collections.OrderedDict()
self._open = 0 # counted open connections, idle + leased
self._waiters = 0
self._closed = False
self.stats = collections.Counter()
@ -168,7 +183,7 @@ class ReadConnectionPool:
def run(self, db, fn):
"""Call fn(conn) with a pooled connection leased for the duration."""
entry = self.acquire(db)
lease = LeasedConnection(entry.conn, db.name)
lease = LeasedConnection(entry.conn, db.name, "read")
try:
result = fn(lease)
finally:
@ -191,19 +206,19 @@ class ReadConnectionPool:
from .database import DatasetteClosedError
raise DatasetteClosedError(f"Database {db.name!r} has been closed")
now = time.monotonic()
self._collect_expired_locked(now, to_close)
deadline = None
self._collect_expired_locked(time.monotonic(), to_close)
while True:
if deadline is not None:
self.stats["wait_ms_max"] = max(
self.stats["wait_ms_max"],
int((time.monotonic() - now) * 1000),
)
if state.idle:
entry = state.idle.pop()
if state.counted:
del self._lru[entry]
if entry.generation != db._conn_generation:
# Opened before the file was replaced or deleted
self._forget_locked(entry)
to_close.append(entry)
self.stats["discarded_stale"] += 1
entry = None
continue
state.leased += 1
self.stats["reused"] += 1
break
@ -219,33 +234,16 @@ class ReadConnectionPool:
to_close.append(victim)
self.stats["evicted_lru"] += 1
continue
# Cap reached and every connection is checked out
if self.wait_s > 0:
if deadline is None:
deadline = now + self.wait_s
self.stats["waits"] += 1
remaining = deadline - time.monotonic()
if remaining > 0:
self._waiters += 1
t0 = time.monotonic()
try:
self._cond.wait(remaining)
finally:
self._waiters -= 1
waited_ms = int((time.monotonic() - t0) * 1000)
self.stats["wait_ms_total"] += waited_ms
continue
self.stats["wait_timeouts"] += 1
if deadline is not None:
self.stats["wait_ms_max"] = max(
self.stats["wait_ms_max"],
int((time.monotonic() - now) * 1000),
)
# Soft cap: open one more. Bounded by the number of threads
# that can hold a lease at once (num_sql_threads).
# Cap reached and every connection is leased: open one more
# (soft cap). Bounded by the number of threads that can hold
# a lease at once, and closed again on release.
self.stats["exceeded_cap"] += 1
break
if entry is None:
# Read before connecting: if the generation moves while we
# connect, the connection is discarded on release - one
# connection too many, never a stale one handed out twice
generation = db._conn_generation
state.open += 1
state.leased += 1
self.stats["opened"] += 1
@ -255,7 +253,7 @@ class ReadConnectionPool:
self._close_entries(to_close)
if entry is not None:
return entry
# Open and prepare outside the lock
# Open and prepare outside the lock, exactly once per connection
conn = None
try:
conn = db.connect()
@ -268,21 +266,19 @@ class ReadConnectionPool:
self.stats["open_failed"] += 1
if state.counted:
self._open -= 1
if self._waiters:
self._cond.notify()
if conn is not None:
self._close_conn(state, conn)
raise
return _Entry(conn, state)
return _Entry(conn, state, generation)
def release(self, entry):
state = entry.state
conn = entry.conn
discard = False
rolled_back = False
try:
# Hygiene: a callback that began a transaction and did not end it
# must not hand that transaction to the next borrower
rolled_back = False
if conn.in_transaction:
conn.rollback()
rolled_back = True
@ -296,14 +292,17 @@ class ReadConnectionPool:
elif rolled_back:
self.stats["rolled_back"] += 1
state.leased -= 1
stale = entry.generation != state.db._conn_generation
over_cap = state.counted and self.max_open and self._open > self.max_open
if discard or over_cap or self._closed or state.closed or not self.enabled:
if discard or stale or over_cap or self._closed or state.closed:
state.open -= 1
if state.counted:
self._open -= 1
discard = True
if over_cap:
self.stats["closed_over_cap"] += 1
if stale:
self.stats["discarded_stale"] += 1
else:
entry.last_used = time.monotonic()
state.idle.append(entry)
@ -311,8 +310,6 @@ class ReadConnectionPool:
self._lru[entry] = None
if self.idle_timeout > 0:
notify_reaper = entry.last_used + self.idle_timeout
if self._waiters:
self._cond.notify()
if discard:
self._close_conn(state, conn)
if notify_reaper is not None:
@ -331,6 +328,23 @@ class ReadConnectionPool:
self._close_entries(to_close)
return next_expiry
def invalidate_database(self, db):
"""db's file was replaced or deleted and its ``_conn_generation`` has
been bumped: close its idle connections now. Leased connections are
closed by the thread using them, when they are released."""
state = getattr(db, "_read_pool_state", None)
if state is None:
return
with self._lock:
stale = [e for e in state.idle if e.generation != db._conn_generation]
for entry in stale:
state.idle.remove(entry)
if state.counted:
self._lru.pop(entry, None)
self._forget_locked(entry)
self.stats["discarded_stale"] += 1
self._close_entries(stale)
def close_database(self, db):
"""Close idle connections for db; leased ones are closed on release."""
state = getattr(db, "_read_pool_state", None)
@ -343,8 +357,7 @@ class ReadConnectionPool:
for entry in to_close:
if state.counted:
self._lru.pop(entry, None)
self._open -= 1
state.open -= 1
self._forget_locked(entry)
self._close_entries(to_close)
def close(self):
@ -354,9 +367,7 @@ class ReadConnectionPool:
self._lru.clear()
for entry in to_close:
entry.state.idle.remove(entry)
entry.state.open -= 1
self._open -= 1
self._cond.notify_all()
self._forget_locked(entry)
self._close_entries(to_close)
def snapshot(self):
@ -380,6 +391,13 @@ class ReadConnectionPool:
state = db._read_pool_state = _DatabaseState(db)
return state
def _forget_locked(self, entry):
# Caller holds the lock and has already removed entry from the idle
# stack and the LRU
entry.state.open -= 1
if entry.state.counted:
self._open -= 1
def _collect_expired_locked(self, now, to_close):
if self.idle_timeout <= 0 or not self._lru:
return
@ -391,8 +409,7 @@ class ReadConnectionPool:
break
lru.popitem(last=False)
head.state.idle.remove(head)
head.state.open -= 1
self._open -= 1
self._forget_locked(head)
to_close.append(head)
self.stats["expired"] += 1
@ -423,21 +440,30 @@ class _Reaper:
self._pools = weakref.WeakSet()
self._thread = None
self._dirty = False
# Set when this platform cannot start threads (Pyodide): pools then
# rely on reaping during checkout only
self.unavailable = False
# monotonic time of the next planned scan, inf while sleeping
# indefinitely or while scanning
self.wake_at = float("inf")
def wake(self, pool, expiry):
# Unlocked pre-check: most releases do not need to wake the reaper
if expiry >= self.wake_at and pool in self._pools:
if self.unavailable or (expiry >= self.wake_at and pool in self._pools):
return
with self._cond:
self._pools.add(pool)
if self._thread is None or not self._thread.is_alive():
self._thread = threading.Thread(
thread = threading.Thread(
target=self._run, name="datasette-read-pool-reaper", daemon=True
)
self._thread.start()
try:
thread.start()
except RuntimeError:
# "can't start new thread": no threads on this platform
self.unavailable = True
return
self._thread = thread
if expiry < self.wake_at:
self._dirty = True
self._cond.notify()

309
datasette/database.py
View file

@ -68,8 +68,6 @@ from .utils import (
from .utils.sql_analysis import SQLAnalysis, analyze_sql_tables
from .utils.sqlite import sqlite_derived_table_dependencies, sqlite_hidden_table_names
connections = threading.local()
EXECUTE_WRITE_RETURNING_LIMIT = 10
AttachedDatabase = namedtuple("AttachedDatabase", ("seq", "name", "file"))
@ -85,7 +83,6 @@ _SHUTDOWN = object()
class Database:
# For table counts stop at this many rows:
count_limit = 10000
_thread_local_id_counter = 1
def __init__(
self,
@ -98,8 +95,6 @@ class Database:
is_temp_disk=False,
):
self.name = None
self._thread_local_id = f"x{self._thread_local_id_counter}"
Database._thread_local_id_counter += 1
self.route = None
self.ds = ds
self.path = path
@ -140,14 +135,25 @@ class Database:
self._closed = False
self._pending_execute_futures = set()
self._pending_execute_futures_lock = threading.Lock()
# Used when in non-threaded mode (reads go through ds._read_pool):
self._read_connection = None
# The write connection in non-threaded mode (reads go through
# ds._read_pool in both modes)
self._write_connection = None
self._write_connection_generation = 0
# Bookkeeping for ds._read_pool, created on first read
self._read_pool_state = None
# Track file and memory connections, including reads on worker threads,
# so close() can release all of them from the calling thread.
self._all_connections = []
# Set by SchemaWatcher.register(); None for the internal database
self._watch_state = None
# Bumped when the file behind this database is replaced or deleted,
# so pooled read connections and the write connection still pointing
# at the old inode are discarded and reopened
self._conn_generation = 0
# True once Datasette has seen this database's file exist (a
# connection opened, or the SchemaWatcher stat()ed it). From then on
# write connections never create the file - see connect()
self._file_seen = False
if not is_temp_disk:
self.mode = mode
@ -159,6 +165,80 @@ class Database:
with self._pending_execute_futures_lock:
self._pending_execute_futures.discard(future)
def _invalidate_connections(self):
"""The file behind this database was replaced or deleted.
Bumps the connection generation: pooled read connections are closed
(idle ones now, leased ones when their callback returns) and the
write connection is reopened before the next write task. Nothing is
closed from this thread while another thread may be using it.
"""
with self._write_thread_lock:
self._conn_generation += 1
read_pool = getattr(self.ds, "_read_pool_or_none", None)
if read_pool is not None:
read_pool.invalidate_database(self)
def _forget_connection(self, conn):
try:
conn.close()
except Exception: # noqa: BLE001, S110
pass
try:
self._all_connections.remove(conn)
except ValueError:
pass
def _schema_check(self, conn, loop=None):
# Runs on the thread that owns conn, right after a write task
state = self._watch_state
if state is not None:
self.ds._schema_watcher.check_after_write(state, conn, loop)
async def _execute_on_write_connection(self, fn):
"""Run fn(conn) on the write connection, serialized with writes,
without write_wrapper hooks, events or a transaction. Used by the
SchemaWatcher for shared-cache memory databases, where a second
connection reading sqlite_master would make concurrent DDL fail
with "database table is locked" (shared-cache locks do not wait)."""
self._check_not_closed()
if self.ds.executor is None:
return fn(self._non_threaded_write_connection())
return await self._send_to_write_thread(fn, transaction=False)
def _non_threaded_write_connection(self):
"""num_sql_threads=0: the single write connection, used inline on
the event loop. Reopened if the file was replaced or deleted. Unlike
the threaded write connection it is not closed when idle - there is
no thread to time it out (Pyodide has none at all)."""
conn = self._write_connection
if (
conn is not None
and self._write_connection_generation != self._conn_generation
):
self._write_connection = None
self._forget_connection(conn)
conn = None
if conn is None:
generation = self._conn_generation
conn = self.connect(write=True)
try:
self.ds._prepare_connection(conn, self.name)
except BaseException:
# Never cache a half-prepared connection
self._forget_connection(conn)
raise
self._write_connection = conn
self._write_connection_generation = generation
return conn
async def _after_write(self):
# Read-your-writes for the catalog: if that write changed the schema,
# wait for the catalog rows for this database to be rebuilt
state = self._watch_state
if state is not None and (state.needs_scan or state.scan_future is not None):
await self.ds._schema_watcher.after_write(state)
@property
def cached_table_counts(self):
if self._cached_table_counts is not None:
@ -187,7 +267,11 @@ class Database:
else:
return "db"
def connect(self, write=False):
def connect(self, write=False, track=True):
"""Open a new connection. ``track=False`` leaves it out of
``_all_connections``: the caller closes it itself and close() must
never close it from another thread while it is in use (the
SchemaWatcher's short-lived scan connections)."""
extra_kwargs = {}
if write:
extra_kwargs["isolation_level"] = "IMMEDIATE"
@ -198,11 +282,13 @@ class Database:
)
if not write:
conn.execute("PRAGMA query_only=1")
self._all_connections.append(conn)
if track:
self._all_connections.append(conn)
return conn
if self.is_memory:
conn = sqlite3.connect(":memory:", uri=True, check_same_thread=False)
self._all_connections.append(conn)
if track:
self._all_connections.append(conn)
return conn
# mode=ro or immutable=1?
@ -214,13 +300,20 @@ class Database:
qs = "?immutable=1"
assert not (write and not self.is_mutable)
if write:
qs = ""
# The first write connection may create the file (a database
# added with add_database() before its file exists). Once
# Datasette has seen the file exist, writes use mode=rw: if it
# is deleted from under us later, writes fail instead of
# silently creating a new empty database in its place.
qs = "?mode=rw" if self._file_seen else ""
if self.mode is not None:
qs = f"?mode={self.mode}"
conn = sqlite3.connect(
f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs
)
self._all_connections.append(conn)
self._file_seen = True
if track:
self._all_connections.append(conn)
if self.is_temp_disk and not self._wal_enabled:
conn.execute("PRAGMA journal_mode=WAL")
self._wal_enabled = True
@ -273,18 +366,7 @@ class Database:
except Exception: # noqa: BLE001, S110
pass
self._all_connections = []
# Drop per-thread cached read connections we can reach
try:
delattr(connections, self._thread_local_id)
except AttributeError:
pass
# Close non-threaded-mode cached connections if still open
if self._read_connection is not None:
try:
self._read_connection.close()
except Exception: # noqa: BLE001, S110
pass
self._read_connection = None
# Close the non-threaded-mode write connection if still open
if self._write_connection is not None:
try:
self._write_connection.close()
@ -417,6 +499,8 @@ class Database:
try:
return fn(isolated_connection)
finally:
if write:
self._schema_check(isolated_connection)
isolated_connection.close()
try:
self._all_connections.remove(isolated_connection)
@ -432,7 +516,9 @@ class Database:
with record_operation_duration(self.name, "write" if write else "read"):
if self.ds.executor is None:
# non-threaded mode
return _run()
result = _run()
await self._after_write()
return result
if not write:
# Immutable database - no writes can ever occur, so there
# is no write queue to block; run against a fresh
@ -442,7 +528,9 @@ class Database:
self.ds.executor, ctx.run, _run
)
# Threaded mode - send to write thread
return await self._send_to_write_thread(fn, isolated_connection=True)
result = await self._send_to_write_thread(fn, isolated_connection=True)
await self._after_write()
return result
async def analyze_sql(self, sql, params=None) -> SQLAnalysis:
self._check_not_closed()
@ -480,15 +568,16 @@ class Database:
fn = self._wrap_fn_with_hooks(fn, request, transaction, track_event)
if self.ds.executor is None:
# non-threaded mode
if self._write_connection is None:
self._write_connection = self.connect(write=True)
self.ds._prepare_connection(self._write_connection, self.name)
if transaction:
with self._write_connection:
self._write_connection.execute("BEGIN IMMEDIATE")
result = fn(self._write_connection)
else:
result = fn(self._write_connection)
conn = self._non_threaded_write_connection()
try:
if transaction:
with conn:
conn.execute("BEGIN IMMEDIATE")
result = fn(conn)
else:
result = fn(conn)
finally:
self._schema_check(conn)
if not block:
# There is no write thread here, so the write has already
# finished. Hand back the same (task_id, reply_future) shape
@ -503,6 +592,7 @@ class Database:
fn, block=block, transaction=transaction
)
if block:
await self._after_write()
for event in pending_events:
await self.ds.track_event(event)
else:
@ -618,48 +708,25 @@ class Database:
# Closing the last connection to a memory database discards
# its contents, so its write connection must stay open
return None
setting = getattr(self.ds, "setting", None)
timeout_ms = setting("write_thread_idle_timeout_ms") if setting else None
if not timeout_ms or timeout_ms <= 0:
return None
return timeout_ms / 1000
get_timeout = getattr(self.ds, "_connection_idle_timeout_s", None)
timeout_s = get_timeout() if get_timeout is not None else 0
return timeout_s or None
def _execute_writes(self):
# Thread that protects the single write connection to this database.
# It runs until close(), or until it has waited idle_timeout seconds
# with no task, in which case it closes its connection and exits.
# The next write starts a new thread.
idle_timeout = self._write_thread_idle_timeout()
conn_exception = None
conn = None
writer = _WriteConnection(self)
writer.open()
deregistered = False
try:
conn = self.connect(write=True)
# Threads do not inherit the caller's context, so any spans
# created by prepare_connection hooks here are root spans
self.ds._prepare_connection(conn, self.name)
except Exception as e: # noqa: BLE001
# Stored and re-raised to every write already queued. With an
# idle timeout the thread then exits as soon as the queue is
# empty, so the next write tries to connect again.
conn_exception = e
if idle_timeout is not None:
idle_timeout = 0
try:
self._process_write_queue(conn, conn_exception, idle_timeout)
self._process_write_queue(writer)
deregistered = True
finally:
if conn is not None:
try:
conn.close()
except Exception: # noqa: BLE001, S110
# Best-effort close as the write thread exits
pass
try:
self._all_connections.remove(conn)
except ValueError:
# May already have been cleared by close().
pass
# Close outside _write_thread_lock: closing the last connection
# to a WAL database runs a checkpoint, which can be slow
writer.close()
current = threading.current_thread()
with self._write_thread_lock:
self._retiring_write_threads.discard(current)
@ -671,7 +738,18 @@ class Database:
if not self._closed and not self._write_queue.empty():
self._start_write_thread()
def _process_write_queue(self, conn, conn_exception, idle_timeout):
def _process_write_queue(self, writer):
configured_idle_timeout = self._write_thread_idle_timeout()
def current_idle_timeout():
if writer.exception is not None and configured_idle_timeout is not None:
# The connect error has been reported to every write already
# queued: exit as soon as the queue is empty, so the next
# write starts a new thread that tries to connect again
return 0
return configured_idle_timeout
idle_timeout = current_idle_timeout()
while True:
try:
task = self._write_queue.get(timeout=idle_timeout)
@ -690,6 +768,13 @@ class Database:
return
if task is _SHUTDOWN:
return
if writer.generation != self._conn_generation:
# 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.
writer.reopen()
idle_timeout = current_idle_timeout()
# block=True: the caller awaits the result, so the write spans
# are children of the caller's span. The token must be detached
# in the finally block or the context leaks into later writes.
@ -712,8 +797,8 @@ class Database:
**write_span_kwargs,
).end(end_time=dequeued_at_ns)
record_write_queue_wait(self.name, dequeued_at_ns - task.enqueued_at_ns)
if conn_exception is not None:
exception = conn_exception
if writer.exception is not None:
exception = writer.exception
elif task.isolated_connection:
try:
with tracer.start_as_current_span(
@ -728,18 +813,15 @@ class Database:
try:
result = task.fn(isolated_connection)
finally:
isolated_connection.close()
try:
self._all_connections.remove(isolated_connection)
except ValueError:
# May already have been cleared by close().
pass
self._schema_check(isolated_connection, task.loop)
self._forget_connection(isolated_connection)
except Exception 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
else:
conn = writer.conn
try:
with tracer.start_as_current_span(
DB_WRITE_EXECUTE, **write_span_kwargs
@ -749,12 +831,15 @@ class Database:
task.isolated_connection,
)
span.set_attribute(TRANSACTION, task.transaction)
if task.transaction:
with conn:
conn.execute("BEGIN IMMEDIATE")
try:
if task.transaction:
with conn:
conn.execute("BEGIN IMMEDIATE")
result = task.fn(conn)
else:
result = task.fn(conn)
else:
result = task.fn(conn)
finally:
self._schema_check(conn, task.loop)
except Exception as e: # noqa: BLE001
sys.stderr.write(f"{e}\n")
sys.stderr.flush()
@ -1131,7 +1216,22 @@ class Database:
async def derived_table_dependencies(self):
"""Return implementation tables and the tables they derive from."""
schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
state = self._watch_state
if (
state is not None
and state.catalog_version is not None
and not state.needs_scan
and not state.missing
):
# O(1): the schema_version the SchemaWatcher last stored in the
# catalog. allowed_resources() calls this for every database on
# every page, so a PRAGMA here was a per-request sweep of all
# databases (one pooled connection each). The cache is now as
# fresh as the catalog itself. The scan count is part of the key
# because a replaced file can have the same schema_version.
schema_version = ("catalog", state.catalog_version, state.stats["scans"])
else:
schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
if (
self._cached_derived_table_dependencies is None
or self._cached_derived_table_dependencies[0] != schema_version
@ -1233,6 +1333,51 @@ def _apply_write_wrapper(fn, wrapper_factory, track_event):
return wrapped
class _WriteConnection:
"""The write thread's connection to its database.
Reopened by the write thread when the file is replaced or deleted
(``Database._conn_generation`` moves). A failure to open or prepare the
connection is kept in ``exception`` and handed to every write that runs
while it is set.
"""
__slots__ = ("conn", "db", "exception", "generation")
def __init__(self, db):
self.db = db
self.conn = None
self.exception = None
self.generation = None
def open(self):
db = self.db
# Read the generation before connecting: if it moves while we
# connect, the next task reopens - one connection too many, never
# a stale one
self.generation = db._conn_generation
self.exception = None
try:
self.conn = db.connect(write=True)
# 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)
except Exception as e: # noqa: BLE001
self.exception = e
self.close()
def reopen(self):
self.close()
self.open()
def close(self):
conn, self.conn = self.conn, None
if conn is not None:
# Also drops it from _all_connections, so that list does not
# grow by one closed connection per write thread restart
self.db._forget_connection(conn)
class WriteTask:
__slots__ = (
"block",

971
datasette/schema_watcher.py Normal file
View file

@ -0,0 +1,971 @@
"""
Keeps the ``_internal`` catalog (``catalog_databases``, ``catalog_tables``,
``catalog_columns`` ...) current without checking every attached database on
the request path.
Every attached database has a *schema watch mode*:
``owned``
Every schema change goes through Datasette. After each write task the
write thread reads ``PRAGMA schema_version`` on the connection it just
used (microseconds, same thread). If the value moved, the catalog for
that one database is rebuilt before the write call returns, so callers
get read-your-writes on the catalog. Owned databases are never polled.
``external``
The file can also be changed by other processes (sqlite-utils, cron
jobs...). On top of the write-path check, a background task stats the
database file, its ``-wal`` and its ``-journal`` every
``schema_watch_interval_ms``. Only a database whose fingerprint
``(st_dev, st_ino, st_size, st_mtime_ns, st_ctime_ns)`` changed - or
whose previous fingerprint was *racily clean* - gets a
``PRAGMA schema_version`` check on a short-lived connection, and only a
changed ``schema_version`` (or a replaced or deleted file) triggers a
catalog rebuild for that database.
``immutable``
Scanned once at startup and never again.
Defaults: files passed to ``Datasette(files=...)`` / the CLI 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
``databases: {name: {schema_watch: owned|external|immutable}}`` in
``datasette.yaml`` or ``add_database(..., schema_watch=...)``.
Replaced and deleted files
--------------------------
A sweep that sees a new inode or a missing file calls
``Database._invalidate_connections()``: the database's connection generation
is bumped, idle pooled read connections are closed, leased ones are
discarded when their callback returns, and the write thread reopens its
connection before the next write. Once Datasette has seen a file exist,
write connections open it with ``mode=rw``, so a deleted file is never
silently recreated.
Racily clean fingerprints
-------------------------
Like git's index: a fingerprint whose newest timestamp is within
``RACY_WINDOW_NS`` of the moment it was taken could be followed by another
write inside the same timestamp tick that leaves size, inode and mtime
identical. Such a fingerprint is flagged ``racy`` and the next sweep runs
the ``PRAGMA schema_version`` check even though the fingerprint looks
unchanged. The stat is always taken *before* the pragma, so a write landing
between the two either shows up in the pragma or changes the next stat.
All catalog reads use one short-lived connection per database inside a
single read transaction, so ``schema_version`` and the rows written to the
catalog describe the same snapshot. The watcher never keeps a connection
open.
"""
import asyncio
import json
import logging
import os
import time
from .utils import sqlite3
from .utils.internal_db import CATALOG_TABLES, collect_schema, write_catalog_entries
logger = logging.getLogger("datasette.schema_watcher")
MODES = ("owned", "external", "immutable")
# Valid values for the default_schema_watch setting
DEFAULT_MODES = ("external", "owned")
# Passed by Datasette.__init__ for files=/CLI databases: resolved to the
# default_schema_watch setting once settings are available
FILES_DEFAULT = "_files_default"
# Timestamps closer than this to the time of the stat are treated as racy.
# 2s covers coarse filesystems (FAT: 2s, HFS+: 1s) and clock skew on
# network filesystems; ext4/xfs/tmpfs on Linux are far finer than this.
RACY_WINDOW_NS = 2_000_000_000
# Databases per thread hop / per internal-DB transaction for bulk scans
SCAN_CHUNK = 64
SCAN_CONCURRENCY = 1
# Directories with at least this many watched files are listed once per
# sweep with os.scandir() instead of probing -wal/-journal with stat()
SCANDIR_THRESHOLD = 8
PRAGMA_BUSY_TIMEOUT_MS = 200
# Read the schema cookie from the file header instead of opening a
# connection when no -wal/-journal has content
USE_HEADER_CHECK = True
class WatchState:
__slots__ = (
"catalog_version",
"db",
"error",
"fp",
"fp_racy",
"missing",
"mode",
"name",
"needs_scan",
"notified_version",
"removed",
"requested_mode",
"scan_future",
"stats",
)
def __init__(self, db, name, requested_mode):
self.db = db
self.name = name
self.requested_mode = requested_mode
self.mode = None
self.fp = None
self.fp_racy = True
self.catalog_version = None
self.notified_version = None
self.needs_scan = True
self.scan_future = None
self.missing = False
self.removed = False
self.error = None
self.stats = {"checks": 0, "scans": 0, "write_detections": 0}
@property
def is_file(self):
db = self.db
return bool(db.path) and not db.is_memory
def __repr__(self):
return f"<WatchState {self.name!r} mode={self.mode} version={self.catalog_version}>"
def _fp_main(st):
return (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns)
def _fp_side(st):
# No ctime for -wal/-journal: when running as root SQLite fchown()s
# these files to match the database every time a connection opens
# them, so their ctime moves on every open - including our own checks
return (st.st_ino, st.st_size, st.st_mtime_ns)
def fingerprint(path, present=None):
"""(main, wal, journal) stat tuples; None for a file that does not exist.
``present`` is an optional set of file names in the directory (from
scandir) used to skip stat() calls for -wal/-journal files that are not
there.
"""
path = os.fspath(path)
try:
main = _fp_main(os.stat(path))
except FileNotFoundError:
main = None
parts = [main]
base = os.path.basename(path) if present is not None else None
for suffix in ("-wal", "-journal"):
if present is not None and (base + suffix) not in present:
parts.append(None)
continue
try:
st = os.stat(path + suffix)
except FileNotFoundError:
parts.append(None)
continue
# An empty -wal holds no frames (and an empty -journal nothing to
# roll back): the database is exactly the main file, same as when
# the file is absent. Read-only opens of a WAL database create an
# empty -wal, which must not look like a change.
parts.append(_fp_side(st) if st.st_size else None)
return tuple(parts)
def newest_timestamp(fp):
main = fp[0]
newest = max(main[3], main[4]) if main is not None else 0
for part in fp[1:]:
if part is not None:
newest = max(newest, part[2])
return newest
def is_racy(fp, t_ns):
return newest_timestamp(fp) >= t_ns - RACY_WINDOW_NS
SQLITE_HEADER = b"SQLite format 3\x00"
def header_schema_version(path):
"""The schema cookie (offset 40 of the database header) read straight
from the file - what PRAGMA schema_version returns - without opening a
SQLite connection or taking any lock.
Only meaningful when there is no non-empty -wal or -journal: then the
main file is the whole database. The caller must have taken the
fingerprint *before* calling this; a commit in flight at that moment has
already bumped the mtime it saw, so the fingerprint is racy and the next
sweep looks again.
"""
try:
fd = os.open(path, os.O_RDONLY)
except OSError:
return None
try:
header = os.pread(fd, 100, 0)
except OSError:
return None
finally:
os.close(fd)
if len(header) < 100 or not header.startswith(SQLITE_HEADER):
return None
return int.from_bytes(header[40:44], "big")
def _fp_to_json(fp, t_ns):
return json.dumps({"fp": fp, "t": t_ns})
def _fp_from_json(value):
try:
data = json.loads(value)
fp = tuple(tuple(p) if p is not None else None for p in data["fp"])
return fp, int(data["t"])
except Exception: # noqa: BLE001
return None, None
class SchemaWatcher:
def __init__(self, ds):
self.ds = ds
self.states = {}
self._pending = set() # states registered after startup, not yet scanned
self._pending_removals = set() # names whose catalog rows must go
self._task = None
self._spawned = set()
self._last_sweep = 0.0
self.counters = {
"sweeps": 0,
"sweep_seconds": 0.0,
"stats": 0,
"pragma_checks": 0,
"header_checks": 0,
"scans": 0,
"write_detections": 0,
"replaced": 0,
"missing": 0,
"busy": 0,
"restored_from_persisted": 0,
}
# ------------------------------------------------------------------
# configuration
# ------------------------------------------------------------------
@property
def interval_s(self):
ms = self.ds.setting("schema_watch_interval_ms")
return (ms or 0) / 1000.0
def _config_mode(self, name):
config = getattr(self.ds, "config", None) or {}
db_config = (config.get("databases") or {}).get(name) or {}
return db_config.get("schema_watch")
def _default_mode_for_files(self):
from .utils import StartupError
mode = self.ds.setting("default_schema_watch") or "external"
if mode not in DEFAULT_MODES:
raise StartupError(
f"Invalid default_schema_watch setting {mode!r}, expected one of "
f"{', '.join(DEFAULT_MODES)}"
)
return mode
def _resolve_mode(self, state):
db = state.db
mode = self._config_mode(state.name) or state.requested_mode
if mode is None:
mode = "owned"
elif mode == FILES_DEFAULT:
mode = self._default_mode_for_files()
if mode not in MODES:
from .utils import StartupError
raise StartupError(
f"Invalid schema_watch mode {mode!r} for database {state.name!r}, "
f"expected one of {', '.join(MODES)}"
)
if not db.is_mutable:
# Nothing can change an immutable database
mode = "immutable"
elif mode == "external" and not state.is_file:
# There is no file to stat: rely on the write path
mode = "owned"
return mode
def configure(self):
"""Resolve modes once ds.config is available (end of __init__)."""
# Validate the setting even when no database uses it
self._default_mode_for_files()
for state in self.states.values():
state.mode = self._resolve_mode(state)
# ------------------------------------------------------------------
# registration
# ------------------------------------------------------------------
def register(self, db, requested_mode=None):
if requested_mode not in (None, FILES_DEFAULT) and requested_mode not in MODES:
raise ValueError(
f"schema_watch must be one of {', '.join(MODES)}, not {requested_mode!r}"
)
state = WatchState(db, db.name, requested_mode)
if hasattr(self.ds, "config"):
state.mode = self._resolve_mode(state)
old = self.states.get(db.name)
if old is not None:
old.removed = True
self.states[db.name] = state
db._watch_state = state
self._pending_removals.discard(db.name)
if self.ds.internal_db_created:
self._pending.add(state)
self._spawn(self.refresh([state]))
return state
def unregister(self, name):
state = self.states.pop(name, None)
if state is None:
return
state.removed = True
state.db._watch_state = None
self._pending.discard(state)
if self.ds.internal_db_created:
self._pending_removals.add(name)
self._spawn(self._delete_catalog([name]))
# ------------------------------------------------------------------
# task helpers
# ------------------------------------------------------------------
def _spawn(self, coro):
try:
loop = asyncio.get_running_loop()
except RuntimeError:
# No loop (sync caller): the work stays pending and is picked
# up by the next flush/sweep
coro.close()
return None
task = loop.create_task(coro)
self._spawned.add(task)
def _done(t):
self._spawned.discard(t)
if not t.cancelled() and t.exception() is not None:
logger.error("schema watcher task failed", exc_info=t.exception())
task.add_done_callback(_done)
return task
def loop_running(self):
task = self._task
if task is None or task.done():
return False
try:
return task.get_loop() is asyncio.get_running_loop()
except RuntimeError:
return False
async def start(self):
"""Start the polling loop (idempotent). Runs one sweep first so the
catalog is current when the first request is served."""
if self.interval_s <= 0 or self.ds._closed or self.loop_running():
return
loop = asyncio.get_running_loop()
# Claim the slot before awaiting so concurrent start() calls no-op
self._task = loop.create_task(self._run(), name="datasette-schema-watcher")
if self.ds.internal_db_created:
try:
await self.ds._refresh_schemas(background=True)
except Exception:
logger.exception("schema watcher initial sweep failed")
async def _run(self):
# Wait one interval first: start() has just swept
while not self.ds._closed:
await asyncio.sleep(self.interval_s)
if self.ds._closed:
return
try:
await self.ds._refresh_schemas(background=True)
except asyncio.CancelledError:
raise
except Exception:
logger.exception("schema watcher sweep failed")
def stop(self):
tasks = [t for t in [self._task, *self._spawned] if t is not None]
self._task = None
for task in tasks:
if task.done():
continue
loop = task.get_loop()
try:
running = asyncio.get_running_loop()
except RuntimeError:
running = None
if running is loop:
task.cancel()
elif not loop.is_closed():
try:
loop.call_soon_threadsafe(task.cancel)
except RuntimeError:
pass
async def astop(self):
tasks = [t for t in [self._task, *self._spawned] if t is not None]
self.stop()
loop = asyncio.get_running_loop()
mine = [t for t in tasks if t.get_loop() is loop and not t.done()]
if mine:
await asyncio.wait(mine, timeout=5)
# ------------------------------------------------------------------
# request path: O(1) unless add/remove_database left work pending
# ------------------------------------------------------------------
async def on_request(self):
if (self._pending or self._pending_removals) and self.ds.internal_db_created:
await self.flush_pending()
if (
not self.loop_running()
and self.interval_s > 0
and self.ds.internal_db_created
):
await self.start()
async def flush_pending(self):
if self._pending_removals:
await self._delete_catalog(list(self._pending_removals))
if self._pending:
await self.refresh(list(self._pending))
# ------------------------------------------------------------------
# write path
# ------------------------------------------------------------------
def check_after_write(self, state, conn, loop=None):
"""Called on the thread that owns ``conn`` right after a write task.
``loop`` is the event loop to notify (None when already on it)."""
try:
version = conn.execute("PRAGMA schema_version").fetchone()[0]
except Exception: # noqa: BLE001
return
if version == state.catalog_version or version == state.notified_version:
return
state.notified_version = version
if loop is None:
self._on_write_schema_change(state)
else:
try:
loop.call_soon_threadsafe(self._on_write_schema_change, state)
except RuntimeError:
pass
def _on_write_schema_change(self, state):
if state.removed:
return
state.needs_scan = True
state.stats["write_detections"] += 1
self.counters["write_detections"] += 1
if self.ds.internal_db_created:
self._spawn(self.refresh([state]))
async def after_write(self, state):
"""Awaited by blocking writes: gives read-your-writes on the catalog."""
if not self.ds.internal_db_created or state.removed:
return
try:
await self.refresh([state])
except Exception:
logger.exception("catalog refresh after write to %r failed", state.name)
# ------------------------------------------------------------------
# sweeps (stat prefilter)
# ------------------------------------------------------------------
async def sweep(self, *, background):
"""Check databases for changes made outside Datasette.
background=True (the polling loop): external-mode files only.
background=False (explicit refresh_schemas()): every mutable
database - files via the stat prefilter, named in-memory databases
via a PRAGMA schema_version check.
"""
t0 = time.perf_counter()
if self._pending_removals:
await self._delete_catalog(list(self._pending_removals))
candidates = []
for state in list(self.states.values()):
if state.mode == "immutable" or state.mode is None:
if state.needs_scan:
candidates.append(state)
continue
if background and state.mode != "external":
continue
candidates.append(state)
outcomes = []
file_candidates = [s for s in candidates if s.is_file]
if file_candidates:
outcomes = await asyncio.to_thread(self._sweep_sync, file_candidates)
for state in candidates:
if not state.is_file and state.db.memory_name and not state.removed:
outcomes.append(await self._memory_pragma_check(state))
to_scan = [s for s in self._pending if not s.removed]
to_clear = []
for state, kind, fp, t_ns, version in outcomes:
if state.removed:
continue
self.counters["stats"] += 1
if fp is not None:
state.fp = fp
state.fp_racy = is_racy(fp, t_ns)
if fp[0] is not None:
state.db._file_seen = True
if kind == "clean":
pass
elif kind == "missing":
self.counters["missing"] += 1
if not state.missing:
state.missing = True
state.db._invalidate_connections()
to_clear.append(state)
elif kind == "replaced":
self.counters["replaced"] += 1
state.missing = False
state.db._invalidate_connections()
state.needs_scan = True
elif kind == "busy":
self.counters["busy"] += 1
state.fp_racy = True # check again next sweep
elif kind in ("checked", "header"):
self.counters[
"pragma_checks" if kind == "checked" else "header_checks"
] += 1
state.stats["checks"] += 1
if version != state.catalog_version:
state.needs_scan = True
elif kind == "error":
state.fp_racy = True
if state.needs_scan:
to_scan.append(state)
if to_clear:
await self._clear_missing(to_clear)
if to_scan:
await self.refresh(to_scan)
self.counters["sweeps"] += 1
self.counters["sweep_seconds"] += time.perf_counter() - t0
self._last_sweep = time.monotonic()
def _sweep_sync(self, states):
"""Runs in a worker thread. Reads state, never writes it."""
by_dir = {}
for state in states:
if state.is_file:
by_dir.setdefault(
os.path.dirname(os.path.abspath(state.db.path)), []
).append(state)
out = []
for directory, dir_states in by_dir.items():
present = None
if len(dir_states) >= SCANDIR_THRESHOLD:
try:
with os.scandir(directory) as it:
present = {entry.name for entry in it}
except OSError:
present = None
for state in dir_states:
t_ns = time.time_ns()
fp = fingerprint(state.db.path, present)
prev = state.fp
if fp == prev and not state.fp_racy and not state.needs_scan:
out.append((state, "clean", fp, t_ns, None))
continue
if fp[0] is None:
out.append((state, "missing", fp, t_ns, None))
continue
if prev is None or prev[0] is None or prev[0][:2] != fp[0][:2]:
# New inode/device (atomic replace, delete+recreate) or
# first sight: the schema_version alone cannot be trusted
out.append((state, "replaced", fp, t_ns, None))
continue
if USE_HEADER_CHECK and fp[1] is None and fp[2] is None:
# No -wal/-journal content: read the cookie from the
# header. No connection, no SHARED lock that could make
# a concurrent writer back off.
version = header_schema_version(os.fspath(state.db.path))
if version is not None:
out.append((state, "header", fp, t_ns, version))
continue
out.append(self._pragma_check(state, fp, t_ns))
return out
async def _memory_pragma_check(self, state):
def _version(conn):
return conn.execute("PRAGMA schema_version").fetchone()[0]
try:
version = await state.db._execute_on_write_connection(_version)
except Exception: # noqa: BLE001
return (state, "error", None, None, None)
return (state, "checked", None, None, version)
def _pragma_check(self, state, fp, t_ns):
db = state.db
try:
conn = self._connect(db, prepare=False)
except Exception: # noqa: BLE001
return (state, "error", fp, t_ns, None)
try:
conn.execute(f"PRAGMA busy_timeout={PRAGMA_BUSY_TIMEOUT_MS}")
version = conn.execute("PRAGMA schema_version").fetchone()[0]
return (state, "checked", fp, t_ns, version)
except sqlite3.OperationalError as e:
msg = str(e)
if "locked" in msg or "busy" in msg:
return (state, "busy", fp, t_ns, None)
return (state, "error", fp, t_ns, None)
except Exception: # noqa: BLE001
return (state, "error", fp, t_ns, None)
finally:
self._close(db, conn)
# ------------------------------------------------------------------
# scanning + storing
# ------------------------------------------------------------------
def _connect(self, db, prepare):
# Untracked: Database.close() on the event loop thread must not close
# this connection while a worker thread is using it (that segfaults).
# It is always closed by _close() in the same thread that opened it.
conn = db.connect(track=False)
try:
if prepare:
self.ds._prepare_connection(conn, db.name)
else:
conn.row_factory = None
except Exception:
conn.close()
raise
return conn
@staticmethod
def _close(db, conn):
conn.close()
def _scan_sync(self, state):
"""Read schema_version + full schema in one read transaction."""
db = state.db
fp = t_ns = None
if state.is_file:
t_ns = time.time_ns()
fp = fingerprint(db.path)
if fp[0] is None:
return {"state": state, "missing": True, "fp": fp, "t": t_ns}
conn = self._connect(db, prepare=True)
try:
conn.row_factory = sqlite3.Row
conn.execute("BEGIN")
try:
version = conn.execute("PRAGMA schema_version").fetchone()[0]
schema = collect_schema(conn, db.name)
finally:
conn.rollback()
finally:
self._close(db, conn)
return {
"state": state,
"fp": fp,
"t": t_ns,
"version": version,
"schema": schema,
}
async def _scan_on_write_connection(self, state):
name = state.db.name
def _scan(conn):
version = conn.execute("PRAGMA schema_version").fetchone()[0]
# The write thread's post-task schema check must not re-notify
# for the version this scan is about to store
state.notified_version = version
return version, collect_schema(conn, name)
try:
version, schema = await state.db._execute_on_write_connection(_scan)
except Exception as e: # noqa: BLE001
return {"state": state, "error": e}
return {
"state": state,
"fp": None,
"t": None,
"version": version,
"schema": schema,
}
def _scan_chunk_sync(self, states):
results = []
for state in states:
try:
results.append(self._scan_sync(state))
except Exception as e: # noqa: BLE001
results.append({"state": state, "error": e})
return results
async def refresh(self, states):
"""Rebuild the catalog for every state that needs it, coalescing with
scans already in flight for the same database."""
loop = asyncio.get_running_loop()
pending = list(states)
for _ in range(10):
mine = []
waits = []
for state in pending:
fut = state.scan_future
if fut is not None and not fut.done():
if fut.get_loop() is loop:
waits.append(fut)
continue
# Scan started on an event loop that has gone away
state.scan_future = None
state.needs_scan = True
if state.needs_scan and not state.removed:
state.needs_scan = False
state.scan_future = loop.create_future()
mine.append(state)
if mine:
try:
await self._scan_and_store(mine)
finally:
for state in mine:
fut = state.scan_future
if fut is not None and not fut.done():
fut.set_result(None)
state.scan_future = None
if waits:
await asyncio.wait(waits)
pending = [
s
for s in pending
if not s.removed
and (
s.needs_scan
or (s.scan_future is not None and not s.scan_future.done())
)
and s.error is None
]
if not pending:
break
for state in states:
# Failed scans are retried by sweeps, not by every request
if not state.needs_scan or state.error is not None:
self._pending.discard(state)
async def _scan_and_store(self, states):
memory = [s for s in states if not s.is_file and s.db.memory_name]
if memory:
results = []
for state in memory:
results.append(await self._scan_on_write_connection(state))
await self._store(results)
states = [s for s in states if s not in memory]
if not states:
return
chunks = [states[i : i + SCAN_CHUNK] for i in range(0, len(states), SCAN_CHUNK)]
semaphore = asyncio.Semaphore(SCAN_CONCURRENCY)
async def one(chunk):
async with semaphore:
results = await asyncio.to_thread(self._scan_chunk_sync, chunk)
await self._store(results)
if len(chunks) == 1:
await one(chunks[0])
else:
await asyncio.gather(*(one(c) for c in chunks))
async def _store(self, results):
entries = []
missing = []
stored = []
for r in results:
state = r["state"]
if state.removed:
continue
if "error" in r:
state.error = r["error"]
state.needs_scan = True
logger.warning(
"Could not read schema of database %r: %s", state.name, r["error"]
)
continue
state.error = None
if r.get("missing"):
missing.append(state)
state.fp = r["fp"]
state.fp_racy = True
continue
db = state.db
fp_json = _fp_to_json(r["fp"], r["t"]) if r["fp"] is not None else None
entries.append(
(
state.name,
str(db.path) if db.path is not None else None,
db.is_memory,
r["version"],
fp_json,
r["schema"],
)
)
stored.append(r)
if entries:
watcher = self
def _write(conn):
# Skip databases removed while we were scanning
live = [e for e in entries if e[0] in watcher.states]
write_catalog_entries(conn, live)
await self.ds.get_internal_database().execute_write_fn(_write)
for r in stored:
state = r["state"]
state.catalog_version = r["version"]
state.missing = False
if r["fp"] is not None:
state.fp = r["fp"]
state.fp_racy = is_racy(r["fp"], r["t"])
if r["fp"][0] is not None:
state.db._file_seen = True
state.stats["scans"] += 1
self.counters["scans"] += 1
if missing:
for state in missing:
if not state.missing:
state.missing = True
state.db._invalidate_connections()
await self._clear_missing(missing)
async def _clear_missing(self, states):
names = [s.name for s in states]
def _clear(conn):
for name in names:
for table in CATALOG_TABLES[:-1]:
conn.execute(f"DELETE FROM {table} WHERE database_name = ?", [name])
conn.execute(
"UPDATE catalog_databases SET schema_version = NULL, fingerprint = NULL "
"WHERE database_name = ?",
[name],
)
await self.ds.get_internal_database().execute_write_fn(_clear)
for state in states:
state.catalog_version = None
async def _delete_catalog(self, names):
watcher = self
def _delete(conn):
for name in names:
if name in watcher.states:
# Re-added under the same name; its own scan replaces rows
continue
for table in CATALOG_TABLES:
conn.execute(f"DELETE FROM {table} WHERE database_name = ?", [name])
await self.ds.get_internal_database().execute_write_fn(_delete)
for name in names:
self._pending_removals.discard(name)
# ------------------------------------------------------------------
# startup
# ------------------------------------------------------------------
async def initial_scan(self):
"""Full catalog build at startup.
Deletes catalog rows for databases that are not attached (a
persistent --internal database can carry rows from a previous run),
then scans every database - except those whose persisted fingerprint
proves the file is unchanged since the catalog was written.
"""
internal = self.ds.get_internal_database()
names = list(self.states.keys())
names_json = json.dumps(names)
def _delete_stale(conn):
for table in CATALOG_TABLES:
conn.execute(
f"DELETE FROM {table} WHERE database_name IS NULL OR database_name NOT IN "
"(SELECT value FROM json_each(?))",
[names_json],
)
await internal.execute_write_fn(_delete_stale)
self._pending_removals.clear()
states = [s for s in self.states.values() if not s.removed]
persisted = {}
if not internal.is_temp_disk:
rows = await internal.execute(
"select database_name, path, schema_version, fingerprint from catalog_databases"
)
persisted = {row["database_name"]: row for row in rows.rows}
to_scan = []
if persisted:
reuse = await asyncio.to_thread(self._reusable_sync, states, persisted)
for state in states:
hit = reuse.get(state.name)
if hit is None:
to_scan.append(state)
continue
fp, t_ns, version = hit
state.db._file_seen = True
state.fp = fp
state.fp_racy = is_racy(fp, t_ns)
state.catalog_version = version
state.needs_scan = False
self.counters["restored_from_persisted"] += 1
else:
to_scan = states
for state in to_scan:
state.needs_scan = True
if to_scan:
await self.refresh(to_scan)
self._pending.clear()
def _reusable_sync(self, states, persisted):
"""Databases whose stored fingerprint matches the file right now and
was not racy when it was stored. Stat calls only - no connections."""
out = {}
for state in states:
row = persisted.get(state.name)
if row is None or not state.is_file or row["fingerprint"] is None:
continue
if row["path"] != str(state.db.path) or row["schema_version"] is None:
continue
stored_fp, stored_t = _fp_from_json(row["fingerprint"])
if stored_fp is None or is_racy(stored_fp, stored_t):
continue
t_ns = time.time_ns()
fp = fingerprint(state.db.path)
if fp == stored_fp:
out[state.name] = (fp, t_ns, row["schema_version"])
return out
# ------------------------------------------------------------------
# introspection
# ------------------------------------------------------------------
def status(self):
modes = {}
for state in self.states.values():
modes[state.mode] = modes.get(state.mode, 0) + 1
return {
"interval_s": self.interval_s,
"loop_running": self.loop_running(),
"modes": modes,
"pending": len(self._pending),
"pending_removals": len(self._pending_removals),
**self.counters,
}

282
datasette/utils/internal_db.py
View file

@ -3,7 +3,8 @@ import textwrap
from sqlite_utils import Database as SQLiteUtilsDatabase
from sqlite_utils import Migrations
from datasette.utils import escape_sqlite, table_column_details
from datasette.utils import escape_sqlite, sqlite3, table_column_details
from datasette.utils.sqlite import supports_table_xinfo
INTERNAL_DB_SCHEMA_TABLES = {
"catalog_databases",
@ -180,87 +181,179 @@ async def init_internal_db(db):
await db.execute_write_fn(apply_migrations, transaction=False)
async def populate_schema_tables(internal_db, db, schema_version):
database_name = db.name
# Children first, catalog_databases last, so deletes work with FK enforcement
CATALOG_TABLES = (
"catalog_columns",
"catalog_foreign_keys",
"catalog_indexes",
"catalog_views",
"catalog_tables",
"catalog_databases",
)
tables = (await db.execute("select * from sqlite_master WHERE type = 'table'")).rows
views = (await db.execute("select * from sqlite_master WHERE type = 'view'")).rows
def collect_info(conn):
tables_to_insert = []
views_to_insert = []
columns_to_insert = []
foreign_keys_to_insert = []
indexes_to_insert = []
@internal_migrations(name="0002_catalog_fingerprint")
def catalog_fingerprint_column(db):
# stat() fingerprint of the database file when the catalog rows were
# written, so a restart with a persistent internal database can skip
# unchanged files without opening them
if "fingerprint" not in db["catalog_databases"].columns_dict:
db.execute("ALTER TABLE catalog_databases ADD COLUMN fingerprint TEXT")
for view in views:
view_name = view["name"]
views_to_insert.append(
(database_name, view_name, view["rootpage"], view["sql"])
)
for table in tables:
table_name = table["name"]
tables_to_insert.append(
(database_name, table_name, table["rootpage"], table["sql"])
)
columns = table_column_details(conn, table_name)
columns_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**column._asdict(),
}
for column in columns
)
foreign_keys = conn.execute(
f"PRAGMA foreign_key_list({escape_sqlite(table_name)})"
).fetchall()
foreign_keys_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**dict(foreign_key),
}
for foreign_key in foreign_keys
)
indexes = conn.execute(
f"PRAGMA index_list({escape_sqlite(table_name)})"
).fetchall()
indexes_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**dict(index),
}
for index in indexes
)
return (
tables_to_insert,
views_to_insert,
columns_to_insert,
foreign_keys_to_insert,
indexes_to_insert,
_COLUMNS_SQL = """
SELECT m.name, p.cid, p.name, p.type, p."notnull", p.dflt_value, p.pk, p.hidden
FROM sqlite_master m, pragma_table_xinfo(m.name) p WHERE m.type = 'table'
"""
_FOREIGN_KEYS_SQL = """
SELECT m.name, p.id, p.seq, p."table", p."from", p."to", p.on_update, p.on_delete, p."match"
FROM sqlite_master m, pragma_foreign_key_list(m.name) p WHERE m.type = 'table'
"""
_INDEXES_SQL = """
SELECT m.name, p.seq, p.name, p."unique", p.origin, p.partial
FROM sqlite_master m, pragma_index_list(m.name) p WHERE m.type = 'table'
"""
def collect_schema(conn, database_name):
"""Read everything the catalog needs from one connection.
Run it inside a read transaction so it sees a single snapshot. Uses
three joined pragma table-valued-function queries per database instead
of three PRAGMA calls per table, falling back to the per-table loop if
that fails (e.g. a virtual table whose module is not loaded).
"""
if supports_table_xinfo():
try:
return _collect_schema_joined(conn, database_name)
except sqlite3.DatabaseError:
pass
return _collect_schema_per_table(conn, database_name)
def _collect_schema_joined(conn, database_name):
tables = conn.execute("select * from sqlite_master WHERE type = 'table'").fetchall()
views = conn.execute("select * from sqlite_master WHERE type = 'view'").fetchall()
columns = [
{
"database_name": database_name,
"table_name": r[0],
"cid": r[1],
"name": r[2],
"type": r[3],
"notnull": r[4],
"default_value": r[5],
"is_pk": r[6],
"hidden": r[7],
}
for r in conn.execute(_COLUMNS_SQL).fetchall()
]
foreign_keys = [
{
"database_name": database_name,
"table_name": r[0],
"id": r[1],
"seq": r[2],
"table": r[3],
"from": r[4],
"to": r[5],
"on_update": r[6],
"on_delete": r[7],
"match": r[8],
}
for r in conn.execute(_FOREIGN_KEYS_SQL).fetchall()
]
indexes = [
{
"database_name": database_name,
"table_name": r[0],
"seq": r[1],
"name": r[2],
"unique": r[3],
"origin": r[4],
"partial": r[5],
}
for r in conn.execute(_INDEXES_SQL).fetchall()
]
return {
"tables": [(database_name, t["name"], t["rootpage"], t["sql"]) for t in tables],
"views": [(database_name, v["name"], v["rootpage"], v["sql"]) for v in views],
"columns": columns,
"foreign_keys": foreign_keys,
"indexes": indexes,
}
def _collect_schema_per_table(conn, database_name):
tables = conn.execute("select * from sqlite_master WHERE type = 'table'").fetchall()
views = conn.execute("select * from sqlite_master WHERE type = 'view'").fetchall()
tables_to_insert = []
views_to_insert = []
columns_to_insert = []
foreign_keys_to_insert = []
indexes_to_insert = []
for view in views:
views_to_insert.append(
(database_name, view["name"], view["rootpage"], view["sql"])
)
(
tables_to_insert,
views_to_insert,
columns_to_insert,
foreign_keys_to_insert,
indexes_to_insert,
) = await db.execute_fn(collect_info)
for table in tables:
table_name = table["name"]
tables_to_insert.append(
(database_name, table_name, table["rootpage"], table["sql"])
)
columns = table_column_details(conn, table_name)
columns_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**column._asdict(),
}
for column in columns
)
foreign_keys = conn.execute(
f"PRAGMA foreign_key_list({escape_sqlite(table_name)})"
).fetchall()
foreign_keys_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**dict(foreign_key),
}
for foreign_key in foreign_keys
)
indexes = conn.execute(
f"PRAGMA index_list({escape_sqlite(table_name)})"
).fetchall()
indexes_to_insert.extend(
{
"database_name": database_name,
"table_name": table_name,
**dict(index),
}
for index in indexes
)
return {
"tables": tables_to_insert,
"views": views_to_insert,
"columns": columns_to_insert,
"foreign_keys": foreign_keys_to_insert,
"indexes": indexes_to_insert,
}
def replace_catalog(conn):
def write_catalog_entries(conn, entries):
"""Replace the catalog rows for each database in ``entries``.
Each entry is (database_name, path, is_memory, schema_version,
fingerprint_json, schema) where schema comes from collect_schema().
Runs on the internal database write connection, in one transaction.
"""
for database_name, path, is_memory, schema_version, fingerprint, schema in entries:
# Delete child rows before their catalog_tables parents so this also
# works if a prepare_connection plugin enables foreign key enforcement.
for table in (
"catalog_columns",
"catalog_foreign_keys",
"catalog_indexes",
"catalog_views",
"catalog_tables",
):
for table in CATALOG_TABLES[:-1]:
conn.execute(
f"DELETE FROM {table} WHERE database_name = ?",
[database_name],
@ -268,29 +361,24 @@ async def populate_schema_tables(internal_db, db, schema_version):
conn.execute(
"""
INSERT OR REPLACE INTO catalog_databases (
database_name, path, is_memory, schema_version
) VALUES (?, ?, ?, ?)
database_name, path, is_memory, schema_version, fingerprint
) VALUES (?, ?, ?, ?, ?)
""",
[
database_name,
str(db.path) if db.path is not None else None,
db.is_memory,
schema_version,
],
[database_name, path, is_memory, schema_version, fingerprint],
)
conn.executemany(
"""
INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
values (?, ?, ?, ?)
""",
tables_to_insert,
schema["tables"],
)
conn.executemany(
"""
INSERT INTO catalog_views (database_name, view_name, rootpage, sql)
values (?, ?, ?, ?)
""",
views_to_insert,
schema["views"],
)
conn.executemany(
"""
@ -300,7 +388,7 @@ async def populate_schema_tables(internal_db, db, schema_version):
:database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden
)
""",
columns_to_insert,
schema["columns"],
)
conn.executemany(
"""
@ -310,7 +398,7 @@ async def populate_schema_tables(internal_db, db, schema_version):
:database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match
)
""",
foreign_keys_to_insert,
schema["foreign_keys"],
)
conn.executemany(
"""
@ -320,7 +408,29 @@ async def populate_schema_tables(internal_db, db, schema_version):
:database_name, :table_name, :seq, :name, :unique, :origin, :partial
)
""",
indexes_to_insert,
schema["indexes"],
)
async def populate_schema_tables(internal_db, db, schema_version):
"""Rebuild the catalog rows for one database (kept for compatibility -
the SchemaWatcher reads the schema on its own short-lived connection)."""
database_name = db.name
def _collect(conn):
return collect_schema(conn, database_name)
schema = await db.execute_fn(_collect)
entry = (
database_name,
str(db.path) if db.path is not None else None,
db.is_memory,
schema_version,
None,
schema,
)
def replace_catalog(conn):
write_catalog_entries(conn, [entry])
await internal_db.execute_write_fn(replace_catalog)

2
datasette/views/base.py
View file

@ -116,7 +116,7 @@ class BaseView:
async def dispatch_request(self, request):
if self.ds:
await self.ds.refresh_schemas()
await self.ds._schema_watcher.on_request()
handler = getattr(self, request.method.lower(), None)
response = await handler(request)
if self.ds.cors:

4
datasette/views/database.py
View file

@ -80,7 +80,7 @@ class DatabaseView(View):
async def get(self, request, datasette):
format_ = request.url_vars.get("format") or "html"
await datasette.refresh_schemas()
await datasette._schema_watcher.on_request()
db = await datasette.resolve_database(request)
database = db.name
@ -711,7 +711,7 @@ class QueryView(View):
async def get(self, request, datasette):
from datasette.app import TableNotFound
await datasette.refresh_schemas()
await datasette._schema_watcher.on_request()
db = await datasette.resolve_database(request)
database = db.name

4
datasette/views/special.py
View file

@ -312,7 +312,7 @@ class AllowedResourcesView(BaseView):
async def get(self, request):
await self.ds.ensure_permission(action="view-instance", actor=request.actor)
await self.ds.refresh_schemas()
await self.ds._schema_watcher.on_request()
# Check if user has permissions-debug (to show sensitive fields)
has_debug_permission = await self.ds.allowed(
@ -530,7 +530,7 @@ class PermissionRulesView(BaseView):
union_sql, union_params, _restriction_sqls = await build_permission_rules_sql(
self.ds, actor, action
)
await self.ds.refresh_schemas()
await self.ds._schema_watcher.on_request()
db = self.ds.get_internal_database()
count_query = f"""

2
datasette/views/table.py
View file

@ -1738,7 +1738,7 @@ async def _sort_order(table_metadata, sortable_columns, request, order_by):
async def table_view(datasette, request):
await datasette.refresh_schemas()
await datasette._schema_watcher.on_request()
with tracer.trace_child_tasks():
response = await table_view_traced(datasette, request)

24
docs/cli-reference.rst
View file

@ -252,18 +252,18 @@ These can be passed to ``datasette serve`` using ``datasette serve --setting nam
num_sql_threads Number of threads in the thread pool for
executing SQLite queries (default=3)
max_open_connections Maximum number of pooled read connections open
across all databases - least recently used idle
connections are closed first, 0 for no limit
(default=128)
connection_idle_timeout Close pooled read connections that have been idle
for this many seconds - 0 to keep them open
(default=60)
connection_pool_wait_ms When max_open_connections is reached and every
connection is in use, wait this long for one to
be returned before opening one more (default=0)
pool_read_connections Reuse read connections between queries - turn off
to open a new connection for every query
(default=True)
across all databases (at least 4 x
num_sql_threads) - 0 for no limit (default=128)
connection_idle_timeout_ms Close read connections, and stop write threads,
that have been idle for this many milliseconds -
0 to keep them open (default=30000)
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=external)
sql_time_limit_ms Time limit for a SQL query in milliseconds
(default=1000)
default_facet_size Number of values to return for requested facets

2
docs/internals.rst
View file

@ -2828,7 +2828,7 @@ The internal database schema is as follows:
path TEXT,
is_memory INTEGER,
schema_version INTEGER
);
, fingerprint TEXT);
CREATE TABLE catalog_tables (
database_name TEXT,
table_name TEXT,

134
docs/settings.rst
View file

@ -227,81 +227,103 @@ Default: ``128``
.. [[[end]]]
Read queries borrow a connection from a pool for the duration of one query and return it afterwards. This setting caps how many pooled read connections can be open at once across all attached file databases. When a new connection is needed and the cap has been reached, the least recently used idle connection is closed first.
Read queries borrow a connection from a pool for the duration of one query (or one :ref:`execute_fn() <database_execute_fn>` callback) and return it afterwards. This setting caps how many pooled read connections can be open at once, across all attached file databases. When a new connection is needed and the cap has been reached, the least recently used idle connection - which may belong to any database - is closed first.
If every connection is in use the pool waits up to :ref:`setting_connection_pool_wait_ms` and then opens one more anyway, so the cap can be exceeded by at most :ref:`setting_num_sql_threads` connections. Set this to 0 for no limit.
This is what lets one Datasette instance serve hundreds or thousands of database files without holding a connection, and its file descriptors, open to every one of them.
Every SQL thread can hold one read connection at a time, so a cap smaller than the number of threads would close and reopen connections for nearly every query. Datasette therefore raises the cap it uses to at least four times :ref:`setting_num_sql_threads` (12 with the default of 3 threads). ``/-/settings.json`` shows the value you configured.
Set this to 0 for no limit, in which case idle connections are only closed by :ref:`setting_connection_idle_timeout_ms`.
::
datasette *.db --setting max_open_connections 64
.. _setting_connection_idle_timeout:
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.
connection_idle_timeout
~~~~~~~~~~~~~~~~~~~~~~~
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.
.. _setting_connection_idle_timeout_ms:
connection_idle_timeout_ms
~~~~~~~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "connection_idle_timeout")
.. ]]]
Default: ``60``
.. [[[end]]]
Pooled read connections that have not been used for this many seconds are closed. Set to 0 to keep idle connections open until they are evicted by :ref:`setting_max_open_connections`.
.. _setting_connection_pool_wait_ms:
connection_pool_wait_ms
~~~~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "connection_pool_wait_ms")
.. ]]]
Default: ``0``
.. [[[end]]]
How long a query waits for a connection to be returned when :ref:`setting_max_open_connections` has been reached and every pooled connection is in use, before opening one more connection anyway.
.. _setting_pool_read_connections:
pool_read_connections
~~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "pool_read_connections")
.. ]]]
Default: ``on``
.. [[[end]]]
Reuse read connections between queries. Turn this off to open, prepare and close a new connection for every read query.
.. _setting_write_thread_idle_timeout_ms:
write_thread_idle_timeout_ms
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "write_thread_idle_timeout_ms")
setting_default(cog, "connection_idle_timeout_ms")
.. ]]]
Default: ``30000``
.. [[[end]]]
Each database that receives writes gets its own write thread, holding a single write connection. Writes to that database are queued and executed one at a time by that thread.
Connections that have not been used for this many milliseconds are closed, so a database that stops receiving traffic stops holding file descriptors, memory and threads.
If a write thread has had no work for this many milliseconds it closes its connection and exits. The next write to that database starts a new thread and opens a new connection. This keeps an instance with many databases from holding a thread and a file descriptor open for every database it has ever written to.
- **Read connections** in the pool are closed once they have been idle this long. The next read to that database opens a new connection.
- **Write threads**: each database that receives writes gets its own write thread, holding a single write connection, and writes to that database are run one at a time by that thread. A write thread that has had no work for this long closes its connection and exits. The next write to that database starts a new thread and opens a new connection.
Anything a write function attaches to the write connection itself - temporary tables, ``ATTACH`` statements, ``PRAGMA`` settings - is lost when the thread exits. In-memory databases are exempt, since closing their only write connection would discard their contents.
Opening a new connection costs around a millisecond (more if :ref:`plugin_hook_prepare_connection` hooks or extensions such as SpatiaLite have to be loaded again), which is negligible for a database that has been idle for 30 seconds, so the default keeps busy databases warm and releases forgotten ones quickly.
Set this to 0 to keep write threads running until Datasette shuts down::
Anything a function attaches to a connection itself - temporary tables, ``ATTACH`` statements, ``PRAGMA`` settings, functions registered with ``conn.create_function()`` - is lost when that connection is closed, and is never shared between pooled connections. Set that up in the :ref:`plugin_hook_prepare_connection` plugin hook instead, which runs once for every new connection.
In-memory databases are exempt, since closing their only write connection would discard their contents. With :ref:`setting_num_sql_threads` set to 0 there are no threads to time out, so the write connection stays open until Datasette shuts down; read connections are still closed when idle.
Set this to 0 to keep read connections open until they are evicted by :ref:`setting_max_open_connections`, and write threads running until Datasette shuts down::
datasette mydatabase.db --setting connection_idle_timeout_ms 0
.. _setting_schema_watch_interval_ms:
schema_watch_interval_ms
~~~~~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "schema_watch_interval_ms")
.. ]]]
Default: ``1000``
.. [[[end]]]
Datasette keeps a catalog of every attached database's tables, columns, indexes and foreign keys in its :ref:`internal database <internals_internal>`. Each database has a *schema watch mode* that decides how that catalog is kept current:
``owned``
Only Datasette changes this database's schema. After every write Datasette makes to it, Datasette checks whether the schema changed and, if it did, updates the catalog before the write call returns. Owned databases are never polled.
``external``
Other processes may change the file too - for example ``sqlite-utils`` adding a table while Datasette is serving it. On top of the checks after Datasette's own writes, the file is polled every ``schema_watch_interval_ms`` milliseconds. Each check is a ``stat()`` of the database file and its ``-wal`` and ``-journal`` files; a database is only opened if those changed, and its catalog is only rebuilt if its schema changed. Databases whose file is replaced (for example by renaming another file over it) or deleted are noticed too: open connections to the old file are discarded and the catalog is updated.
``immutable``
The catalog is built once at startup. Immutable databases (opened with ``-i``) always use this mode.
Set this to ``0`` to turn polling off. External databases are then only checked after Datasette's own writes to them::
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:
.. code-block:: yaml
databases:
mydatabase:
schema_watch: owned # or external, or immutable
.. _setting_default_schema_watch:
default_schema_watch
~~~~~~~~~~~~~~~~~~~~
.. [[[cog
setting_default(cog, "default_schema_watch")
.. ]]]
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 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::
datasette *.db --setting default_schema_watch owned
datasette mydatabase.db --setting write_thread_idle_timeout_ms 0
.. _setting_allow_facet:

7
tests/test_api.py
View file

@ -760,10 +760,9 @@ async def test_settings_json(ds_client):
"default_cache_ttl": 5,
"num_sql_threads": 1,
"max_open_connections": 128,
"connection_idle_timeout": 60,
"connection_pool_wait_ms": 0,
"pool_read_connections": True,
"write_thread_idle_timeout_ms": 30000,
"connection_idle_timeout_ms": 30000,
"schema_watch_interval_ms": 1000,
"default_schema_watch": "external",
"cache_size_kb": 0,
"allow_csv_stream": True,
"max_csv_mb": 100,

6
tests/test_internal_db.py
View file

@ -29,7 +29,8 @@ async def test_internal_migrations_recorded(ds_client):
order by id
""")
assert [tuple(row) for row in migrations.rows] == [
("datasette_internal", "0001_initial")
("datasette_internal", "0001_initial"),
("datasette_internal", "0002_catalog_fingerprint"),
]
@ -61,7 +62,8 @@ async def test_internal_migrations_adopt_existing_internal_db(tmp_path):
order by id
""")
assert [tuple(row) for row in migrations.rows] == [
("datasette_internal", "0001_initial")
("datasette_internal", "0001_initial"),
("datasette_internal", "0002_catalog_fingerprint"),
]
ds.close()

170
tests/test_read_pool.py
View file

@ -105,11 +105,10 @@ async def test_leased_connection_compatibility(tmp_path):
@pytest.mark.parametrize("num_sql_threads", [0, 3])
async def test_global_cap_evicts_least_recently_used(tmp_path, num_sql_threads):
paths = _make_dbs(tmp_path, 6)
ds = Datasette(
paths,
settings={"num_sql_threads": num_sql_threads, "max_open_connections": 2},
)
ds = Datasette(paths, settings={"num_sql_threads": num_sql_threads})
pool = ds._read_pool
# Below the 4 x num_sql_threads clamp, so set the private attribute
pool.max_open = 2
for i in range(6):
await ds.get_database(f"db{i}").execute("select 1")
assert pool.snapshot()["open"] <= 2
@ -130,7 +129,8 @@ async def test_global_cap_evicts_least_recently_used(tmp_path, num_sql_threads):
@pytest.mark.asyncio
async def test_prepare_connection_runs_for_every_pooled_connection(tmp_path):
paths = _make_dbs(tmp_path, 3)
ds = Datasette(paths, settings={"max_open_connections": 1})
ds = Datasette(paths)
ds._read_pool.max_open = 1
calls = []
original = ds._prepare_connection
@ -174,54 +174,61 @@ def test_idle_reaper_closes_connections_without_event_loop(tmp_path):
ds.close()
@pytest.mark.asyncio
async def test_pooling_disabled_opens_connection_per_query(tmp_path):
(path,) = _make_dbs(tmp_path, 1)
ds = Datasette([path], settings={"pool_read_connections": False})
db = ds.get_database("db0")
for _ in range(5):
await db.execute("select 1")
pool = ds._read_pool
assert pool.stats["opened"] == 5
assert pool.snapshot()["open"] == 0
assert _file_conns(db) == []
@pytest.mark.parametrize(
"num_sql_threads,configured,expected",
[
(3, 128, 128),
(3, 5, 12),
(8, 20, 32),
(0, 1, 1),
(3, 0, 0),
],
)
def test_max_open_connections_clamped_to_four_times_threads(
num_sql_threads, configured, expected
):
ds = Datasette(
settings={
"num_sql_threads": num_sql_threads,
"max_open_connections": configured,
}
)
assert ds._read_pool.max_open == expected
# /-/settings.json reports the configured value
assert ds.setting("max_open_connections") == configured
ds.close()
def test_connection_idle_timeout_ms_setting():
ds = Datasette(settings={"connection_idle_timeout_ms": 1500})
assert ds._read_pool.idle_timeout == 1.5
ds.close()
ds = Datasette(settings={"connection_idle_timeout_ms": 0})
assert ds._read_pool.idle_timeout == 0
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("wait_ms,expect_exceeded", [(0, True), (10000, False)])
async def test_cap_exhausted_policy(tmp_path, wait_ms, expect_exceeded):
async def test_soft_cap_when_every_connection_is_leased(tmp_path):
# Only reachable with a cap below the thread count, which the setting
# clamp prevents: open one more connection rather than wait, and close
# it again on release
paths = _make_dbs(tmp_path, 3)
ds = Datasette(
paths,
settings={
"num_sql_threads": 3,
"max_open_connections": 1,
"connection_pool_wait_ms": wait_ms,
},
)
barrier = threading.Barrier(3, timeout=0.5)
ds = Datasette(paths, settings={"num_sql_threads": 3})
pool = ds._read_pool
pool.max_open = 1
barrier = threading.Barrier(3, timeout=5)
def slow(conn):
try:
# With a hard cap only one callback can run at a time
barrier.wait()
except threading.BrokenBarrierError:
pass
time.sleep(0.05)
barrier.wait()
return conn.execute("select 1").fetchone()[0]
results = await asyncio.gather(
*[ds.get_database(f"db{i}").execute_fn(slow) for i in range(3)]
)
assert results == [1, 1, 1]
pool = ds._read_pool
if expect_exceeded:
assert pool.stats["exceeded_cap"] >= 1
assert pool.stats["peak_open"] > 1
else:
assert pool.stats["exceeded_cap"] == 0
assert pool.stats["peak_open"] == 1
assert pool.stats["exceeded_cap"] >= 2
assert pool.stats["peak_open"] == 3
# Back under the cap once everything has been returned
assert pool.snapshot()["open"] <= 1
ds.close()
@ -282,3 +289,86 @@ async def test_open_transaction_rolled_back_on_release(tmp_path):
assert ds._read_pool.stats["rolled_back"] == 1
assert not await db.execute_fn(lambda conn: conn.in_transaction)
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", [0, 2])
async def test_invalidate_database_discards_stale_connections(
tmp_path, num_sql_threads
):
paths = _make_dbs(tmp_path, 2)
ds = Datasette(paths, settings={"num_sql_threads": num_sql_threads})
db0, db1 = ds.get_database("db0"), ds.get_database("db1")
pool = ds._read_pool
await db0.execute("select 1")
await db1.execute("select 1")
old = db0._read_pool_state.idle[-1].conn
assert db0._read_pool_state.open == 1
# What the SchemaWatcher does when it sees db0's file replaced
db0._invalidate_connections()
# Idle connections for db0 are closed at once; db1 is untouched
assert db0._read_pool_state.open == 0
assert db1._read_pool_state.open == 1
assert old not in db0._all_connections
with pytest.raises(sqlite3.ProgrammingError):
old.execute("select 1")
# The next read opens a new connection
await db0.execute("select 1")
assert db0._read_pool_state.idle[-1].conn is not old
assert pool.stats["discarded_stale"] == 1
ds.close()
@pytest.mark.asyncio
async def test_leased_connection_discarded_on_release_after_invalidation(tmp_path):
(path,) = _make_dbs(tmp_path, 1)
ds = Datasette([path], settings={"num_sql_threads": 2})
db = ds.get_database("db0")
pool = ds._read_pool
seen = {}
def callback(conn):
seen["raw"] = conn.execute("select 1").connection
# The file is replaced while this callback holds the connection:
# it must not be closed under us, only discarded on release
db._invalidate_connections()
return conn.execute("select count(*) from t").fetchone()[0]
assert await db.execute_fn(callback) == 1
assert db._read_pool_state.open == 0
assert pool.stats["discarded_stale"] == 1
with pytest.raises(sqlite3.ProgrammingError):
seen["raw"].execute("select 1")
assert (await db.execute("select count(*) from t")).single_value() == 1
ds.close()
def test_reaper_unavailable_falls_back_to_checkout_reaping(tmp_path, monkeypatch):
# Pyodide cannot start threads: idle connections must then be reaped by
# the next checkout instead, and reads must keep working
from datasette import connection_pool
reaper = connection_pool._Reaper()
monkeypatch.setattr(connection_pool, "_reaper", reaper)
def no_threads(self):
raise RuntimeError("can't start new thread")
monkeypatch.setattr(threading.Thread, "start", no_threads)
paths = _make_dbs(tmp_path, 2)
ds = Datasette(paths, settings={"num_sql_threads": 0})
pool = ds._read_pool
pool.idle_timeout = 0.05
async def read(name):
return (await ds.get_database(name).execute("select 1")).single_value()
assert asyncio.run(read("db0")) == 1
assert reaper.unavailable
assert pool.snapshot()["open"] == 1
time.sleep(0.1)
assert asyncio.run(read("db1")) == 1
# db0's expired connection was closed during db1's checkout
assert pool.stats["expired"] == 1
assert ds.get_database("db0")._read_pool_state.open == 0
ds.close()

693
tests/test_schema_watcher.py Normal file
View file

@ -0,0 +1,693 @@
"""
SchemaWatcher tests. Changes to "external" databases are made from a
separate OS process (a child Python interpreter using plain sqlite3, or the
sqlite-utils CLI), the way other tools change files Datasette is serving.
"""
import asyncio
import os
import sqlite3
import subprocess
import sys
import textwrap
import time
import pytest
from datasette import schema_watcher as schema_watcher_module
from datasette.app import Datasette
from datasette.database import Database
INTERVAL_MS = 50
def run_external(path, sql, journal_mode=None, keep_wal=False):
"""Run sql against path in a separate Python process; returns the
child's time.monotonic() right after COMMIT."""
script = textwrap.dedent("""
import sqlite3, sys, time
path, sql, journal_mode, keep_wal = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]
conn = sqlite3.connect(path, isolation_level=None, timeout=10)
if journal_mode:
conn.execute("PRAGMA journal_mode=" + journal_mode)
if keep_wal == "1":
# Never checkpoint: changes stay in the -wal file
conn.execute("PRAGMA wal_autocheckpoint=0")
conn.executescript(sql)
print(time.monotonic(), flush=True)
if keep_wal == "1":
# Keep the connection open so close() does not checkpoint
sys.stdin.read()
conn.close()
""")
if keep_wal:
proc = subprocess.Popen(
[sys.executable, "-c", script, path, sql, journal_mode or "", "1"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
)
line = proc.stdout.readline()
return float(line), proc
out = subprocess.run(
[sys.executable, "-c", script, path, sql, journal_mode or "", "0"],
check=True,
capture_output=True,
text=True,
)
return float(out.stdout.strip())
def run_process(*args):
subprocess.run([sys.executable, *args], check=True)
def make_db(path, sql="create table t (id integer primary key, a text)", wal=False):
conn = sqlite3.connect(path)
if wal:
conn.execute("PRAGMA journal_mode=wal")
conn.executescript(sql)
conn.execute("insert into t (a) values ('one')")
conn.commit()
conn.close()
async def catalog_columns(ds, db_name, table):
rows = await ds.get_internal_database().execute(
"select name from catalog_columns where database_name = ? and table_name = ? order by cid",
[db_name, table],
)
return [r[0] for r in rows.rows]
async def catalog_tables(ds, db_name):
rows = await ds.get_internal_database().execute(
"select table_name from catalog_tables where database_name = ? order by table_name",
[db_name],
)
return [r[0] for r in rows.rows]
async def wait_for(check, timeout=5.0):
t0 = time.monotonic()
while True:
result = await check()
if result:
return time.monotonic() - t0
if time.monotonic() - t0 > timeout:
raise AssertionError(f"timed out after {timeout}s")
await asyncio.sleep(0.01)
@pytest.fixture
def db_path(tmp_path):
path = str(tmp_path / "ext.db")
make_db(path)
return path
async def start_ds(paths, interval_ms=INTERVAL_MS, settings=None, **kwargs):
ds = Datasette(
files=paths,
settings={"schema_watch_interval_ms": interval_ms, **(settings or {})},
**kwargs,
)
await ds.invoke_startup()
await ds.start_background_tasks()
return ds
@pytest.mark.asyncio
async def test_modes_and_defaults(tmp_path, db_path):
immutable_path = str(tmp_path / "imm.db")
make_db(immutable_path)
ds = Datasette(files=[db_path], immutables=[immutable_path])
await ds.invoke_startup()
watcher = ds._schema_watcher
assert watcher.states["ext"].mode == "external"
assert watcher.states["imm"].mode == "immutable"
added = ds.add_database(Database(ds, memory_name="sw_owned_mem"), name="added")
assert watcher.states["added"].mode == "owned"
assert added._watch_state is watcher.states["added"]
ds.close()
@pytest.mark.asyncio
async def test_mode_from_config(tmp_path, db_path):
ds = Datasette(
files=[db_path], config={"databases": {"ext": {"schema_watch": "owned"}}}
)
assert ds._schema_watcher.states["ext"].mode == "owned"
ds.close()
@pytest.mark.asyncio
async def test_no_request_path_sweep(db_path):
ds = await start_ds([db_path], interval_ms=0)
watcher = ds._schema_watcher
sweeps = watcher.counters["sweeps"]
for _ in range(5):
response = await ds.client.get("/ext/t.json")
assert response.status_code == 200
assert watcher.counters["sweeps"] == sweeps
ds.close()
@pytest.mark.asyncio
async def test_owned_write_path_detects_schema_change(tmp_path):
ds = Datasette(settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
db = ds.add_database(Database(ds, memory_name="sw_owned"), name="owned")
await ds.refresh_schemas(force=True)
watcher = ds._schema_watcher
scans = watcher.counters["scans"]
# Data-only writes: no catalog rebuild
await db.execute_write("create table if not exists t (id integer primary key)")
assert await catalog_tables(ds, "owned") == ["t"]
scans = watcher.counters["scans"]
for i in range(5):
await db.execute_write("insert into t default values")
assert watcher.counters["scans"] == scans
# Each write API: catalog is current as soon as the call returns
await db.execute_write("create table t2 (id integer primary key)")
assert "t2" in await catalog_tables(ds, "owned")
await db.execute_write_script("create table t3 (id integer); create table t4 (id);")
assert {"t3", "t4"} <= set(await catalog_tables(ds, "owned"))
scans = watcher.counters["scans"]
await db.execute_write_many("insert into t2 (id) values (?)", [[1], [2]])
assert watcher.counters["scans"] == scans
def plugin_write(conn):
conn.execute("alter table t add column extra text")
await db.execute_write_fn(plugin_write)
assert await catalog_columns(ds, "owned", "t") == ["id", "extra"]
def isolated(conn):
conn.execute("create table t6 (id)")
await db.execute_isolated_fn(isolated)
assert "t6" in await catalog_tables(ds, "owned")
# block=False: refreshed in the background
await db.execute_write("create table t7 (id)", block=False)
await wait_for(lambda: _contains(ds, "owned", "t7"))
ds.close()
async def _contains(ds, db_name, table):
return table in await catalog_tables(ds, db_name)
@pytest.mark.asyncio
async def test_owned_write_path_non_threaded(tmp_path):
path = str(tmp_path / "nt.db")
make_db(path)
ds = Datasette(settings={"num_sql_threads": 0, "schema_watch_interval_ms": 0})
await ds.invoke_startup()
db = ds.add_database(Database(ds, path=path), name="nt")
await ds._schema_watcher.flush_pending()
await db.execute_write("create table t2 (id)")
assert await catalog_tables(ds, "nt") == ["t", "t2"]
ds.close()
@pytest.mark.asyncio
async def test_external_data_only_insert_does_not_rescan(db_path):
ds = await start_ds([db_path])
watcher = ds._schema_watcher
state = watcher.states["ext"]
scans = state.stats["scans"]
checks = state.stats["checks"]
for i in range(3):
run_external(db_path, f"insert into t (a) values ('x{i}')")
await asyncio.sleep(INTERVAL_MS * 3 / 1000)
# The fingerprint changed, so schema_version was checked...
assert state.stats["checks"] > checks
# ... but the catalog was not rebuilt
assert state.stats["scans"] == scans
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("use_header", (True, False))
async def test_external_add_column(db_path, monkeypatch, use_header):
monkeypatch.setattr(schema_watcher_module, "USE_HEADER_CHECK", use_header)
ds = await start_ds([db_path])
run_external(db_path, "alter table t add column b text")
async def has_b():
return await catalog_columns(ds, "ext", "t") == ["id", "a", "b"]
await wait_for(has_b)
ds.close()
@pytest.mark.asyncio
async def test_external_sqlite_utils_transform(db_path):
ds = await start_ds([db_path])
run_process(
"-m", "sqlite_utils", "transform", db_path, "t", "--rename", "a", "renamed"
)
async def renamed():
return await catalog_columns(ds, "ext", "t") == ["id", "renamed"]
await wait_for(renamed)
# And the table page uses the new column too
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "renamed": "one"}]
ds.close()
@pytest.mark.asyncio
async def test_external_create_table_in_wal_before_checkpoint(tmp_path):
path = str(tmp_path / "walled.db")
make_db(path, wal=True)
ds = await start_ds([path])
_, proc = run_external(
path, "create table in_wal (id integer)", journal_mode="wal", keep_wal=True
)
try:
assert os.path.getsize(path + "-wal") > 0
async def seen():
return "in_wal" in await catalog_tables(ds, "walled")
await wait_for(seen)
# Main file still does not contain it: it was only in the WAL
assert os.path.exists(path + "-wal")
finally:
proc.stdin.close()
proc.wait()
ds.close()
@pytest.mark.asyncio
async def test_external_atomic_replace_by_rename(tmp_path, db_path):
ds = await start_ds([db_path])
# Warm a cached read connection on the old inode
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "a": "one"}]
replacement = str(tmp_path / "replacement.db")
conn = sqlite3.connect(replacement)
# Same schema_version as the original file (1 CREATE TABLE) but a
# different table: only the inode change gives it away
conn.execute("create table t (id integer primary key, different text)")
conn.execute("insert into t (different) values ('new file')")
conn.commit()
assert conn.execute("pragma schema_version").fetchone()[0] == 1
conn.close()
old_inode = os.stat(db_path).st_ino
run_process(
"-c",
"import os, sys; os.replace(sys.argv[1], sys.argv[2])",
replacement,
db_path,
)
assert os.stat(db_path).st_ino != old_inode
async def replaced():
return await catalog_columns(ds, "ext", "t") == ["id", "different"]
await wait_for(replaced)
# Cached connections were invalidated, so queries see the new file
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "different": "new file"}]
ds.close()
@pytest.mark.asyncio
async def test_old_inode_connections_without_watcher(tmp_path, db_path):
# Documents the hazard the watcher fixes: with polling off, a cached
# read connection keeps reading the replaced (unlinked) file
# One SQL thread, so every request reuses the same cached connection
ds = await start_ds([db_path], interval_ms=0, settings={"num_sql_threads": 1})
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "a": "one"}]
replacement = str(tmp_path / "replacement.db")
conn = sqlite3.connect(replacement)
conn.execute("create table t (id integer primary key, a text)")
conn.execute("insert into t (a) values ('new file')")
conn.commit()
conn.close()
os.replace(replacement, db_path)
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "a": "one"}] # stale!
# An explicit refresh notices the inode change and fixes it
await ds.refresh_schemas(force=True)
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "a": "new file"}]
ds.close()
@pytest.mark.asyncio
async def test_external_vacuum(db_path):
run_external(
db_path,
"create table big (x); insert into big select zeroblob(1000) from "
"(with recursive c(i) as (select 1 union all select i+1 from c where i < 200) select i from c);"
"drop table big;",
)
ds = await start_ds([db_path])
before = await ds.get_internal_database().execute(
"select schema_version from catalog_databases where database_name = 'ext'"
)
run_external(db_path, "vacuum")
async def bumped():
after = await ds.get_internal_database().execute(
"select schema_version from catalog_databases where database_name = 'ext'"
)
return after.first()[0] == before.first()[0] + 1
await wait_for(bumped)
ds.close()
@pytest.mark.asyncio
async def test_external_rapid_changes(db_path):
ds = await start_ds([db_path])
for i in range(10):
run_external(db_path, f"create table rapid_{i} (id)")
async def all_seen():
tables = await catalog_tables(ds, "ext")
return all(f"rapid_{i}" in tables for i in range(10))
await wait_for(all_seen)
ds.close()
def _coarse_fingerprint(real):
# Simulate a filesystem with 1 second timestamp granularity
def fingerprint(path, present=None):
parts = real(path, present)
out = []
for i, part in enumerate(parts):
if part is None:
out.append(None)
continue
part = list(part)
# main: (dev, ino, size, mtime, ctime); side files: (ino, size, mtime)
for idx in (3, 4) if i == 0 else (2,):
part[idx] = part[idx] // 10**9 * 10**9
out.append(tuple(part))
return tuple(out)
return fingerprint
@pytest.mark.parametrize("journal_mode", ("delete", "wal", "truncate"))
def test_header_schema_version_matches_pragma(tmp_path, journal_mode):
path = str(tmp_path / "h.db")
conn = sqlite3.connect(path, isolation_level=None)
conn.execute(f"PRAGMA journal_mode={journal_mode}")
for i in range(300):
conn.execute(f"create table t{i} (id integer)")
conn.execute("alter table t1 add column b")
version = conn.execute("PRAGMA schema_version").fetchone()[0]
if journal_mode == "wal":
# Pages still in the -wal: the main file header is stale, which is
# why the watcher only reads it when the -wal is empty
assert schema_watcher_module.fingerprint(path)[1] is not None
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
assert schema_watcher_module.fingerprint(path)[1:] == (None, None)
assert schema_watcher_module.header_schema_version(path) == version == 301
conn.close()
assert (
schema_watcher_module.header_schema_version(str(tmp_path / "nope.db")) is None
)
(tmp_path / "junk.db").write_bytes(b"not a database" * 20)
assert (
schema_watcher_module.header_schema_version(str(tmp_path / "junk.db")) is None
)
@pytest.mark.asyncio
@pytest.mark.parametrize("use_header", (True, False))
@pytest.mark.parametrize("racy_rule", (True, False))
async def test_racily_clean_same_tick_change(
db_path, monkeypatch, racy_rule, use_header
):
"""Two schema changes inside one timestamp tick, same file size and
inode: only the racy-clean rule catches the second one."""
monkeypatch.setattr(schema_watcher_module, "USE_HEADER_CHECK", use_header)
monkeypatch.setattr(
schema_watcher_module,
"fingerprint",
_coarse_fingerprint(schema_watcher_module.fingerprint),
)
if not racy_rule:
monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", -(10**18))
ds = Datasette(files=[db_path], settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
# Line the first change up with the start of a fresh second
await asyncio.sleep(1 - (time.time() % 1) + 0.05)
size = os.path.getsize(db_path)
run_external(db_path, "alter table t add column b text")
await ds._refresh_schemas() # sweep: sees change 1
assert await catalog_columns(ds, "ext", "t") == ["id", "a", "b"]
run_external(db_path, "alter table t add column c text")
assert os.path.getsize(db_path) == size
assert time.time() % 1 < 0.9, "test machine too slow to stay in one tick"
await ds._refresh_schemas() # sweep: fingerprint identical to last one
columns = await catalog_columns(ds, "ext", "t")
if racy_rule:
assert columns == ["id", "a", "b", "c"]
else:
# Without the rule the change is missed
assert columns == ["id", "a", "b"]
ds.close()
@pytest.mark.asyncio
async def test_external_file_deleted_and_recreated(tmp_path, db_path):
ds = await start_ds([db_path])
assert await catalog_tables(ds, "ext") == ["t"]
run_process("-c", "import os, sys; os.unlink(sys.argv[1])", db_path)
async def gone():
return await catalog_tables(ds, "ext") == []
await wait_for(gone)
response = await ds.client.get("/ext/t.json")
assert response.status_code in (404, 500)
# Writes fail instead of silently creating a new empty file
with pytest.raises(sqlite3.OperationalError):
await ds.get_database("ext").execute_write("insert into t (a) values ('x')")
assert not os.path.exists(db_path)
make_db(db_path, sql="create table t (id integer primary key, a text, back text)")
async def back():
return await catalog_columns(ds, "ext", "t") == ["id", "a", "back"]
await wait_for(back)
response = await ds.client.get("/ext/t.json?_shape=array")
assert response.json() == [{"id": 1, "a": "one", "back": None}]
ds.close()
@pytest.mark.asyncio
async def test_remove_database_deletes_catalog_rows(tmp_path, db_path):
other = str(tmp_path / "other.db")
make_db(other)
ds = Datasette(files=[db_path, other])
await ds.invoke_startup()
assert await catalog_tables(ds, "other") == ["t"]
ds.remove_database("other")
await ds._schema_watcher.on_request()
internal = ds.get_internal_database()
for table in (
"catalog_databases",
"catalog_tables",
"catalog_columns",
"catalog_indexes",
):
rows = await internal.execute(
f"select count(*) from {table} where database_name = 'other'"
)
assert rows.first()[0] == 0
assert await catalog_tables(ds, "ext") == ["t"]
ds.close()
@pytest.mark.asyncio
async def test_add_database_visible_on_next_request(tmp_path):
ds = Datasette(settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
await ds.client.get("/-/versions.json")
path = str(tmp_path / "late.db")
make_db(path)
ds.add_database(Database(ds, path=path), name="late")
response = await ds.client.get("/late.json")
assert response.status_code == 200
assert await catalog_tables(ds, "late") == ["t"]
ds.close()
@pytest.mark.asyncio
async def test_persisted_fingerprints_skip_unchanged_files(tmp_path, monkeypatch):
# Shrink the racy window so freshly written files count as stable
monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", 1_000_000)
internal = str(tmp_path / "internal.db")
paths = []
for i in range(3):
path = str(tmp_path / f"p{i}.db")
make_db(path)
paths.append(path)
await asyncio.sleep(0.05)
ds = Datasette(files=paths, internal=internal)
await ds.invoke_startup()
assert ds._schema_watcher.counters["scans"] == 3
ds.close()
# Change one file while Datasette is "down"
run_external(paths[1], "create table added_offline (id)")
ds2 = Datasette(files=paths, internal=internal)
await ds2.invoke_startup()
counters = ds2._schema_watcher.counters
assert counters["restored_from_persisted"] == 2
assert counters["scans"] == 1
assert await catalog_tables(ds2, "p1") == ["added_offline", "t"]
assert await catalog_tables(ds2, "p0") == ["t"]
ds2.close()
@pytest.mark.asyncio
async def test_remove_database_during_catalog_scan(tmp_path):
# remove_database() closes the Database while the watcher may still be
# scanning it in a worker thread. The scan connection is not tracked by
# the Database, so close() must not pull it out from under the scan
# (that used to segfault).
path = str(tmp_path / "big.db")
conn = sqlite3.connect(path)
for i in range(150):
conn.execute(f"create table t{i} (a, b, c integer references t0(a))")
conn.execute(f"create index i{i} on t{i}(a)")
conn.commit()
conn.close()
ds = Datasette(settings={"schema_watch_interval_ms": 0})
await ds.invoke_startup()
for i in range(60):
ds.add_database(Database(ds, path=path, is_mutable=i % 2 == 0), name=f"x{i}")
await asyncio.sleep(0.0005 * (i % 5))
ds.remove_database(f"x{i}")
await asyncio.sleep(0.2)
rows = await ds.get_internal_database().execute(
"select count(*) from catalog_tables where database_name like 'x%'"
)
assert rows.first()[0] == 0
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"setting,expected", (("external", "external"), ("owned", "owned"))
)
async def test_default_schema_watch_setting(tmp_path, db_path, setting, expected):
immutable_path = str(tmp_path / "imm.db")
make_db(immutable_path)
ds = Datasette(
files=[db_path],
immutables=[immutable_path],
settings={"default_schema_watch": setting},
)
watcher = ds._schema_watcher
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
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"
polled = ds.add_database(
Database(ds, path=other), name="polled", schema_watch="external"
)
assert polled._watch_state.mode == "external"
ds.close()
def test_default_schema_watch_config_overrides(db_path):
ds = Datasette(
files=[db_path],
settings={"default_schema_watch": "owned"},
config={"databases": {"ext": {"schema_watch": "external"}}},
)
assert ds._schema_watcher.states["ext"].mode == "external"
ds.close()
def test_default_schema_watch_invalid(db_path):
from datasette.utils import StartupError
with pytest.raises(StartupError, match="default_schema_watch"):
Datasette(files=[db_path], settings={"default_schema_watch": "immutable"})
@pytest.mark.asyncio
async def test_owned_default_is_not_polled(db_path):
ds = await start_ds([db_path], settings={"default_schema_watch": "owned"})
run_external(db_path, "create table not_polled (id)")
await asyncio.sleep(INTERVAL_MS * 4 / 1000)
assert await catalog_tables(ds, "ext") == ["t"]
# Datasette's own writes are still detected immediately
await ds.get_database("ext").execute_write("create table mine (id)")
assert await catalog_tables(ds, "ext") == ["mine", "not_polled", "t"]
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", (0, 2))
async def test_scratch_database_file_created_by_first_write(tmp_path, num_sql_threads):
# A database added before its file exists (a plugin's scratch
# database): the first write creates it. Once it has existed, deleting
# it makes writes fail rather than silently creating a new empty file.
ds = Datasette(
settings={"num_sql_threads": num_sql_threads, "schema_watch_interval_ms": 0}
)
await ds.invoke_startup()
path = str(tmp_path / "scratch.db")
db = ds.add_database(Database(ds, path=path), name="scratch")
await ds._schema_watcher.flush_pending()
assert not os.path.exists(path)
await db.execute_write("create table t (id integer primary key)")
assert os.path.exists(path)
assert await catalog_tables(ds, "scratch") == ["t"]
await db.execute_write("insert into t default values")
os.unlink(path)
await ds.refresh_schemas(force=True)
assert await catalog_tables(ds, "scratch") == []
with pytest.raises(sqlite3.OperationalError):
await db.execute_write("insert into t default values")
assert not os.path.exists(path)
ds.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("num_sql_threads", (0, 2))
async def test_write_connection_follows_replaced_file(
tmp_path, db_path, num_sql_threads
):
ds = await start_ds(
[db_path], interval_ms=0, settings={"num_sql_threads": num_sql_threads}
)
db = ds.get_database("ext")
# Warm the write connection on the old inode
await db.execute_write("insert into t (a) values ('old')")
replacement = str(tmp_path / "replacement.db")
make_db(replacement)
os.replace(replacement, db_path)
await ds.refresh_schemas(force=True)
await db.execute_write("insert into t (a) values ('new')")
conn = sqlite3.connect(db_path)
try:
rows = conn.execute("select a from t order by id").fetchall()
finally:
conn.close()
# The write went to the new file, not the unlinked one
assert rows == [("one",), ("new",)]
assert [r["a"] for r in (await db.execute("select a from t order by id")).rows] == [
"one",
"new",
]
ds.close()

20
tests/test_write_thread_idle.py
View file

@ -1,5 +1,5 @@
"""
Tests for the write_thread_idle_timeout_ms setting: a database's write thread
Tests for the connection_idle_timeout_ms setting: a database's write thread
closes its connection and exits after sitting idle, and the next write
starts a new one.
"""
@ -42,7 +42,7 @@ def _write_threads(name):
@pytest.mark.asyncio
async def test_write_thread_exits_when_idle_and_restarts(tmp_path):
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=50)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=50)
try:
await db.execute_write("insert into t (writer, seq) values (1, 1)")
thread = db._write_thread
@ -66,7 +66,7 @@ async def test_write_thread_exits_when_idle_and_restarts(tmp_path):
@pytest.mark.asyncio
async def test_write_thread_idle_timeout_zero_keeps_thread(tmp_path):
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=0)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=0)
try:
await db.execute_write("insert into t (writer, seq) values (1, 1)")
thread = db._write_thread
@ -82,7 +82,7 @@ async def test_write_thread_idle_timeout_zero_keeps_thread(tmp_path):
@pytest.mark.parametrize("named", (True, False))
async def test_memory_database_write_thread_never_idles(named):
# Closing the write connection of a memory database would lose its data
ds = Datasette(settings={"write_thread_idle_timeout_ms": 10})
ds = Datasette(settings={"connection_idle_timeout_ms": 10})
if named:
db = ds.add_memory_database("idle_mem")
else:
@ -103,7 +103,7 @@ async def test_memory_database_write_thread_never_idles(named):
@pytest.mark.asyncio
async def test_close_after_idle_exit(tmp_path):
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=10)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=10)
await db.execute_write("insert into t (writer, seq) values (1, 1)")
thread = db._write_thread
assert await _wait_for(lambda: db._write_thread is None)
@ -116,7 +116,7 @@ async def test_close_after_idle_exit(tmp_path):
@pytest.mark.asyncio
async def test_close_drains_queued_writes_with_idle_timeout(tmp_path):
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=10)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=10)
release = threading.Event()
def slow(conn):
@ -149,7 +149,7 @@ async def test_close_drains_queued_writes_with_idle_timeout(tmp_path):
async def test_connect_failure_retried_after_queue_drains(tmp_path):
# With an idle timeout, a failed connect is reported to the writes that
# were queued, then the thread exits and the next write tries again
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=30000)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=30000)
class ConnectError(Exception):
pass
@ -198,7 +198,7 @@ class _RacyQueue(queue.Queue):
@pytest.mark.asyncio
async def test_task_enqueued_between_timeout_and_exit_is_not_stranded(tmp_path):
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=60000)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=60000)
racy = _RacyQueue()
db._write_queue = racy
try:
@ -246,7 +246,7 @@ async def test_idle_exit_race_stress(tmp_path, idle_timeout_ms):
Every task must complete, and each writer's tasks must run in the order
they were queued.
"""
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=idle_timeout_ms)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=idle_timeout_ms)
idle_timeout = idle_timeout_ms / 1000
rng = random.Random(42)
writers = 12
@ -310,7 +310,7 @@ def test_idle_exit_race_stress_many_event_loops(tmp_path):
ordering between producers). Rounds are synchronised with a barrier,
then each producer waits a random 0.5x-1.5x of the idle timeout.
"""
ds, db = _make_db(tmp_path, write_thread_idle_timeout_ms=1)
ds, db = _make_db(tmp_path, connection_idle_timeout_ms=1)
loops = 6
rounds = 200
barrier = threading.Barrier(loops)