Keep num_sql_threads=0 thread-free and refresh_schemas loop-agnostic

- SchemaWatcher ran its stat sweeps and catalog scans with
  asyncio.to_thread(), which starts threads: with num_sql_threads=0 they
  now run inline, so Datasette still starts where threads cannot be
  started at all (Pyodide). New test simulates that in a subprocess with
  Thread.start() raising, covering reads, writes, isolated functions,
  owned and external schema detection and a scratch database.
- refresh_schemas(force=True) waited on an asyncio.Lock, which binds to
  whichever event loop first contends for it. It is now a threading.Lock
  only ever acquired without blocking; a forced refresh that finds one
  running runs its own sweep instead of waiting.
- Database.close() no longer copies the set of in-use write connections
  (membership tests only), which could race a running write thread.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv

datasette/app.py

@@ -487,8 +487,11 @@ class Datasette:
         self._column_types = {}  # .invoke_startup() will populate this
         self._setup_db_done = False
         self._suppress_background_tasks = False
+        # A threading.Lock only ever acquired without blocking: one
+        # Datasette can be driven by several event loops, and an
+        # asyncio.Lock binds to whichever loop first waits on it
+        self._refresh_schemas_lock = threading.Lock()
         try:
-            self._refresh_schemas_lock = asyncio.Lock()
             self._startup_lock = asyncio.Lock()
         except RuntimeError as rex:
             # Workaround for intermittent test failure, see:
@@ -496,7 +499,6 @@ class Datasette:
             if "There is no current event loop in thread" in str(rex):
                 loop = asyncio.new_event_loop()
                 asyncio.set_event_loop(loop)
-                self._refresh_schemas_lock = asyncio.Lock()
                 self._startup_lock = asyncio.Lock()
             else:
                 raise
@@ -755,10 +757,20 @@ class Datasette:
         ):
             return
         self._last_schema_refresh = time.monotonic()
-        if self._refresh_schemas_lock.locked() and not force:
-            return
-        async with self._refresh_schemas_lock:
+        if not self._refresh_schemas_lock.acquire(blocking=False):
+            if not force:
+                return
+            # A refresh is already running, maybe on another event loop.
+            # A forced refresh must see changes made before it was called,
+            # so run another sweep now rather than wait for that one (sweeps
+            # are safe to overlap: catalog writes go through the internal
+            # database's write queue)
             await self._refresh_schemas()
+            return
+        try:
+            await self._refresh_schemas()
+        finally:
+            self._refresh_schemas_lock.release()
 
     async def _refresh_schemas(self, *, background=False):
         internal_db = self.get_internal_database()

datasette/database.py

@@ -354,7 +354,8 @@ class Database:
                 sys.stderr.flush()
         # A write thread that is still running owns its connection and closes
         # it when it finishes; closing it here could free it mid-query
-        in_use = set(self._write_thread_connections) if still_running else set()
+        # (membership tests only: copying the set could race the thread)
+        in_use = self._write_thread_connections if still_running else ()
         for future in pending_execute_futures:
             try:
                 future.result()

datasette/schema_watcher.py

@@ -344,6 +344,14 @@ class SchemaWatcher:
     # ------------------------------------------------------------------
     # task helpers
     # ------------------------------------------------------------------
+    async def _off_loop(self, fn, *args):
+        """Run blocking stat()/scan work in a worker thread - or inline with
+        num_sql_threads=0, which must work where threads cannot be started
+        at all (Pyodide)."""
+        if self.ds.executor is None:
+            return fn(*args)
+        return await asyncio.to_thread(fn, *args)
+
     def _spawn(self, coro):
         try:
             loop = asyncio.get_running_loop()
@@ -511,7 +519,7 @@ class SchemaWatcher:
         outcomes = []
         file_candidates = [s for s in candidates if s.is_file]
         if file_candidates:
-            outcomes = await asyncio.to_thread(self._sweep_sync, file_candidates)
+            outcomes = await self._off_loop(self._sweep_sync, file_candidates)
         for state in candidates:
             if not state.is_file and state.db.memory_name and not state.removed:
                 outcomes.append(await self._memory_pragma_check(state))
@@ -779,7 +787,7 @@ class SchemaWatcher:
 
         async def one(chunk):
             async with semaphore:
-                results = await asyncio.to_thread(self._scan_chunk_sync, chunk)
+                results = await self._off_loop(self._scan_chunk_sync, chunk)
             await self._store(results)
 
         if len(chunks) == 1:
@@ -914,7 +922,7 @@ class SchemaWatcher:
             persisted = {row["database_name"]: row for row in rows.rows}
         to_scan = []
         if persisted:
-            reuse = await asyncio.to_thread(self._reusable_sync, states, persisted)
+            reuse = await self._off_loop(self._reusable_sync, states, persisted)
             for state in states:
                 hit = reuse.get(state.name)
                 if hit is None:

tests/test_connection_leases.py

@@ -187,3 +187,97 @@ async def test_close_waits_for_immutable_isolated_fn(tmp_path):
     db.close()
     assert await task == 1
     ds.close()
+
+
+NO_THREADS_SCRIPT = r"""
+import asyncio, os, sqlite3, sys, threading
+
+def no_threads(self):
+    raise RuntimeError("can't start new thread")
+
+threading.Thread.start = no_threads
+
+from datasette.app import Datasette
+from datasette.database import Database
+
+path, scratch = sys.argv[1], sys.argv[2]
+conn = sqlite3.connect(path)
+conn.execute("create table t (id integer primary key, v text)")
+conn.execute("insert into t (v) values ('x')")
+conn.commit()
+conn.close()
+
+
+async def main():
+    ds = Datasette(
+        [path],
+        settings={
+            "num_sql_threads": 0,
+            "connection_idle_timeout_ms": 50,
+            "schema_watch_interval_ms": 50,
+        },
+    )
+    await ds.invoke_startup()
+    db = ds.get_database("nothreads")
+    assert (await db.execute("select count(*) from t")).single_value() == 1
+    await db.execute_write("insert into t (v) values ('y')")
+    await db.execute_write("create table t2 (id)")
+    tables = await ds.get_internal_database().execute(
+        "select table_name from catalog_tables where database_name = 'nothreads' "
+        "order by table_name"
+    )
+    assert [r[0] for r in tables.rows] == ["t", "t2"], tables.rows
+    assert await db.execute_isolated_fn(
+        lambda conn: conn.execute("select count(*) from t").fetchone()[0]
+    ) == 2
+    await db.execute_write("insert into t (v) values ('z')", block=False)
+    response = await ds.client.get("/nothreads/t.json?_shape=array")
+    assert response.status_code == 200, response.text
+    assert len(response.json()) == 3
+    # External change detected by the polling task, inline on the loop
+    other = sqlite3.connect(path)
+    other.execute("create table external_t (id)")
+    other.commit()
+    other.close()
+    for _ in range(100):
+        await asyncio.sleep(0.02)
+        tables = await ds.get_internal_database().execute(
+            "select 1 from catalog_tables where table_name = 'external_t'"
+        )
+        if tables.rows:
+            break
+    else:
+        raise AssertionError("external change not seen")
+    # Scratch database added later, created by its first write
+    sdb = ds.add_database(Database(ds, path=scratch), name="scratch")
+    await sdb.execute_write("create table s (id)")
+    assert os.path.exists(scratch)
+    ds.close()
+    print("ok", threading.active_count())
+
+
+asyncio.run(main())
+"""
+
+
+def test_num_sql_threads_zero_starts_no_threads(tmp_path):
+    # Simulates Pyodide, where no thread can be started: nothing in the
+    # read pool, write path or schema watcher may need one
+    import subprocess
+    import sys
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            NO_THREADS_SCRIPT,
+            str(tmp_path / "nothreads.db"),
+            str(tmp_path / "scratch.db"),
+        ],
+        capture_output=True,
+        text=True,
+        timeout=60,
+        check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert result.stdout.strip() == "ok 1"