diff --git a/datasette/app.py b/datasette/app.py index a00841b2..187700b3 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -116,6 +116,7 @@ from .utils.asgi import ( asgi_static, ) from .utils.catalog import all_derived_table_dependencies +from .utils.inflight import InFlight, wait_for_concurrent from .utils.internal_db import init_internal_db from .utils.sqlite import ( sqlite3, @@ -499,17 +500,11 @@ class Datasette: # Datasette can be driven by several event loops, and an # asyncio.Lock binds to whichever loop first waits on it self._refresh_schemas_lock = threading.Lock() - try: - self._startup_lock = asyncio.Lock() - except RuntimeError as rex: - # Workaround for intermittent test failure, see: - # https://github.com/simonw/datasette/issues/1802 - if "There is no current event loop in thread" in str(rex): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - self._startup_lock = asyncio.Lock() - else: - raise + # invoke_startup() in progress (datasette.utils.inflight.InFlight), + # claimed under this lock: concurrent callers on any event loop wait + # for it. Not an asyncio.Lock, which binds to one event loop + self._startup_running = None + self._startup_state_lock = threading.Lock() self._background_tasks = BackgroundTaskSupervisor(self) self.crossdb = crossdb self.nolock = nolock @@ -834,9 +829,40 @@ class Datasette: return pm async def invoke_startup(self): - # This must be called for Datasette to be in a usable state - if self._startup_invoked: + # This must be called for Datasette to be in a usable state. + # Several event loops (threads) may call it at once: one runs + # startup, the others wait for it to finish - and try again + # themselves if it failed or its event loop went away. + while not self._startup_invoked: + loop = asyncio.get_running_loop() + task = asyncio.current_task() + with self._startup_state_lock: + running = self._startup_running + if running is not None and (running.done() or running.abandoned()): + running.finish() + running = self._startup_running = None + if running is None: + running = self._startup_running = InFlight(loop, task) + mine = True + elif running.task is task: + # Called from inside startup itself (a startup hook + # making a request through datasette.client) + return + else: + mine = False + if not mine: + await wait_for_concurrent(running.future) + continue + try: + await self._invoke_startup() + finally: + with self._startup_state_lock: + if self._startup_running is running: + self._startup_running = None + running.finish() return + + async def _invoke_startup(self): # Group spans created during startup under a single parent span with tracer.start_as_current_span(STARTUP): # Register event classes @@ -3293,16 +3319,13 @@ ORDER BY allowed.parent, allowed.child """ if self._startup_invoked and self._setup_db_done: return - async with self._startup_lock: - if self._startup_invoked and self._setup_db_done: - return - if not self._setup_db_done: - # Immutable databases used to have their table counts - # computed here, opening every one of them at startup. They - # are now computed the first time a page needs them (see - # Database.table_counts()) and cached from then on. - self._setup_db_done = True - await self.invoke_startup() + # Immutable databases used to have their table counts computed + # here, opening every one of them at startup. They are now computed + # the first time a page needs them (see Database.table_counts()) and + # cached from then on. invoke_startup() makes concurrent callers - + # on any event loop - wait for the one running it. + self._setup_db_done = True + await self.invoke_startup() def add_background_task(self, func, name=None) -> BackgroundTask: """Register a piece of supervised background work, typically from diff --git a/datasette/connection_pool.py b/datasette/connection_pool.py index a0a71e2b..4a5160fb 100644 --- a/datasette/connection_pool.py +++ b/datasette/connection_pool.py @@ -154,6 +154,58 @@ class LeasedConnection: return f"" +def close_cursors(value): + """Close the cursors in a callback's result (a cursor, or a tuple or + list holding one), in a generator's locals, or - for an exception - in + the locals of the frames it was raised through. + + Call it on the thread that owns the connection, before the connection + can be used by anyone else. A cursor keeps a reference to the + connection's cached statement for its SQL; on Python before 3.12, + deallocating the cursor resets that statement. If the cursor dies on + another thread (the caller's event loop) while the connection is running + the same SQL again - the write thread's next execute_write_many(), or + another worker that has leased the connection - the statement is reset + mid-step and that query fails with "bad parameter or other API misuse". + A closed cursor keeps its rowcount, lastrowid and description. + """ + if isinstance(value, BaseException): + tb = value.__traceback__ + while tb is not None: + try: + values = list(tb.tb_frame.f_locals.values()) + except Exception: # noqa: BLE001 + values = [] + for item in values: + if isinstance(item, sqlite3.Cursor): + _close_cursor(item) + tb = tb.tb_next + return + if isinstance(value, sqlite3.Cursor): + _close_cursor(value) + elif isinstance(value, (tuple, list)): + for item in value: + if isinstance(item, sqlite3.Cursor): + _close_cursor(item) + elif inspect.isgenerator(value): + try: + values = list(inspect.getgeneratorlocals(value).values()) + except Exception: # noqa: BLE001 + values = [] + for item in values: + if isinstance(item, sqlite3.Cursor): + _close_cursor(item) + value.close() + + +def _close_cursor(cursor): + try: + cursor.close() + except Exception: # noqa: BLE001, S110 + # Its connection is closed already + pass + + def _generator_uses_connection(result): """True for a generator that would step the leased connection after the callback returned: one holding the connection (a closure over @@ -219,12 +271,19 @@ class ReadConnectionPool: """Call fn(conn) with a pooled connection leased for the duration.""" entry = self.acquire(db) lease = LeasedConnection(entry.conn, db.name, "read") + rejected = False try: result = fn(lease) + if isinstance(result, sqlite3.Cursor) or _generator_uses_connection(result): + rejected = True + close_cursors(result) + except BaseException as e: + close_cursors(e) + raise finally: lease._expire() self.release(entry) - if isinstance(result, sqlite3.Cursor) or _generator_uses_connection(result): + if rejected: raise ConnectionLeaseError( f"An execute_fn() callback for database {db.name!r} returned a {type(result).__name__}, which " "would keep using the pooled connection after the callback " diff --git a/datasette/database.py b/datasette/database.py index e7dad6e6..e505dae5 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -16,7 +16,11 @@ import sqlite_utils from opentelemetry import context as otel_context_api from opentelemetry.trace import Status, StatusCode -from .connection_pool import ConnectionLeaseError, LeasedConnection # noqa: F401 +from .connection_pool import ( # noqa: F401 + ConnectionLeaseError, + LeasedConnection, + close_cursors, +) from .inspect import inspect_hash from .telemetry import ( callback_name, @@ -1119,7 +1123,12 @@ class Database: elif self.is_memory: return 0 elif self.is_mutable: - return Path(self.path).stat().st_size + try: + return Path(self.path).stat().st_size + except FileNotFoundError: + # Deleted from under Datasette: report it as empty rather + # than failing every page that lists databases + return 0 elif self.ds.inspect_data and self.ds.inspect_data.get(self.name): self.cached_size = self.ds.inspect_data[self.name]["size"] return self.cached_size @@ -1440,12 +1449,22 @@ def _call_with_lease(fn, conn, db_name, kind): Unlike read callbacks, write callbacks may return a cursor (for example ``lambda conn: conn.execute(...)``, and execute_write_many() and - execute_write_script() return theirs); reading attributes such as - ``lastrowid`` from it afterwards is fine, fetching rows is not. + execute_write_script() return theirs); it is closed before it leaves + this thread, so reading attributes such as ``lastrowid`` from it + afterwards is fine, fetching rows is not. """ lease = LeasedConnection(conn, db_name, kind) try: - return fn(lease) + result = fn(lease) + except BaseException as e: + close_cursors(e) + raise + else: + # A returned cursor is closed here, on the thread that owns the + # connection (see close_cursors()); its rowcount and lastrowid can + # still be read + close_cursors(result) + return result finally: lease._expire() diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index 0db4bc48..42aa18ef 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -74,7 +74,6 @@ open. """ import asyncio -import concurrent.futures import json import logging import os @@ -83,6 +82,7 @@ import time from .utils import sqlite3 from .utils.catalog import remember_derived_table_dependencies +from .utils.inflight import InFlight, wait_for_concurrent from .utils.internal_db import CATALOG_TABLES, collect_schema, write_catalog_entries logger = logging.getLogger("datasette.schema_watcher") @@ -144,7 +144,8 @@ class WatchState: self.catalog_version = None self.notified_version = None self.needs_scan = True - # An _InflightScan while a catalog scan of this database runs + # An InFlight (datasette.utils.inflight) while a catalog scan of + # this database runs self.scan_future = None # time.monotonic() before which a failed scan is not retried, and # the delay used for the next failure (doubles, see RETRY_DELAY_S) @@ -248,61 +249,6 @@ def header_schema_version(path): return int.from_bytes(header[40:44], "big") -class _InflightScan: - """One catalog scan in progress for one database. - - Not bound to an event loop: one Datasette is driven by several loops - (``--get``, TestClient, pytest-asyncio), each in its own thread, and a - write on any of them may need to wait for a scan started on another. - Waiters use :func:`_wait_for` on ``future``, a ``concurrent.futures`` - future that the scan's owner resolves in a ``finally`` block. - """ - - __slots__ = ("future", "loop") - - def __init__(self, loop): - self.future = concurrent.futures.Future() - self.loop = loop - - def done(self): - return self.future.done() - - def abandoned(self): - """True if nothing will ever finish this scan: the loop running it - has closed, or has stopped (``run_until_complete()`` returned while - the scan task was still pending).""" - loop = self.loop - return loop.is_closed() or not loop.is_running() - - def finish(self): - try: - self.future.set_result(None) - except concurrent.futures.InvalidStateError: - pass - - -def _wait_for(cf_future): - """An asyncio future on the running loop that completes when the - concurrent future does. Cancelling it does not cancel cf_future, which - other waiters (maybe on other loops) share.""" - loop = asyncio.get_running_loop() - waiter = loop.create_future() - - def _set(): - if not waiter.done(): - waiter.set_result(None) - - def _done(_): - try: - loop.call_soon_threadsafe(_set) - except RuntimeError: - # That loop has closed; nobody is waiting any more - pass - - cf_future.add_done_callback(_done) - return waiter - - def _is_transient(error): if isinstance(error, sqlite3.OperationalError): message = str(error) @@ -987,7 +933,7 @@ class SchemaWatcher: inflight.finish() if state.needs_scan and not state.removed and state not in failed: state.needs_scan = False - state.scan_future = _InflightScan(loop) + state.scan_future = InFlight(loop) mine.append(state) if mine: records = {state: state.scan_future for state in mine} @@ -1018,7 +964,7 @@ class SchemaWatcher: for record in records.values(): record.finish() if waits: - await asyncio.wait([_wait_for(f) for f in waits]) + await asyncio.wait([wait_for_concurrent(f) for f in waits]) pending = [ s for s in pending @@ -1042,7 +988,7 @@ class SchemaWatcher: """Wait for a catalog scan of state that is in flight, if any.""" inflight = state.scan_future if inflight is not None and not inflight.done() and not inflight.abandoned(): - await _wait_for(inflight.future) + await wait_for_concurrent(inflight.future) async def _scan_and_store(self, states, records, handled): memory = [s for s in states if not s.is_file and s.db.memory_name] @@ -1069,7 +1015,7 @@ class SchemaWatcher: async def _store(self, results, records=None, handled=None): """Write scan results to the catalog. ``records`` maps each state to - the _InflightScan this scan was started as: results for a state whose + the InFlight this scan was started as: results for a state whose scan has since been taken over by another loop are dropped. States whose outcome was recorded are added to ``handled``; the others (transient failures) are retried later by refresh().""" diff --git a/datasette/utils/inflight.py b/datasette/utils/inflight.py new file mode 100644 index 00000000..f73b5310 --- /dev/null +++ b/datasette/utils/inflight.py @@ -0,0 +1,63 @@ +"""Waiting for work started on another event loop. + +One Datasette can be driven by several event loops at once, each in its own +thread (``datasette --get``, TestClient, pytest-asyncio, an embedding app). +asyncio primitives (``asyncio.Lock``, ``asyncio.Future``) belong to one loop, +so "one call does the work, concurrent callers wait for it" is built from a +``concurrent.futures.Future`` that any loop can wait for, and a +``threading.Lock`` held only for the moment it takes to claim the work. +""" + +import asyncio +import concurrent.futures + + +class InFlight: + """One piece of work in progress, started by ``task`` on ``loop``.""" + + __slots__ = ("future", "loop", "task") + + def __init__(self, loop, task=None): + self.future = concurrent.futures.Future() + self.loop = loop + self.task = task + + def done(self): + return self.future.done() + + def abandoned(self): + """True if nothing will ever finish this work: the loop running it + has closed, or has stopped (``run_until_complete()`` returned while + the task doing it was still pending).""" + loop = self.loop + return loop.is_closed() or not loop.is_running() + + def finish(self): + """Wake every waiter (they check for themselves whether the work + succeeded). Safe to call more than once, from any thread.""" + try: + self.future.set_result(None) + except concurrent.futures.InvalidStateError: + pass + + +def wait_for_concurrent(cf_future): + """An asyncio future on the running loop that completes when + ``cf_future`` does. Cancelling it does not cancel ``cf_future``, which + other waiters - maybe on other loops - share.""" + loop = asyncio.get_running_loop() + waiter = loop.create_future() + + def _set(): + if not waiter.done(): + waiter.set_result(None) + + def _done(_): + try: + loop.call_soon_threadsafe(_set) + except RuntimeError: + # That loop has closed; nobody is waiting any more + pass + + cf_future.add_done_callback(_done) + return waiter diff --git a/datasette/views/index.py b/datasette/views/index.py index bb01fd61..b41ba3f8 100644 --- a/datasette/views/index.py +++ b/datasette/views/index.py @@ -1,5 +1,6 @@ import json +from datasette.database import DatasetteClosedError from datasette.plugins import pm from datasette.utils import ( UNSTABLE_API_MESSAGE, @@ -7,6 +8,8 @@ from datasette.utils import ( add_cors_headers, await_me_maybe, make_slot_function, + sqlite3, + tilde_encode, ) from datasette.utils.asgi import Response from datasette.utils.catalog import ( @@ -46,8 +49,13 @@ class IndexView(BaseView): "view-database", request.actor, include_is_private=True ) allowed_databases = [r async for r in db_page.all()] + # One snapshot for the whole request: add_database() and + # remove_database() replace ds.databases rather than changing it, so + # a database removed while this page is being built is still listed + # instead of failing the page with a KeyError + all_databases = self.ds.databases allowed_db_dict = { - r.parent: r for r in allowed_databases if r.parent in self.ds.databases + r.parent: r for r in allowed_databases if r.parent in all_databases } # Group tables by database @@ -62,7 +70,7 @@ class IndexView(BaseView): names = list(allowed_db_dict) summaries = await catalog_summaries(self.ds, names) - all_counts = await self._table_counts(names) + all_counts = await self._table_counts(names, all_databases) sort_by_relationships = request.args.get("_sort") == "relationships" need_relationships = { name for name in names if sort_by_relationships or not all_counts[name] @@ -149,7 +157,7 @@ class IndexView(BaseView): table_counts, tables_and_views_truncated, ) in prepared: - db = self.ds.databases[name] + db = all_databases[name] tables_and_views_truncated = [ { "name": t["name"], @@ -173,7 +181,8 @@ class IndexView(BaseView): "name": name, "hash": db.hash, "color": db.color, - "path": self.ds.urls.database(name), + # From the snapshot: name may have been removed since + "path": self.ds.urls.path(tilde_encode(db.route)), "tables_and_views_truncated": tables_and_views_truncated, "tables_and_views_more": (len(visible_tables) + len(views)) > TRUNCATE_AT, @@ -189,6 +198,10 @@ class IndexView(BaseView): } ) + # Leave out databases removed while this page was being built: the + # HTML template links to them by name + databases = [d for d in databases if d["name"] in self.ds.databases] + if as_format: headers = {} if self.ds.cors: @@ -235,13 +248,13 @@ class IndexView(BaseView): }, ) - async def _table_counts(self, names): + async def _table_counts(self, names, all_databases): """``{database: {table: count}}`` - empty for a database whose counts are skipped or timed out (see COUNT_MAX_DATABASES).""" counts = {} to_count = [] for name in names: - db = self.ds.databases[name] + db = all_databases[name] if not db.is_mutable and db.cached_table_counts is not None: # Known without opening the database counts[name] = db.cached_table_counts @@ -258,7 +271,12 @@ class IndexView(BaseView): except OSError: # The file has gone: there is nothing to count continue - table_counts = await db.table_counts(10) + try: + table_counts = await db.table_counts(10) + except (DatasetteClosedError, sqlite3.Error): + # Closed or removed while this page was being built, or its + # file deleted or replaced: show it without counts + continue # If any of these are None it means at least one timed out - ignore them all if any(v is None for v in table_counts.values()): table_counts = {} diff --git a/docs/internals.rst b/docs/internals.rst index 9d21cd78..dbcac30d 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -1951,7 +1951,7 @@ Application lifecycle Datasette guarantees a fixed sequence of events between the moment a ``Datasette`` instance is constructed and the moment its resources are released: 1. ``Datasette(...)`` — the constructor runs synchronously and does not run plugin hooks. -2. **Startup** — ``await datasette.invoke_startup()`` runs once: it populates the internal database's catalog of table schemas (:ref:`internals_internal`), loads canned queries and column type configuration, then calls every registered :ref:`plugin_hook_startup` hook, in plugin registration order. When Datasette is being served, table-count precomputation for immutable databases runs immediately before this, as part of the same startup sequence. +2. **Startup** — ``await datasette.invoke_startup()`` runs once, even if several event loops call it at the same time (the others wait for it to finish): it populates the internal database's catalog of table schemas (:ref:`internals_internal`), loads canned queries and column type configuration, then calls every registered :ref:`plugin_hook_startup` hook, in plugin registration order. When Datasette is being served, table-count precomputation for immutable databases runs immediately before this, as part of the same startup sequence. 3. **Background-task launch** — once *every* ``startup`` hook has finished (not before), every task registered with :ref:`datasette_add_background_task` — by any plugin — is launched. A task registered by one plugin's ``startup`` hook can safely depend on state set up by another plugin's ``startup`` hook, because launch only happens after the whole round of hooks completes. 4. **Serving** — the instance handles requests (or, for headless or CLI use, does whatever the embedding program does with it). 5. **Shutdown** — triggered by the ASGI ``lifespan.shutdown`` event (Ctrl-C, ``SIGTERM``) or the end of a ``datasette serve`` process: every :ref:`plugin_hook_shutdown` hook runs first, while background tasks are still alive, so a plugin can tell its own task to wind down gracefully; every still-running background task is then cancelled and given a five-second grace period to actually stop; finally every database connection is released via :ref:`datasette_close`. @@ -2239,7 +2239,7 @@ The write connection belongs to Datasette: calling ``conn.close()`` inside an `` Datasette cannot detect every misuse. These are unsupported and may fail unpredictably, or read from a connection that is now being used by someone else: -- Iterating over or fetching from a **cursor** created inside the function after the function has returned. Write functions may return a cursor (``lambda conn: conn.execute(...)`` is a common pattern) and reading attributes such as ``.lastrowid`` and ``.rowcount`` from it afterwards is fine; fetching rows from it is not. +- Iterating over or fetching from a **cursor** created inside the function after the function has returned. Write functions may return a cursor (``lambda conn: conn.execute(...)`` is a common pattern) and reading attributes such as ``.lastrowid`` and ``.rowcount`` from it afterwards is fine; fetching rows from it is not. A cursor returned by a write or isolated function - including the ones returned by ``execute_write_many()`` and ``execute_write_script()`` - is closed before it is handed back, on the thread that owns the connection, and so are cursors left in the local variables of a function that raised an exception. (On Python versions before 3.12, a cursor that was freed on another thread could reset the same statement while the connection was running it again, making that query fail with "bad parameter or other API misuse".) - Reaching the underlying connection through ``cursor.connection``. - Relying on state set up on the connection by an earlier call (see above). diff --git a/tests/connection_torture.py b/tests/connection_torture.py new file mode 100644 index 00000000..1c450cd5 --- /dev/null +++ b/tests/connection_torture.py @@ -0,0 +1,1297 @@ +""" +Randomized torture harness for Datasette's connection management. + +Run one scenario per process, so that a segfault or a hang is a result rather +than the end of the test run:: + + python tests/connection_torture.py SCENARIO --seed N [--duration S] + +Several event loops, each in its own thread (and each running several +tasks), drive one Datasette with extreme settings while a separate process +changes some of the database files. The last line of output is a JSON +summary. Exit status: 0 every check passed, 1 a check failed, 3 something +hung (an operation or a thread did not finish in time); a crash shows up as +death by a signal (faulthandler prints the stacks). + +Checks: + +* no operation hangs, no crash; +* only documented exceptions (see ``Torture.expected()``); +* per database, each writer's rows are in the order it wrote them, and + every acknowledged blocking write (and every non-blocking one, once the + queue has drained) is in the file; +* reads see the reader's own earlier blocking writes; +* the catalog lists a table as soon as a blocking DDL write returns (when + the checkout has a SchemaWatcher), and matches the files after a forced + refresh; +* after ``Datasette.close()`` and ``gc.collect()``: no file descriptor open + on any database file, no write or SQL thread left, the instance itself + collected. + +Works against older checkouts too: features they lack (settings, scratch +databases, schema watch modes) are detected and skipped. +""" + +import argparse +import asyncio +import collections +import faulthandler +import gc +import inspect +import json +import os +import random +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +import traceback +import weakref + +from datasette.app import Datasette +from datasette.database import Database, QueryInterrupted + +try: + from datasette.app import DEFAULT_SETTINGS +except ImportError: # pragma: no cover + DEFAULT_SETTINGS = {} +try: + from datasette.database import DatasetteClosedError +except ImportError: # pragma: no cover + DatasetteClosedError = RuntimeError +try: + from datasette.connection_pool import ConnectionLeaseError +except ImportError: + ConnectionLeaseError = None +try: + from datasette import scratch as scratch_module +except ImportError: + scratch_module = None + +OP_TIMEOUT = 30.0 +SLOW_SQL = ( + "with recursive c(x) as (select 1 union all select x + 1 from c " + "where x < 5000000) select count(*) from c" +) + +SCENARIOS = { + # Reads, writes of every kind, DDL, isolated functions, analyze_sql, + # cancellation, external schema changes - on databases that stay put + "mixed": { + "loops": 3, + "tasks": 4, + "threads": 3, + "duration": 6.0, + "stable": 4, + "ext": 2, + "external_process": True, + "ops": { + "read": 6, + "execute_fn": 3, + "http": 3, + "write": 6, + "write_nonblocking": 3, + "write_many": 2, + "write_script": 1, + "ddl": 2, + "isolated": 1, + "analyze": 1, + "cancel_read": 1, + "cancel_write": 1, + }, + }, + # Databases come and go while they are being used + "lifecycle": { + "loops": 3, + "tasks": 3, + "threads": 3, + "duration": 6.0, + "stable": 3, + "ext": 1, + "churn": 3, + "external_process": True, + "ops": { + "read": 4, + "write": 4, + "write_nonblocking": 1, + "http": 2, + "isolated": 1, + "churn_read": 4, + "churn_write": 3, + "churn_slow": 1, + "add_remove": 2, + "replace_delete": 1, + "db_close": 1, + "scratch": 3, + "cancel_read": 1, + }, + }, + # Datasette.close() from one loop while the others keep going + "close": { + "loops": 3, + "tasks": 4, + "threads": 3, + "duration": 4.0, + "stable": 3, + "ext": 1, + "churn": 2, + "external_process": True, + "close_midflight": True, + "ops": { + "read": 4, + "write": 4, + "write_nonblocking": 2, + "http": 2, + "ddl": 1, + "isolated": 1, + "churn_slow": 1, + "scratch": 1, + "cancel_read": 1, + }, + }, + # Slow queries with a progress handler, cancelled or with their database + # closed underneath them (the close-while-stepping segfault) + "cancel": { + "loops": 2, + "tasks": 6, + "threads": 2, + "duration": 5.0, + "stable": 2, + "churn": 3, + "ops": { + "cancel_read": 4, + "cancel_write": 2, + "churn_slow": 4, + "churn_isolated_slow": 3, + "add_remove": 3, + "write": 2, + "read": 2, + }, + }, + # Databases removed (closed) while isolated functions with a progress + # handler step them - on main, immutable isolated functions were not + # tracked as pending work, and close() could free a connection mid-step + # (segfault) + "closerace": { + "loops": 3, + "tasks": 6, + "threads": 3, + "duration": 6.0, + "stable": 1, + "churn": 4, + "churn_mutable_every": 1, + "isolated_rows": 20000, + "ops": { + "churn_isolated_slow": 8, + "churn_slow": 2, + "churn_read": 2, + "add_remove": 6, + }, + }, + # Write-heavy: execute_write_many() and execute_write() from several + # loops at once (a returned cursor freed on the caller's thread used to + # reset the write thread's statement: "bad parameter or other API + # misuse" on Python < 3.12) + "writes": { + "loops": 3, + "tasks": 6, + "threads": 3, + "duration": 6.0, + "stable": 2, + "ext": 1, + "external_process": True, + "ops": { + "write_many": 6, + "write": 4, + "write_nonblocking": 3, + "cancel_write": 2, + "write_script": 1, + "ddl": 1, + "read": 3, + "isolated": 1, + }, + }, + # Many databases, the smallest pool, 1ms idle timeout, crossdb + "pool": { + "loops": 3, + "tasks": 6, + "threads": 3, + "duration": 5.0, + "stable": 12, + "crossdb": True, + "private_cap": 1, + "ops": { + "read": 6, + "execute_fn": 3, + "http": 3, + "crossdb_read": 2, + "write": 3, + }, + }, + # num_sql_threads=0: one thread, a new event loop every round (the + # TestClient pattern), every kind of operation + "nothreads": { + "loops": 1, + "rounds": True, + "tasks": 6, + "threads": 0, + "duration": 5.0, + "stable": 3, + "ext": 1, + "churn": 2, + "external_process": True, + "ops": { + "read": 4, + "execute_fn": 2, + "http": 2, + "write": 4, + "write_nonblocking": 2, + "write_many": 1, + "write_script": 1, + "ddl": 2, + "isolated": 1, + "analyze": 1, + "churn_read": 2, + "add_remove": 1, + "replace_delete": 1, + "scratch": 2, + }, + }, +} + +EXTERNAL_WRITER = r""" +import random, sqlite3, sys, time +paths = sys.argv[2:] +rng = random.Random(int(sys.argv[1])) +n = 0 +while True: + path = rng.choice(paths) + try: + conn = sqlite3.connect(path, timeout=1, isolation_level=None) + n += 1 + kind = rng.random() + if kind < 0.4: + conn.execute(f"create table if not exists ext_{n % 7} (id integer, v text)") + elif kind < 0.6: + conn.execute(f"drop table if exists ext_{rng.randrange(7)}") + elif kind < 0.8: + try: + conn.execute(f"alter table ext_{rng.randrange(7)} add column c{n} text") + except sqlite3.OperationalError: + pass + else: + conn.execute("insert into data (v) values ('external')") + conn.close() + except sqlite3.Error: + pass + time.sleep(rng.uniform(0.005, 0.05)) +""" + + +class Hang(Exception): + pass + + +def make_file(path, wal=False, extra_sql=""): + conn = sqlite3.connect(path) + if wal: + conn.execute("PRAGMA journal_mode=wal") + conn.executescript( + "create table if not exists log (id integer primary key, writer text, " + "seq integer, kind text);" + "create index if not exists log_writer on log (writer, seq);" + "create table if not exists data (id integer primary key, v text);" + "insert into data (v) values ('one'), ('two');" + extra_sql + ) + conn.commit() + conn.close() + + +def describe(exc): + return f"{type(exc).__module__}.{type(exc).__name__}: {exc}"[:300] + + +class Writer: + """One task's identity: its own sequence numbers per database.""" + + def __init__(self, name, rng): + self.name = name + self.rng = rng + self.seq = collections.Counter() + self.ddl = collections.Counter() + + +class Torture: + def __init__(self, scenario, seed, duration=None): + self.name = scenario + self.spec = dict(SCENARIOS[scenario]) + if duration is not None: + self.spec["duration"] = duration + self.seed = seed + self.rng = random.Random(seed) + self.tmp = tempfile.mkdtemp(prefix="datasette_torture_") + self.lock = threading.Lock() + self.problems = [] + self.counts = collections.Counter() + self.expected_errors = collections.Counter() + # (db name, writer) -> list of seqs acknowledged by blocking writes + self.acked = collections.defaultdict(list) + # (db name, writer) -> seqs issued as non-blocking writes, and seqs + # whose write was cancelled (may or may not have committed) + self.nonblocking = collections.defaultdict(list) + self.maybe = collections.defaultdict(list) + self.closing = False + self.closed_at = None + self.stop_at = None + self.churn_seq = 0 + self.scratch_seq = 0 + self.features = {} + self.stable_names = [] + self.ext_names = [] + self.churn_paths = {} + self.ds = None + self.external = None + self.writers = {} + + # -- bookkeeping ----------------------------------------------------- + + def problem(self, kind, message): + with self.lock: + self.problems.append({"kind": kind, "message": message[:2000]}) + + def count(self, key): + with self.lock: + self.counts[key] += 1 + + def expected(self, op, group, exc): + """Is exc a documented outcome of op on a database in group? + Returns a label for the counts, or None for a problem.""" + message = str(exc) + if isinstance(exc, QueryInterrupted): + return "QueryInterrupted" + if isinstance(exc, DatasetteClosedError) and ( + group in ("churn", "scratch") or self.closing + ): + return type(exc).__name__ + if self.closing and isinstance(exc, RuntimeError): + # Work submitted to the executor after close() shut it down + return "RuntimeError after close" + if scratch_module is not None and isinstance( + exc, + ( + scratch_module.ScratchDatabaseNotFound, + scratch_module.ScratchDatabaseExists, + ), + ): + return type(exc).__name__ + if isinstance(exc, KeyError) and group in ("churn", "scratch"): + return "KeyError (database removed)" + if isinstance(exc, sqlite3.Error): + if group in ("churn", "scratch"): + # Files replaced, deleted and closed underneath + return f"sqlite3 on {group}: {message[:40]}" + if group == "mem" and "is locked" in message: + # Shared-cache in-memory databases do not wait for table or + # schema locks (documented) + return "mem: shared-cache lock" + if group == "ext" and ( + "no such table" in message + or "no such column" in message + or "database is locked" in message + or "schema has changed" in message + ): + return f"ext: {message[:30]}" + if self.closing and "closed database" in message: + return "closed database after close" + return None + + async def call(self, op, group, awaitable, timeout=OP_TIMEOUT): + """Await with a timeout; classify exceptions. Returns (ok, value).""" + try: + value = await asyncio.wait_for(awaitable, timeout) + except asyncio.TimeoutError: + self.problem("hang", f"{op} on {group} did not finish in {timeout}s") + return False, None + except asyncio.CancelledError: + raise + except Exception as e: # noqa: BLE001 + label = self.expected(op, group, e) + if label is None: + if ConnectionLeaseError is not None and isinstance( + e, ConnectionLeaseError + ): + kind = "lease" + else: + kind = "unexpected" + self.problem( + kind, + f"{op} on {group}: {describe(e)}\n" + + "".join(traceback.format_exception(e)[-6:]), + ) + else: + with self.lock: + self.expected_errors[label] += 1 + return False, None + self.count(op) + return True, value + + # -- setup ----------------------------------------------------------- + + def settings(self): + wanted = { + "num_sql_threads": self.spec["threads"], + "max_open_connections": 1, + "connection_idle_timeout_ms": 1, + "schema_watch_interval_ms": 10, + } + return {k: v for k, v in wanted.items() if k in DEFAULT_SETTINGS} + + def build(self): + spec = self.spec + files = [] + for i in range(spec.get("stable", 0)): + path = os.path.join(self.tmp, f"stable_{i}.db") + make_file(path, wal=i % 2 == 0) + files.append(path) + ext_paths = [] + for i in range(spec.get("ext", 0)): + path = os.path.join(self.tmp, f"ext_{i}.db") + make_file(path, wal=i % 2 == 1) + ext_paths.append(path) + imm = os.path.join(self.tmp, "imm.db") + make_file(imm) + self.template = os.path.join(self.tmp, "template.db") + make_file(self.template) + kwargs = {} + if spec.get("crossdb"): + kwargs["crossdb"] = True + scratch_dir = os.path.join(self.tmp, "scratch") + self.features["scratch"] = ( + "scratch_dir" in inspect.signature(Datasette.__init__).parameters + ) + if self.features["scratch"]: + kwargs["scratch_dir"] = scratch_dir + self.features["schema_watch"] = ( + "schema_watch" in inspect.signature(Datasette.add_database).parameters + ) + self.ds = Datasette( + files + ext_paths, immutables=[imm], settings=self.settings(), **kwargs + ) + ds = self.ds + self.features["watcher"] = hasattr(ds, "_schema_watcher") + self.stable_names = [os.path.basename(p)[:-3] for p in files] + self.ext_names = [os.path.basename(p)[:-3] for p in ext_paths] + # Half the stable databases owned (when the checkout has modes): + # added with add_database() rather than on the "command line" + for i in range(spec.get("stable", 0)): + if i % 2 == 1 and self.features["schema_watch"]: + name = f"stable_{i}" + ds.remove_database(name) + ds.add_database( + Database(ds, path=files[i]), name=name, schema_watch="owned" + ) + ds.add_memory_database(f"torture_mem_{os.getpid()}_{self.seed}", name="mem") + self.groups = {name: "stable" for name in self.stable_names} + self.groups.update({name: "ext" for name in self.ext_names}) + self.groups["mem"] = "mem" + self.groups["imm"] = "imm" + for i in range(spec.get("churn", 0)): + self.add_churn() + if spec.get("external_process") and ext_paths: + self.external = subprocess.Popen( + [sys.executable, "-c", EXTERNAL_WRITER, str(self.seed), *ext_paths], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + def add_churn(self): + with self.lock: + self.churn_seq += 1 + n = self.churn_seq + name = f"churn_{n}" + path = os.path.join(self.tmp, f"{name}.db") + shutil.copy(self.template, path) + # Every nth churn database is immutable (all of them with 1) + mutable = n % self.spec.get("churn_mutable_every", 4) != 0 + db = Database(self.ds, path=path, is_mutable=mutable) + kwargs = {"schema_watch": "external"} if self.features["schema_watch"] else {} + self.ds.add_database(db, name=name, **kwargs) + with self.lock: + self.churn_paths[name] = path + self.groups[name] = "churn" + return name + + # -- operations ------------------------------------------------------ + + def pick(self, rng, groups): + names = [n for n, g in list(self.groups.items()) if g in groups] + if not names: + return None, None + name = rng.choice(names) + return name, self.groups.get(name) + + def db(self, name): + return self.ds.databases.get(name) + + async def op_read(self, w): + name, group = self.pick(w.rng, ("stable", "ext", "mem", "imm")) + db = self.db(name) + if db is None: + return + if group == "stable" and w.seq[name] and w.rng.random() < 0.5: + # Read-your-writes: this writer's last acknowledged seq + last = self.acked.get((name, w.name)) + ok, results = await self.call( + "read", + group, + db.execute("select max(seq) from log where writer = ?", [w.name]), + ) + if ok and last: + got = results.first()[0] + if got is None or got < last[-1]: + self.problem( + "read-your-writes", + f"{name}: writer {w.name} read max(seq)={got} after " + f"its write {last[-1]} was acknowledged", + ) + return + sql = w.rng.choice( + [ + "select count(*) from data", + "select * from data order by id desc limit 3", + "select name from sqlite_master", + "PRAGMA schema_version", + ] + ) + await self.call("read", group, db.execute(sql)) + + async def op_execute_fn(self, w): + name, group = self.pick(w.rng, ("stable", "ext", "mem", "imm")) + db = self.db(name) + if db is None: + return + + def fn(conn): + n = conn.execute("select count(*) from data").fetchone()[0] + tables = [r[0] for r in conn.execute("select name from sqlite_master")] + return n, tables + + await self.call("execute_fn", group, db.execute_fn(fn)) + + async def op_http(self, w): + name, group = self.pick(w.rng, ("stable", "ext", "imm", "mem")) + path = w.rng.choice( + [ + f"/{name}.json", + f"/{name}/data.json?_size=2", + "/.json", + "/-/databases.json", + ] + ) + ok, response = await self.call("http", group, self.ds.client.get(path)) + if ok and response.status_code != 200: + if self.closing or (group == "ext" and response.status_code == 404): + # 404: data or the database page during an external change + self.count(f"http {response.status_code} (expected)") + return + if response.status_code == 500 and ( + "schema is locked" in response.text + or "table is locked" in response.text + ): + # SQLITE_LOCKED from the shared-cache in-memory database: it + # does not wait for locks (documented, the same on main) + with self.lock: + self.expected_errors["http 500: shared-cache lock"] += 1 + return + self.problem( + "http", f"GET {path} -> {response.status_code}: {response.text[:300]}" + ) + + def _next_seq(self, w, name): + w.seq[name] += 1 + return w.seq[name] + + async def _write(self, w, name, group, block): + db = self.db(name) + if db is None: + return + seq = self._next_seq(w, name) + ok, _ = await self.call( + "write" if block else "write_nonblocking", + group, + db.execute_write( + "insert into log (writer, seq, kind) values (?, ?, ?)", + [w.name, seq, "write"], + block=block, + ), + ) + # Recorded once the call has returned: a block=False write is in the + # queue by then, so a later drain write is behind it + if ok and block: + self.acked[(name, w.name)].append(seq) + elif ok: + self.nonblocking[(name, w.name)].append(seq) + elif not block: + # Failed to queue, or maybe queued then failed: either way it + # may or may not be in the file + self.maybe[(name, w.name)].append(seq) + + async def op_write(self, w): + name, group = self.pick(w.rng, ("stable",)) + if name: + await self._write(w, name, group, True) + + async def op_write_nonblocking(self, w): + name, group = self.pick(w.rng, ("stable",)) + if name: + await self._write(w, name, group, False) + + async def op_write_many(self, w): + name, group = self.pick(w.rng, ("stable",)) + db = self.db(name) + if db is None: + return + seqs = [self._next_seq(w, name) for _ in range(w.rng.randint(1, 4))] + ok, _ = await self.call( + "write_many", + group, + db.execute_write_many( + "insert into log (writer, seq, kind) values (?, ?, 'many')", + [(w.name, s) for s in seqs], + ), + ) + if ok: + self.acked[(name, w.name)].extend(seqs) + + async def op_write_script(self, w): + name, group = self.pick(w.rng, ("stable",)) + db = self.db(name) + if db is None: + return + seqs = [self._next_seq(w, name) for _ in range(2)] + script = "".join( + f"insert into log (writer, seq, kind) values ('{w.name}', {s}, 'script');" + for s in seqs + ) + ok, _ = await self.call("write_script", group, db.execute_write_script(script)) + if ok: + self.acked[(name, w.name)].extend(seqs) + + async def op_ddl(self, w): + name, group = self.pick(w.rng, ("stable", "mem")) + db = self.db(name) + if db is None: + return + w.ddl[name] += 1 + n = w.ddl[name] + if n % 3 == 0: + sql = f"drop table if exists ddl_{w.name}_{n - 1}" + table = None + else: + table = f"ddl_{w.name}_{n}" + sql = f"create table if not exists {table} (id integer primary key)" + ok, _ = await self.call("ddl", group, db.execute_write(sql)) + if ok and table and self.features["watcher"] and not self.closing: + # The catalog lists a table once a blocking DDL write returns + ok2, rows = await self.call( + "ddl_catalog", + group, + self.ds.get_internal_database().execute( + "select 1 from catalog_tables where database_name = ? " + "and table_name = ?", + [name, table], + ), + ) + if ok2 and not rows.rows and not self.closing: + self.problem( + "catalog", f"{name}: {table} not in the catalog after its DDL" + ) + + async def op_isolated(self, w): + name, group = self.pick(w.rng, ("stable", "imm")) + db = self.db(name) + if db is None: + return + + def fn(conn): + return conn.execute("select count(*) from data").fetchone()[0] + + await self.call("isolated", group, db.execute_isolated_fn(fn)) + + async def op_analyze(self, w): + name, group = self.pick(w.rng, ("stable",)) + db = self.db(name) + if db is None or not hasattr(db, "analyze_sql"): + return + await self.call( + "analyze", group, db.analyze_sql("select * from log where writer = 'x'") + ) + + async def op_cancel_read(self, w): + name, group = self.pick(w.rng, ("stable", "imm")) + db = self.db(name) + if db is None: + return + task = asyncio.ensure_future(db.execute(SLOW_SQL, custom_time_limit=300)) + await asyncio.sleep(w.rng.uniform(0, 0.05)) + task.cancel() + try: + await asyncio.wait_for(task, OP_TIMEOUT) + self.count("cancel_read finished") + except asyncio.CancelledError: + self.count("cancel_read cancelled") + except asyncio.TimeoutError: + self.problem("hang", f"cancelled read on {name} did not finish") + except Exception as e: # noqa: BLE001 + if self.expected("cancel_read", group, e) is None: + self.problem("unexpected", f"cancel_read on {name}: {describe(e)}") + + async def op_cancel_write(self, w): + name, group = self.pick(w.rng, ("stable",)) + db = self.db(name) + if db is None: + return + seq = self._next_seq(w, name) + self.maybe[(name, w.name)].append(seq) + task = asyncio.ensure_future( + db.execute_write( + "insert into log (writer, seq, kind) values (?, ?, 'cancelled')", + [w.name, seq], + ) + ) + await asyncio.sleep(w.rng.uniform(0, 0.005)) + task.cancel() + try: + await asyncio.wait_for(task, OP_TIMEOUT) + self.count("cancel_write finished") + except asyncio.CancelledError: + self.count("cancel_write cancelled") + except asyncio.TimeoutError: + self.problem("hang", f"cancelled write on {name} did not finish") + except Exception as e: # noqa: BLE001 + if self.expected("cancel_write", group, e) is None: + self.problem("unexpected", f"cancel_write on {name}: {describe(e)}") + + async def op_crossdb_read(self, w): + db = self.db("_memory") + name, group = self.pick(w.rng, ("stable",)) + if db is None or name is None: + return + attached = {r.name for r in await db.attached_databases()} + if name not in attached: + return + await self.call( + "crossdb_read", group, db.execute(f"select count(*) from [{name}].data") + ) + + async def op_churn_read(self, w): + name, group = self.pick(w.rng, ("churn",)) + db = self.db(name) + if db is None: + return + if w.rng.random() < 0.5: + await self.call( + "churn_read", group, db.execute("select count(*) from data") + ) + else: + ok, response = await self.call( + "churn_http", group, self.ds.client.get(f"/{name}/data.json?_size=1") + ) + # 400: SQL errors on a replaced or deleted file are reported as + # "Invalid SQL" by the table page + if ok and response.status_code not in (200, 400, 404, 500): + self.problem("http", f"GET /{name}/data.json -> {response.status_code}") + + async def op_churn_write(self, w): + name, group = self.pick(w.rng, ("churn",)) + db = self.db(name) + if db is None or not db.is_mutable: + return + await self.call( + "churn_write", + group, + db.execute_write( + "insert into data (v) values ('churn')", block=w.rng.random() < 0.7 + ), + ) + + async def op_churn_slow(self, w): + # A slow query with Datasette's progress-handler time limit, on a + # database that another task may close or remove meanwhile + name, group = self.pick(w.rng, ("churn",)) + db = self.db(name) + if db is None: + return + await self.call( + "churn_slow", group, db.execute(SLOW_SQL, custom_time_limit=200) + ) + + async def op_churn_isolated_slow(self, w): + # The close-while-stepping crash on main: an isolated function on an + # immutable database (not tracked as pending work there) with a + # progress handler installed, while the database is removed + name, group = self.pick(w.rng, ("churn",)) + db = self.db(name) + if db is None: + return + + rows = self.spec.get("isolated_rows", 300000) + + def fn(conn): + conn.set_progress_handler(lambda: 0, 100) + try: + return conn.execute( + "with recursive c(x) as (select 1 union all select x + 1 " + f"from c where x < {rows}) select count(*) from c" + ).fetchone()[0] + finally: + conn.set_progress_handler(None, 0) + + await self.call("churn_isolated_slow", group, db.execute_isolated_fn(fn)) + + async def op_add_remove(self, w): + churn = [n for n, g in list(self.groups.items()) if g == "churn"] + if len(churn) < 2 or (len(churn) < 6 and w.rng.random() < 0.5): + if not self.closing: + self.add_churn() + self.count("add_database") + return + name = w.rng.choice(churn) + with self.lock: + if self.groups.get(name) != "churn": + return + del self.groups[name] + try: + self.ds.remove_database(name) + self.count("remove_database") + except KeyError: + pass + except Exception as e: # noqa: BLE001 + if self.expected("remove_database", "churn", e) is None: + self.problem("unexpected", f"remove_database({name}): {describe(e)}") + + async def op_replace_delete(self, w): + name, _group = self.pick(w.rng, ("churn",)) + path = self.churn_paths.get(name) + db = self.db(name) + if path is None or db is None or not db.is_mutable: + return + if w.rng.random() < 0.6: + new = f"{path}.{w.name}.{w.rng.randrange(10**9)}.new" + make_file(new, extra_sql="create table replaced (id integer);") + os.replace(new, path) + self.count("file replaced") + else: + for suffix in ("", "-wal", "-shm", "-journal"): + try: + os.unlink(path + suffix) + except OSError: + pass + self.count("file deleted") + + async def op_db_close(self, w): + name, _group = self.pick(w.rng, ("churn",)) + db = self.db(name) + if db is None or not hasattr(db, "close"): + return + with self.lock: + if self.groups.get(name) != "churn": + return + self.groups.pop(name, None) + db.close() + self.count("Database.close") + + async def op_scratch(self, w): + ds = self.ds + if not self.features["scratch"] or self.closing: + return + scratch = [n for n, g in list(self.groups.items()) if g == "scratch"] + roll = w.rng.random() + if not scratch or (roll < 0.3 and len(scratch) < 6): + with self.lock: + self.scratch_seq += 1 + name = f"s{self.seed}_{self.scratch_seq}" + ok, db = await self.call( + "scratch_create", "scratch", ds.create_scratch_database(name) + ) + if ok: + with self.lock: + self.groups[db.name] = "scratch" + await self.call( + "scratch_write", + "scratch", + db.execute_write("create table t (id integer primary key, v text)"), + ) + return + name = w.rng.choice(scratch) + db = self.db(name) + if db is None: + return + if roll < 0.75: + if w.rng.random() < 0.5: + await self.call( + "scratch_write", + "scratch", + db.execute_write("insert into t (v) values ('x')"), + ) + else: + await self.call( + "scratch_read", "scratch", db.execute("select count(*) from t") + ) + elif roll < 0.9: + with self.lock: + self.scratch_seq += 1 + new_name = f"s{self.seed}_{self.scratch_seq}" + ok, new_db = await self.call( + "scratch_rename", "scratch", ds.rename_scratch_database(name, new_name) + ) + if ok: + with self.lock: + self.groups.pop(name, None) + self.groups[new_db.name] = "scratch" + else: + ok, _ = await self.call( + "scratch_delete", "scratch", ds.delete_scratch_database(name) + ) + if ok: + with self.lock: + self.groups.pop(name, None) + + # -- driving --------------------------------------------------------- + + async def worker(self, w, deadline): + ops = self.spec["ops"] + names = list(ops) + weights = [ops[n] for n in names] + while time.monotonic() < deadline: + op = w.rng.choices(names, weights)[0] + if self.closing and op in ("add_remove", "scratch", "replace_delete"): + op = "read" + try: + await getattr(self, f"op_{op}")(w) + except asyncio.CancelledError: + raise + except Exception as e: # noqa: BLE001 + self.problem( + "harness", + f"{op}: {describe(e)}\n" + + "".join(traceback.format_exception(e)[-5:]), + ) + await asyncio.sleep(0) + + def writer(self, loop_index, t): + # One Writer per task slot for the whole run: in "rounds" mode each + # round's event loop reuses the same writers (and their sequences) + key = (loop_index, t) + with self.lock: + w = self.writers.get(key) + if w is None: + rng = random.Random(f"{self.seed}-{loop_index}-{t}") + w = self.writers[key] = Writer(f"w{loop_index}_{t}", rng) + return w + + async def loop_main(self, loop_index, deadline): + tasks = [] + for t in range(self.spec["tasks"]): + tasks.append(self.worker(self.writer(loop_index, t), deadline)) + if self.spec.get("close_midflight") and loop_index == 0: + tasks.append(self.close_midflight(deadline)) + await asyncio.gather(*tasks) + + async def close_midflight(self, deadline): + await asyncio.sleep( + self.rng.uniform(0.3, max(0.4, deadline - time.monotonic() - 1)) + ) + await self.drain_and_verify() + self.closing = True + self.closed_at = time.monotonic() + t0 = time.monotonic() + self.ds.close() + self.counts["close seconds x100"] = int((time.monotonic() - t0) * 100) + + def run_loops(self): + duration = self.spec["duration"] + deadline = time.monotonic() + duration + if self.spec.get("rounds"): + # One thread, a fresh event loop per round + while time.monotonic() < deadline: + round_end = min(deadline, time.monotonic() + 0.4) + asyncio.run(self.loop_main(0, round_end)) + self.count("rounds") + return + errors = [] + + def runner(i): + try: + asyncio.run(self.loop_main(i, deadline)) + except BaseException as e: # noqa: BLE001 + errors.append(e) + + threads = [ + threading.Thread( + target=runner, args=(i,), name=f"torture-loop-{i}", daemon=True + ) + for i in range(self.spec["loops"]) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(duration + OP_TIMEOUT + 15) + if thread.is_alive(): + raise Hang(f"{thread.name} still running") + for e in errors: + self.problem("harness", f"loop died: {describe(e)}") + + # -- checks ---------------------------------------------------------- + + async def drain_and_verify(self): + """Wait for every write queue to drain, then check ordering and that + nothing acknowledged is missing. Runs before close().""" + ds = self.ds + # Other loops may still be writing (close_midflight): only writes + # acknowledged or queued before the drain must be visible to it + with self.lock: + acked = {k: list(v) for k, v in self.acked.items()} + nonblocking = {k: list(v) for k, v in self.nonblocking.items()} + for name in self.stable_names: + db = ds.databases.get(name) + await self.call("drain", "stable", db.execute_write("select 1")) + for name in self.stable_names: + db = ds.databases.get(name) + ok, results = await self.call( + "verify", + "stable", + db.execute( + "select writer, seq from log order by id", custom_time_limit=None + ), + ) + if ok: + self.check_log( + name, + [(r[0], r[1]) for r in results.rows], + "before close", + acked, + nonblocking, + ) + + def check_log(self, name, rows, phase, acked=None, nonblocking=None): + if acked is None: + acked = self.acked + if nonblocking is None: + nonblocking = self.nonblocking + by_writer = collections.defaultdict(list) + for writer, seq in rows: + by_writer[writer].append(seq) + for (db_name, writer), seqs in list(acked.items()): + if db_name != name: + continue + present = set(by_writer.get(writer, [])) + missing = [s for s in seqs if s not in present] + if missing: + self.problem( + "lost-write", + f"{name} ({phase}): {len(missing)} acknowledged writes by " + f"{writer} missing, e.g. seq {missing[:5]}", + ) + for (db_name, writer), seqs in list(nonblocking.items()): + if db_name != name: + continue + present = set(by_writer.get(writer, [])) + missing = [s for s in seqs if s not in present] + if missing: + self.problem( + "lost-write", + f"{name} ({phase}): {len(missing)} block=False writes by " + f"{writer} missing after the queue drained, e.g. seq {missing[:5]}", + ) + for writer, seqs in by_writer.items(): + if seqs != sorted(seqs) or len(seqs) != len(set(seqs)): + self.problem( + "write-order", + f"{name} ({phase}): writer {writer} rows out of order: {seqs[:20]}", + ) + self.count("verified databases") + + async def check_catalog(self): + """After a forced refresh, the catalog matches each stable and + external database file.""" + ds = self.ds + ok, _ = await self.call("refresh", "stable", ds.refresh_schemas(force=True)) + if not ok: + return + for name in self.stable_names + self.ext_names: + db = ds.databases.get(name) + internal = ds.get_internal_database() + ok1, catalog = await self.call( + "catalog", + "stable", + internal.execute( + "select table_name from catalog_tables where database_name = ?", + [name], + ), + ) + ok2, live = await self.call( + "catalog", + self.groups.get(name, "stable"), + db.execute("select name from sqlite_master where type = 'table'"), + ) + if ok1 and ok2: + catalog_set = {r[0] for r in catalog.rows} + live_set = {r[0] for r in live.rows} + if catalog_set != live_set: + if name in self.ext_names: + # The external process may have changed it since; + # one more forced refresh must converge + await ds.refresh_schemas(force=True) + catalog = await internal.execute( + "select table_name from catalog_tables where database_name = ?", + [name], + ) + live = await db.execute( + "select name from sqlite_master where type = 'table'" + ) + catalog_set = {r[0] for r in catalog.rows} + live_set = {r[0] for r in live.rows} + if catalog_set == live_set or self.external is not None: + continue + self.problem( + "catalog", + f"{name}: catalog {sorted(catalog_set ^ live_set)} differ " + "from the file after refresh_schemas(force=True)", + ) + + def check_files_after_close(self): + for name in self.stable_names: + path = os.path.join(self.tmp, f"{name}.db") + conn = sqlite3.connect(path) + try: + rows = conn.execute( + "select writer, seq from log order by id" + ).fetchall() + finally: + conn.close() + self.check_log(name, rows, "file after close") + + def check_resources(self, baseline_threads): + deadline = time.monotonic() + 15 + while True: + gc.collect() + leaked_fds = [] + for fd in os.listdir("/proc/self/fd"): + try: + target = os.readlink(f"/proc/self/fd/{fd}") + except OSError: + continue + if target.startswith(self.tmp) or "datasette_temp_" in target: + leaked_fds.append(target) + threads = [ + t.name + for t in threading.enumerate() + if t.name.startswith("_execute_writes") + or t.name.startswith("ThreadPoolExecutor") + or t.name.startswith("datasette-") + and t.name != "datasette-read-pool-reaper" + ] + if (not leaked_fds and not threads) or time.monotonic() > deadline: + break + time.sleep(0.1) + if leaked_fds: + self.problem( + "fd-leak", + f"{len(leaked_fds)} fds still open after close: " + + ", ".join(sorted({os.path.basename(p) for p in leaked_fds}))[:500], + ) + if threads: + self.problem("thread-leak", f"threads after close: {sorted(set(threads))}") + extra = threading.active_count() - baseline_threads + self.counts["threads above baseline after close"] = max(extra, 0) + + def run(self): + baseline_threads = threading.active_count() + self.build() + ds = self.ds + + async def start(): + await ds.invoke_startup() + await ds.get_database("mem").execute_write_script( + "create table data (id integer primary key, v text);" + "insert into data (v) values ('one'), ('two');" + ) + if self.spec.get("private_cap") and hasattr(ds, "_read_pool"): + # Below the public minimum: the pool's soft-cap path + ds._read_pool.max_open = self.spec["private_cap"] + + asyncio.run(start()) + hung = None + try: + self.run_loops() + except Hang as e: + hung = e + self.problem("hang", str(e)) + if hung is None and not self.closing: + + async def finish(): + await self.drain_and_verify() + await self.check_catalog() + + asyncio.run(finish()) + if self.external is not None: + self.external.kill() + self.external.wait() + if hung is None: + if not self.closing: + ds.close() + ref = weakref.ref(ds) + self.ds = ds = None + self.check_files_after_close() + self.check_resources(baseline_threads) + gc.collect() + if ref() is not None: + self.problem( + "instance-leak", + "Datasette still alive after close() + gc.collect()", + ) + shutil.rmtree(self.tmp, ignore_errors=True) + return hung is None + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("scenario", choices=sorted(SCENARIOS)) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--duration", type=float, default=None) + args = parser.parse_args() + faulthandler.enable() + spec = SCENARIOS[args.scenario] + duration = args.duration if args.duration is not None else spec["duration"] + # A last resort: dump every thread's stack and exit if the whole run + # takes far longer than it should + faulthandler.dump_traceback_later(duration + OP_TIMEOUT + 60, exit=True) + torture = Torture(args.scenario, args.seed, args.duration) + t0 = time.monotonic() + finished = torture.run() + summary = { + "scenario": args.scenario, + "seed": args.seed, + "python": sys.version.split()[0], + "gil": getattr(sys, "_is_gil_enabled", lambda: True)(), + "seconds": round(time.monotonic() - t0, 2), + "features": torture.features, + "ok": finished and not torture.problems, + "problems": torture.problems[:50], + "problem_kinds": dict(collections.Counter(p["kind"] for p in torture.problems)), + "counts": dict(torture.counts), + "expected_errors": dict(torture.expected_errors), + } + print(json.dumps(summary), flush=True) + faulthandler.cancel_dump_traceback_later() + if not finished or any(p["kind"] == "hang" for p in torture.problems): + os._exit(3) + os._exit(0 if summary["ok"] else 1) + + +if __name__ == "__main__": + main() diff --git a/tests/test_connection_review_fixes.py b/tests/test_connection_review_fixes.py index 7b99d3b4..39153368 100644 --- a/tests/test_connection_review_fixes.py +++ b/tests/test_connection_review_fixes.py @@ -668,22 +668,22 @@ async def test_stale_owned_catalog_is_not_persisted_at_close(tmp_path): make_db(path) config = {"databases": {"data": {"schema_watch": "owned"}}} - async def tables_after_start(change=None): + async def tables_after_start(change_before_close=None): ds = Datasette([path], internal=internal, config=config) await ds.invoke_startup() - if change: - change() response = await ds.client.get("/data.json") + if change_before_close: + # Behind Datasette's back, after the last request: an owned + # database is not polled, so the catalog is now stale + change_before_close() ds.close() return [t["name"] for t in response.json()["tables"]] assert await tables_after_start() == ["t"] - # Owned: a change made behind Datasette's back is not noticed while - # it runs (documented) ... assert await tables_after_start( lambda: make_db(path, "create table t2 (id integer)") ) == ["t"] - # ... but it is not persisted as current either + # The stale catalog was not persisted as current: the restart rescans assert await tables_after_start() == ["t", "t2"] @@ -744,3 +744,164 @@ async def test_execute_fn_may_return_generator_over_fetched_rows(): with pytest.raises(ConnectionLeaseError): await db.execute_fn(closure) ds.close() + + +# ---------------------------------------------------------------------- +# Found by the torture harness (tests/connection_torture.py) +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_index_survives_database_removed_mid_request(tmp_path, monkeypatch): + # The index page looked databases up by name again after listing them: + # one removed meanwhile (remove_database(), a scratch delete or rename) + # failed the whole page with a KeyError + from datasette.views import index as index_module + + paths = [] + for name in ("keep", "gone"): + path = str(tmp_path / f"{name}.db") + make_db(path) + paths.append(path) + ds = Datasette(paths) + await ds.invoke_startup() + original = index_module.catalog_summaries + + async def remove_during(datasette, names): + result = await original(datasette, names) + if "gone" in datasette.databases: + datasette.remove_database("gone") + return result + + monkeypatch.setattr(index_module, "catalog_summaries", remove_during) + response = await ds.client.get("/.json") + assert response.status_code == 200 + assert [d["name"] for d in response.json()["databases"]] == ["keep"] + ds.close() + + +@pytest.mark.asyncio +async def test_index_and_databases_with_closed_or_deleted_database(tmp_path): + paths = [] + for name in ("keep", "closed", "deleted"): + path = str(tmp_path / f"{name}.db") + make_db(path) + paths.append(path) + ds = Datasette(paths, settings={"schema_watch_interval_ms": 0}) + await ds.invoke_startup() + # Closed by a plugin but still attached + ds.get_database("closed").close() + # Deleted from under Datasette before any sweep noticed + os.unlink(paths[2]) + for path in ("/.json", "/", "/-/databases.json"): + response = await ds.client.get(path) + assert response.status_code == 200, (path, response.text[:300]) + databases = (await ds.client.get("/-/databases.json")).json() + if isinstance(databases, dict): + databases = databases["databases"] + sizes = {d["name"]: d["size"] for d in databases} + assert sizes["deleted"] == 0 + ds.close() + + +@pytest.mark.asyncio +async def test_cursors_are_closed_before_leaving_their_thread(owned_db): + # A cursor still references its connection's cached statement, and on + # Python < 3.12 freeing it resets that statement. execute_write_many() + # handed its cursor to the caller's event loop thread; freed there while + # the write thread ran the same SQL again, the write failed with + # "bad parameter or other API misuse" + _ds, db, _path = owned_db + cursor = await db.execute_write_many( + "insert into t (v) values (?)", [("a",), ("b",)] + ) + assert cursor.rowcount == 2 + with pytest.raises(sqlite3.ProgrammingError): + cursor.fetchall() + cursor = await db.execute_write_fn( + lambda conn: conn.execute("insert into t (v) values ('c')") + ) + assert cursor.lastrowid == 3 + with pytest.raises(sqlite3.ProgrammingError): + cursor.fetchone() + + # Cursors held by the frames of an exception raised by a callback are + # closed before the connection goes back to the pool (or write thread) + captured = [] + + def read_then_fail(conn): + cursor = conn.execute("select * from t") + captured.append(cursor) + raise ValueError("boom") + + with pytest.raises(ValueError): + await db.execute_fn(read_then_fail) + with pytest.raises(ValueError): + await db.execute_write_fn(read_then_fail) + for cursor in captured: + with pytest.raises(sqlite3.ProgrammingError): + cursor.fetchone() + + +def test_startup_from_several_event_loops_runs_once(monkeypatch): + # _startup_sequence() serialized startup with an asyncio.Lock: callers + # on a second event loop got "is bound to a different event loop" (or, + # free-threaded, waited forever), and invoke_startup() itself let two + # loops run every startup hook twice + ds = Datasette(memory=True) + calls = [] + original = Datasette._apply_column_types_config + + async def slow(self): + calls.append(threading.current_thread().name) + await asyncio.sleep(0.3) + return await original(self) + + monkeypatch.setattr(Datasette, "_apply_column_types_config", slow) + + def starter(use_sequence): + async def run(): + if use_sequence: + await ds._startup_sequence() + else: + await ds.invoke_startup() + assert ds._startup_invoked + return (await ds.client.get("/_memory.json")).status_code + + return run + + statuses = run_in_threads(starter(True), starter(True), starter(False)) + assert statuses == [200, 200, 200] + assert len(calls) == 1 + ds.close() + + +@pytest.mark.asyncio +async def test_startup_hook_can_make_requests(): + # A request made from inside startup (same task) must not wait for + # startup to finish - that would wait forever + from datasette import hookimpl + from datasette.plugins import pm + + statuses = [] + + class Plugin: + __name__ = "StartupRequestPlugin" + + @hookimpl + def startup(self, datasette): + async def inner(): + statuses.append( + (await datasette.client.get("/-/versions.json")).status_code + ) + + return inner + + pm.register(Plugin(), name="startup_request_plugin") + try: + ds = Datasette(memory=True) + await asyncio.wait_for(ds.invoke_startup(), 10) + assert statuses == [200] + ds.close() + finally: + pm.unregister(name="startup_request_plugin") diff --git a/tests/test_connection_torture.py b/tests/test_connection_torture.py new file mode 100644 index 00000000..f6a06569 --- /dev/null +++ b/tests/test_connection_torture.py @@ -0,0 +1,69 @@ +""" +A fast, fixed-seed subset of the connection torture harness +(tests/connection_torture.py). Each scenario runs in its own process, so a +segfault or a hang fails one test instead of killing the test run. + +Several event loops (threads) drive one Datasette with extreme settings +(smallest pool, 1ms idle timeout, 10ms schema polling) through reads, +writes of every kind, DDL, isolated functions, analyze_sql, cancellation, +add/remove_database, scratch databases, external schema changes, files +replaced and deleted, Database.close() and Datasette.close() mid-flight, +and non-threaded mode. See the harness docstring for what is checked. +""" + +import json +import os +import signal +import subprocess +import sys + +import pytest + +HARNESS = os.path.join(os.path.dirname(__file__), "connection_torture.py") + + +@pytest.mark.parametrize( + "scenario,seed", + [ + ("mixed", 101), + ("lifecycle", 102), + ("close", 103), + ("cancel", 104), + ("pool", 105), + ("nothreads", 106), + ("writes", 107), + ], +) +def test_connection_torture(scenario, seed): + try: + proc = subprocess.run( + [ + sys.executable, + HARNESS, + scenario, + "--seed", + str(seed), + "--duration", + "1.5", + ], + capture_output=True, + text=True, + timeout=150, + check=False, + ) + except subprocess.TimeoutExpired as e: # pragma: no cover + pytest.fail(f"{scenario} hung:\n{(e.stderr or '')[-4000:]}") + if proc.returncode < 0: # pragma: no cover + pytest.fail( + f"{scenario} crashed with {signal.Signals(-proc.returncode).name}:\n" + f"{proc.stderr[-4000:]}" + ) + summary = json.loads(proc.stdout.strip().splitlines()[-1]) + problems = "\n".join( + f"[{p['kind']}] {p['message'][:600]}" for p in summary["problems"] + ) + assert proc.returncode == 0, f"{scenario} seed {seed}:\n{problems}" + assert summary["ok"] + # It did something (the slow-query scenarios do a few dozen operations + # in 1.5s on a busy machine, the others hundreds) + assert sum(summary["counts"].values()) > 10 diff --git a/tests/test_schema_watcher.py b/tests/test_schema_watcher.py index ad24c10d..fe9b6613 100644 --- a/tests/test_schema_watcher.py +++ b/tests/test_schema_watcher.py @@ -625,7 +625,9 @@ async def test_default_schema_watch_setting(tmp_path, db_path, setting, expected Database(ds, path=other), name="polled", schema_watch="external" ) assert polled._watch_state.mode == "external" - owned = ds.add_database(Database(ds, path=other), name="owned", schema_watch="owned") + owned = ds.add_database( + Database(ds, path=other), name="owned", schema_watch="owned" + ) assert owned._watch_state.mode == "owned" ds.close()