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