Bound database reads, writers, and queued work

datasette/app.py

@@ -240,8 +240,15 @@ SETTINGS = (
     ),
     Setting(
         "max_open_connections",
-        128,
-        "Maximum number of pooled read connections open across all databases (at least 4 x num_sql_threads) - 0 for no limit",
+        32,
+        "Maximum number of pooled read connections open across all databases - 0 for no limit",
+    ),
+    Setting("max_write_connections", 8, "Maximum concurrent user database writers"),
+    Setting(
+        "max_pending_writes", 256, "Maximum queued user writes across all databases"
+    ),
+    Setting(
+        "write_queue_timeout_ms", 5000, "Maximum wait before a queued write starts"
     ),
     Setting(
         "connection_idle_timeout_ms",
@@ -629,6 +636,13 @@ class Datasette:
             self.executor = futures.ThreadPoolExecutor(
                 max_workers=self.setting("num_sql_threads")
             )
+        for name in (
+            "max_write_connections",
+            "max_pending_writes",
+            "write_queue_timeout_ms",
+        ):
+            if self.setting(name) <= 0:
+                raise StartupError(f"{name} must be greater than zero")
         self.max_returned_rows = self.setting("max_returned_rows")
         self.sql_time_limit_ms = self.setting("sql_time_limit_ms")
         self.page_size = self.setting("default_page_size")
@@ -842,6 +856,7 @@ class Datasette:
             return
 
     async def _invoke_startup(self):
+        self._warn_connection_headroom()
         # Group spans created during startup under a single parent span
         with tracer.start_as_current_span(STARTUP):
             # Register event classes
@@ -1107,6 +1122,20 @@ class Datasette:
     def setting(self, key):
         return self._settings.get(key, None)
 
+    @property
+    def _write_budget(self):
+        from .write_budget import WriteBudget
+
+        budget = self.__dict__.get("_write_budget_instance")
+        if budget is None:
+            budget = WriteBudget(
+                self.setting("max_write_connections"),
+                self.setting("max_pending_writes"),
+                self.setting("write_queue_timeout_ms"),
+            )
+            budget = self.__dict__.setdefault("_write_budget_instance", budget)
+        return budget
+
     @property
     def _read_pool(self):
         pool = self.__dict__.get("_read_pool_instance")
@@ -1121,17 +1150,32 @@ class Datasette:
             pool = self.__dict__.setdefault("_read_pool_instance", pool)
         return pool
 
-    def _effective_max_open_connections(self):
-        """max_open_connections, raised to at least 4 x num_sql_threads.
+    def _warn_connection_headroom(self):
+        try:
+            import resource
 
-        Every executor thread can hold one read connection at a time, so a
-        cap below the thread count would close and reopen a connection for
-        nearly every query. 0 means no limit."""
-        configured = self.setting("max_open_connections") or 0
-        if configured <= 0:
-            return 0
-        threads = self.setting("num_sql_threads") or 0
-        return max(configured, 4 * threads)
+            limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
+        except (ImportError, OSError, ValueError):
+            return
+        reads = self._effective_max_open_connections()
+        writes = self.setting("max_write_connections")
+        attachments = min(10, len(self.databases)) if self.crossdb else 0
+        estimate = 64 + 3 * (reads * (1 + attachments) + writes + 2)
+        if limit != resource.RLIM_INFINITY and (not reads or estimate > limit):
+            logging.getLogger(__name__).warning(
+                "Connection settings may exceed the file descriptor limit (%s): "
+                "read limit %s, writer limit %s, estimated descriptors including "
+                "WAL, attachments and headroom %s. Reduce connection limits or "
+                "increase the OS limit. This estimate excludes plugin allocations.",
+                limit,
+                reads,
+                writes,
+                estimate,
+            )
+
+    def _effective_max_open_connections(self):
+        """Configured read cap; do not silently increase it with thread count."""
+        return max(self.setting("max_open_connections") or 0, 0)
 
     def _connection_idle_timeout_s(self):
         "connection_idle_timeout_ms in seconds; 0 means never close idle ones."

datasette/connection_pool.py

@@ -16,20 +16,19 @@ callback and returns it afterwards.
   reaper thread, so it works with or without a running event loop
   (non-threaded mode, sync callers, closed loops in tests). Where threads
   cannot be started at all (Pyodide) only the opportunistic reaping runs.
-* Every connection is opened with ``Database.connect()`` and prepared with
-  ``Datasette._prepare_connection()`` exactly once, outside the pool lock.
+* Reusable connections are opened with ``Database.connect()`` and prepared
+  with ``Datasette._prepare_connection()`` exactly once, outside the pool
+  lock. Immutable isolated callbacks use fresh, unprepared connections
+  counted against the same cap and closed when the callback finishes.
 * Each pooled connection remembers the database's ``_conn_generation`` when
   it was opened. The SchemaWatcher bumps that generation when a file is
   replaced or deleted and calls ``invalidate_database()``: idle connections
   are closed at once, and leased ones are discarded by their own thread when
   they are returned. A connection is never closed while another thread may
   be stepping it (that segfaults).
-* The cap is soft. Each lease occupies one executor thread (or the event
-  loop in non-threaded mode), and Datasette clamps the cap to at least
-  ``4 x num_sql_threads``, so "cap reached and every connection leased"
-  cannot happen with the public settings. If it does (a cap lowered through
-  the private attribute) the pool opens one more connection and closes it
-  on release.
+* The cap is hard. When every connection is leased, admission waits for
+  release instead of exceeding it. Retiring connections remain counted
+  until they have actually closed.
 * Callbacks receive a ``LeasedConnection`` proxy. Once the callback returns
   the proxy is expired and any further use raises ``ConnectionLeaseError``.
   Write callbacks get the same proxy (see ``Database._execute_writes``).
@@ -48,6 +47,7 @@ import time
 import weakref
 
 from .utils import sqlite3
+from .write_budget import DatabaseResourceError, callback_scope
 
 
 class ConnectionLeaseError(RuntimeError):
@@ -79,7 +79,7 @@ class LeasedConnection:
     on the owning thread before the connection is returned to the pool.
     """
 
-    __slots__ = ("_conn", "_db_name", "_kind", "_cursors")
+    __slots__ = ("_conn", "_cursors", "_db_name", "_kind")
 
     def __init__(self, conn, db_name, kind="read"):
         object.__setattr__(self, "_conn", conn)
@@ -340,11 +340,11 @@ class _DatabaseState:
 
 
 class ReadConnectionPool:
-    def __init__(self, prepare_connection, max_open=128, idle_timeout=30.0):
+    def __init__(self, prepare_connection, max_open=32, idle_timeout=30.0):
         self._prepare_connection = prepare_connection
         self.max_open = max_open  # 0 = no limit
         self.idle_timeout = idle_timeout  # seconds, 0 = never
-        self._lock = threading.Lock()
+        self._lock = threading.Condition()
         # Idle, counted entries, least recently used first
         self._lru = collections.OrderedDict()
         self._open = 0  # counted open connections, idle + leased
@@ -353,22 +353,27 @@ class ReadConnectionPool:
 
     # -- public API ---------------------------------------------------------
 
-    def run(self, db, fn):
+    def run(self, db, fn, isolated=False):
         """Call fn(conn) with a pooled connection leased for the duration."""
-        entry = self.acquire(db)
-        lease = LeasedConnection(entry.conn, db.name, "read")
+        entry = self.acquire(db, fresh=isolated)
+        lease = LeasedConnection(
+            entry.conn, db.name, "isolated" if isolated else "read"
+        )
         rejected = False
+        discard = isolated
         try:
-            result = fn(lease)
+            with callback_scope():
+                result = fn(lease)
             if _contains_cursor(result) or _generator_uses_connection(result):
                 rejected = True
                 close_cursors(result)
         except BaseException as e:
             close_cursors(e)
+            discard = True
             raise
         finally:
             lease._expire()
-            self.release(entry)
+            self.release(entry, discard=discard)
         if rejected:
             raise ConnectionLeaseError(
                 f"An execute_fn() callback for database {db.name!r} returned a {result.__class__.__name__} containing database resources, which "
@@ -377,84 +382,96 @@ class ReadConnectionPool:
             )
         return result
 
-    def acquire(self, db):
+    def acquire(self, db, fresh=False):
         state = self._state(db)
-        to_close = []
-        entry = None
-        with self._lock:
-            if self._closed or state.closed:
-                from .database import DatasetteClosedError
+        deadline = time.monotonic() + 5
+        while True:
+            to_close = []
+            entry = None
+            reserved = False
+            with self._lock:
+                if self._closed or state.closed or db._closed:
+                    from .database import DatasetteClosedError
 
-                raise DatasetteClosedError(f"Database {db.name!r} has been closed")
-            self._collect_expired_locked(time.monotonic(), to_close)
-            while True:
-                if state.idle:
+                    raise DatasetteClosedError(f"Database {db.name!r} has been closed")
+                self._collect_expired_locked(time.monotonic(), to_close)
+                if to_close:
+                    pass  # Close before spending the capacity it releases.
+                elif state.idle and not fresh:
                     entry = state.idle.pop()
                     if state.counted:
                         del self._lru[entry]
                     if entry.generation != db._conn_generation:
-                        # Opened before the file was replaced or deleted
-                        self._forget_locked(entry)
                         to_close.append(entry)
                         self.stats["discarded_stale"] += 1
                         entry = None
-                        continue
+                    else:
+                        state.leased += 1
+                        self.stats["reused"] += 1
+                elif state.counted and self.max_open and self._open >= self.max_open:
+                    if self._lru:
+                        victim, _ = self._lru.popitem(last=False)
+                        victim.state.idle.remove(victim)
+                        to_close.append(victim)
+                        self.stats["evicted_lru"] += 1
+                    else:
+                        remaining = deadline - time.monotonic()
+                        if remaining <= 0:
+                            raise DatabaseResourceError(
+                                "Timed out waiting for a read connection"
+                            )
+                        self._lock.wait(min(remaining, 0.1))
+                else:
+                    generation = db._conn_generation
+                    state.open += 1
                     state.leased += 1
-                    self.stats["reused"] += 1
-                    break
-                if not state.counted or not self.max_open or self._open < self.max_open:
-                    break
-                if self._lru:
-                    # Make room by closing the least recently used idle
-                    # connection, which may belong to any database
-                    victim, _ = self._lru.popitem(last=False)
-                    victim.state.idle.remove(victim)
-                    victim.state.open -= 1
-                    self._open -= 1
-                    to_close.append(victim)
-                    self.stats["evicted_lru"] += 1
-                    continue
-                # Cap reached and every connection is leased: open one more
-                # (soft cap). Bounded by the number of threads that can hold
-                # a lease at once, and closed again on release.
-                self.stats["exceeded_cap"] += 1
+                    self.stats["opened"] += 1
+                    if state.counted:
+                        self._open += 1
+                        self.stats["peak_open"] = max(
+                            self.stats["peak_open"], self._open
+                        )
+                    reserved = True
+            if to_close:
+                self._close_entries(to_close)
+            if entry is not None:
+                return entry
+            if reserved:
                 break
-            if entry is None:
-                # Read before connecting: if the generation moves while we
-                # connect, the connection is discarded on release - one
-                # connection too many, never a stale one handed out twice
-                generation = db._conn_generation
-                state.open += 1
-                state.leased += 1
-                self.stats["opened"] += 1
-                if state.counted:
-                    self._open += 1
-                    self.stats["peak_open"] = max(self.stats["peak_open"], self._open)
-        self._close_entries(to_close)
-        if entry is not None:
-            return entry
-        # Open and prepare outside the lock, exactly once per connection
         conn = None
         try:
             conn = db.connect()
-            self._prepare_connection(conn, db.name)
+            if not fresh:
+                with callback_scope():
+                    self._prepare_connection(conn, db.name)
         except BaseException:
+            if conn is not None:
+                self._close_conn(state, conn)
             with self._lock:
-                state.open -= 1
+                if conn is None:
+                    state.open -= 1
+                    if state.counted:
+                        self._open -= 1
                 state.leased -= 1
                 self.stats["opened"] -= 1
                 self.stats["open_failed"] += 1
-                if state.counted:
-                    self._open -= 1
-            if conn is not None:
-                self._close_conn(state, conn)
+                self._lock.notify_all()
             raise
         return _Entry(conn, state, generation)
 
-    def release(self, entry):
+    def evict_idle(self):
+        """Reclaim idle file connections after external descriptor pressure."""
+        with self._lock:
+            entries = list(self._lru)
+            self._lru.clear()
+            for entry in entries:
+                entry.state.idle.remove(entry)
+        self._close_entries(entries)
+        return len(entries)
+
+    def release(self, entry, discard=False):
         state = entry.state
         conn = entry.conn
-        discard = False
         rolled_back = False
         try:
             # Hygiene: a callback that began a transaction and did not end it
@@ -475,9 +492,6 @@ class ReadConnectionPool:
             stale = entry.generation != state.db._conn_generation
             over_cap = state.counted and self.max_open and self._open > self.max_open
             if discard or stale or over_cap or self._closed or state.closed:
-                state.open -= 1
-                if state.counted:
-                    self._open -= 1
                 discard = True
                 if over_cap:
                     self.stats["closed_over_cap"] += 1
@@ -490,6 +504,7 @@ class ReadConnectionPool:
                     self._lru[entry] = None
                     if self.idle_timeout > 0:
                         notify_reaper = entry.last_used + self.idle_timeout
+            self._lock.notify_all()
         if discard:
             self._close_conn(state, conn)
         if notify_reaper is not None:
@@ -521,7 +536,6 @@ class ReadConnectionPool:
                 state.idle.remove(entry)
                 if state.counted:
                     self._lru.pop(entry, None)
-                self._forget_locked(entry)
                 self.stats["discarded_stale"] += 1
         self._close_entries(stale)
 
@@ -537,7 +551,6 @@ class ReadConnectionPool:
             for entry in to_close:
                 if state.counted:
                     self._lru.pop(entry, None)
-                self._forget_locked(entry)
         self._close_entries(to_close)
 
     def close(self):
@@ -547,7 +560,6 @@ class ReadConnectionPool:
             self._lru.clear()
             for entry in to_close:
                 entry.state.idle.remove(entry)
-                self._forget_locked(entry)
         self._close_entries(to_close)
 
     def snapshot(self):
@@ -571,13 +583,6 @@ class ReadConnectionPool:
                     state = db._read_pool_state = _DatabaseState(db)
         return state
 
-    def _forget_locked(self, entry):
-        # Caller holds the lock and has already removed entry from the idle
-        # stack and the LRU
-        entry.state.open -= 1
-        if entry.state.counted:
-            self._open -= 1
-
     def _collect_expired_locked(self, now, to_close):
         if self.idle_timeout <= 0 or not self._lru:
             return
@@ -589,7 +594,6 @@ class ReadConnectionPool:
                 break
             lru.popitem(last=False)
             head.state.idle.remove(head)
-            self._forget_locked(head)
             to_close.append(head)
             self.stats["expired"] += 1
 
@@ -603,7 +607,11 @@ class ReadConnectionPool:
         except Exception:  # noqa: BLE001, S110
             pass
         with self._lock:
+            state.open -= 1
+            if state.counted:
+                self._open -= 1
             self.stats["closed"] += 1
+            self._lock.notify_all()
         try:
             state.db._all_connections.remove(conn)
         except ValueError:

datasette/database.py

@@ -1,6 +1,8 @@
 import asyncio
+import contextlib
 import contextvars
 import inspect
+import logging
 import os
 import queue
 import sys
@@ -16,7 +18,7 @@ import sqlite_utils
 from opentelemetry import context as otel_context_api
 from opentelemetry.trace import Status, StatusCode
 
-from .connection_pool import (  # noqa: F401
+from .connection_pool import (
     ConnectionLeaseError,
     LeasedConnection,
     close_cursors,
@@ -56,6 +58,7 @@ from .telemetry_registry import (
 )
 from .tracer import trace
 from .utils import (
+    await_me_maybe,
     call_with_supported_arguments,
     detect_fts,
     detect_primary_keys,
@@ -72,6 +75,15 @@ from .utils import (
 from .utils.catalog import SPATIALITE_HIDDEN_TABLES
 from .utils.sql_analysis import SQLAnalysis, analyze_sql_tables
 from .utils.sqlite import sqlite_derived_table_dependencies, sqlite_hidden_table_names
+from .write_budget import (
+    RETIRE_WRITER,
+    DatabaseResourceError,
+    callback_scope,
+    check_reentrancy,
+    connect_with_retry,
+)
+
+logger = logging.getLogger(__name__)
 
 EXECUTE_WRITE_RETURNING_LIMIT = 10
 
@@ -181,6 +193,10 @@ class Database:
         if not is_temp_disk:
             self.mode = mode
 
+    @property
+    def _budgeted_writer(self):
+        return self is not getattr(self.ds, "_internal_database", None)
+
     def _check_not_closed(self):
         if self._closed:
             raise DatasetteClosedError(f"Database {self.name!r} has been closed")
@@ -253,7 +269,12 @@ class Database:
         with "database table is locked" (shared-cache locks do not wait)."""
         self._check_not_closed()
         if self.ds.executor is None:
-            return fn(self._non_threaded_write_connection())
+            with (
+                self.ds._write_budget.inline(self)
+                if self._budgeted_writer
+                else contextlib.nullcontext()
+            ):
+                return fn(self._non_threaded_write_connection())
         return await self._send_to_write_thread(fn, transaction=False)
 
     def _non_threaded_write_connection(self):
@@ -274,7 +295,8 @@ class Database:
             conn = self.connect(write=True)
             try:
                 _pin_write_connection_pragmas(conn)
-                self.ds._prepare_connection(conn, self.name)
+                with callback_scope():
+                    self.ds._prepare_connection(conn, self.name)
             except BaseException:
                 # Never cache a half-prepared connection
                 self._forget_connection(conn)
@@ -288,7 +310,12 @@ class Database:
         # wait for the catalog rows for this database to be rebuilt
         state = self._watch_state
         if state is not None and (state.needs_scan or state.scan_future is not None):
-            await self.ds._schema_watcher.after_write(state)
+            try:
+                await self.ds._schema_watcher.after_write(state)
+            except DatabaseResourceError as error:
+                raise RuntimeError(
+                    "Write completed, but catalog refresh failed; do not retry the write automatically"
+                ) from error
 
     @property
     def cached_table_counts(self):
@@ -359,8 +386,14 @@ class Database:
             qs = "?mode=rw" if self._file_seen else ""
         if self.mode is not None:
             qs = f"?mode={self.mode}"
-        conn = sqlite3.connect(
-            f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs
+        conn = connect_with_retry(
+            self,
+            lambda: sqlite3.connect(
+                f"file:{self.path}{qs}",
+                uri=True,
+                check_same_thread=False,
+                **extra_kwargs,
+            ),
         )
         self._file_seen = True
         if track:
@@ -390,8 +423,10 @@ class Database:
         with self._write_thread_lock:
             write_thread = self._write_thread
             retiring_threads = tuple(self._retiring_write_threads)
-            if write_thread is not None:
+            if write_thread is not None and write_thread.is_alive():
                 self._write_queue.put(_SHUTDOWN)
+            elif self._budgeted_writer:
+                self.ds._write_budget.cancel_database(self)
         still_running = False
         for thread in ((write_thread,) if write_thread else ()) + retiring_threads:
             thread.join(timeout=10)
@@ -544,6 +579,7 @@ class Database:
         return results
 
     async def execute_isolated_fn(self, fn):
+        check_reentrancy()
         self._check_not_closed()
         # Open a new connection just for the duration of this function,
         # blocking the write queue to avoid any writes occurring during it
@@ -569,7 +605,24 @@ class Database:
             with record_operation_duration(self.name, "write" if write else "read"):
                 if self.ds.executor is None:
                     # non-threaded mode
-                    result = _run()
+                    if write:
+                        with (
+                            self.ds._write_budget.inline(self)
+                            if self._budgeted_writer
+                            else contextlib.nullcontext()
+                        ):
+                            if (
+                                self._write_connection is not None
+                                and not self.is_memory
+                            ):
+                                old, self._write_connection = (
+                                    self._write_connection,
+                                    None,
+                                )
+                                self._forget_connection(old)
+                            result = _run()
+                    else:
+                        result = self.ds._read_pool.run(self, fn, isolated=True)
                     await self._after_write()
                     return result
                 if not write:
@@ -577,7 +630,9 @@ class Database:
                     # is no write queue to block; run against a fresh
                     # read-only connection on the SQL thread pool, tracked so
                     # close() waits for it
-                    return await self._run_in_executor(_run)
+                    return await self._run_in_executor(
+                        lambda: self.ds._read_pool.run(self, fn, isolated=True)
+                    )
                 # Threaded mode - send to write thread
                 result = await self._send_to_write_thread(fn, isolated_connection=True)
                 await self._after_write()
@@ -610,6 +665,7 @@ class Database:
                 )
 
     async def _execute_write_fn(self, fn, block=True, transaction=True, request=None):
+        check_reentrancy()
         self._check_not_closed()
         pending_events = []
 
@@ -619,16 +675,21 @@ class Database:
         fn = self._wrap_fn_with_hooks(fn, request, transaction, track_event)
         if self.ds.executor is None:
             # non-threaded mode
-            conn = self._non_threaded_write_connection()
-            try:
-                if transaction:
-                    with conn:
-                        conn.execute("BEGIN IMMEDIATE")
-                        result = _call_with_lease(fn, conn, self.name, "write")
-                else:
-                    result = _call_with_lease(fn, conn, self.name, "write")
-            finally:
-                self._schema_check(conn)
+            with (
+                self.ds._write_budget.inline(self)
+                if self._budgeted_writer
+                else contextlib.nullcontext()
+            ):
+                conn = self._non_threaded_write_connection()
+                try:
+                    if transaction:
+                        with conn:
+                            conn.execute("BEGIN IMMEDIATE")
+                            result = _call_with_lease(fn, conn, self.name, "write")
+                    else:
+                        result = _call_without_transaction(fn, conn, self.name)
+                finally:
+                    self._schema_check(conn)
             if not block:
                 # There is no write thread here, so the write has already
                 # finished. Hand back the same (task_id, reply_future) shape
@@ -644,8 +705,13 @@ class Database:
             )
         if block:
             await self._after_write()
-            for event in pending_events:
-                await self.ds.track_event(event)
+            try:
+                for event in pending_events:
+                    await self.ds.track_event(event)
+            except DatabaseResourceError as error:
+                raise RuntimeError(
+                    "Write completed, but event dispatch failed; do not retry the write automatically"
+                ) from error
         else:
             # For non-blocking writes, spawn a background task to
             # dispatch events after the write thread completes
@@ -654,10 +720,19 @@ class Database:
             async def _dispatch_events_after_write():
                 try:
                     await reply_future
-                except Exception:  # noqa: BLE001
-                    # The write failed; skip success events regardless of why
-                    # if the write failed, don't emit success events
+                except Exception as error:
+                    logger.exception(
+                        "Nonblocking write %s failed",
+                        task_id,
+                        extra={
+                            "database": self.name,
+                            "task_id": str(task_id),
+                            "error_code": getattr(error, "code", None),
+                        },
+                    )
+                    await self._notify_write_completed(task_id, error)
                     return
+                await self._notify_write_completed(task_id, None)
                 for event in pending_events:
                     await self.ds.track_event(event)
 
@@ -665,6 +740,20 @@ class Database:
             result = task_id
         return result
 
+    async def _notify_write_completed(self, task_id, exception):
+        from .plugins import pm
+
+        try:
+            for result in pm.hook.write_task_completed(
+                datasette=self.ds,
+                database=self.name,
+                task_id=task_id,
+                exception=exception,
+            ):
+                await await_me_maybe(result)
+        except Exception:
+            logger.exception("write_task_completed hook failed for %s", task_id)
+
     def _wrap_fn_with_hooks(self, fn, request, transaction, track_event):
         from .plugins import pm
 
@@ -702,6 +791,7 @@ class Database:
     async def _send_to_write_thread(
         self, fn, block=True, isolated_connection=False, transaction=True
     ):
+        check_reentrancy()
         task_id = uuid.uuid4()
         loop = asyncio.get_running_loop()
         reply_future = loop.create_future()
@@ -716,24 +806,23 @@ class Database:
                 raise DatasetteClosedError(f"Database {self.name!r} has been closed")
             if self._write_queue is None:
                 self._write_queue = queue.Queue()
-            if self._write_thread is None:
-                # Start the thread before enqueueing: if start() fails (for
-                # example "can't start new thread") nothing is stranded
-                self._start_write_thread()
-            # Capture the OpenTelemetry context and enqueue time for the write thread
-            self._write_queue.put(
-                WriteTask(
-                    fn,
-                    task_id,
-                    loop,
-                    reply_future,
-                    isolated_connection,
-                    transaction,
-                    otel_context_api.get_current(),
-                    time.time_ns(),
-                    block,
-                )
+            task = WriteTask(
+                fn,
+                task_id,
+                loop,
+                reply_future,
+                isolated_connection,
+                transaction,
+                otel_context_api.get_current(),
+                time.time_ns(),
+                block,
             )
+            if self._budgeted_writer:
+                self.ds._write_budget.submit(self, task)
+            else:
+                if self._write_thread is None:
+                    self._start_write_thread()
+                self._write_queue.put(task)
         if block:
             return await reply_future
         else:
@@ -769,7 +858,6 @@ class Database:
         # with no task, in which case it closes its connection and exits.
         # The next write starts a new thread.
         writer = _WriteConnection(self)
-        writer.open()
         deregistered = False
         try:
             self._process_write_queue(writer)
@@ -781,7 +869,11 @@ class Database:
             current = threading.current_thread()
             with self._write_thread_lock:
                 self._retiring_write_threads.discard(current)
-                if not deregistered and self._write_thread is current:
+                if self._budgeted_writer:
+                    if self._write_thread is current:
+                        self._write_thread = None
+                    self.ds._write_budget.wake()
+                elif not deregistered and self._write_thread is current:
                     # Died from an unexpected BaseException. Deregister so
                     # the next write starts a fresh thread, and start one
                     # now if tasks are already waiting.
@@ -793,7 +885,7 @@ class Database:
         configured_idle_timeout = self._write_thread_idle_timeout()
 
         def current_idle_timeout():
-            if writer.exception is not None and configured_idle_timeout is not None:
+            if writer.exception is not None:
                 # The connect error has been reported to every write already
                 # queued: exit as soon as the queue is empty, so the next
                 # write starts a new thread that tries to connect again
@@ -817,9 +909,19 @@ class Database:
                         self._write_thread = None
                     self._retiring_write_threads.add(current)
                 return
-            if task is _SHUTDOWN:
+            if task is _SHUTDOWN or task is RETIRE_WRITER:
                 return
-            if writer.generation != self._conn_generation or not writer.usable():
+            if self._budgeted_writer and not self.ds._write_budget.take(self, task):
+                continue
+            if (
+                task.isolated_connection
+                and writer.conn is not None
+                and not self.is_memory
+            ):
+                writer.close()
+            if not task.isolated_connection and (
+                writer.generation != self._conn_generation or not writer.usable()
+            ):
                 # The file was replaced or deleted since this connection was
                 # opened (the SchemaWatcher bumps the generation): reopen by
                 # path. connect() uses mode=rw for a file Datasette has seen,
@@ -850,7 +952,7 @@ class Database:
                     **write_span_kwargs,
                 ).end(end_time=dequeued_at_ns)
                 record_write_queue_wait(self.name, dequeued_at_ns - task.enqueued_at_ns)
-                if writer.exception is not None:
+                if writer.exception is not None and not task.isolated_connection:
                     exception = writer.exception
                 elif task.isolated_connection:
                     try:
@@ -901,8 +1003,8 @@ class Database:
                                             task.fn, conn, self.name, "write"
                                         )
                                 else:
-                                    result = _call_with_lease(
-                                        task.fn, conn, self.name, "write"
+                                    result = _call_without_transaction(
+                                        task.fn, conn, self.name
                                     )
                             finally:
                                 self._schema_check(conn, task.loop)
@@ -912,6 +1014,10 @@ class Database:
             finally:
                 if token is not None:
                     otel_context_api.detach(token)
+                if self._budgeted_writer:
+                    self.ds._write_budget.completed(self)
+            if self._budgeted_writer and self.ds._write_budget.should_yield(self):
+                return
 
     def _write_task_failed(self, e):
         """The exception to deliver to the caller of a write task that
@@ -954,6 +1060,7 @@ class Database:
                 return await self._execute_fn(fn_in_execute_span)
 
     async def _execute_fn(self, fn):
+        check_reentrancy()
         self._check_not_closed()
         read_pool = self.ds._read_pool
         if self.ds.executor is None:
@@ -1439,6 +1546,22 @@ def _pin_write_connection_pragmas(conn):
     conn.execute("PRAGMA recursive_triggers = on")
 
 
+def _call_without_transaction(fn, conn, db_name):
+    try:
+        result = _call_with_lease(fn, conn, db_name, "write")
+    except BaseException:
+        if conn.in_transaction:
+            conn.rollback()
+        raise
+    if conn.in_transaction:
+        conn.rollback()
+        raise ConnectionLeaseError(
+            "transaction=False callbacks must commit or roll back before returning; "
+            "the unfinished transaction was rolled back"
+        )
+    return result
+
+
 def _call_with_lease(fn, conn, db_name, kind):
     """Call fn with a LeasedConnection for conn that expires when it returns.
 
@@ -1450,9 +1573,14 @@ def _call_with_lease(fn, conn, db_name, kind):
     """
     lease = LeasedConnection(conn, db_name, kind)
     try:
-        result = fn(lease)
+        with callback_scope():
+            result = fn(lease)
     except BaseException as e:
         close_cursors(e)
+        if isinstance(e, DatabaseResourceError):
+            raise RuntimeError(  # noqa: TRY004 - execution failure, not invalid input type
+                "Database callback started before a resource error; do not retry the write automatically"
+            ) from e
         raise
     else:
         # A returned cursor is closed here, on the thread that owns the
@@ -1494,9 +1622,10 @@ class _WriteConnection:
             _pin_write_connection_pragmas(self.conn)
             # Threads do not inherit the caller's context, so any spans
             # created by prepare_connection hooks here are root spans
-            db.ds._prepare_connection(self.conn, db.name)
-        except Exception as e:  # noqa: BLE001
-            self.exception = e
+            with callback_scope():
+                db.ds._prepare_connection(self.conn, db.name)
+        except BaseException as e:  # noqa: BLE001
+            self.exception = db._write_task_failed(e)
             self.close()
 
     def reopen(self):

datasette/handle_exception.py

@@ -9,6 +9,7 @@ from .utils.asgi import (
     Base400,
 )
 from .views.base import DatasetteError
+from .write_budget import DatabaseResourceError
 
 # Debugger imports are deliberate - they back the "pdb" setting, which drops
 # into a debugger on unhandled exceptions
@@ -34,7 +35,11 @@ def handle_exception(datasette, request, exception):
 
         title = None
         plain_message = None
-        if isinstance(exception, Base400):
+        if isinstance(exception, DatabaseResourceError):
+            status = 503
+            info = {"code": exception.code, "execution_started": False}
+            message = str(exception)
+        elif isinstance(exception, Base400):
             status = exception.status
             info = {}
             message = exception.args[0]
@@ -52,7 +57,9 @@ def handle_exception(datasette, request, exception):
             message = str(exception)
             traceback.print_exc()
         templates = [f"{status}.html", "error.html"]
-        headers = {}
+        headers = (
+            {"Retry-After": "1"} if isinstance(exception, DatabaseResourceError) else {}
+        )
         if datasette.cors:
             add_cors_headers(headers)
         if request.path.split("?")[0].endswith(".json"):

datasette/hookspecs.py

@@ -267,3 +267,8 @@ def write_wrapper(datasette, database, request, transaction):
 
     Return ``None`` to skip wrapping.
     """
+
+
+@hookspec
+def write_task_completed(datasette, database, task_id, exception):
+    """Completion of an accepted block=False write, including admission expiry."""

datasette/schema_watcher.py

@@ -251,6 +251,10 @@ def header_schema_version(path):
 
 
 def _is_transient(error):
+    from .write_budget import DatabaseResourceError
+
+    if isinstance(error, DatabaseResourceError):
+        return True
     if isinstance(error, OSError):
         return error.errno in (errno.EMFILE, errno.ENFILE, errno.ENOMEM, errno.EAGAIN)
     if isinstance(error, sqlite3.OperationalError):
@@ -312,6 +316,7 @@ class SchemaWatcher:
     def __init__(self, ds):
         self.ds = ds
         self.states = {}
+        self._scan_resource_lock = threading.Lock()
         # States whose catalog must be (re)built: registered after startup,
         # changed by a write whose notification may never run, or whose last
         # scan failed for a transient reason. Flushed by the next request or
@@ -735,6 +740,10 @@ class SchemaWatcher:
         self._last_sweep = time.monotonic()
 
     def _sweep_sync(self, states, background=False):
+        with self._scan_resource_lock:
+            return self._sweep_serial(states, background)
+
+    def _sweep_serial(self, states, background=False):
         """Runs in a worker thread. Reads state, never writes it."""
         by_dir = {}
         for state in states:
@@ -911,6 +920,10 @@ class SchemaWatcher:
         }
 
     def _scan_chunk_sync(self, states):
+        with self._scan_resource_lock:
+            return self._scan_chunk_serial(states)
+
+    def _scan_chunk_serial(self, states):
         results = []
         for state in states:
             try:

datasette/write_budget.py

@@ -0,0 +1,344 @@
+"""Bound retained writers and queued writes without changing per-database FIFO.
+
+The coordinator only admits work and retires idle workers. SQLite connections
+are opened, used and closed by their owning writer. A retiring thread keeps
+its slot until it has actually exited. Internal catalog writes use their own
+single writer, so user saturation cannot prevent catalog progress.
+"""
+
+import collections
+import contextlib
+import contextvars
+import errno
+import os
+import threading
+import time
+
+in_database_callback = contextvars.ContextVar("in_database_callback", default=False)
+RETIRE_WRITER = object()
+_STARTING = object()
+
+
+class DatabaseResourceError(RuntimeError):
+    code = "database_resource_unavailable"
+    execution_started = False
+
+
+class DatabaseQueueFull(DatabaseResourceError):
+    code = "database_queue_full"
+
+
+class DatabaseAdmissionTimeout(DatabaseResourceError):
+    code = "database_admission_timeout"
+
+
+class DatabaseReentrancyError(RuntimeError):
+    code = "database_reentrant_operation"
+
+
+def check_reentrancy():
+    if in_database_callback.get():
+        raise DatabaseReentrancyError(
+            "Schedule dependent database work after the current callback returns"
+        )
+
+
+@contextlib.contextmanager
+def callback_scope():
+    token = in_database_callback.set(True)
+    try:
+        yield
+    finally:
+        in_database_callback.reset(token)
+
+
+def connect_with_retry(db, connect):
+    """Retry only opening a connection, never preparation or a callback."""
+    from .utils import sqlite3
+
+    for attempt in range(2):
+        try:
+            return connect()
+        except (OSError, sqlite3.OperationalError) as error:
+            pressure = isinstance(error, OSError) and error.errno in (
+                errno.EMFILE,
+                errno.ENFILE,
+                errno.ENOMEM,
+            )
+            cantopen = (
+                getattr(error, "sqlite_errorcode", None) == sqlite3.SQLITE_CANTOPEN
+            )
+            if not pressure and (not cantopen or not os.path.exists(db.path)):
+                raise
+            if attempt:
+                if pressure:
+                    raise DatabaseResourceError(
+                        "Unable to open a database connection; try again later"
+                    ) from error
+                raise
+            pool = db.ds._read_pool_or_none
+            if pool is not None:
+                pool.evict_idle()
+            # Writers own their connections: ask them to retire rather
+            # than closing SQLite objects from this thread.
+            budget = db.ds.__dict__.get("_write_budget_instance")
+            if budget is not None:
+                budget.retire_idle()
+
+
+class WriteBudget:
+    def __init__(self, max_writers, max_pending, timeout_ms):
+        self.max_writers = max_writers
+        self.max_pending = max_pending
+        self.timeout = timeout_ms / 1000
+        self._condition = threading.Condition()
+        self._active = {}  # db -> starting marker or thread (including retiring)
+        self._pending = {}  # task -> (db, monotonic deadline)
+        self._ready = collections.OrderedDict()
+        self._busy = set()
+        self._retirement_requested = set()
+        self._thread = None
+        self._inline_lock = threading.RLock()
+        self._inline = collections.OrderedDict()
+        self.stats = collections.Counter()
+
+    def submit(self, db, task):
+        # Caller holds db._write_thread_lock. Never take a database lock
+        # while holding _condition in the coordinator.
+        with self._condition:
+            if len(self._pending) >= self.max_pending:
+                self.stats["rejected"] += 1
+                raise DatabaseQueueFull(
+                    "Too many queued database writes; try again later"
+                )
+            self._pending[task] = (db, time.monotonic() + self.timeout)
+            db._write_queue.put(task)
+            self._ready.setdefault(db, None)
+            self.stats["peak_pending"] = max(
+                self.stats["peak_pending"], len(self._pending)
+            )
+            if self._thread is None:
+                thread = threading.Thread(
+                    target=self._run, name="datasette-write-scheduler", daemon=True
+                )
+                try:
+                    thread.start()
+                except BaseException:
+                    self._pending.pop(task)
+                    self._remove_queued(db, task)
+                    self._ready.pop(db, None)
+                    raise
+                self._thread = thread
+            self._condition.notify_all()
+
+    @staticmethod
+    def _remove_queued(db, task):
+        # Remove expired entries instead of retaining unbounded tombstones.
+        q = db._write_queue
+        with q.mutex:
+            try:
+                q.queue.remove(task)
+            except ValueError:
+                pass  # A worker took it; take() will reject the expired task.
+            q.not_full.notify()
+
+    @staticmethod
+    def _fail(task, error):
+        from .database import _deliver_write_result
+
+        _deliver_write_result(task, None, error)
+
+    def take(self, db, task):
+        with self._condition:
+            item = self._pending.pop(task, None)
+            self._condition.notify_all()
+            if item is None:
+                return False
+            if time.monotonic() >= item[1]:
+                self.stats["expired"] += 1
+                expired = True
+            else:
+                self._busy.add(db)
+                expired = False
+        if expired:
+            self._fail(
+                task, DatabaseAdmissionTimeout("Write expired before execution started")
+            )
+        return not expired
+
+    def completed(self, db):
+        with self._condition:
+            self._busy.discard(db)
+            self._condition.notify_all()
+
+    def should_yield(self, db):
+        # Named in-memory databases need a keeper connection. Until a
+        # separate keeper is introduced their writer slot is pinned.
+        if db.is_memory or db._closed:
+            return False
+        with self._condition:
+            return any(
+                other is not db and other not in self._active
+                for other, _ in self._pending.values()
+            )
+
+    def wake(self):
+        with self._condition:
+            self._condition.notify_all()
+
+    def cancel_database(self, db):
+        from .database import DatasetteClosedError
+
+        cancelled = []
+        with self._condition:
+            for task, (owner, _) in list(self._pending.items()):
+                if owner is db:
+                    self._pending.pop(task)
+                    self._remove_queued(db, task)
+                    cancelled.append(task)
+            self._ready.pop(db, None)
+            self._condition.notify_all()
+        for task in cancelled:
+            self._fail(
+                task, DatasetteClosedError("Database closed before write started")
+            )
+        with self._inline_lock:
+            self._inline.pop(db, None)
+
+    def _run(self):
+        while True:
+            expired = []
+            start = []
+            retire = []
+            with self._condition:
+                for db, thread in list(self._active.items()):
+                    if thread is not _STARTING and not thread.is_alive():
+                        del self._active[db]
+                        self._retirement_requested.discard(db)
+                        self._busy.discard(db)
+                        self._ready.pop(db, None)
+                        self._ready[db] = None
+                now = time.monotonic()
+                for task, (db, deadline) in list(self._pending.items()):
+                    if now >= deadline:
+                        self._pending.pop(task)
+                        self._remove_queued(db, task)
+                        self.stats["expired"] += 1
+                        expired.append(task)
+                pending_dbs = {db for db, _ in self._pending.values()}
+                for db in list(self._ready):
+                    if db not in pending_dbs:
+                        del self._ready[db]
+                    elif (
+                        db not in self._active and len(self._active) < self.max_writers
+                    ):
+                        del self._ready[db]
+                        self._active[db] = _STARTING
+                        start.append(db)
+                        self.stats["peak_writers"] = max(
+                            self.stats["peak_writers"], len(self._active)
+                        )
+                if any(db not in self._active for db in pending_dbs):
+                    for db in self._active:
+                        if (
+                            db not in self._busy
+                            and not db.is_memory
+                            and db not in self._retirement_requested
+                        ):
+                            self._retirement_requested.add(db)
+                            retire.append(db)
+                if not self._pending and not self._active:
+                    self._ready.clear()
+                    self._thread = None
+                    stop = True
+                else:
+                    stop = False
+            for task in expired:
+                self._fail(
+                    task,
+                    DatabaseAdmissionTimeout("Write expired before execution started"),
+                )
+            if stop:
+                return
+            for db in start:
+                try:
+                    with db._write_thread_lock:
+                        if db._closed:
+                            self.cancel_database(db)
+                            thread = None
+                        else:
+                            db._start_write_thread()
+                            thread = db._write_thread
+                    with self._condition:
+                        if thread is None:
+                            self._active.pop(db, None)
+                        else:
+                            self._active[db] = thread
+                except Exception as error:  # noqa: BLE001
+                    failures = []
+                    with self._condition:
+                        self._active.pop(db, None)
+                        for task, (owner, _) in list(self._pending.items()):
+                            if owner is db:
+                                self._pending.pop(task)
+                                self._remove_queued(db, task)
+                                failures.append(task)
+                    for task in failures:
+                        self._fail(task, error)
+            for db in retire:
+                with db._write_thread_lock:
+                    if db._write_thread is not None and not db._closed:
+                        db._write_queue.put(RETIRE_WRITER)
+            # Timed wake also observes completed thread retirement and
+            # expires queued work while all slots are pinned or busy.
+            with self._condition:
+                self._condition.wait(0.01)
+            # No completed database is retained in this long-lived frame.
+            del start, retire, expired, pending_dbs
+            db = thread = task = owner = failures = None
+
+    @contextlib.contextmanager
+    def inline(self, db):
+        """Non-threaded mode: bounded idle connections, no helper threads."""
+        check_reentrancy()
+        with self._inline_lock:
+            for other in list(self._inline):
+                if other._write_connection is None or other._closed:
+                    self._inline.pop(other)
+            if db not in self._inline and len(self._inline) >= self.max_writers:
+                victim = next((d for d in self._inline if not d.is_memory), None)
+                if victim is None:
+                    raise DatabaseResourceError(
+                        "Writer capacity is held by in-memory databases"
+                    )
+                self._inline.pop(victim)
+                conn, victim._write_connection = victim._write_connection, None
+                victim._forget_connection(conn)
+            self._inline.pop(db, None)
+            self._inline[db] = None
+            yield
+
+    def retire_idle(self):
+        with self._condition:
+            candidates = [
+                db
+                for db in self._active
+                if db not in self._busy
+                and not db.is_memory
+                and db not in self._retirement_requested
+            ]
+            self._retirement_requested.update(candidates)
+        for db in candidates:
+            with db._write_thread_lock:
+                if db._write_thread is not None:
+                    db._write_queue.put(RETIRE_WRITER)
+
+    def snapshot(self):
+        with self._condition:
+            return {
+                "writers": len(self._active),
+                "pending": len(self._pending),
+                "max_writers": self.max_writers,
+                "max_pending": self.max_pending,
+                "stats": dict(self.stats),
+            }

docs/cli-reference.rst

@@ -252,8 +252,14 @@ These can be passed to ``datasette serve`` using ``datasette serve --setting nam
       num_sql_threads              Number of threads in the thread pool for
                                    executing SQLite queries (default=3)
       max_open_connections         Maximum number of pooled read connections open
-                                   across all databases (at least 4 x
-                                   num_sql_threads) - 0 for no limit (default=128)
+                                   across all databases - 0 for no limit
+                                   (default=32)
+      max_write_connections        Maximum concurrent user database writers
+                                   (default=8)
+      max_pending_writes           Maximum queued user writes across all databases
+                                   (default=256)
+      write_queue_timeout_ms       Maximum wait before a queued write starts
+                                   (default=5000)
       connection_idle_timeout_ms   Close read connections, and stop write threads,
                                    that have been idle for this many milliseconds -
                                    0 to keep them open (default=30000)

docs/internals.rst

@@ -2063,7 +2063,7 @@ Datasette is designed to serve anything from one database to thousands of databa
 
 **Reads** run on a shared pool of :ref:`setting_num_sql_threads` threads. Each read borrows a connection from a pool for the duration of one query or one ``execute_fn()`` callback, then returns it. At most :ref:`setting_max_open_connections` pooled read connections are open at once across all databases; when a new one is needed the least recently used idle connection is closed first, and connections idle for longer than :ref:`setting_connection_idle_timeout_ms` are closed. Read connections to the database files Datasette serves are read-only.
 
-**Writes** to each database are run one at a time, in the order they were submitted, by a write thread dedicated to that database and holding its single write connection. The thread is started by the first write and stops, closing its connection, after :ref:`setting_connection_idle_timeout_ms` without writes; the next write starts a new one.
+**Writes** to each database are run one at a time, in the order they were submitted, by a write thread dedicated to that database and holding its single write connection. At most :ref:`setting_max_write_connections` user writers exist at once. The thread stops, closing its connection, after :ref:`setting_connection_idle_timeout_ms` without writes, or after a callback when another database needs its slot; the next write waits for capacity and starts a new one. Queued writes are bounded by :ref:`setting_max_pending_writes` and :ref:`setting_write_queue_timeout_ms`. The internal catalog uses its own writer.
 
 **Isolated functions** (:ref:`database_execute_isolated_fn`) get a connection of their own that is opened and closed for that one call.
 
@@ -2334,6 +2334,8 @@ For example:
     except Exception as e:
         print("An error occurred:", e)
 
+With ``transaction=False``, finish any transaction you start before returning. An unfinished transaction is rolled back and raises ``ConnectionLeaseError``. A connection can be retired between callbacks, so transaction state and connection-local state must not span callbacks. Do not synchronously wait for another Datasette database operation inside a callback: finite connection limits can deadlock that pattern. Straightforward nested calls raise ``DatabaseReentrancyError``; schedule dependent operations after the callback has returned.
+
 Your function can optionally accept a ``track_event`` parameter in addition to ``conn``.  If it does, it will be passed a callable that can be used to queue events for dispatch after the write transaction commits successfully.  Events queued this way are discarded if the write raises an exception.
 
 .. code-block:: python

docs/plugin_hooks.rst

@@ -2604,3 +2604,13 @@ Tokens can then be created and verified using :ref:`datasette.create_token() <da
     actor = await datasette.verify_token(token)
 
 If no handlers are registered, ``create_token()`` raises ``RuntimeError``. If the requested ``handler`` name is not found, it raises ``ValueError``.
+
+
+.. _plugin_hook_write_task_completed:
+
+write_task_completed(datasette, database, task_id, exception)
+-------------------------------------------------------------
+
+Called after an accepted ``execute_write_fn(..., block=False)`` task finishes, fails, or expires before execution. ``database`` is the database name and ``task_id`` is the UUID returned on acceptance. ``exception`` is ``None`` on success, otherwise the exception describing the failure. ``DatabaseAdmissionTimeout`` means execution never started; other execution failures must not be assumed safe to retry. An immediately rejected submission does not trigger this hook because no task was accepted.
+
+The hook may return an awaitable. Delivery runs on the submitting event loop and is not durable: stopping that loop or the process can prevent notification. Failures are also logged with ``database``, ``task_id`` and ``error_code`` fields. A failing completion hook is logged and does not retry the write or change its result.

docs/settings.rst

@@ -223,25 +223,64 @@ max_open_connections
     setting_default(cog, "max_open_connections")
 .. ]]]
 
-Default: ``128``
+Default: ``32``
 
 .. [[[end]]]
 
-Read queries borrow a connection from a pool for the duration of one query (or one :ref:`execute_fn() <database_execute_fn>` callback) and return it afterwards. This setting caps how many pooled read connections can be open at once, across all attached file databases. When a new connection is needed and the cap has been reached, the least recently used idle connection - which may belong to any database - is closed first.
+Maximum pooled read connections across file-backed databases. When all connections are busy, reads wait up to five seconds for capacity instead of exceeding the cap. Idle connections are evicted before opening a connection to another database. The configured limit is respected even when it is smaller than :ref:`setting_num_sql_threads`. Set this to 0 for no limit.
 
-This is what lets one Datasette instance serve hundreds or thousands of database files without holding a connection, and its file descriptors, open to every one of them.
+Immutable isolated callbacks use fresh connections counted against this same cap. In-memory read connections are exempt, except for the ``_memory`` database under ``--crossdb``, whose connections attach database files and count against the cap.
 
-Every SQL thread can hold one read connection at a time, so a cap smaller than the number of threads would close and reopen connections for nearly every query. Datasette therefore raises the cap it uses to at least four times :ref:`setting_num_sql_threads` (12 with the default of 3 threads). ``/-/settings.json`` shows the value you configured.
+This limits connections, not exact file descriptors or RSS. WAL files, cross-database attachments, HTTP sockets, temporary files and plugins consume additional resources. Datasette logs a warning at startup if a conservative estimate of configured connection costs and headroom exceeds the OS file descriptor limit. Read queries waiting for executor threads, result sizes, callback payload sizes and plugin memory remain outside these limits.
 
-Set this to 0 for no limit, in which case idle connections are only closed by :ref:`setting_connection_idle_timeout_ms`.
+.. _setting_max_write_connections:
 
-::
+max_write_connections
+~~~~~~~~~~~~~~~~~~~~~
 
-    datasette *.db --setting max_open_connections 64
+.. [[[cog
+    setting_default(cog, "max_write_connections")
+.. ]]]
 
-The cap does not cover write connections (one per database that is being written to, see :ref:`setting_connection_idle_timeout_ms`), the short-lived connections used by :ref:`database_execute_isolated_fn` and by schema checks, or in-memory databases, which use no file descriptors. The exception is the ``_memory`` database under ``--crossdb``: each of its connections ``ATTACH``\ es up to ten database files, so those connections count against the cap and are closed when idle like file connections. Adding or removing a database closes them too, so the next cross-database query sees the current set of databases.
+Default: ``8``
 
-Each pooled connection to a file in WAL mode uses three file descriptors (the database, its ``-wal`` and its ``-shm`` file); other databases use one. Under a small file descriptor limit such as ``ulimit -n 256``, a cap of 64 keeps pooled read connections below about 200 descriptors even if every database uses WAL, leaving room for write connections and Datasette itself.
+.. [[[end]]]
+
+Maximum simultaneous user database writer threads and their retained connections. Must be greater than zero. Each database preserves FIFO write order. When other databases are waiting, a writer closes its connection and releases its slot after its current callback completes. A retiring writer counts against the limit until its thread exits. Idle writers also release their slots under pressure, even if :ref:`setting_connection_idle_timeout_ms` is 0.
+
+The internal catalog has a separate writer, and file schema scans are serialized. Mutable isolated callbacks occupy a writer slot without retaining an additional ordinary file connection. Named in-memory writers occupy non-evictable slots to preserve their data; waiting writes can time out if these consume all capacity. In non-threaded mode, the same setting bounds retained writer connections, evicting an idle file connection before opening another.
+
+.. _setting_max_pending_writes:
+
+max_pending_writes
+~~~~~~~~~~~~~~~~~~
+
+.. [[[cog
+    setting_default(cog, "max_pending_writes")
+.. ]]]
+
+Default: ``256``
+
+.. [[[end]]]
+
+Maximum queued user writes across all databases, excluding callbacks already executing. Must be greater than zero. A full queue immediately raises ``datasette.write_budget.DatabaseQueueFull`` without accepting the write. Callers are not queued waiting for queue space. The internal catalog writer is separate.
+
+.. _setting_write_queue_timeout_ms:
+
+write_queue_timeout_ms
+~~~~~~~~~~~~~~~~~~~~~~
+
+.. [[[cog
+    setting_default(cog, "write_queue_timeout_ms")
+.. ]]]
+
+Default: ``5000``
+
+.. [[[end]]]
+
+Maximum milliseconds an accepted write can wait before execution starts. Must be greater than zero. Expired tasks are removed and raise ``datasette.write_budget.DatabaseAdmissionTimeout``. Queue-full and admission-timeout errors have ``execution_started=False`` and become HTTP 503 responses with ``Retry-After: 1`` and an error ``code`` in JSON responses.
+
+After execution starts, cancellation of the calling request does not cancel the write. Datasette never automatically retries a write callback. A connection-open failure can reclaim idle connections and retry opening once, before any preparation hook or callback runs. ``block=False`` calls return a task ID on acceptance; later failures are logged and reported to :ref:`plugin_hook_write_task_completed` while the calling event loop remains running.
 
 .. _setting_connection_idle_timeout_ms:
 
@@ -265,9 +304,9 @@ Opening a new connection costs around a millisecond (more if :ref:`plugin_hook_p
 
 Anything a function attaches to a connection itself - temporary tables, ``ATTACH`` statements, ``PRAGMA`` settings, functions registered with ``conn.create_function()`` - is lost when that connection is closed, and is never shared between pooled connections. Set that up in the :ref:`plugin_hook_prepare_connection` plugin hook instead, which runs once for every new connection.
 
-In-memory databases are exempt, since closing their only write connection would discard their contents. With :ref:`setting_num_sql_threads` set to 0 there are no threads to time out, so the write connection stays open until Datasette shuts down; read connections are still closed when idle.
+In-memory databases are exempt, since closing their only write connection would discard their contents. With :ref:`setting_num_sql_threads` set to 0 there are no threads to time out, so idle write connections are closed under capacity pressure or when Datasette shuts down; read connections are still closed when idle.
 
-Set this to 0 to keep read connections open until they are evicted by :ref:`setting_max_open_connections`, and write threads running until Datasette shuts down::
+Set this to 0 to keep read connections open until they are evicted by :ref:`setting_max_open_connections`, and write threads running until capacity pressure or Datasette shutdown::
 
     datasette mydatabase.db --setting connection_idle_timeout_ms 0
 

tests/test_api.py

@@ -759,7 +759,10 @@ async def test_settings_json(ds_client):
         "suggest_facets": True,
         "default_cache_ttl": 5,
         "num_sql_threads": 1,
-        "max_open_connections": 128,
+        "max_open_connections": 32,
+        "max_write_connections": 8,
+        "max_pending_writes": 256,
+        "write_queue_timeout_ms": 5000,
         "connection_idle_timeout_ms": 30000,
         "schema_watch_interval_ms": 1000,
         "default_schema_watch": "external",

tests/test_connection_limits.py

@@ -0,0 +1,92 @@
+"""Exercise real HTTP sockets and WAL files under a small OS descriptor limit."""
+
+import json
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="Requires RLIMIT_NOFILE")
+def test_500_wal_databases_with_http_and_writes_under_256_fds():
+    script = r"""
+import asyncio, concurrent.futures, json, resource, socket, sqlite3, tempfile, threading, time, urllib.request
+from pathlib import Path
+import psutil
+import uvicorn
+from datasette.app import Datasette
+
+async def main():
+    with tempfile.TemporaryDirectory() as directory:
+        paths = []
+        for i in range(500):
+            path = str(Path(directory) / f"db{i}.db")
+            conn = sqlite3.connect(path)
+            conn.execute("pragma journal_mode=wal")
+            conn.execute("create table t(id integer primary key)")
+            conn.close()
+            paths.append(path)
+        _, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (256, hard))
+        ds = Datasette(paths)
+        await ds.invoke_startup()
+        process = psutil.Process()
+        baseline_rss = process.memory_info().rss
+        peak = {"fds": 0, "rss": 0, "writers": 0}
+        stopped = threading.Event()
+        def monitor():
+            while not stopped.wait(0.002):
+                peak["fds"] = max(peak["fds"], process.num_fds())
+                peak["rss"] = max(peak["rss"], process.memory_info().rss)
+                peak["writers"] = max(peak["writers"], ds._write_budget.snapshot()["writers"])
+        watcher = threading.Thread(target=monitor)
+        watcher.start()
+        sock = socket.socket()
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+        server = uvicorn.Server(uvicorn.Config(ds.app(), log_level="error"))
+        thread = threading.Thread(target=lambda: server.run(sockets=[sock]))
+        thread.start()
+        try:
+            deadline = time.monotonic() + 10
+            while not server.started:
+                assert time.monotonic() < deadline
+                await asyncio.sleep(0.01)
+            def request(i):
+                with urllib.request.urlopen(f"http://127.0.0.1:{port}/db{i}/-/query.json?sql=select+count(*)+from+t", timeout=20) as response:
+                    assert response.status == 200
+                    assert json.load(response)["ok"]
+            def reads():
+                with concurrent.futures.ThreadPoolExecutor(max_workers=8) as clients:
+                    list(clients.map(request, range(500)))
+            async def writes():
+                for start in range(0, 500, 16):
+                    await asyncio.gather(*(ds.get_database(f"db{i}").execute_write("insert into t values (1)") for i in range(start, min(start + 16, 500))))
+            await asyncio.gather(asyncio.to_thread(reads), writes())
+            for i in range(500):
+                assert (await ds.get_database(f"db{i}").execute("select count(*) from t")).single_value() == 1
+            assert ds._read_pool.stats["peak_open"] <= 32
+            assert ds._write_budget.snapshot()["stats"]["peak_writers"] <= 8
+            print(json.dumps({"peak": peak, "baseline_rss": baseline_rss, "writes": 500, "http_reads": 500}), flush=True)
+        finally:
+            server.should_exit = True
+            await asyncio.to_thread(thread.join, 15)
+            assert not thread.is_alive()
+            stopped.set()
+            watcher.join()
+            sock.close()
+            ds.close()
+asyncio.run(main())
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        timeout=90,
+        check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    measurements = json.loads(result.stdout.strip().splitlines()[-1])
+    assert measurements["peak"]["fds"] < 256
+    assert measurements["peak"]["writers"] <= 8
+    print(measurements)

tests/test_connection_review_fixes.py

@@ -333,12 +333,11 @@ async def test_write_callback_cannot_close_write_connection(owned_db):
         await db.execute_write_fn(close_it)
     await db.execute_write("insert into t (v) values ('after')")
 
-    # Reaching the raw connection through a cursor still closes it: the
-    # write thread notices and reopens before the next write
+    # cursor.connection has the same lease protections as the connection.
     def close_raw(conn):
         conn.execute("select 1").connection.close()
 
-    with pytest.raises(sqlite3.ProgrammingError):
+    with pytest.raises(ConnectionLeaseError):
         await db.execute_write_fn(close_raw)
     for _ in range(3):
         await db.execute_write("insert into t (v) values ('again')")
@@ -747,13 +746,13 @@ async def test_cursors_are_closed_before_leaving_their_thread(owned_db):
         "insert into t (v) values (?)", [("a",), ("b",)]
     )
     assert cursor.rowcount == 2
-    with pytest.raises(sqlite3.ProgrammingError):
+    with pytest.raises(ConnectionLeaseError):
         cursor.fetchall()
     cursor = await db.execute_write_fn(
         lambda conn: conn.execute("insert into t (v) values ('c')")
     )
     assert cursor.lastrowid == 3
-    with pytest.raises(sqlite3.ProgrammingError):
+    with pytest.raises(ConnectionLeaseError):
         cursor.fetchone()
 
     # Cursors held by the frames of an exception raised by a callback are
@@ -770,7 +769,7 @@ async def test_cursors_are_closed_before_leaving_their_thread(owned_db):
     with pytest.raises(ValueError):
         await db.execute_write_fn(read_then_fail)
     for cursor in captured:
-        with pytest.raises(sqlite3.ProgrammingError):
+        with pytest.raises(ConnectionLeaseError):
             cursor.fetchone()
 
 

tests/test_internals_database.py

@@ -1264,10 +1264,11 @@ async def test_database_close_shuts_down_write_thread(tmpdir):
     await db.execute_write("insert into t (id) values (1)")
     assert db._write_thread is not None
     assert db._write_thread.is_alive()
+    thread = db._write_thread
     db.close()
-    # Wait briefly for the thread to exit — the sentinel should cause it to return.
-    db._write_thread.join(timeout=5)
-    assert not db._write_thread.is_alive()
+    thread.join(timeout=5)
+    assert not thread.is_alive()
+    assert db._write_thread is None
     ds._internal_database.close()
 
 
@@ -1314,21 +1315,17 @@ async def test_close_releases_memory_connections(num_sql_threads, named):
     ds = Datasette(memory=True, settings={"num_sql_threads": num_sql_threads})
     db = ds.add_memory_database(uuid.uuid4().hex) if named else ds.get_database()
     leased = await db.execute_fn(lambda conn: conn)
-    # Read connections are leased: the proxy is unusable once the callback
-    # returns, so reach the underlying connection through a cursor
-    read_connection = await db.execute_fn(
-        lambda conn: conn.execute("select 1").connection
-    )
     leased_write = await db.execute_write_fn(lambda conn: conn)
-    write_connection = await db.execute_write_fn(
-        lambda conn: conn.execute("select 1").connection
-    )
+    # Inspect the owned raw connections to verify physical closure as well
+    # as lease expiry. cursor.connection intentionally no longer leaks them.
+    connections = list(db._all_connections)
+    assert len(connections) >= 2
     with pytest.raises(ConnectionLeaseError):
         leased.execute("select 1")
     with pytest.raises(ConnectionLeaseError):
         leased_write.execute("select 1")
     ds.close()
-    for conn in (read_connection, write_connection):
+    for conn in connections:
         with pytest.raises(sqlite3.ProgrammingError, match="closed"):
             conn.execute("select 1")
 

tests/test_plugins.py

@@ -48,6 +48,11 @@ def test_plugin_hooks_have_tests(plugin_hook):
     assert ok, f"Plugin hook is missing tests: {plugin_hook}"
 
 
+def test_hook_write_task_completed():
+    # Success and expiry delivery are exercised in tests/test_write_budget.py.
+    assert "write_task_completed" in dir(pm.hook)
+
+
 def test_hook_jump_items_sql():
     # Detailed behavior is covered in tests/test_jump.py.
     assert "jump_items_sql" in dir(pm.hook)

tests/test_read_pool.py

@@ -178,13 +178,13 @@ def test_idle_reaper_closes_connections_without_event_loop(tmp_path):
     "num_sql_threads,configured,expected",
     [
         (3, 128, 128),
-        (3, 5, 12),
-        (8, 20, 32),
+        (3, 5, 5),
+        (8, 20, 20),
         (0, 1, 1),
         (3, 0, 0),
     ],
 )
-def test_max_open_connections_clamped_to_four_times_threads(
+def test_max_open_connections_respects_configured_limit(
     num_sql_threads, configured, expected
 ):
     ds = Datasette(
@@ -209,29 +209,33 @@ def test_connection_idle_timeout_ms_setting():
 
 
 @pytest.mark.asyncio
-async def test_soft_cap_when_every_connection_is_leased(tmp_path):
-    # Only reachable with a cap below the thread count, which the setting
-    # clamp prevents: open one more connection rather than wait, and close
-    # it again on release
+async def test_hard_cap_when_every_connection_is_leased(tmp_path):
     paths = _make_dbs(tmp_path, 3)
-    ds = Datasette(paths, settings={"num_sql_threads": 3})
-    pool = ds._read_pool
-    pool.max_open = 1
-    barrier = threading.Barrier(3, timeout=5)
+    ds = Datasette(paths, settings={"num_sql_threads": 3, "max_open_connections": 1})
+    entered = threading.Event()
+    release = threading.Event()
 
     def slow(conn):
-        barrier.wait()
+        entered.set()
+        assert release.wait(5)
         return conn.execute("select 1").fetchone()[0]
 
-    results = await asyncio.gather(
-        *[ds.get_database(f"db{i}").execute_fn(slow) for i in range(3)]
-    )
-    assert results == [1, 1, 1]
-    assert pool.stats["exceeded_cap"] >= 2
-    assert pool.stats["peak_open"] == 3
-    # Back under the cap once everything has been returned
-    assert pool.snapshot()["open"] <= 1
-    ds.close()
+    tasks = [
+        asyncio.create_task(ds.get_database(f"db{i}").execute_fn(slow))
+        for i in range(3)
+    ]
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        await asyncio.sleep(0.05)
+        assert ds._read_pool.snapshot()["open"] == 1
+        assert not any(task.done() for task in tasks)
+        release.set()
+        assert await asyncio.gather(*tasks) == [1, 1, 1]
+        assert ds._read_pool.stats["peak_open"] == 1
+    finally:
+        release.set()
+        await asyncio.gather(*tasks, return_exceptions=True)
+        ds.close()
 
 
 @pytest.mark.asyncio

tests/test_write_budget.py

@@ -0,0 +1,388 @@
+import asyncio
+import sqlite3
+import threading
+import time
+
+import pytest
+
+from datasette.app import Datasette
+from datasette.database import DatasetteClosedError
+from datasette.write_budget import (
+    DatabaseAdmissionTimeout,
+    DatabaseQueueFull,
+    DatabaseReentrancyError,
+    DatabaseResourceError,
+)
+
+
+def make_databases(tmp_path, count=3, **settings):
+    paths = []
+    for i in range(count):
+        path = tmp_path / f"budget{i}.db"
+        with sqlite3.connect(path) as conn:
+            conn.execute("create table t(id)")
+        conn.close()
+        paths.append(str(path))
+    ds = Datasette(paths, settings={"schema_watch_interval_ms": 0, **settings})
+    return ds, [ds.get_database(f"budget{i}") for i in range(count)]
+
+
+async def wait_for(predicate):
+    deadline = time.monotonic() + 5
+    while not predicate():
+        assert time.monotonic() < deadline
+        await asyncio.sleep(0.005)
+
+
+@pytest.mark.asyncio
+async def test_many_writers_bounded_fifo_and_fairness(tmp_path):
+    ds, dbs = make_databases(
+        tmp_path, 30, max_write_connections=2, max_pending_writes=200
+    )
+    entered, release = threading.Event(), threading.Event()
+    order = []
+
+    def slow(conn):
+        entered.set()
+        assert release.wait(5)
+
+    first = asyncio.create_task(dbs[0].execute_write_fn(slow))
+    tasks = []
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        for seq in range(4):
+            for db in dbs:
+
+                def write(conn, name=db.name, seq=seq):
+                    order.append((name, seq))
+                    conn.execute("insert into t values (?)", (seq,))
+
+                tasks.append(asyncio.create_task(db.execute_write_fn(write)))
+        await asyncio.sleep(0.03)
+        release.set()
+        await asyncio.gather(first, *tasks)
+        assert ds._write_budget.snapshot()["stats"]["peak_writers"] == 2
+        for db in dbs:
+            assert [seq for name, seq in order if name == db.name] == list(range(4))
+            assert [
+                r[0] for r in (await db.execute("select id from t order by rowid")).rows
+            ] == list(range(4))
+        assert order.index((dbs[-1].name, 0)) < order.index((dbs[0].name, 3))
+    finally:
+        release.set()
+        await asyncio.gather(first, *tasks, return_exceptions=True)
+        ds.close()
+    await wait_for(lambda: ds._write_budget.snapshot()["writers"] == 0)
+
+
+@pytest.mark.asyncio
+async def test_queue_full_expiration_and_internal_progress(tmp_path):
+    ds, dbs = make_databases(
+        tmp_path,
+        max_write_connections=1,
+        max_pending_writes=1,
+        write_queue_timeout_ms=100,
+    )
+    entered, release = threading.Event(), threading.Event()
+    ran = []
+
+    def slow(conn):
+        entered.set()
+        assert release.wait(5)
+        conn.execute("insert into t values (1)")
+
+    first = asyncio.create_task(dbs[0].execute_write_fn(slow))
+    queued = None
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        queued = asyncio.create_task(
+            dbs[1].execute_write_fn(lambda conn: ran.append(True))
+        )
+        await wait_for(lambda: ds._write_budget.snapshot()["pending"] == 1)
+        with pytest.raises(DatabaseQueueFull) as exc:
+            await dbs[2].execute_write_fn(lambda conn: ran.append(True))
+        assert exc.value.execution_started is False
+        # Internal catalog work must not compete for the single user slot.
+        assert (
+            await asyncio.wait_for(
+                ds._internal_database.execute_write_fn(lambda conn: 42), 2
+            )
+            == 42
+        )
+        with pytest.raises(DatabaseAdmissionTimeout):
+            await queued
+        assert not ran
+        assert dbs[1]._write_queue.qsize() == 0
+        release.set()
+        await first
+        await dbs[2].execute_write("insert into t values (2)")
+        assert ds._write_budget.snapshot()["stats"]["peak_pending"] == 1
+    finally:
+        release.set()
+        await asyncio.gather(
+            first, *([queued] if queued else []), return_exceptions=True
+        )
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_close_waiting_database_releases_pending(tmp_path):
+    ds, dbs = make_databases(tmp_path, max_write_connections=1)
+    entered, release = threading.Event(), threading.Event()
+
+    def slow(conn):
+        entered.set()
+        assert release.wait(5)
+
+    first = asyncio.create_task(dbs[0].execute_write_fn(slow))
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        pending = asyncio.create_task(
+            dbs[1].execute_write_fn(lambda conn: pytest.fail("closed callback ran"))
+        )
+        await wait_for(lambda: ds._write_budget.snapshot()["pending"] == 1)
+        dbs[1].close()
+        with pytest.raises(DatasetteClosedError):
+            await pending
+        assert ds._write_budget.snapshot()["pending"] == 0
+    finally:
+        release.set()
+        await first
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("threads", [0, 3])
+async def test_memory_pins_capacity_and_recovers_after_close(tmp_path, threads):
+    ds, dbs = make_databases(
+        tmp_path,
+        max_write_connections=1,
+        write_queue_timeout_ms=50,
+        num_sql_threads=threads,
+    )
+    memory = ds.add_memory_database("budget_memory")
+    try:
+        await memory.execute_write("create table t(id)")
+        with pytest.raises(DatabaseResourceError):
+            await dbs[0].execute_write("insert into t values (1)")
+        assert (
+            await memory.execute_write_fn(
+                lambda conn: conn.execute("select count(*) from t").fetchone()[0]
+            )
+            == 0
+        )
+        memory.close()
+        await dbs[0].execute_write("insert into t values (1)")
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("threads", [0, 3])
+async def test_isolated_writer_does_not_hold_extra_file_connection(tmp_path, threads):
+    ds, dbs = make_databases(tmp_path, max_write_connections=1, num_sql_threads=threads)
+    try:
+        for db in dbs:
+            await db.execute_write("insert into t values (1)")
+            assert (
+                await db.execute_isolated_fn(
+                    lambda conn, db=db: len(db._all_connections)
+                )
+                == 1
+            )
+        if threads == 0:
+            assert sum(db._write_connection is not None for db in dbs) <= 1
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["read", "write"])
+async def test_reentrant_database_call_fails_without_deadlock(tmp_path, kind):
+    ds, dbs = make_databases(tmp_path, max_write_connections=1, max_open_connections=1)
+    try:
+
+        def callback(conn):
+            return asyncio.run(dbs[1].execute_write_fn(lambda conn: 1))
+
+        method = dbs[0].execute_fn if kind == "read" else dbs[0].execute_write_fn
+        with pytest.raises(DatabaseReentrancyError):
+            await asyncio.wait_for(method(callback), 2)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_admission_errors_are_http_503(tmp_path, monkeypatch):
+    ds, dbs = make_databases(tmp_path)
+    await ds.invoke_startup()
+
+    async def fail(*args, **kwargs):
+        raise DatabaseQueueFull("Too many queued database writes")
+
+    monkeypatch.setattr(dbs[0], "execute", fail)
+    try:
+        response = await ds.client.get("/budget0/-/query.json?sql=select+1")
+        assert response.status_code == 503, response.text
+        assert response.headers["retry-after"] == "1"
+        assert response.json()["code"] == "database_queue_full"
+        assert response.json()["execution_started"] is False
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("threads", [0, 3])
+async def test_unfinished_explicit_transaction_is_rolled_back(tmp_path, threads):
+    from datasette.connection_pool import ConnectionLeaseError
+
+    ds, dbs = make_databases(tmp_path, num_sql_threads=threads)
+    try:
+        with pytest.raises(ConnectionLeaseError, match="rolled back"):
+            await dbs[0].execute_write_fn(
+                lambda conn: conn.execute("insert into t values (1)"), transaction=False
+            )
+        await dbs[0].execute_write("insert into t values (2)")
+        assert [row[0] for row in (await dbs[0].execute("select id from t")).rows] == [
+            2
+        ]
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_nonblocking_expiration_notifies_plugin(tmp_path):
+    from datasette import hookimpl
+    from datasette.plugins import pm
+
+    notifications = []
+
+    class Plugin:
+        @hookimpl
+        async def write_task_completed(self, database, task_id, exception):
+            notifications.append((database, task_id, exception))
+
+    plugin = Plugin()
+    pm.register(plugin)
+    ds, dbs = make_databases(
+        tmp_path, max_write_connections=1, write_queue_timeout_ms=50
+    )
+    entered, release = threading.Event(), threading.Event()
+
+    def slow(conn):
+        entered.set()
+        assert release.wait(5)
+
+    first = asyncio.create_task(dbs[0].execute_write_fn(slow))
+    try:
+        assert await asyncio.to_thread(entered.wait, 5)
+        task_id = await dbs[1].execute_write_fn(
+            lambda conn: pytest.fail("expired callback ran"), block=False
+        )
+        await wait_for(lambda: notifications)
+        assert notifications[0][:2] == (dbs[1].name, task_id)
+        assert isinstance(notifications[0][2], DatabaseAdmissionTimeout)
+        release.set()
+        await first
+        successful_id = await dbs[1].execute_write_fn(lambda conn: 42, block=False)
+        await wait_for(lambda: len(notifications) == 2)
+        assert notifications[1] == (dbs[1].name, successful_id, None)
+    finally:
+        release.set()
+        await first
+        ds.close()
+        pm.unregister(plugin)
+
+
+@pytest.mark.asyncio
+async def test_connection_open_pressure_retries_before_callback(tmp_path, monkeypatch):
+    import errno
+
+    from datasette.utils import sqlite3 as sqlite_module
+
+    ds, dbs = make_databases(tmp_path)
+    await ds.invoke_startup()
+    await dbs[1].execute("select 1")
+    original = sqlite_module.connect
+    calls = []
+
+    def connect(*args, **kwargs):
+        if "budget0.db" in str(args[0]):
+            calls.append(1)
+            if len(calls) == 1:
+                raise OSError(errno.EMFILE, "too many files")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(sqlite_module, "connect", connect)
+    ran = []
+    try:
+        await dbs[0].execute_write_fn(lambda conn: ran.append(1))
+        assert calls == [1, 1]
+        assert ran == [1]
+        assert not dbs[1]._read_pool_state.idle
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("threads", [0, 3])
+async def test_resource_error_after_execution_is_not_an_admission_error(
+    tmp_path, threads, monkeypatch
+):
+    ds, dbs = make_databases(tmp_path, num_sql_threads=threads)
+    db = dbs[0]
+    await ds.invoke_startup()
+    try:
+
+        def committed_then_failed(conn):
+            conn.execute("insert into t values (1)")
+            conn.commit()
+            raise DatabaseQueueFull("A different operation could not be admitted")
+
+        with pytest.raises(RuntimeError, match="callback started") as exc:
+            await db.execute_write_fn(committed_then_failed, transaction=False)
+        assert not isinstance(exc.value, DatabaseResourceError)
+        assert isinstance(exc.value.__cause__, DatabaseQueueFull)
+        assert (await db.execute("select count(*) from t")).single_value() == 1
+
+        async def refresh_failed(*args):
+            raise DatabaseResourceError("Catalog connection unavailable")
+
+        monkeypatch.setattr(ds._schema_watcher, "after_write", refresh_failed)
+        db._watch_state.needs_scan = True
+        with pytest.raises(RuntimeError, match="Write completed") as exc:
+            await db.execute_write("insert into t values (2)")
+        assert not isinstance(exc.value, DatabaseResourceError)
+        assert (await db.execute("select count(*) from t")).single_value() == 2
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_retiring_writer_keeps_its_slot_until_connection_closes(
+    tmp_path, monkeypatch
+):
+    ds, dbs = make_databases(tmp_path, max_write_connections=1)
+    closing, release = threading.Event(), threading.Event()
+    original = dbs[0]._forget_connection
+
+    def slow_close(conn):
+        closing.set()
+        assert release.wait(5)
+        original(conn)
+
+    await dbs[0].execute_write("insert into t values (1)")
+    monkeypatch.setattr(dbs[0], "_forget_connection", slow_close)
+    pending = asyncio.create_task(dbs[1].execute_write("insert into t values (2)"))
+    try:
+        assert await asyncio.to_thread(closing.wait, 5)
+        assert not pending.done()
+        assert dbs[1]._write_thread is None
+        assert ds._write_budget.snapshot()["writers"] == 1
+        release.set()
+        await pending
+        assert ds._write_budget.snapshot()["stats"]["peak_writers"] == 1
+    finally:
+        release.set()
+        await asyncio.gather(pending, return_exceptions=True)
+        ds.close()

tests/test_write_thread_idle.py

@@ -323,6 +323,14 @@ def test_idle_exit_race_stress_many_event_loops(tmp_path):
         async def go():
             for seq in range(rounds):
                 barrier.wait(timeout=30)
+                # Force some completed retirements as well as racing the
+                # boundary. Scheduler latency must not decide coverage.
+                if seq % 4 == 0:
+                    deadline = time.monotonic() + 5
+                    while db._write_thread is not None:
+                        assert time.monotonic() < deadline
+                        await asyncio.sleep(0.001)
+                    barrier.wait(timeout=30)
                 await asyncio.sleep(0.001 * rng.uniform(0.5, 1.5))
                 await asyncio.wait_for(
                     db.execute_write_fn(_recorder(log, w, seq)), timeout=10