execute_write(transaction=False) parameter, plus fix for errors inside tasks

Ensure a write inside a failing Datasette task never becomes visible. Refs #2831

datasette/database.py

@@ -246,6 +246,7 @@ class Database:
         request=None,
         return_all=False,
         returning_limit=EXECUTE_WRITE_RETURNING_LIMIT,
+        transaction=True,
     ):
         self._check_not_closed()
         if returning_limit < 0:
@@ -258,7 +259,9 @@ class Database:
             )
 
         with trace("sql", database=self.name, sql=sql.strip(), params=params):
-            results = await self.execute_write_fn(_inner, block=block, request=request)
+            results = await self.execute_write_fn(
+                _inner, block=block, request=request, transaction=transaction
+            )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
@@ -348,6 +351,7 @@ class Database:
                 self.ds._prepare_connection(self._write_connection, self.name)
             if transaction:
                 with self._write_connection:
+                    self._write_connection.execute("BEGIN IMMEDIATE")
                     result = fn(self._write_connection)
             else:
                 result = fn(self._write_connection)
@@ -477,6 +481,7 @@ class Database:
                 try:
                     if task.transaction:
                         with conn:
+                            conn.execute("BEGIN IMMEDIATE")
                             result = task.fn(conn)
                     else:
                         result = task.fn(conn)

datasette/views/database.py

@@ -643,8 +643,15 @@ class QueryView(View):
         ok = None
         redirect_url = None
         try:
+            execute_write_kwargs = {"request": request}
+            if stored_query.is_trusted:
+                analysis = await db.analyze_sql(stored_query.sql, params_for_query)
+                if any(
+                    operation.operation == "vacuum" for operation in analysis.operations
+                ):
+                    execute_write_kwargs["transaction"] = False
             cursor = await db.execute_write(
-                stored_query.sql, params_for_query, request=request
+                stored_query.sql, params_for_query, **execute_write_kwargs
             )
             # success message can come from on_success_message or on_success_message_sql
             message = None

docs/internals.rst

@@ -2023,8 +2023,8 @@ Example usage:
 
 .. _database_execute_write:
 
-await db.execute_write(sql, params=None, block=True, request=None, return_all=False, returning_limit=10)
---------------------------------------------------------------------------------------------------------
+await db.execute_write(sql, params=None, block=True, request=None, return_all=False, returning_limit=10, transaction=True)
+--------------------------------------------------------------------------------------------------------------------------
 
 SQLite only allows one database connection to write at a time. Datasette handles this for you by maintaining a queue of writes to be executed against a given database. Plugins can submit write operations to this queue and they will be executed in the order in which they are received.
 
@@ -2059,7 +2059,9 @@ If you need to retrieve every row returned by a statement, pass ``return_all=Tru
 
 If you pass ``block=False`` this behavior changes to "fire and forget" - queries will be added to the write queue and executed in a separate thread while your code can continue to do other things. The method will return a UUID representing the queued task.
 
-Each call to ``execute_write()`` will be executed inside a transaction.
+Each call to ``execute_write()`` will be executed inside a transaction. Pass
+``transaction=False`` for statements such as ``VACUUM`` that cannot run inside
+a transaction.
 
 .. _database_execute_write_script:
 

tests/test_internals_database.py

@@ -11,6 +11,7 @@ from datasette.database import _deliver_write_result
 from datasette.utils.sqlite import sqlite3, supports_returning
 from datasette.utils import Column
 import pytest
+import sqlite_utils
 import time
 import uuid
 
@@ -718,6 +719,41 @@ async def test_execute_write_fn_exception(db):
         await db.execute_write_fn(write_fn)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("num_sql_threads", (0, 1))
+async def test_execute_write_fn_sqlite_utils_transaction(tmp_path, num_sql_threads):
+    # A write inside a failing Datasette task must never become visible or
+    # survive the rollback. Exercise both the synchronous and writer-thread
+    # paths against a file-backed database so a second connection can observe
+    # committed state independently.
+    db_path = tmp_path / "test.db"
+    sqlite3.connect(db_path).close()
+    ds = Datasette([str(db_path)], settings={"num_sql_threads": num_sql_threads})
+    db = ds.get_database("test")
+    await db.execute_write("create table items (id integer primary key)")
+    # This reader is used inside the write callback, which may run on another
+    # thread, but it is never accessed concurrently.
+    reader = sqlite3.connect(db_path, check_same_thread=False)
+
+    def insert_then_fail(conn):
+        # Datasette must open the outer transaction before sqlite-utils writes.
+        assert conn.in_transaction
+        sqlite_utils.Database(conn)["items"].insert({"id": 1})
+        # If sqlite-utils committed its own transaction, this would return 1.
+        assert reader.execute("select count(*) from items").fetchone()[0] == 0
+        # Simulate a later step failing after the sqlite-utils write succeeded.
+        raise ValueError("deliberate")
+
+    try:
+        with pytest.raises(ValueError, match="deliberate"):
+            await db.execute_write_fn(insert_then_fail)
+        # The outer transaction must roll back the sqlite-utils write as well.
+        assert reader.execute("select count(*) from items").fetchone()[0] == 0
+    finally:
+        reader.close()
+        db.close()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("param_name", ["conn", "connection", "db", "c"])
 async def test_execute_write_fn_accepts_any_single_param_name(db, param_name):