From 1f154f1aa468a5e1dda8c0e64b0e50d3ad5f9aae Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 05:56:03 +0000 Subject: [PATCH] Scratch databases: file-based, crash-consistent, survive restart datasette.create_scratch_database() / delete_scratch_database() / rename_scratch_database() / list_scratch_databases() manage SQLite files in a scratch directory (--scratch-dir, config_dir/scratch, or a temporary directory created on first use). - Registry: a SQLite sidecar (.datasette-scratch.db) with intent states (creating/deleting/renaming); startup reconciles it with the directory, which is the source of truth (hand-copied files are adopted, hand-deleted ones dropped, orphan -wal/-shm/-journal removed). flock() per directory. - Scratch databases are attached at startup without opening them, use schema watch mode "owned", and track last_used in memory (flushed to the registry at most once a minute and on close). - Delete/rename: queued reads and writes fail with ScratchDatabaseDeleted, running ones finish; files are unlinked only once no connection (including the SchemaWatcher's, now counted) has them open. - Datasette.close() records final fingerprints of owned file databases in a persistent internal database, so restarts reuse their catalog rows without opening them. - Scratch databases are not attached to crossdb _memory and are skipped by check_databases(). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/app.py | 89 +++- datasette/cli.py | 11 + datasette/database.py | 39 +- datasette/schema_watcher.py | 96 +++- datasette/scratch.py | 1000 +++++++++++++++++++++++++++++++++++ docs/cli-reference.rst | 2 + docs/internals.rst | 153 +++++- docs/settings.rst | 1 + tests/test_scratch.py | 952 +++++++++++++++++++++++++++++++++ 9 files changed, 2326 insertions(+), 17 deletions(-) create mode 100644 datasette/scratch.py create mode 100644 tests/test_scratch.py diff --git a/datasette/app.py b/datasette/app.py index e090a4f1..fbbfdce9 100644 --- a/datasette/app.py +++ b/datasette/app.py @@ -49,7 +49,12 @@ 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 .schema_watcher import ( + FILES_DEFAULT, + SchemaWatcher, + store_closing_fingerprints, +) +from .scratch import ScratchDatabases from .telemetry import ( TelemetryMiddleware, _in_datasette_client, @@ -451,6 +456,7 @@ class Datasette: nolock=False, internal=None, default_deny=False, + scratch_dir=None, ): self._startup_invoked = False self._shutdown_invoked = False @@ -526,6 +532,14 @@ class Datasette: self._internal_database = Database(self, path=internal, mode="rwc") self._internal_database.name = INTERNAL_DB_NAME + # Scratch databases: an explicit directory, else config_dir/scratch + # if it exists, else a temporary directory created on first use + if scratch_dir is None and config_dir and (config_dir / "scratch").is_dir(): + scratch_dir = config_dir / "scratch" + self._scratch = ScratchDatabases(self, scratch_dir) + # Attaches existing scratch databases without opening any of them + self._scratch.load() + self.cache_headers = cache_headers self._static_asset_hashes = {} self.cors = cors @@ -989,6 +1003,11 @@ class Datasette: ) def remove_database(self, name): + db = self._detach_database(name) + db.close() + + def _detach_database(self, name): + # remove_database() without the close() db = self.get_database(name) new_databases = self.databases.copy() new_databases.pop(name) @@ -996,7 +1015,40 @@ class Datasette: # Deletes this database's catalog rows (explicitly - there is no # periodic stale-catalog scan any more) self._schema_watcher.unregister(name) - db.close() + return db + + @property + def scratch_dir(self): + """The scratch directory in use, or None if scratch databases are + temporary and none has been created yet.""" + return self._scratch.directory + + async def create_scratch_database(self, name=None, *, actor=None, metadata=None): + """Create a new, empty scratch database and attach it. + + ``name`` defaults to a random ``scratch_xxxxxxxx`` name. ``actor`` + (an actor dictionary or an actor ID) is recorded as the owner and + ``metadata`` is any JSON-serializable dictionary, both returned by + :meth:`list_scratch_databases`. Returns the + :class:`~datasette.scratch.ScratchDatabase`. + """ + return await self._scratch.create(name, actor=actor, metadata=metadata) + + async def delete_scratch_database(self, name): + """Detach a scratch database and delete its files. + + Waits for reads and writes that are already running; calls that are + queued or made later raise + :class:`~datasette.scratch.ScratchDatabaseDeleted`.""" + await self._scratch.delete(name) + + async def rename_scratch_database(self, name, new_name): + "Rename a scratch database. Returns the new Database object." + return await self._scratch.rename(name, new_name) + + async def list_scratch_databases(self): + "List scratch databases as ScratchDatabaseInfo objects, sorted by name." + return await self._scratch.list() def close(self): """Release all resources held by this Datasette instance. @@ -1012,14 +1064,32 @@ class Datasette: # Stop reporting metrics before closing databases unregister_datasette(self) first_exception = None - dbs = list(self.databases.values()) + [self._internal_database] - for db in dbs: + for db in list(self.databases.values()): try: db.close() except Exception as e: # noqa: BLE001 # Collect the first failure and re-raise after every close() has run if first_exception is None: first_exception = e + internal = self._internal_database + closing_fingerprints = [] + if self.internal_db_created and not internal.is_temp_disk: + # Owned databases are closed now: record their final + # fingerprints so a restart does not rescan them + try: + closing_fingerprints = self._schema_watcher.closing_fingerprints() + except Exception: + logger.exception("Could not fingerprint databases at close") + try: + internal.close() + except Exception as e: # noqa: BLE001 + if first_exception is None: + first_exception = e + if closing_fingerprints: + try: + store_closing_fingerprints(internal.path, closing_fingerprints) + except Exception: + logger.exception("Could not store fingerprints at close") read_pool = self._read_pool_or_none if read_pool is not None: read_pool.close() @@ -1029,6 +1099,13 @@ class Datasette: except Exception as e: # noqa: BLE001 if first_exception is None: first_exception = e + try: + # Records last_used, unlocks the directory, removes a temporary + # scratch directory + self._scratch.close() + except Exception as e: # noqa: BLE001 + if first_exception is None: + first_exception = e if first_exception is not None: raise first_exception @@ -1651,7 +1728,9 @@ class Datasette: if self.crossdb and database == "_memory": count = 0 for db_name, db in self.databases.items(): - if count >= SQLITE_LIMIT_ATTACHED or db.is_memory: + # Scratch databases are never attached: they can be deleted + # while a pooled _memory connection would still hold them + if count >= SQLITE_LIMIT_ATTACHED or db.is_memory or db.is_scratch: continue sql = 'ATTACH DATABASE "file:{path}?{qs}" AS [{name}];'.format( path=db.path, diff --git a/datasette/cli.py b/datasette/cli.py index e83de93a..24cc4c93 100644 --- a/datasette/cli.py +++ b/datasette/cli.py @@ -503,6 +503,12 @@ def uninstall(packages, yes): help="Path to a persistent Datasette internal SQLite database", envvar="DATASETTE_INTERNAL", ) +@click.option( + "--scratch-dir", + type=click.Path(file_okay=False), + help="Directory for scratch databases, which then survive a restart", + envvar="DATASETTE_SCRATCH_DIR", +) def serve( files, immutable, @@ -537,6 +543,7 @@ def serve( ssl_keyfile, ssl_certfile, internal, + scratch_dir=None, return_instance=False, ): """Serve up specified SQLite database files with a web UI""" @@ -604,6 +611,7 @@ def serve( "nolock": nolock, "internal": internal, "default_deny": default_deny, + "scratch_dir": scratch_dir, } # Separate directories from files @@ -899,6 +907,9 @@ async def check_databases(ds): # Run check_connection against every connected database # to confirm they are all usable for database in list(ds.databases.values()): + if database.is_scratch: + # Created by Datasette; checking them would open every one + continue try: await database.execute_fn(check_connection) except SpatialiteConnectionProblem: diff --git a/datasette/database.py b/datasette/database.py index ef60bc57..c4db3c73 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -83,6 +83,9 @@ _SHUTDOWN = object() class Database: # For table counts stop at this many rows: count_limit = 10000 + # True for ScratchDatabase (datasette.scratch): created and deleted by + # Datasette in its scratch directory + is_scratch = False def __init__( self, @@ -156,6 +159,12 @@ class Database: # connection opened, or the SchemaWatcher stat()ed it). From then on # write connections never create the file - see connect() self._file_seen = False + # Short-lived connections that are not in _all_connections (the + # SchemaWatcher's scans and pragma checks): counted, so that deleting + # the file can wait until none is open - see + # _untracked_connection_opened() + self._untracked_open = 0 + self._untracked_cond = threading.Condition() if not is_temp_disk: self.mode = mode @@ -191,6 +200,32 @@ class Database: except ValueError: pass + def _untracked_connection_opened(self): + """Call before opening a connection with ``connect(track=False)``. + + Raises DatasetteClosedError once close() has started, so after + close() no new untracked connection can be opened and + _wait_for_untracked_connections() only has to wait for the ones + already counted.""" + with self._untracked_cond: + if self._closed: + raise DatasetteClosedError(f"Database {self.name!r} has been closed") + self._untracked_open += 1 + + def _untracked_connection_closed(self): + with self._untracked_cond: + self._untracked_open -= 1 + if self._untracked_open <= 0: + self._untracked_cond.notify_all() + + def _wait_for_untracked_connections(self, timeout=None): + """Wait until every untracked connection has been closed. Only + meaningful after close(). Returns False on timeout.""" + with self._untracked_cond: + return self._untracked_cond.wait_for( + lambda: self._untracked_open <= 0, timeout=timeout + ) + def _schema_check(self, conn, loop=None): # Runs on the thread that owns conn, right after a write task state = self._watch_state @@ -329,7 +364,7 @@ class Database: """ if self._closed: return - with self._pending_execute_futures_lock: + with self._pending_execute_futures_lock, self._untracked_cond: if self._closed: return self._closed = True @@ -1306,6 +1341,8 @@ class Database: tags.append("memory") if self.is_temp_disk: tags.append("temp_disk") + if self.is_scratch: + tags.append("scratch") if self.hash: tags.append(f"hash={self.hash}") if self.size is not None: diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index d7c47bbf..cff81aff 100644 --- a/datasette/schema_watcher.py +++ b/datasette/schema_watcher.py @@ -219,17 +219,42 @@ def header_schema_version(path): 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_to_json(fp, t_ns, closed=False): + data = {"fp": fp, "t": t_ns} + if closed: + # Taken by Datasette.close() after every connection to an owned + # database had been closed: nothing in this process can change the + # file after the stat, so the racily-clean rule does not apply + data["closed"] = True + return json.dumps(data) def _fp_from_json(value): + """(fp, t_ns, closed) - (None, None, False) if unreadable.""" 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"]) + return fp, int(data["t"]), bool(data.get("closed")) except Exception: # noqa: BLE001 - return None, None + return None, None, False + + +def store_closing_fingerprints(internal_path, rows): + """Write SchemaWatcher.closing_fingerprints() to a persistent internal + database that has already been closed (Datasette.close() is synchronous + and the internal database's write thread is gone by then). Only rows + whose stored schema_version still matches are updated.""" + conn = sqlite3.connect(internal_path, isolation_level=None) + try: + conn.execute("BEGIN IMMEDIATE") + conn.executemany( + "UPDATE catalog_databases SET fingerprint = ? " + "WHERE database_name = ? AND schema_version = ?", + [(fp_json, name, version) for name, version, fp_json in rows], + ) + conn.execute("COMMIT") + finally: + conn.close() class SchemaWatcher: @@ -649,20 +674,30 @@ class SchemaWatcher: # 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) + # It is counted instead, so that deleting a scratch database's files + # can wait for it (and raises once the database has been closed). + db._untracked_connection_opened() + try: + conn = db.connect(track=False) + except BaseException: + db._untracked_connection_closed() + raise try: if prepare: self.ds._prepare_connection(conn, db.name) else: conn.row_factory = None - except Exception: - conn.close() + except BaseException: + self._close(db, conn) raise return conn @staticmethod def _close(db, conn): - conn.close() + try: + conn.close() + finally: + db._untracked_connection_closed() def _scan_sync(self, state): """Read schema_version + full schema in one read transaction.""" @@ -804,6 +839,9 @@ class SchemaWatcher: if state.removed: continue if "error" in r: + if state.db._closed: + # Closed (or deleted) while the scan was running + continue state.error = r["error"] state.needs_scan = True logger.warning( @@ -953,8 +991,8 @@ class SchemaWatcher: 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): + stored_fp, stored_t, closed = _fp_from_json(row["fingerprint"]) + if stored_fp is None or (is_racy(stored_fp, stored_t) and not closed): continue t_ns = time.time_ns() fp = fingerprint(state.db.path) @@ -962,6 +1000,44 @@ class SchemaWatcher: out[state.name] = (fp, t_ns, row["schema_version"]) return out + # ------------------------------------------------------------------ + # shutdown + # ------------------------------------------------------------------ + def closing_fingerprints(self): + """Called by Datasette.close() once every attached database has been + closed. For each owned file database whose catalog is current, + returns ``(name, schema_version, fingerprint_json)`` for the file as + it is now. Owned databases are only changed through Datasette, so a + restart with a persistent internal database can reuse their catalog + rows without opening them - even if data was written (or the -wal + checkpointed by the final close) after the catalog was last built. + Stat calls only.""" + out = [] + for state in list(self.states.values()): + if ( + state.mode != "owned" + or not state.is_file + or state.removed + or state.missing + or state.needs_scan + or state.scan_future is not None + or state.error is not None + or state.catalog_version is None + or state.notified_version not in (None, state.catalog_version) + ): + continue + t_ns = time.time_ns() + try: + fp = fingerprint(state.db.path) + except OSError: + continue + if fp[0] is None: + continue + out.append( + (state.name, state.catalog_version, _fp_to_json(fp, t_ns, closed=True)) + ) + return out + # ------------------------------------------------------------------ # introspection # ------------------------------------------------------------------ diff --git a/datasette/scratch.py b/datasette/scratch.py new file mode 100644 index 00000000..e75d7007 --- /dev/null +++ b/datasette/scratch.py @@ -0,0 +1,1000 @@ +"""Scratch databases: SQLite files that plugins create, fill, change and +throw away, and that survive a restart. + +The scratch directory looks like this:: + + / + .datasette-scratch.db the registry: one row per scratch database + .datasette-scratch.lock flock()ed by the Datasette instance using it + .db one SQLite file per scratch database + .db-wal, -shm while the database is open (WAL mode) + +Why a registry file next to the databases (and not metadata inside each +file, or rows in the internal database): see "Scratch databases" in +docs/internals.rst. In short: + +* Each registry change is one SQLite transaction. create, delete and rename + first record an intent (``state`` = creating / deleting / renaming), then + change the files, then finish the row. ``_reconcile()`` at startup rolls + back an interrupted create and finishes an interrupted delete or rename, + so a crash at any point leaves either the old or the new state, never a + database without its metadata or a stray ``-wal`` file. +* Startup reads one small file and lists one directory. No scratch database + is opened, and no thread is started, until something uses it. +* The registry lives with the files, so it survives a temporary internal + database, and moving or copying the directory keeps the metadata. +* ``last_used`` is tracked in memory on every read and write and written to + the registry at most once per ``LAST_USED_FLUSH_INTERVAL_S`` (and on + close, and with any other registry change) - never a write per read. + +The directory is the source of truth for which databases exist: a ``.db`` +file copied in by hand is adopted at the next startup (without owner or +metadata), and a row whose file was deleted by hand is dropped. +""" + +import asyncio +import dataclasses +import errno +import json +import logging +import os +import re +import secrets +import shutil +import tempfile +import threading +import time +import weakref +from pathlib import Path + +from .database import Database, DatasetteClosedError +from .utils import StartupError, sqlite3 + +logger = logging.getLogger("datasette.scratch") + +REGISTRY_FILENAME = ".datasette-scratch.db" +LOCK_FILENAME = ".datasette-scratch.lock" +SUFFIX = ".db" +# Removal order: -wal first. A -wal left next to a missing main file would be +# replayed into the next database file of that name (SQLite says "database +# disk image is malformed") +SIDECAR_SUFFIXES = ("-wal", "-journal", "-shm") +NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}") +# Characters that would change the meaning of the file: URI Datasette opens +# databases with +UNSAFE_PATH_CHARS = ("?", "#", "%") +LAST_USED_FLUSH_INTERVAL_S = 60.0 +SQLITE_HEADER = b"SQLite format 3\x00" + +REGISTRY_SCHEMA = """ +CREATE TABLE IF NOT EXISTS scratch_databases ( + name TEXT PRIMARY KEY, + created REAL NOT NULL, + last_used REAL, + owner TEXT, + metadata TEXT, + -- ready | creating | deleting | renaming + state TEXT NOT NULL DEFAULT 'ready', + rename_to TEXT +) +""" + + +# Registry statements. create, delete and rename record their intent first +# (state = creating / deleting / renaming), then change the files, then +# finish the row - see _reconcile() +SQL_INSERT_CREATING = ( + "insert into scratch_databases " + "(name, created, last_used, owner, metadata, state) " + "values (?, ?, ?, ?, ?, 'creating')" +) +SQL_MARK_READY = "update scratch_databases set state = 'ready' where name = ?" +SQL_MARK_DELETING = "update scratch_databases set state = 'deleting' where name = ?" +SQL_MARK_RENAMING = ( + "update scratch_databases set state = 'renaming', rename_to = ? where name = ?" +) +SQL_FINISH_RENAME = ( + "update scratch_databases set name = ?, state = 'ready', rename_to = null " + "where name = ?" +) +SQL_CANCEL_RENAME = ( + "update scratch_databases set state = 'ready', rename_to = null where name = ?" +) +SQL_DELETE = "delete from scratch_databases where name = ?" +SQL_INSERT_ADOPTED = "insert into scratch_databases (name, created) values (?, ?)" + + +class ScratchDatabaseError(Exception): + """A scratch database operation could not be carried out.""" + + +class ScratchDatabaseExists(ScratchDatabaseError): + """A database with that name already exists.""" + + +class ScratchDatabaseNotFound(ScratchDatabaseError): + """There is no scratch database with that name.""" + + +class ScratchDatabaseDeleted(DatasetteClosedError): + """Raised to callers using a scratch database that has been deleted or + renamed, including calls that were queued when that happened.""" + + +@dataclasses.dataclass(frozen=True) +class ScratchDatabaseInfo: + name: str + path: str + created: float + last_used: float | None + size: int + owner: str | None + metadata: dict + attached: bool + + +class _Entry: + __slots__ = ( + "created", + "db", + "dirty", + "last_used", + "metadata", + "name", + "owner", + "state", + ) + + def __init__(self, name, created, owner=None, metadata=None, state="ready"): + self.name = name + self.created = created + self.last_used = None + self.owner = owner + self.metadata = metadata or {} + self.state = state + self.db = None + self.dirty = False + + +class ScratchDatabase(Database): + """A database in the scratch directory, created with + ``datasette.create_scratch_database()``. + + Behaves like any other mutable file database. Its catalog is kept current + by its own writes (schema watch mode ``owned``). Once it has been deleted + or renamed every call - including reads and writes that were already + queued - raises :class:`ScratchDatabaseDeleted`. + """ + + is_scratch = True + + def __init__(self, ds, path, *, entry, manager): + super().__init__(ds, path=str(path), is_mutable=True) + self._scratch_entry = entry + self._scratch_manager = manager + # None while usable, otherwise why it is not ("was deleted") + self._scratch_gone = None + # Datasette created this file or found it in the scratch directory: + # writes use mode=rw and never recreate it if it is deleted by hand + self._file_seen = True + + @property + def owner(self): + return self._scratch_entry.owner + + @property + def created(self): + return self._scratch_entry.created + + @property + def last_used(self): + return self._scratch_entry.last_used + + @property + def scratch_metadata(self): + return dict(self._scratch_entry.metadata) + + @property + def size(self): + # 0 once deleted, rather than FileNotFoundError (repr() uses this) + try: + return super().size + except FileNotFoundError: + return 0 + + def _gone_error(self): + return ScratchDatabaseDeleted( + f"Scratch database {self.name!r} {self._scratch_gone}" + ) + + def _check_not_closed(self): + if self._scratch_gone is not None: + raise self._gone_error() + super()._check_not_closed() + + def _guard(self, fn): + # Runs when the queued call is picked up by a thread: anything queued + # before a delete fails here instead of touching the file + def guarded(conn): + if self._scratch_gone is not None: + raise self._gone_error() + return fn(conn) + + return guarded + + def _translate(self, e): + if self._scratch_gone is not None and not isinstance(e, ScratchDatabaseDeleted): + return self._gone_error() + return None + + async def _execute_fn(self, fn): + self._check_not_closed() + self._scratch_manager._touch(self) + try: + return await super()._execute_fn(self._guard(fn)) + except DatasetteClosedError as e: + raise (self._translate(e) or e) from None + + async def _execute_write_fn(self, fn, block=True, transaction=True, request=None): + self._check_not_closed() + self._scratch_manager._touch(self) + return await super()._execute_write_fn( + fn, block=block, transaction=transaction, request=request + ) + + async def execute_isolated_fn(self, fn): + self._check_not_closed() + self._scratch_manager._touch(self) + return await super().execute_isolated_fn(fn) + + async def _send_to_write_thread(self, fn, **kwargs): + try: + return await super()._send_to_write_thread(self._guard(fn), **kwargs) + except DatasetteClosedError as e: + raise (self._translate(e) or e) from None + + +def _owner_id(actor): + if actor is None: + return None + if isinstance(actor, dict): + actor_id = actor.get("id") + return None if actor_id is None else str(actor_id) + return str(actor) + + +def _fsync_dir(directory): + try: + fd = os.open(directory, os.O_RDONLY) + except OSError: + return + try: + os.fsync(fd) + except OSError: + pass + finally: + os.close(fd) + + +def _looks_like_sqlite(path): + try: + with open(path, "rb") as fp: + header = fp.read(16) + except OSError: + return False + # An empty file is a valid, empty SQLite database + return header == b"" or header == SQLITE_HEADER + + +class ScratchDatabases: + """Owns the scratch directory and its registry. One per Datasette, + available as ``datasette._scratch``; plugins use the Datasette methods + (``create_scratch_database()`` etc.).""" + + def __init__(self, ds, directory=None): + self.ds = ds + self.configured_dir = ( + Path(directory).expanduser().resolve() if directory else None + ) + # The directory in use: configured_dir, or a temporary directory + # created by the first create_scratch_database() call + self.directory = None + self._entries = {} + # Protects _entries and _dirty. Never held across a blocking call + self._lock = threading.Lock() + # Serializes registry transactions (they run on worker threads) + self._registry_lock = threading.Lock() + self._registry_initialized = False + self._dirty = set() + self._last_flush = time.monotonic() + self._flush_scheduled = False + self._closed = False + self._finalizers = [] + self.counters = { + "registry_transactions": 0, + "last_used_flushes": 0, + "adopted": 0, + "dropped_missing": 0, + "rolled_back_creates": 0, + "finished_deletes": 0, + "finished_renames": 0, + "stray_files_removed": 0, + } + + @property + def persistent(self): + return self.configured_dir is not None + + # ------------------------------------------------------------------ + # startup + # ------------------------------------------------------------------ + def load(self): + """Called from Datasette.__init__. Attaches every scratch database + in the configured directory without opening any of them.""" + if self.configured_dir is None: + return + self._open_directory(self.configured_dir) + for entry in self._reconcile(): + self._attach(entry) + + def _open_directory(self, directory): + directory = Path(directory) + if any(c in str(directory) for c in UNSAFE_PATH_CHARS): + raise StartupError( + f"Scratch directory path {str(directory)!r} must not contain any of " + f"{' '.join(UNSAFE_PATH_CHARS)}" + ) + try: + directory.mkdir(parents=True, exist_ok=True) + except OSError as e: + raise StartupError(f"Cannot create scratch directory {directory}: {e}") + self._acquire_directory_lock(directory) + self.directory = str(directory) + + def _acquire_directory_lock(self, directory): + try: + import fcntl + except ImportError: # Windows, Pyodide: no advisory lock + return + fd = os.open(os.path.join(directory, LOCK_FILENAME), os.O_RDWR | os.O_CREAT) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as e: + os.close(fd) + if e.errno in (errno.EWOULDBLOCK, errno.EAGAIN, errno.EACCES): + raise StartupError( + f"Scratch directory {directory} is in use by another Datasette instance" + ) + # flock() not supported here (some network file systems, + # Emscripten): carry on without the lock + logger.warning("Could not lock scratch directory %s: %s", directory, e) + return + # Released by close(), or when this object is garbage collected + self._finalizers.append(weakref.finalize(self, os.close, fd)) + + def _ensure_directory(self): + if self.directory is not None: + return + with self._lock: + if self.directory is not None: + return + path = tempfile.mkdtemp(prefix="datasette_scratch_") + # Removed by close(), or at exit. Holds only the path, so it + # does not keep this Datasette alive (unlike atexit.register of a + # bound method) + self._finalizers.append( + weakref.finalize(self, shutil.rmtree, path, ignore_errors=True) + ) + self.directory = path + + def _reconcile(self): + """Bring the registry and the directory into agreement. Returns the + entries to attach. Runs once, at startup, before anything else can + use the directory (the directory lock is held).""" + directory = self.directory + files = set(os.listdir(directory)) + on_disk = set() + for filename in files: + if not filename.endswith(SUFFIX) or filename.startswith("."): + continue + name = filename[: -len(SUFFIX)] + if NAME_RE.fullmatch(name): + on_disk.add(name) + else: + logger.warning( + "Ignoring %s in scratch directory %s: not a valid scratch database name", + filename, + directory, + ) + statements = [] + entries = {} + with self._registry_lock: + conn = self._registry_connection() + try: + rows = conn.execute( + "select name, created, last_used, owner, metadata, state, rename_to " + "from scratch_databases" + ).fetchall() + finally: + conn.close() + for name, created, last_used, owner, metadata, state, rename_to in rows: + if state == "creating": + # The create never returned to its caller: roll it back + self._unlink_files(name) + on_disk.discard(name) + statements.append((SQL_DELETE, [name])) + self.counters["rolled_back_creates"] += 1 + continue + if state == "deleting": + self._unlink_files(name) + on_disk.discard(name) + statements.append((SQL_DELETE, [name])) + self.counters["finished_deletes"] += 1 + continue + final_name = name + if state == "renaming": + old_exists, new_exists = name in on_disk, rename_to in on_disk + if old_exists and new_exists and self._same_file(name, rename_to): + # Crashed between link() and unlink() + self._unlink_files(name) + old_exists = False + on_disk.discard(name) + if new_exists and not old_exists: + final_name = rename_to + statements.append((SQL_FINISH_RENAME, [rename_to, name])) + self.counters["finished_renames"] += 1 + else: + statements.append((SQL_CANCEL_RENAME, [name])) + if final_name not in on_disk: + # Deleted by hand + statements.append((SQL_DELETE, [name])) + self.counters["dropped_missing"] += 1 + continue + on_disk.discard(final_name) + entry = _Entry( + final_name, created, owner, json.loads(metadata) if metadata else {} + ) + entry.last_used = last_used + entries[final_name] = entry + for name in sorted(on_disk): + path = self._path(name) + if not _looks_like_sqlite(path): + logger.warning("Ignoring %s: not a SQLite database", path) + continue + # Copied in by hand: adopt it, with no owner or metadata + created = os.stat(path).st_mtime + statements.append((SQL_INSERT_ADOPTED, [name, created])) + entries[name] = _Entry(name, created) + self.counters["adopted"] += 1 + # -wal/-shm/-journal files whose database is gone would be replayed + # into the next database created under that name + files = set(os.listdir(directory)) + for filename in files: + for suffix in SIDECAR_SUFFIXES: + if ( + filename.endswith(SUFFIX + suffix) + and not filename.startswith(".") + and filename[: -len(suffix)] not in files + ): + try: + os.unlink(os.path.join(directory, filename)) + self.counters["stray_files_removed"] += 1 + except OSError: + pass + if statements: + self._registry_write(statements, flush_dirty=False) + with self._lock: + self._entries.update(entries) + return list(entries.values()) + + def _same_file(self, a, b): + try: + return os.path.samefile(self._path(a), self._path(b)) + except OSError: + return False + + # ------------------------------------------------------------------ + # registry + # ------------------------------------------------------------------ + def _registry_connection(self): + # Short-lived, plain connection: the registry is never held open + conn = sqlite3.connect( + os.path.join(self.directory, REGISTRY_FILENAME), + isolation_level=None, + check_same_thread=False, + ) + if not self._registry_initialized: + conn.execute(REGISTRY_SCHEMA) + self._registry_initialized = True + return conn + + def _registry_write(self, statements, flush_dirty=True): + """Run statements in one transaction, plus any pending last_used + updates. Blocking: call from a worker thread (or inline when there + are no threads).""" + with self._registry_lock: + dirty = self._take_dirty() if flush_dirty else [] + if not statements and not dirty: + return + conn = self._registry_connection() + try: + conn.execute("BEGIN IMMEDIATE") + for sql, params in statements: + conn.execute(sql, params) + if dirty: + conn.executemany( + "update scratch_databases set last_used = ? where name = ?", + [(entry.last_used, entry.name) for entry in dirty], + ) + conn.execute("COMMIT") + except BaseException: + if conn.in_transaction: + conn.execute("ROLLBACK") + self._restore_dirty(dirty) + raise + finally: + conn.close() + self.counters["registry_transactions"] += 1 + if dirty: + self.counters["last_used_flushes"] += 1 + + def _take_dirty(self): + with self._lock: + dirty = [e for e in self._dirty if e.last_used is not None] + for entry in self._dirty: + entry.dirty = False + self._dirty.clear() + self._last_flush = time.monotonic() + return dirty + + def _restore_dirty(self, dirty): + with self._lock: + for entry in dirty: + entry.dirty = True + self._dirty.add(entry) + + def flush(self): + """Write in-memory last_used times to the registry (blocking).""" + if self.directory is None: + return + self._registry_write([]) + + def _touch(self, db): + # On every read and write: a clock read and two attribute stores. + # The registry is written at most once per LAST_USED_FLUSH_INTERVAL_S + entry = db._scratch_entry + entry.last_used = time.time() + if not entry.dirty: + with self._lock: + entry.dirty = True + self._dirty.add(entry) + if ( + not self._flush_scheduled + and time.monotonic() - self._last_flush >= LAST_USED_FLUSH_INTERVAL_S + ): + self._schedule_flush() + + def _schedule_flush(self): + with self._lock: + if self._flush_scheduled or self._closed: + return + self._flush_scheduled = True + executor = self.ds.executor + if executor is None: + # No threads (Pyodide): one small transaction, inline + self._flush_quietly() + return + try: + executor.submit(self._flush_quietly) + except RuntimeError: + # Executor already shut down: close() flushes + self._flush_scheduled = False + + def _flush_quietly(self): + try: + self.flush() + except Exception: + logger.exception("Could not record scratch database last_used times") + finally: + self._last_flush = time.monotonic() + self._flush_scheduled = False + + # ------------------------------------------------------------------ + # helpers + # ------------------------------------------------------------------ + def _path(self, name): + return os.path.join(self.directory, name + SUFFIX) + + def _unlink_files(self, name, main=True): + path = self._path(name) + for suffix in SIDECAR_SUFFIXES + (("",) if main else ()): + try: + os.unlink(path + suffix) + except FileNotFoundError: + pass + + async def _run_blocking(self, fn, *args): + """Blocking file and registry work: on a worker thread - not + ds.executor, whose threads may be the very ones a close() waits for + - or inline when there are no threads (num_sql_threads=0).""" + if self.ds.executor is None: + return fn(*args) + return await asyncio.to_thread(fn, *args) + + def _check_open(self): + if self._closed or self.ds._closed: + raise DatasetteClosedError("Datasette has been closed") + + @staticmethod + def validate_name(name): + if not isinstance(name, str) or not NAME_RE.fullmatch(name): + raise ValueError( + f"Invalid scratch database name {name!r}: use 1-64 letters, digits, " + "'_' or '-', starting with a letter or digit" + ) + + def _check_available_locked(self, name, ignore=None): + if name in self.ds.databases and name != ignore: + raise ScratchDatabaseExists(f"A database called {name!r} already exists") + lowered = name.lower() + for other in self._entries: + # Case-insensitive: Foo.db and foo.db are one file on macOS and + # Windows + if other.lower() == lowered and other != ignore: + raise ScratchDatabaseExists( + f"A scratch database called {other!r} already exists" + ) + + def _generate_name_locked(self): + while True: + name = f"scratch_{secrets.token_hex(4)}" + try: + self._check_available_locked(name) + except ScratchDatabaseExists: + continue + return name + + def _attach(self, entry): + db = ScratchDatabase(self.ds, self._path(entry.name), entry=entry, manager=self) + entry.db = db + if entry.name in self.ds.databases: + logger.warning( + "Scratch database %r not attached: a database with that name already exists", + entry.name, + ) + return db + self.ds.add_database(db, name=entry.name, schema_watch="owned") + return db + + def _is_attached(self, entry): + return entry.db is not None and self.ds.databases.get(entry.name) is entry.db + + def _get_entry_locked(self, name): + entry = self._entries.get(name) + if entry is None or entry.state != "ready": + raise ScratchDatabaseNotFound(f"No scratch database called {name!r}") + return entry + + # ------------------------------------------------------------------ + # create + # ------------------------------------------------------------------ + async def create(self, name=None, *, actor=None, metadata=None): + self._check_open() + if name is not None: + self.validate_name(name) + owner = _owner_id(actor) + metadata = dict(metadata or {}) + metadata_json = json.dumps(metadata) # fail early if not JSON + if self.directory is None: + await self._run_blocking(self._ensure_directory) + with self._lock: + if name is None: + name = self._generate_name_locked() + else: + self._check_available_locked(name) + # Reserves the name until the create finishes or fails + entry = _Entry(name, time.time(), owner, metadata, state="creating") + self._entries[name] = entry + try: + await self._run_blocking(self._create_files, entry, metadata_json) + except BaseException: + with self._lock: + self._entries.pop(name, None) + raise + with self._lock: + # add_database() may have taken the name while we were creating + # the file - it does not know about scratch reservations + taken = name in self.ds.databases + if not taken: + entry.state = "ready" + entry.last_used = entry.created + if taken: + await self._run_blocking(self._delete_files, entry) + with self._lock: + self._entries.pop(name, None) + raise ScratchDatabaseExists(f"A database called {name!r} already exists") + db = self._attach(entry) + # Read-your-writes for the catalog: catalog_databases has a row for + # the new database when this returns + await db._after_write() + return db + + def _create_files(self, entry, metadata_json): + name = entry.name + path = self._path(name) + # Sidecars left by a crash would be replayed into the new file + self._unlink_files(name, main=False) + self._registry_write( + [ + ( + SQL_INSERT_CREATING, + [name, entry.created, entry.created, entry.owner, metadata_json], + ) + ] + ) + created_file = False + try: + # O_EXCL: never adopt (and later delete) a file that is + # already there + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + os.close(fd) + created_file = True + conn = sqlite3.connect(path) + try: + if self.ds.nolock: + # nolock=1 readers cannot open WAL databases + conn.execute("VACUUM") + else: + # Readers never wait for writers (copying data in while + # the database is being browsed) + conn.execute("PRAGMA journal_mode=WAL").fetchone() + finally: + conn.close() + _fsync_dir(self.directory) + self._registry_write([(SQL_MARK_READY, [name])]) + except BaseException: + if created_file: + self._unlink_files(name) + try: + self._registry_write( + [(SQL_DELETE, [name])], + flush_dirty=False, + ) + except Exception: + # Left as 'creating': rolled back at the next startup + logger.exception("Could not roll back scratch database %r", name) + raise + + # ------------------------------------------------------------------ + # delete + # ------------------------------------------------------------------ + async def delete(self, name): + self._check_open() + with self._lock: + entry = self._get_entry_locked(name) + entry.state = "deleting" + try: + await self._run_blocking( + self._registry_write, + [(SQL_MARK_DELETING, [name])], + ) + except BaseException: + with self._lock: + entry.state = "ready" + raise + # From here on the delete always completes: if this process dies, + # the 'deleting' row makes the next startup finish it + await self._detach_and_close(entry, "was deleted") + await self._run_blocking(self._delete_files, entry) + await self._delete_catalog(name) + with self._lock: + if self._entries.get(name) is entry: + del self._entries[name] + + async def _detach_and_close(self, entry, reason): + db = entry.db + db._scratch_gone = reason + state = db._watch_state + if state is not None: + fut = state.scan_future + if fut is not None and not fut.done(): + try: + same_loop = fut.get_loop() is asyncio.get_running_loop() + except RuntimeError: + same_loop = False + if same_loop: + # A catalog scan in flight: let it finish, its rows are + # deleted below + await asyncio.wait([fut]) + if self.ds.databases.get(entry.name) is db: + self.ds._detach_database(entry.name) + await self._run_blocking(self._close_and_wait, db) + + @staticmethod + def _close_and_wait(db): + """Close db and wait until nothing has its files open: close() waits + for queued and running reads and writes, but gives up on a write + thread after 10s; the files must not be removed while it runs.""" + db.close() + with db._write_thread_lock: + threads = [db._write_thread, *db._retiring_write_threads] + for thread in threads: + if thread is not None and thread is not threading.current_thread(): + thread.join() + # The SchemaWatcher's short-lived scan connections + db._wait_for_untracked_connections() + + def _delete_files(self, entry): + self._unlink_files(entry.name) + _fsync_dir(self.directory) + self._registry_write([(SQL_DELETE, [entry.name])]) + + async def _delete_catalog(self, name): + if self.ds.internal_db_created: + # unregister() already spawned this as a task; awaiting it here + # means the rows are gone when delete returns (it is idempotent) + await self.ds._schema_watcher._delete_catalog([name]) + + # ------------------------------------------------------------------ + # rename + # ------------------------------------------------------------------ + async def rename(self, name, new_name): + self._check_open() + self.validate_name(new_name) + with self._lock: + entry = self._get_entry_locked(name) + if new_name == name: + return entry.db + # ignore=name allows a case-only rename (foo -> Foo) + self._check_available_locked(new_name, ignore=name) + entry.state = "renaming" + # Reserve the new name + target = _Entry( + new_name, entry.created, entry.owner, entry.metadata, "creating" + ) + target.last_used = entry.last_used + self._entries[new_name] = target + try: + await self._run_blocking( + self._registry_write, + [(SQL_MARK_RENAMING, [new_name, name])], + ) + except BaseException: + with self._lock: + entry.state = "ready" + if self._entries.get(new_name) is target: + del self._entries[new_name] + raise + await self._detach_and_close(entry, f"was renamed to {new_name!r}") + try: + await self._run_blocking(self._rename_files, name, new_name) + except BaseException: + # Nothing was renamed: re-attach under the old name + with self._lock: + if self._entries.get(new_name) is target: + del self._entries[new_name] + entry.state = "ready" + await self._run_blocking( + self._registry_write, + [(SQL_CANCEL_RENAME, [name])], + ) + self._attach(entry) + raise + await self._delete_catalog(name) + with self._lock: + if self._entries.get(name) is entry: + del self._entries[name] + target.state = "ready" + self._entries[new_name] = target + db = self._attach(target) + await db._after_write() + return db + + def _rename_files(self, name, new_name): + old, new = self._path(name), self._path(new_name) + # Datasette's last connection to close is often a read-only pooled + # one, which cannot checkpoint: fold the -wal back into the main + # file so the rename moves a single, complete file + conn = sqlite3.connect(old) + try: + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchall() + finally: + conn.close() + for suffix in ("-wal", "-journal"): + try: + if os.path.getsize(old + suffix): + # After a clean close there is none: another process has + # this database open, or it needs crash recovery + raise ScratchDatabaseError( + f"Cannot rename scratch database {name!r}: {old + suffix} " + "is not empty (is another process using it?)" + ) + except FileNotFoundError: + pass + self._unlink_files(name, main=False) + self._unlink_files(new_name, main=False) + if name.lower() == new_name.lower() and name != new_name: + # Case-only change: link() would fail on case-insensitive file + # systems + os.rename(old, new) + else: + try: + # link + unlink never replaces an existing file (rename does) + os.link(old, new) + except FileExistsError: + raise ScratchDatabaseExists(f"{new} already exists") + except OSError: + # No hard links on this file system + if os.path.exists(new): + raise ScratchDatabaseExists(f"{new} already exists") + os.rename(old, new) + else: + os.unlink(old) + _fsync_dir(self.directory) + self._registry_write([(SQL_FINISH_RENAME, [new_name, name])]) + + # ------------------------------------------------------------------ + # list + # ------------------------------------------------------------------ + async def list(self): + with self._lock: + entries = sorted( + (e for e in self._entries.values() if e.state == "ready"), + key=lambda e: e.name, + ) + if not entries: + return [] + sizes = await self._run_blocking(self._sizes, [e.name for e in entries]) + return [ + ScratchDatabaseInfo( + name=entry.name, + path=self._path(entry.name), + created=entry.created, + last_used=entry.last_used, + size=size, + owner=entry.owner, + metadata=dict(entry.metadata), + attached=self._is_attached(entry), + ) + for entry, size in zip(entries, sizes) + ] + + def _sizes(self, names): + sizes = [] + for name in names: + size = 0 + for suffix in ("", "-wal"): + try: + size += os.stat(self._path(name) + suffix).st_size + except OSError: + pass + sizes.append(size) + return sizes + + # ------------------------------------------------------------------ + # shutdown + # ------------------------------------------------------------------ + def close(self): + """Called by Datasette.close() after every database has been closed + and the executor shut down: records last_used, releases the + directory lock, removes a temporary directory.""" + if self._closed: + return + self._closed = True + with self._lock: + entries = list(self._entries.values()) + for entry in entries: + # Detached ones (name clash, remove_database()) are not closed + # by Datasette.close() + if entry.db is not None and not entry.db._closed: + try: + entry.db.close() + except Exception: + logger.exception("Closing scratch database %r failed", entry.name) + if self.directory is not None and self.persistent: + try: + self.flush() + except Exception: + logger.exception("Could not record scratch database last_used times") + for finalizer in self._finalizers: + finalizer() diff --git a/docs/cli-reference.rst b/docs/cli-reference.rst index f64fd289..187255cf 100644 --- a/docs/cli-reference.rst +++ b/docs/cli-reference.rst @@ -140,6 +140,8 @@ Use ``--internal PATH`` or the ``DATASETTE_INTERNAL`` environment variable to pe --ssl-certfile TEXT SSL certificate file --internal PATH Path to a persistent Datasette internal SQLite database + --scratch-dir DIRECTORY Directory for scratch databases, which then + survive a restart --help Show this message and exit. diff --git a/docs/internals.rst b/docs/internals.rst index 4312372a..7d8102e8 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -1395,12 +1395,163 @@ The ``name`` and ``route`` parameters are optional and work the same way as they This removes a database that has been previously added. ``name=`` is the unique name of that database. +.. _datasette_scratch_databases: + +Scratch databases +----------------- + +A *scratch database* is a SQLite file that a plugin (or a user, through a plugin) creates, copies data into, changes, uses for a while and eventually throws away. Scratch databases are served like any other database - ``/name``, the JSON API, the write API - and they survive a restart of Datasette. + +They live in the **scratch directory**: + +- ``datasette --scratch-dir path/to/scratch`` (or the ``DATASETTE_SCRATCH_DIR`` environment variable, or ``Datasette(scratch_dir=...)``). The directory is created if it does not exist. +- In :ref:`config_dir`, a ``scratch/`` directory inside the configuration directory is used if it exists. +- Otherwise scratch databases are **temporary**: they go in a temporary directory that is created the first time a plugin creates one and deleted when Datasette shuts down, the same way the internal database is temporary unless ``--internal`` is used. ``datasette.scratch_dir`` is ``None`` until then. Plugins can use the same API either way. + +Use a directory just for scratch databases. Only one Datasette instance can use a scratch directory at a time: a second one fails to start with an error. + +The directory contains one ``.db`` file per scratch database, plus a registry (``.datasette-scratch.db``) that records each database's owner, metadata, creation time and when it was last used. Scratch databases are created in WAL mode (rollback journal mode with ``--nolock``). + +**Startup.** Existing scratch databases are attached when Datasette starts without opening any of them: Datasette reads the registry and lists the directory. Like any database, a scratch database gets connections and a write thread only while it is being used, and they are closed again after :ref:`setting_connection_idle_timeout_ms`. With a persistent internal database (``--internal``) and a clean shutdown, the table catalog for scratch databases is restored at startup without opening them either. With the default temporary internal database each scratch database is opened briefly, once, at startup to read its schema - as are, after a crash, the ones that were written to since their schema was last read. + +**The catalog** of a scratch database's tables is kept current by its own writes (schema watch mode ``"owned"``): a table created through any of the write methods is in the catalog before the write returns. If another process writes to a scratch database file directly, Datasette will not notice schema changes. + +**Copying files by hand.** The directory is the source of truth. A ``.db`` file copied into the scratch directory while Datasette is stopped is adopted as a scratch database at the next startup, with no owner or metadata; its name must follow the naming rules below. A scratch database file deleted by hand is dropped from the registry at the next startup. Do not add, remove or rename files while Datasette is running. Copying a scratch database's ``.db`` file out of the directory gives you an ordinary SQLite database: nothing Datasette-specific is stored inside it. + +**Crashes.** Creating, deleting and renaming are recorded in the registry before the files are touched, so if Datasette is killed part-way through, the next startup either finishes the operation (delete, rename) or undoes it (create). It also removes ``-wal``, ``-shm`` and ``-journal`` files whose database file is missing, since SQLite would otherwise apply an old ``-wal`` file to a new database created later under the same name. + +**Names** are 1 to 64 characters long, use only letters, digits, ``_`` and ``-``, and start with a letter or a digit. They are unique among all attached databases, and case-insensitively unique among scratch databases (``Foo`` and ``foo`` would be the same file on macOS and Windows). + +**Permissions.** Scratch databases are subject to the same permission checks as every other database. Recording an owner does not restrict who can see a scratch database: plugins that create scratch databases for one user should also restrict access to them, for example with the :ref:`plugin_hook_permission_resources_sql` plugin hook. + +Scratch databases are not attached to the ``_memory`` database by ``--crossdb``, because they can be deleted at any time. To combine data from several databases in one query, ``ATTACH`` them inside your own write function, as in the recipe below. + +.. _datasette_create_scratch_database: + +await .create_scratch_database(name=None, \*, actor=None, metadata=None) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +``name`` - string, optional + The name of the new database. Defaults to a random name like ``scratch_3fa85f64``. + +``actor`` - dictionary or string, optional + The actor the database is created for. Its ``"id"`` (or the string itself) is recorded as the owner. + +``metadata`` - dictionary, optional + Anything JSON-serializable that the plugin wants to remember about this database, returned by :ref:`datasette_list_scratch_databases`. + +Creates an empty database file in the scratch directory, attaches it and returns its ``ScratchDatabase`` object, a subclass of :ref:`internals_database` with extra ``.owner``, ``.created``, ``.last_used`` and ``.scratch_metadata`` attributes and ``.is_scratch`` set to ``True``. Raises ``ValueError`` for an invalid name and ``datasette.scratch.ScratchDatabaseExists`` if a database with that name already exists. + +.. code-block:: python + + db = await datasette.create_scratch_database( + actor=request.actor, metadata={"source": "my-plugin"} + ) + await db.execute_write( + "create table notes (id integer primary key, body text)" + ) + +.. _datasette_delete_scratch_database: + +await .delete_scratch_database(name) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Detaches the scratch database and deletes its files (including ``-wal`` and ``-shm``), its catalog entries and its registry entry. Raises ``datasette.scratch.ScratchDatabaseNotFound`` if there is no scratch database with that name. + +Reads and writes that are already running when ``delete_scratch_database()`` is called are allowed to finish: the delete waits for them. Reads and writes that are queued but have not started, including ``block=False`` writes, and any call made afterwards using the old ``Database`` object, raise ``datasette.scratch.ScratchDatabaseDeleted`` (a subclass of ``DatasetteClosedError``). + +.. _datasette_rename_scratch_database: + +await .rename_scratch_database(name, new_name) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Renames a scratch database, keeping its data, owner and metadata, and returns the new ``ScratchDatabase`` object. This detaches the database and waits for running queries in the same way as deleting it, so calls using the old object raise ``ScratchDatabaseDeleted``. Raises ``ScratchDatabaseExists`` if ``new_name`` is taken. + +.. _datasette_list_scratch_databases: + +await .list_scratch_databases() +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Returns a list of ``datasette.scratch.ScratchDatabaseInfo`` objects, sorted by name, with these attributes: + +- ``name`` and ``path`` +- ``created`` and ``last_used`` - Unix timestamps. ``last_used`` is the time of the most recent read or write. It is kept in memory and written to the registry about once a minute and when Datasette shuts down, so after a crash it can be up to a minute old. +- ``size`` - in bytes, including the ``-wal`` file +- ``owner`` - the actor ID passed to ``create_scratch_database()``, or ``None`` +- ``metadata`` - the dictionary passed to ``create_scratch_database()`` +- ``attached`` - ``False`` if the database could not be attached because another database already uses its name + +A plugin that cleans up scratch databases nobody has used for a week could do this: + +.. code-block:: python + + import time + + for info in await datasette.list_scratch_databases(): + if info.last_used < time.time() - 7 * 24 * 60 * 60: + await datasette.delete_scratch_database(info.name) + +.. _datasette_scratch_copy_recipe: + +Copying data into a scratch database +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +The fastest way to copy a table into a scratch database is to ``ATTACH`` the source database read-only inside one write function and copy with ``INSERT ... SELECT``. The rows never pass through Python, and 10,000 rows take around 10ms. + +.. code-block:: python + + from pathlib import Path + from datasette.utils import escape_sqlite + + + async def copy_table( + datasette, source_name, table, scratch + ): + source = datasette.get_database(source_name) + source_uri = Path(source.path).resolve().as_uri() + ( + "?mode=ro" if source.is_mutable else "?immutable=1" + ) + + def copy_in(conn): + conn.execute( + "ATTACH DATABASE ? AS source", [source_uri] + ) + try: + with conn: + conn.execute("BEGIN IMMEDIATE") + create_sql = conn.execute( + "select sql from source.sqlite_master " + "where type = 'table' and name = ?", + [table], + ).fetchone()[0] + conn.execute(create_sql) + conn.execute( + "insert into main.{t} select * from source.{t}".format( + t=escape_sqlite(table) + ) + ) + finally: + conn.execute("DETACH DATABASE source") + + await scratch.execute_write_fn( + copy_in, transaction=False + ) + +Notes on this recipe: + +- It works for mutable source databases (``?mode=ro``) and immutable ones (``?immutable=1``). It needs a database file: plain in-memory databases cannot be attached this way. +- Pass ``transaction=False`` and manage the transaction yourself, as here. SQLite cannot ``DETACH`` a database that was read inside a transaction that is still open, so with the default ``transaction=True`` the ``DETACH`` fails and the attached database stays on the write connection. +- ``DETACH`` in a ``finally`` block: an ``ATTACH`` lasts as long as the connection, and the write connection is shared by every later write to that database (see :ref:`database_connections`). +- Reusing the original ``CREATE TABLE`` statement keeps the primary key, column types and constraints. ``CREATE TABLE ... AS SELECT`` would lose them. +- With a mutable source the ``INSERT ... SELECT`` reads one consistent snapshot of it. If the source uses a rollback journal rather than WAL, writes to it wait until the copy has finished. +- The table is in the catalog, and visible in Datasette, when ``execute_write_fn()`` returns. + .. _datasette_close: .close() -------- -Release all resources held by this ``Datasette`` instance. This calls :ref:`database_close` on every attached database (including the internal database), shuts down the thread pool executor used to run SQL queries, and unlinks the temporary file used to back the internal database if one was created. +Release all resources held by this ``Datasette`` instance. This calls :ref:`database_close` on every attached database (including the internal database), shuts down the thread pool executor used to run SQL queries, and unlinks the temporary file used to back the internal database if one was created. It also records when each :ref:`scratch database ` was last used and deletes the scratch directory if it was a temporary one. ``close()`` is synchronous, idempotent and one-way: after a call to ``close()`` any attempt to use the Datasette instance to execute SQL will raise a ``datasette.database.DatasetteClosedError`` exception. A closed ``Datasette`` cannot be reopened — callers that need a fresh instance should construct a new one. diff --git a/docs/settings.rst b/docs/settings.rst index 175bc910..bf07d747 100644 --- a/docs/settings.rst +++ b/docs/settings.rst @@ -56,6 +56,7 @@ The files that can be included in this directory are as follows. All are optiona * ``templates/`` - a directory containing :ref:`customization_custom_templates` * ``plugins/`` - a directory containing plugins, see :ref:`writing_plugins_one_off` * ``static/`` - a directory containing static files - these will be served from ``/static/filename.txt``, see :ref:`customization_static_files` +* ``scratch/`` - a directory for :ref:`scratch databases `, used instead of a temporary directory if it exists Settings -------- diff --git a/tests/test_scratch.py b/tests/test_scratch.py new file mode 100644 index 00000000..36b396c5 --- /dev/null +++ b/tests/test_scratch.py @@ -0,0 +1,952 @@ +""" +Scratch databases: file-based databases that plugins create, fill, change and +delete through the Datasette API, and that survive a restart. +""" + +import asyncio +import json +import os +import sqlite3 +import subprocess +import sys +import threading +import time +from pathlib import Path + +import pytest + +from datasette import schema_watcher as schema_watcher_module +from datasette import scratch as scratch_module +from datasette.app import Datasette +from datasette.database import Database, DatasetteClosedError +from datasette.scratch import ( + REGISTRY_FILENAME, + ScratchDatabase, + ScratchDatabaseDeleted, + ScratchDatabaseExists, + ScratchDatabaseNotFound, +) +from datasette.utils import StartupError + + +def _registry_rows(scratch_dir): + conn = sqlite3.connect(os.path.join(scratch_dir, REGISTRY_FILENAME)) + try: + conn.row_factory = sqlite3.Row + return { + row["name"]: dict(row) + for row in conn.execute("select * from scratch_databases") + } + finally: + conn.close() + + +async def _catalog_tables(ds, name): + rows = await ds.get_internal_database().execute( + "select table_name from catalog_tables where database_name = ? order by table_name", + [name], + ) + return [r[0] for r in rows.rows] + + +async def _catalog_has_database(ds, name): + rows = await ds.get_internal_database().execute( + "select 1 from catalog_databases where database_name = ?", [name] + ) + return bool(rows.rows) + + +def _files(scratch_dir, name): + return sorted(f for f in os.listdir(scratch_dir) if f.startswith(name + ".db")) + + +def _write_threads(name): + return [ + t + for t in threading.enumerate() + if t.name == f"_execute_writes for database {name}" + ] + + +def _make_source(path, rows=10000): + conn = sqlite3.connect(path) + conn.execute( + "create table items (id integer primary key, name text not null, score real)" + ) + conn.executemany( + "insert into items (name, score) values (?, ?)", + [(f"item {i}", i * 0.5) for i in range(rows)], + ) + conn.commit() + conn.close() + return path + + +@pytest.fixture +def scratch_dir(tmp_path): + return str(tmp_path / "scratch") + + +# --------------------------------------------------------------------------- +# create / list / names +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_list_and_catalog(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir) + await ds.invoke_startup() + db = await ds.create_scratch_database( + "work", actor={"id": "alice"}, metadata={"source": "test"} + ) + assert isinstance(db, ScratchDatabase) + assert db.is_scratch + assert ds.get_database("work") is db + assert db.owner == "alice" + assert db.scratch_metadata == {"source": "test"} + assert ds._schema_watcher.states["work"].mode == "owned" + assert os.path.exists(os.path.join(scratch_dir, "work.db")) + # The catalog knows about it before create returns + assert await _catalog_has_database(ds, "work") + # ... and about its tables before each write returns + await db.execute_write("create table t (id integer primary key, v text)") + assert await _catalog_tables(ds, "work") == ["t"] + await db.execute_write("insert into t (v) values ('a')") + # Created in WAL mode + assert (await db.execute("pragma journal_mode")).single_value() == "wal" + infos = await ds.list_scratch_databases() + assert len(infos) == 1 + info = infos[0] + assert info.name == "work" + assert info.path == os.path.join(ds.scratch_dir, "work.db") + assert info.owner == "alice" + assert info.metadata == {"source": "test"} + assert info.attached + assert info.size > 0 + assert info.created <= info.last_used <= time.time() + assert _registry_rows(scratch_dir)["work"]["state"] == "ready" + # Browsable like any other database + response = await ds.client.get("/work/t.json?_shape=array") + assert response.status_code == 200 + assert response.json() == [{"id": 1, "v": "a"}] + ds.close() + + +@pytest.mark.asyncio +async def test_create_generated_name_and_string_actor(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir) + await ds.invoke_startup() + db1 = await ds.create_scratch_database(actor="bob") + db2 = await ds.create_scratch_database() + assert db1.name != db2.name + assert db1.name.startswith("scratch_") + assert db1.owner == "bob" + assert db2.owner is None + assert [i.name for i in await ds.list_scratch_databases()] == sorted( + [db1.name, db2.name] + ) + ds.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "name", + [ + "", + "../escape", + "a/b", + "a\\b", + ".hidden", + "_internal", + "-x", + "a.b", + "a b", + "café", + "x" * 65, + "work\n", + 123, + ], +) +async def test_invalid_names(scratch_dir, name): + ds = Datasette(scratch_dir=scratch_dir) + with pytest.raises(ValueError): + await ds.create_scratch_database(name) + assert [f for f in os.listdir(scratch_dir) if not f.startswith(".")] == [] + ds.close() + + +@pytest.mark.asyncio +async def test_duplicate_names(scratch_dir, tmp_path): + other = str(tmp_path / "existing.db") + sqlite3.connect(other).execute("vacuum") + ds = Datasette([other], scratch_dir=scratch_dir) + await ds.invoke_startup() + await ds.create_scratch_database("Work") + with pytest.raises(ScratchDatabaseExists): + await ds.create_scratch_database("Work") + # One file on case-insensitive file systems + with pytest.raises(ScratchDatabaseExists): + await ds.create_scratch_database("work") + # Clashes with a database that is not a scratch database + with pytest.raises(ScratchDatabaseExists): + await ds.create_scratch_database("existing") + with pytest.raises(TypeError): + await ds.create_scratch_database("bad_metadata", metadata={"x": object()}) + assert "bad_metadata" not in ds.databases + assert sorted(i.name for i in await ds.list_scratch_databases()) == ["Work"] + ds.close() + + +# --------------------------------------------------------------------------- +# restart +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_restart_attaches_without_opening(scratch_dir, monkeypatch): + ds = Datasette(scratch_dir=scratch_dir) + await ds.invoke_startup() + for i in range(5): + db = await ds.create_scratch_database( + f"s{i}", actor={"id": "alice"}, metadata={"i": i} + ) + await db.execute_write("create table t (id integer primary key)") + await db.execute_write("insert into t default values") + ds.close() + + connects = [] + original_connect = Database.connect + + def counting_connect(self, *args, **kwargs): + connects.append(self.name) + return original_connect(self, *args, **kwargs) + + monkeypatch.setattr(Database, "connect", counting_connect) + ds = Datasette(scratch_dir=scratch_dir) + # Registered, nothing opened, no threads + assert sorted(n for n in ds.databases if n.startswith("s")) == [ + f"s{i}" for i in range(5) + ] + assert connects == [] + assert ds._read_pool_or_none is None + assert not any(_write_threads(f"s{i}") for i in range(5)) + await ds.invoke_startup() + # A temporary internal database has no catalog yet: each scratch + # database was scanned once, on a short-lived connection + assert await _catalog_tables(ds, "s3") == ["t"] + for i in range(5): + db = ds.get_database(f"s{i}") + assert db._all_connections == [] + assert db._read_pool_state is None + assert db._write_thread is None + infos = {i.name: i for i in await ds.list_scratch_databases()} + assert infos["s2"].owner == "alice" + assert infos["s2"].metadata == {"i": 2} + assert ( + await ds.get_database("s4").execute("select count(*) from t") + ).single_value() == 1 + ds.close() + + +@pytest.mark.asyncio +async def test_restart_with_persistent_internal_scans_nothing( + scratch_dir, tmp_path, monkeypatch +): + # Fingerprints stored within the racy window would be rescanned + monkeypatch.setattr(schema_watcher_module, "RACY_WINDOW_NS", 0) + internal = str(tmp_path / "internal.db") + ds = Datasette(scratch_dir=scratch_dir, internal=internal) + await ds.invoke_startup() + for i in range(3): + db = await ds.create_scratch_database(f"s{i}") + await db.execute_write("create table t (id integer primary key)") + await db.execute_write("insert into t default values") + ds.close() + ds = Datasette(scratch_dir=scratch_dir, internal=internal) + await ds.invoke_startup() + watcher = ds._schema_watcher + assert watcher.counters["restored_from_persisted"] == 3 + # Data written, and the -wal checkpointed by close(), after the last + # catalog scan: still reused, because close() recorded the final + # fingerprint of each owned database + assert [watcher.states[f"s{i}"].stats["scans"] for i in range(3)] == [0, 0, 0] + assert await _catalog_tables(ds, "s1") == ["t"] + ds.close() + + +@pytest.mark.asyncio +async def test_directory_lock(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir) + with pytest.raises(StartupError, match="in use by another Datasette instance"): + Datasette(scratch_dir=scratch_dir) + ds.close() + Datasette(scratch_dir=scratch_dir).close() + + +@pytest.mark.asyncio +async def test_temporary_scratch_dir_when_not_configured(): + ds = Datasette() + await ds.invoke_startup() + assert ds.scratch_dir is None + assert not ds._scratch.persistent + db = await ds.create_scratch_database("tmp") + await db.execute_write("create table t (id)") + directory = ds.scratch_dir + assert os.path.isdir(directory) + ds.close() + assert not os.path.exists(directory) + + +@pytest.mark.asyncio +async def test_config_dir_scratch_subdirectory(tmp_path): + config_dir = tmp_path / "config" + (config_dir / "scratch").mkdir(parents=True) + ds = Datasette(config_dir=config_dir) + assert ds.scratch_dir == str((config_dir / "scratch").resolve()) + await ds.create_scratch_database("in_config") + ds.close() + assert (config_dir / "scratch" / "in_config.db").exists() + # Not picked up as a config_dir database (only *.db at the top level) + ds = Datasette(config_dir=config_dir) + assert ds.get_database("in_config").is_scratch + ds.close() + + +def test_cli_scratch_dir(tmp_path): + from click.testing import CliRunner + + from datasette.cli import cli + + scratch_dir = str(tmp_path / "scratch") + + async def setup(): + ds = Datasette(scratch_dir=scratch_dir) + db = await ds.create_scratch_database("from_cli") + await db.execute_write("create table t (id integer primary key)") + await db.execute_write("insert into t default values") + ds.close() + + asyncio.run(setup()) + result = CliRunner().invoke( + cli, + [ + "serve", + "--memory", + "--scratch-dir", + scratch_dir, + "--get", + "/from_cli/t.json?_shape=array", + ], + ) + assert result.exit_code == 0, result.output + assert json.loads(result.output) == [{"id": 1}] + + +@pytest.mark.asyncio +async def test_name_clash_at_startup(scratch_dir, tmp_path): + ds = Datasette(scratch_dir=scratch_dir) + await ds.create_scratch_database("clash") + ds.close() + other = str(tmp_path / "clash.db") + sqlite3.connect(other).execute("vacuum") + ds = Datasette([other], scratch_dir=scratch_dir) + assert not ds.get_database("clash").is_scratch + [info] = await ds.list_scratch_databases() + assert info.name == "clash" + assert not info.attached + ds.close() + + +# --------------------------------------------------------------------------- +# last_used +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_last_used_in_memory_then_flushed_on_close(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir) + db = await ds.create_scratch_database("used") + await db.execute_write("create table t (id)") + transactions = ds._scratch.counters["registry_transactions"] + before = db.last_used + for _ in range(50): + await db.execute("select count(*) from t") + await db.execute_fn(lambda conn: conn.execute("select 1").fetchall()) + await db.execute_isolated_fn(lambda conn: None) + assert db.last_used > before + # Never a registry write per read + assert ds._scratch.counters["registry_transactions"] == transactions + in_memory = db.last_used + assert _registry_rows(scratch_dir)["used"]["last_used"] < in_memory + ds.close() + assert _registry_rows(scratch_dir)["used"]["last_used"] == in_memory + ds = Datasette(scratch_dir=scratch_dir) + [info] = await ds.list_scratch_databases() + assert info.last_used == in_memory + ds.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_sql_threads", [0, 2]) +async def test_last_used_periodic_flush(scratch_dir, monkeypatch, num_sql_threads): + monkeypatch.setattr(scratch_module, "LAST_USED_FLUSH_INTERVAL_S", 0.2) + ds = Datasette( + scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads} + ) + db = await ds.create_scratch_database("used") + await db.execute_write("create table t (id)") + flushes = ds._scratch.counters["last_used_flushes"] + deadline = time.monotonic() + 1.1 + reads = 0 + while time.monotonic() < deadline: + await db.execute("select count(*) from t") + reads += 1 + await asyncio.sleep(0.005) + await asyncio.sleep(0.1) + new_flushes = ds._scratch.counters["last_used_flushes"] - flushes + # Several flushes, far fewer than reads + assert 2 <= new_flushes <= 7, (new_flushes, reads) + assert reads > 50 + ds.close() + + +# --------------------------------------------------------------------------- +# delete +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_sql_threads", [0, 3]) +async def test_delete_removes_files_catalog_and_registry(scratch_dir, num_sql_threads): + ds = Datasette( + scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads} + ) + await ds.invoke_startup() + db = await ds.create_scratch_database("doomed") + await db.execute_write("create table t (id integer primary key, v)") + await db.execute_write_many( + "insert into t (v) values (?)", [(i,) for i in range(100)] + ) + assert (await db.execute("select count(*) from t")).single_value() == 100 + if num_sql_threads: + # WAL mode: the sidecars exist while connections are open + assert _files(scratch_dir, "doomed") == [ + "doomed.db", + "doomed.db-shm", + "doomed.db-wal", + ] + assert await _catalog_tables(ds, "doomed") == ["t"] + await ds.delete_scratch_database("doomed") + assert _files(scratch_dir, "doomed") == [] + assert "doomed" not in ds.databases + assert "doomed" not in _registry_rows(scratch_dir) + assert not await _catalog_has_database(ds, "doomed") + assert await _catalog_tables(ds, "doomed") == [] + assert await ds.list_scratch_databases() == [] + with pytest.raises(ScratchDatabaseDeleted, match="'doomed' was deleted"): + await db.execute("select 1") + with pytest.raises(ScratchDatabaseDeleted): + await db.execute_write("insert into t (v) values (1)") + # Still a DatasetteClosedError for code that handles that + with pytest.raises(DatasetteClosedError): + await db.execute_fn(lambda conn: 1) + with pytest.raises(ScratchDatabaseNotFound): + await ds.delete_scratch_database("doomed") + # The name can be used again, for a new empty database + again = await ds.create_scratch_database("doomed") + assert await again.table_names() == [] + ds.close() + + +@pytest.mark.asyncio +async def test_delete_with_reads_and_writes_in_flight(scratch_dir): + # One SQL thread: a slow read occupies it and more reads queue behind it + ds = Datasette(scratch_dir=scratch_dir, settings={"num_sql_threads": 1}) + await ds.invoke_startup() + db = await ds.create_scratch_database("busy") + await db.execute_write("create table t (id integer primary key, v)") + await db.execute_write_many( + "insert into t (v) values (?)", [(i,) for i in range(1000)] + ) + read_started = threading.Event() + write_started = threading.Event() + + def slow_read(conn): + read_started.set() + time.sleep(0.4) + return conn.execute("select count(*) from t").fetchone()[0] + + def slow_write(conn): + write_started.set() + time.sleep(0.4) + conn.execute("insert into t (v) values ('slow')") + return "slow write done" + + running_read = asyncio.ensure_future(db.execute_fn(slow_read)) + running_write = asyncio.ensure_future(db.execute_write_fn(slow_write)) + while not (read_started.is_set() and write_started.is_set()): + await asyncio.sleep(0.01) + queued_reads = [ + asyncio.ensure_future(db.execute("select count(*) from t")) for _ in range(3) + ] + queued_writes = [ + asyncio.ensure_future(db.execute_write("insert into t (v) values (1)")) + for _ in range(3) + ] + non_blocking = await db.execute_write("insert into t (v) values (2)", block=False) + assert non_blocking is not None + await asyncio.sleep(0.05) + await ds.delete_scratch_database("busy") + # Work that was already running finished normally + assert await running_read == 1000 + assert await running_write == "slow write done" + # Work that was queued got a clear error + for future in queued_reads + queued_writes: + with pytest.raises(ScratchDatabaseDeleted, match="'busy' was deleted"): + await future + assert _files(scratch_dir, "busy") == [] + ds.close() + + +@pytest.mark.asyncio +async def test_delete_waits_for_untracked_connections(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir) + await ds.invoke_startup() + db = await ds.create_scratch_database("watched") + await db.execute_write("create table t (id)") + # Simulate a SchemaWatcher scan holding a short-lived connection + watcher = ds._schema_watcher + conn = watcher._connect(db, prepare=False) + conn.execute("select count(*) from t").fetchall() + released = threading.Event() + + def release_later(): + time.sleep(0.3) + # Still there while the scan connection is open + assert os.path.exists(os.path.join(scratch_dir, "watched.db")) + released.set() + watcher._close(db, conn) + + thread = threading.Thread(target=release_later) + thread.start() + start = time.monotonic() + await ds.delete_scratch_database("watched") + assert time.monotonic() - start >= 0.25 + assert released.is_set() + thread.join() + assert _files(scratch_dir, "watched") == [] + # No new untracked connection once the database is closed + with pytest.raises(DatasetteClosedError): + watcher._connect(db, prepare=False) + ds.close() + + +@pytest.mark.asyncio +async def test_delete_stress_no_crash(scratch_dir): + """Delete while reads (with the time-limit progress handler installed) + and writes are running and queued, many times over.""" + ds = Datasette( + scratch_dir=scratch_dir, + settings={"num_sql_threads": 3, "sql_time_limit_ms": 2000}, + ) + await ds.invoke_startup() + for iteration in range(15): + db = await ds.create_scratch_database(f"stress{iteration}") + await db.execute_write("create table t (id integer primary key, v)") + await db.execute_write_many( + "insert into t (v) values (?)", [(i,) for i in range(2000)] + ) + calls = [] + for i in range(12): + calls.append( + db.execute( + "with recursive c(x) as (select 1 union all select x + 1 from c " + "where x < 20000) select count(*) from c, (select 1 from t limit 3)" + ) + ) + calls.append(db.execute_write("insert into t (v) values (?)", [i])) + futures = [asyncio.ensure_future(c) for c in calls] + await asyncio.sleep(0.002 * (iteration % 5)) + await ds.delete_scratch_database(f"stress{iteration}") + results = await asyncio.gather(*futures, return_exceptions=True) + for result in results: + if isinstance(result, BaseException): + assert isinstance(result, ScratchDatabaseDeleted), repr(result) + assert _files(scratch_dir, f"stress{iteration}") == [] + ds.close() + + +# --------------------------------------------------------------------------- +# rename +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +@pytest.mark.parametrize("num_sql_threads", [0, 3]) +async def test_rename(scratch_dir, num_sql_threads): + ds = Datasette( + scratch_dir=scratch_dir, settings={"num_sql_threads": num_sql_threads} + ) + await ds.invoke_startup() + db = await ds.create_scratch_database( + "before", actor={"id": "alice"}, metadata={"a": 1} + ) + await db.execute_write("create table t (id integer primary key, v)") + await db.execute_write("insert into t (v) values ('kept')") + other = await ds.create_scratch_database("other") + with pytest.raises(ScratchDatabaseExists): + await ds.rename_scratch_database("before", "other") + with pytest.raises(ScratchDatabaseNotFound): + await ds.rename_scratch_database("missing", "x") + with pytest.raises(ValueError): + await ds.rename_scratch_database("before", "../x") + renamed = await ds.rename_scratch_database("before", "after") + assert renamed.name == "after" + assert renamed.owner == "alice" + assert renamed.scratch_metadata == {"a": 1} + assert (await renamed.execute("select v from t")).single_value() == "kept" + assert _files(scratch_dir, "before") == [] + assert os.path.exists(os.path.join(scratch_dir, "after.db")) + assert "before" not in ds.databases + assert await _catalog_tables(ds, "before") == [] + assert await _catalog_tables(ds, "after") == ["t"] + with pytest.raises(ScratchDatabaseDeleted, match="renamed to 'after'"): + await db.execute("select 1") + rows = _registry_rows(scratch_dir) + assert set(rows) == {"after", "other"} + assert rows["after"]["owner"] == "alice" + # Case-only rename + renamed = await ds.rename_scratch_database("after", "After") + assert (await renamed.execute("select v from t")).single_value() == "kept" + assert sorted(i.name for i in await ds.list_scratch_databases()) == [ + "After", + "other", + ] + assert other.name == "other" + ds.close() + + +# --------------------------------------------------------------------------- +# copying data in +# --------------------------------------------------------------------------- + + +def copy_table_into(scratch_conn, source_db, table): + # The recipe from docs/internals.rst + from datasette.utils import escape_sqlite + + mode = "mode=ro" if source_db.is_mutable else "immutable=1" + source_uri = Path(source_db.path).resolve().as_uri() + "?" + mode + scratch_conn.execute("ATTACH DATABASE ? AS source", [source_uri]) + try: + with scratch_conn: + scratch_conn.execute("BEGIN IMMEDIATE") + create_sql = scratch_conn.execute( + "select sql from source.sqlite_master where type = 'table' and name = ?", + [table], + ).fetchone()[0] + scratch_conn.execute(create_sql) + scratch_conn.execute( + "insert into main.{t} select * from source.{t}".format( + t=escape_sqlite(table) + ) + ) + finally: + scratch_conn.execute("DETACH DATABASE source") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("immutable", [False, True]) +@pytest.mark.parametrize("num_sql_threads", [0, 3]) +async def test_copy_data_in_recipe(tmp_path, scratch_dir, immutable, num_sql_threads): + source_path = _make_source(str(tmp_path / "source.db")) + kwargs = {"immutables": [source_path]} if immutable else {"files": [source_path]} + ds = Datasette( + scratch_dir=scratch_dir, + settings={"num_sql_threads": num_sql_threads}, + **kwargs, + ) + await ds.invoke_startup() + source = ds.get_database("source") + assert source.is_mutable is not immutable + scratch = await ds.create_scratch_database("copy") + await scratch.execute_write_fn( + lambda conn: copy_table_into(conn, source, "items"), transaction=False + ) + assert (await scratch.execute("select count(*) from items")).single_value() == 10000 + # Same schema, primary key included + assert await scratch.primary_keys("items") == ["id"] + assert await _catalog_tables(ds, "copy") == ["items"] + # Modify the copy; the source is unchanged + await scratch.execute_write("delete from items where id % 2 = 0") + await scratch.execute_write("update items set score = score * 2") + assert (await scratch.execute("select count(*) from items")).single_value() == 5000 + assert (await source.execute("select count(*) from items")).single_value() == 10000 + # Nothing left attached to the write connection + attached = await scratch.execute_write_fn( + lambda conn: [r[1] for r in conn.execute("pragma database_list").fetchall()] + ) + assert attached == ["main"] + + # The alias is free again for the next copy on the same write connection + def attach_again(conn): + conn.execute( + "ATTACH DATABASE ? AS source", + [Path(source.path).resolve().as_uri() + "?mode=ro"], + ) + try: + return conn.execute("select count(*) from source.items").fetchone()[0] + finally: + conn.execute("DETACH DATABASE source") + + assert await scratch.execute_write_fn(attach_again, transaction=False) == 10000 + await ds.delete_scratch_database("copy") + ds.close() + + +@pytest.mark.asyncio +async def test_detach_inside_managed_transaction_fails(tmp_path, scratch_dir): + # Why the recipe uses transaction=False: SQLite refuses to DETACH a + # database that was read inside the still-open transaction, so the + # attachment would stay on the write connection + source_path = _make_source(str(tmp_path / "source.db"), rows=10) + ds = Datasette([source_path], scratch_dir=scratch_dir) + source = ds.get_database("source") + scratch = await ds.create_scratch_database("tx") + uri = Path(source.path).resolve().as_uri() + "?mode=ro" + + def attach_read_detach(conn): + conn.execute("ATTACH DATABASE ? AS source", [uri]) + conn.execute("select count(*) from source.items").fetchall() + conn.execute("DETACH DATABASE source") + + with pytest.raises(sqlite3.OperationalError, match="database source is locked"): + await scratch.execute_write_fn(attach_read_detach) + # ... and the attachment has leaked into the write connection: the next + # callback on it cannot use the alias + with pytest.raises(sqlite3.OperationalError, match="already in use"): + await scratch.execute_write_fn( + lambda conn: copy_table_into(conn, source, "items"), transaction=False + ) + ds.close() + + +# --------------------------------------------------------------------------- +# crash consistency +# --------------------------------------------------------------------------- + +CRASH_SCRIPT = r""" +import asyncio, os, sys +from datasette import scratch +from datasette.app import Datasette + +scratch_dir, op, crash_after = sys.argv[1], sys.argv[2], int(sys.argv[3]) +steps = [0] +original_write = scratch.ScratchDatabases._registry_write +original_fsync = scratch._fsync_dir + +def step(): + steps[0] += 1 + if steps[0] == crash_after: + os._exit(17) + +def registry_write(self, *args, **kwargs): + original_write(self, *args, **kwargs) + step() + +def fsync_dir(d): + original_fsync(d) + step() + +async def main(): + ds = Datasette(scratch_dir=scratch_dir) + await ds.invoke_startup() + if op != "create": + db = await ds.create_scratch_database("victim", actor={"id": "alice"}) + await db.execute_write("create table t (id integer primary key)") + await db.execute_write("insert into t default values") + scratch.ScratchDatabases._registry_write = registry_write + scratch._fsync_dir = fsync_dir + if op == "create": + await ds.create_scratch_database("victim", actor={"id": "alice"}) + elif op == "delete": + await ds.delete_scratch_database("victim") + elif op == "rename": + await ds.rename_scratch_database("victim", "renamed") + print("completed", steps[0]) + os._exit(0) + +asyncio.run(main()) +""" + + +@pytest.mark.parametrize("op", ["create", "delete", "rename"]) +@pytest.mark.parametrize("crash_after", [1, 2, 3, 4, 99]) +def test_crash_consistency(tmp_path, op, crash_after): + """Kill the process after the Nth registry transaction or directory fsync + of an operation; the next startup must see the old or the new state, + with metadata, and no stray files.""" + scratch_dir = str(tmp_path / "scratch") + result = subprocess.run( + [sys.executable, "-c", CRASH_SCRIPT, scratch_dir, op, str(crash_after)], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert result.returncode in (0, 17), result.stderr + ds = Datasette(scratch_dir=scratch_dir) + infos = {i.name: i for i in asyncio.run(ds.list_scratch_databases())} + files = sorted(f for f in os.listdir(scratch_dir) if not f.startswith(".")) + if op == "create" or op == "delete": + assert set(infos) in (set(), {"victim"}) + else: + assert set(infos) in ({"victim"}, {"renamed"}) + for name, info in infos.items(): + assert info.owner == "alice" + assert info.attached + # Only the listed databases and their own -wal/-shm (a process killed + # with connections open leaves those): no orphan file or sidecar + expected = {f"{name}.db" for name in infos} + assert {f for f in files if f.endswith(".db")} == expected + for f in files: + assert f.endswith(".db") or f.rsplit("-", 1)[0] in expected, files + rows = _registry_rows(scratch_dir) + assert set(rows) == set(infos) + assert all(row["state"] == "ready" for row in rows.values()) + for name in infos: + db = ds.get_database(name) + tables = asyncio.run(db.table_names()) + if op != "create": + assert tables == ["t"] + ds.close() + + +@pytest.mark.asyncio +async def test_hand_copied_and_hand_deleted_files(scratch_dir, tmp_path): + ds = Datasette(scratch_dir=scratch_dir) + await ds.create_scratch_database("kept", actor={"id": "alice"}) + await ds.create_scratch_database("removed_by_hand") + ds.close() + os.unlink(os.path.join(scratch_dir, "removed_by_hand.db")) + # Copied in: a SQLite file, a non-SQLite file and an invalid name + _make_source(os.path.join(scratch_dir, "dropped_in.db"), rows=3) + Path(scratch_dir, "notes.db").write_text("not a database") + _make_source(os.path.join(scratch_dir, "bad name.db"), rows=1) + # A -wal whose database is gone would be replayed into the next file + # of that name + Path(scratch_dir, "ghost.db-wal").write_bytes(b"x" * 100) + ds = Datasette(scratch_dir=scratch_dir) + infos = {i.name: i for i in await ds.list_scratch_databases()} + assert set(infos) == {"kept", "dropped_in"} + assert infos["kept"].owner == "alice" + assert infos["dropped_in"].owner is None + assert ( + await ds.get_database("dropped_in").execute("select count(*) from items") + ).single_value() == 3 + assert not os.path.exists(os.path.join(scratch_dir, "ghost.db-wal")) + # Left alone + assert os.path.exists(os.path.join(scratch_dir, "notes.db")) + assert os.path.exists(os.path.join(scratch_dir, "bad name.db")) + counters = ds._scratch.counters + assert counters["adopted"] == 1 + assert counters["dropped_missing"] == 1 + assert counters["stray_files_removed"] == 1 + ds.close() + + +# --------------------------------------------------------------------------- +# modes and isolation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_not_attached_to_crossdb_memory(scratch_dir, tmp_path): + other = str(tmp_path / "regular.db") + sqlite3.connect(other).execute("vacuum") + ds = Datasette([other], scratch_dir=scratch_dir, crossdb=True) + await ds.create_scratch_database("private_scratch") + names = [ + r[1] + for r in (await ds.get_database("_memory").execute("pragma database_list")).rows + ] + assert "regular" in names + assert "private_scratch" not in names + ds.close() + + +@pytest.mark.asyncio +async def test_nolock_creates_rollback_journal_databases(scratch_dir): + ds = Datasette(scratch_dir=scratch_dir, nolock=True) + db = await ds.create_scratch_database("nolock") + await db.execute_write("create table t (id)") + assert (await db.execute("pragma journal_mode")).single_value() == "delete" + assert (await db.execute("select count(*) from t")).single_value() == 0 + ds.close() + + +NO_THREADS_SCRIPT = r""" +import asyncio, os, sys, threading + +def no_threads(self): + raise RuntimeError("can't start new thread") + +threading.Thread.start = no_threads + +from datasette.app import Datasette + +scratch_dir = sys.argv[1] + +async def main(): + ds = Datasette(scratch_dir=scratch_dir, settings={"num_sql_threads": 0}) + await ds.invoke_startup() + db = await ds.create_scratch_database("nothreads", actor={"id": "a"}) + await db.execute_write("create table t (id integer primary key)") + await db.execute_write("insert into t default values") + assert (await db.execute("select count(*) from t")).single_value() == 1 + renamed = await ds.rename_scratch_database("nothreads", "renamed") + assert [i.name for i in await ds.list_scratch_databases()] == ["renamed"] + await ds.delete_scratch_database("renamed") + assert [f for f in os.listdir(scratch_dir) if not f.startswith(".")] == [] + ds.close() + print("ok") + +asyncio.run(main()) +""" + + +def test_no_threads(tmp_path): + # Pyodide: nothing in the scratch database lifecycle may need a thread + result = subprocess.run( + [sys.executable, "-c", NO_THREADS_SCRIPT, str(tmp_path / "scratch")], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert result.stdout.strip() == "ok" + + +@pytest.mark.asyncio +async def test_idle_scratch_database_releases_everything(scratch_dir): + ds = Datasette( + scratch_dir=scratch_dir, settings={"connection_idle_timeout_ms": 200} + ) + await ds.invoke_startup() + db = await ds.create_scratch_database("idle") + await db.execute_write("create table t (id)") + await db.execute("select count(*) from t") + assert db._write_thread is not None + for _ in range(100): + await asyncio.sleep(0.05) + if ( + db._write_thread is None + and not db._retiring_write_threads + and db._all_connections == [] + ): + break + assert db._write_thread is None + assert db._all_connections == [] + assert not _write_threads("idle") + ds.close()