Fix execute_isolated_fn() against immutable databases

execute_isolated_fn() always opened its temporary connection with
write=True, which is not allowed for immutable databases - so APIs
that rely on it, like SQL analysis when storing a query, failed.

An immutable database can never receive writes, so there is no write
queue to block: in that case the function now opens a read-only
connection and runs it on the executor, bypassing the write thread
entirely. Mutable databases keep the existing write-thread behavior.

Also fixed a latent bug in the write thread where a connect() failure
for an isolated task would crash the thread instead of delivering the
exception back to the caller.

Closes #2768

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_internals_database.py

@@ -863,6 +863,39 @@ async def test_execute_isolated(db, disable_threads):
     assert not await db.execute_isolated_fn(table_exists_checker("created_by_isolated"))
 
 
+@pytest.mark.asyncio
+async def test_execute_isolated_connect_failure_does_not_kill_write_thread():
+    # A connect() failure for an isolated task should be returned to the
+    # caller as an exception, not crash the write thread
+    class ConnectError(Exception):
+        pass
+
+    ds = Datasette(memory=True)
+    db = ds.add_memory_database("test_isolated_connect_failure")
+    # Start the write thread with a healthy dedicated write connection
+    await db.execute_write("create table dogs (id integer primary key)")
+
+    original_connect = db.connect
+
+    def broken_connect(write=False):
+        raise ConnectError("Could not connect")
+
+    db.connect = broken_connect
+    try:
+        with pytest.raises(ConnectError):
+            await asyncio.wait_for(db.execute_isolated_fn(lambda conn: None), timeout=2)
+    finally:
+        db.connect = original_connect
+
+    # Write thread should still be alive and processing tasks
+    assert db._write_thread.is_alive()
+    await db.execute_write("insert into dogs (id) values (1)")
+    count = await db.execute_isolated_fn(
+        lambda conn: conn.execute("select count(*) from dogs").fetchone()[0]
+    )
+    assert count == 1
+
+
 @pytest.mark.asyncio
 async def test_analyze_sql():
     ds = Datasette(memory=True)

tests/test_queries.py

@@ -9,7 +9,7 @@ from datasette.app import Datasette
 from datasette.resources import DatabaseResource, QueryResource
 from datasette.stored_queries import StoredQuery, StoredQueryPage
 from datasette.utils.asgi import Forbidden
-from datasette.utils.sqlite import supports_returning
+from datasette.utils.sqlite import sqlite3, supports_returning
 
 requires_sqlite_returning = pytest.mark.skipif(
     not supports_returning(), reason="SQLite does not support RETURNING"
@@ -593,6 +593,38 @@ async def test_query_store_api_creates_read_only_query():
     assert data["query"]["owner_id"] == "root"
 
 
+@pytest.mark.asyncio
+async def test_query_store_api_creates_query_for_immutable_database(tmp_path):
+    db_path = tmp_path / "immutable.db"
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("create table dogs (id integer primary key, name text)")
+    conn.commit()
+    conn.close()
+
+    ds = Datasette([], immutables=[str(db_path)], default_deny=True)
+    ds.root_enabled = True
+    await ds.invoke_startup()
+
+    response = await ds.client.post(
+        "/immutable/-/queries/store",
+        actor={"id": "root"},
+        json={
+            "query": {
+                "name": "by_name",
+                "sql": "select * from dogs where name = :name",
+            }
+        },
+    )
+
+    ds.close()
+    assert response.status_code == 201
+    data = response.json()
+    assert data["ok"] is True
+    assert data["query"]["name"] == "by_name"
+    assert data["query"]["parameters"] == ["name"]
+    assert data["query"]["is_write"] is False
+
+
 @pytest.mark.asyncio
 async def test_query_list_and_definition_api():
     ds = Datasette(memory=True)