From 87ba82de92709932a8add6b36a7d76f66d728598 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 04:57:25 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01EwkLo2rJWidpkDVT3HkvNv --- datasette/app.py | 22 ++++++-- datasette/database.py | 3 +- datasette/schema_watcher.py | 14 +++-- tests/test_connection_leases.py | 94 +++++++++++++++++++++++++++++++++ 4 files changed, 124 insertions(+), 9 deletions(-) diff --git a/datasette/app.py b/datasette/app.py index ad1dd23e..e090a4f1 100644 --- a/datasette/app.py +++ b/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() diff --git a/datasette/database.py b/datasette/database.py index 0faf7e97..ef60bc57 100644 --- a/datasette/database.py +++ b/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() diff --git a/datasette/schema_watcher.py b/datasette/schema_watcher.py index ea0ea3b5..d7c47bbf 100644 --- a/datasette/schema_watcher.py +++ b/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: diff --git a/tests/test_connection_leases.py b/tests/test_connection_leases.py index c803e5e7..550ea280 100644 --- a/tests/test_connection_leases.py +++ b/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"