Add busy_timeout_ms and journal_mode to expected /-/settings.json

Refs #2831

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N76afGMhBRQk528VF1LTpR

datasette/app.py

@@ -220,6 +220,16 @@ SETTINGS = (
         "Number of threads in the thread pool for executing SQLite queries",
     ),
     Setting("sql_time_limit_ms", 1000, "Time limit for a SQL query in milliseconds"),
+    Setting(
+        "busy_timeout_ms",
+        5000,
+        "How long SQLite waits for a locked database file before giving up",
+    ),
+    Setting(
+        "journal_mode",
+        "",
+        "Journal mode to set on mutable database files, e.g. wal - leave blank to use each file's existing mode",
+    ),
     Setting(
         "default_facet_size", 30, "Number of values to return for requested facets"
     ),
@@ -478,7 +488,11 @@ class Datasette:
         if internal is None:
             self._internal_database = Database(self, is_temp_disk=True)
         else:
-            self._internal_database = Database(self, path=internal, mode="rwc")
+            # WAL for the same reason as the temporary internal database:
+            # the catalog can be read while it is being rewritten
+            self._internal_database = Database(
+                self, path=internal, mode="rwc", enable_wal=True
+            )
         self._internal_database.name = INTERNAL_DB_NAME
 
         self.cache_headers = cache_headers
@@ -1565,6 +1579,10 @@ class Datasette:
                     conn.execute("SELECT load_extension(?)", [extension])
         if self.setting("cache_size_kb"):
             conn.execute(f"PRAGMA cache_size=-{self.setting('cache_size_kb')}")
+        # Consistent trigger semantics on every connection - sqlite-utils
+        # would otherwise enable this on just the write connection, as a
+        # side effect of the first sqlite-utils based write
+        conn.execute("PRAGMA recursive_triggers=on")
         # pylint: disable=no-member
         if database != INTERNAL_DB_NAME:
             pm.hook.prepare_connection(conn=conn, database=database, datasette=self)

datasette/database.py

@@ -5,6 +5,7 @@ import inspect
 import os
 from pathlib import Path
 import queue
+import re
 import sqlite_utils
 import sys
 import tempfile
@@ -35,6 +36,8 @@ EXECUTE_WRITE_RETURNING_LIMIT = 10
 
 AttachedDatabase = namedtuple("AttachedDatabase", ("seq", "name", "file"))
 
+ALLOWED_JOURNAL_MODES = {"delete", "truncate", "persist", "wal"}
+
 
 class DatasetteClosedError(RuntimeError):
     """Raised when using a Datasette or Database instance after close()."""
@@ -42,6 +45,77 @@ class DatasetteClosedError(RuntimeError):
 
 _SHUTDOWN = object()
 
+# Statements that SQLite refuses to execute inside a transaction. These run
+# without the task transaction - a single statement needs no extra atomicity.
+_STATEMENTS_DISALLOWED_IN_TRANSACTION = {"vacuum", "attach", "detach", "pragma"}
+
+_FIRST_KEYWORD_RE = re.compile(
+    r"^(?:\s+|--[^\n]*(?:\n|$)|/\*.*?\*/)*([a-zA-Z]+)", re.DOTALL
+)
+
+
+def _can_execute_in_transaction(sql):
+    match = _FIRST_KEYWORD_RE.match(sql)
+    if match is None:
+        return True
+    return match.group(1).lower() not in _STATEMENTS_DISALLOWED_IN_TRANSACTION
+
+
+# Scripts that manage their own transactions also cannot run inside the
+# task transaction
+_SCRIPT_STATEMENTS_DISALLOWED_IN_TRANSACTION = _STATEMENTS_DISALLOWED_IN_TRANSACTION | {
+    "begin",
+    "commit",
+    "end",
+    "rollback",
+    "savepoint",
+    "release",
+}
+
+
+def _script_can_execute_in_transaction(statements):
+    for statement in statements:
+        match = _FIRST_KEYWORD_RE.match(statement)
+        if (
+            match is not None
+            and match.group(1).lower() in _SCRIPT_STATEMENTS_DISALLOWED_IN_TRANSACTION
+        ):
+            return False
+    return True
+
+
+def _iter_sql_statements(sql):
+    # Split a multi-statement SQL string into complete statements using
+    # sqlite3.complete_statement()
+    statement = []
+    for char in sql:
+        statement.append(char)
+        statement_sql = "".join(statement).strip()
+        if statement_sql and sqlite3.complete_statement(statement_sql):
+            yield statement_sql
+            statement = []
+    remainder = "".join(statement).strip()
+    if remainder:
+        yield remainder
+
+
+def _run_write_in_transaction(conn, fn):
+    # Open the transaction explicitly instead of relying on the sqlite3
+    # driver's implicit BEGIN, which only fires on the first data-modifying
+    # statement. With a transaction genuinely open, sqlite-utils write
+    # methods nest inside it as savepoints instead of committing their own
+    # transactions mid-task: https://github.com/simonw/datasette/issues/2831
+    conn.execute("BEGIN IMMEDIATE")
+    try:
+        result = fn(conn)
+    except Exception:
+        if conn.in_transaction:
+            conn.rollback()
+        raise
+    if conn.in_transaction:
+        conn.commit()
+    return result
+
 
 class Database:
     # For table counts stop at this many rows:
@@ -57,6 +131,7 @@ class Database:
         memory_name=None,
         mode=None,
         is_temp_disk=False,
+        enable_wal=False,
     ):
         self.name = None
         self._thread_local_id = f"x{self._thread_local_id_counter}"
@@ -68,6 +143,8 @@ class Database:
         self.is_memory = is_memory
         self.memory_name = memory_name
         self.is_temp_disk = is_temp_disk
+        self.enable_wal = enable_wal or is_temp_disk
+        self._wal_enabled = False
         if memory_name is not None:
             self.is_memory = True
         if is_temp_disk:
@@ -76,10 +153,7 @@ class Database:
             self.path = temp_path
             self.is_mutable = True
             self.mode = "rwc"
-            self._wal_enabled = False
             atexit.register(self._cleanup_temp_file)
-        else:
-            self._wal_enabled = False
         self.cached_hash = None
         self.cached_size = None
         self._cached_table_counts = None
@@ -138,6 +212,10 @@ class Database:
         extra_kwargs = {}
         if write:
             extra_kwargs["isolation_level"] = "IMMEDIATE"
+        # An explicit busy timeout policy rather than the driver's silent
+        # 5 second default - matters when external processes write to the
+        # same database files
+        extra_kwargs["timeout"] = self.ds.setting("busy_timeout_ms") / 1000
         if self.memory_name:
             uri = "file:{}?mode=memory&cache=shared".format(self.memory_name)
             conn = sqlite3.connect(
@@ -165,9 +243,23 @@ class Database:
             f"file:{self.path}{qs}", uri=True, check_same_thread=False, **extra_kwargs
         )
         self._all_file_connections.append(conn)
-        if self.is_temp_disk and not self._wal_enabled:
+        if self.enable_wal and not self._wal_enabled:
             conn.execute("PRAGMA journal_mode=WAL")
             self._wal_enabled = True
+        if write and self.is_mutable and not self.enable_wal:
+            journal_mode = self.ds.setting("journal_mode")
+            if journal_mode:
+                if journal_mode not in ALLOWED_JOURNAL_MODES:
+                    raise ValueError(
+                        "journal_mode setting must be one of: {}".format(
+                            ", ".join(sorted(ALLOWED_JOURNAL_MODES))
+                        )
+                    )
+                conn.execute("PRAGMA journal_mode={}".format(journal_mode))
+                if journal_mode == "wal":
+                    # The standard WAL pairing - fewer fsyncs, application
+                    # level consistency guarantees are unchanged
+                    conn.execute("PRAGMA synchronous=NORMAL")
         return conn
 
     def close(self):
@@ -258,18 +350,31 @@ 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,
+                transaction=_can_execute_in_transaction(sql),
+                request=request,
+            )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
         self._check_not_closed()
+        statements = list(_iter_sql_statements(sql))
+        transaction = _script_can_execute_in_transaction(statements)
 
         def _inner(conn):
-            return conn.executescript(sql)
+            if transaction:
+                # Execute statements one at a time so they run inside the
+                # task transaction - conn.executescript() would commit it
+                for statement in statements:
+                    conn.execute(statement)
+            else:
+                return conn.executescript(sql)
 
         with trace("sql", database=self.name, sql=sql.strip(), executescript=True):
             results = await self.execute_write_fn(
-                _inner, block=block, transaction=False, request=request
+                _inner, block=block, transaction=transaction, request=request
             )
         return results
 
@@ -347,8 +452,7 @@ class Database:
                 self._write_connection = self.connect(write=True)
                 self.ds._prepare_connection(self._write_connection, self.name)
             if transaction:
-                with self._write_connection:
-                    result = fn(self._write_connection)
+                result = _run_write_in_transaction(self._write_connection, fn)
             else:
                 result = fn(self._write_connection)
         else:
@@ -476,8 +580,7 @@ class Database:
             else:
                 try:
                     if task.transaction:
-                        with conn:
-                            result = task.fn(conn)
+                        result = _run_write_in_transaction(conn, task.fn)
                     else:
                         result = task.fn(conn)
                 except Exception as e:
@@ -674,7 +777,7 @@ class Database:
 
         def column_details(conn):
             # Returns {column_name: (type, is_unique)}
-            db = sqlite_utils.Database(conn)
+            db = sqlite_utils.Database(conn, execute_plugins=False)
             columns = db[table].columns_dict
             indexes = db[table].indexes
             details = {}

datasette/utils/internal_db.py

@@ -183,26 +183,6 @@ async def init_internal_db(db):
 async def populate_schema_tables(internal_db, db):
     database_name = db.name
 
-    def delete_everything(conn):
-        conn.execute(
-            "DELETE FROM catalog_tables WHERE database_name = ?", [database_name]
-        )
-        conn.execute(
-            "DELETE FROM catalog_views WHERE database_name = ?", [database_name]
-        )
-        conn.execute(
-            "DELETE FROM catalog_columns WHERE database_name = ?", [database_name]
-        )
-        conn.execute(
-            "DELETE FROM catalog_foreign_keys WHERE database_name = ?",
-            [database_name],
-        )
-        conn.execute(
-            "DELETE FROM catalog_indexes WHERE database_name = ?", [database_name]
-        )
-
-    await internal_db.execute_write_fn(delete_everything)
-
     tables = (await db.execute("select * from sqlite_master WHERE type = 'table'")).rows
     views = (await db.execute("select * from sqlite_master WHERE type = 'view'")).rows
 
@@ -266,47 +246,69 @@ async def populate_schema_tables(internal_db, db):
         indexes_to_insert,
     ) = await db.execute_fn(collect_info)
 
-    await internal_db.execute_write_many(
-        """
-        INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
-        values (?, ?, ?, ?)
-    """,
-        tables_to_insert,
-    )
-    await internal_db.execute_write_many(
-        """
-        INSERT INTO catalog_views (database_name, view_name, rootpage, sql)
-        values (?, ?, ?, ?)
-    """,
-        views_to_insert,
-    )
-    await internal_db.execute_write_many(
-        """
-        INSERT INTO catalog_columns (
-            database_name, table_name, cid, name, type, "notnull", default_value, is_pk, hidden
-        ) VALUES (
-            :database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden
+    def update_catalog(conn):
+        # Delete and reinsert as one write task so internal database readers
+        # never observe a database with missing catalog rows
+        # https://github.com/simonw/datasette/issues/2831
+        conn.execute(
+            "DELETE FROM catalog_tables WHERE database_name = ?", [database_name]
         )
-    """,
-        columns_to_insert,
-    )
-    await internal_db.execute_write_many(
-        """
-        INSERT INTO catalog_foreign_keys (
-            database_name, table_name, "id", seq, "table", "from", "to", on_update, on_delete, match
-        ) VALUES (
-            :database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match
+        conn.execute(
+            "DELETE FROM catalog_views WHERE database_name = ?", [database_name]
         )
-    """,
-        foreign_keys_to_insert,
-    )
-    await internal_db.execute_write_many(
-        """
-        INSERT INTO catalog_indexes (
-            database_name, table_name, seq, name, "unique", origin, partial
-        ) VALUES (
-            :database_name, :table_name, :seq, :name, :unique, :origin, :partial
+        conn.execute(
+            "DELETE FROM catalog_columns WHERE database_name = ?", [database_name]
         )
-    """,
-        indexes_to_insert,
-    )
+        conn.execute(
+            "DELETE FROM catalog_foreign_keys WHERE database_name = ?",
+            [database_name],
+        )
+        conn.execute(
+            "DELETE FROM catalog_indexes WHERE database_name = ?", [database_name]
+        )
+        conn.executemany(
+            """
+            INSERT INTO catalog_tables (database_name, table_name, rootpage, sql)
+            values (?, ?, ?, ?)
+        """,
+            tables_to_insert,
+        )
+        conn.executemany(
+            """
+            INSERT INTO catalog_views (database_name, view_name, rootpage, sql)
+            values (?, ?, ?, ?)
+        """,
+            views_to_insert,
+        )
+        conn.executemany(
+            """
+            INSERT INTO catalog_columns (
+                database_name, table_name, cid, name, type, "notnull", default_value, is_pk, hidden
+            ) VALUES (
+                :database_name, :table_name, :cid, :name, :type, :notnull, :default_value, :is_pk, :hidden
+            )
+        """,
+            columns_to_insert,
+        )
+        conn.executemany(
+            """
+            INSERT INTO catalog_foreign_keys (
+                database_name, table_name, "id", seq, "table", "from", "to", on_update, on_delete, match
+            ) VALUES (
+                :database_name, :table_name, :id, :seq, :table, :from, :to, :on_update, :on_delete, :match
+            )
+        """,
+            foreign_keys_to_insert,
+        )
+        conn.executemany(
+            """
+            INSERT INTO catalog_indexes (
+                database_name, table_name, seq, name, "unique", origin, partial
+            ) VALUES (
+                :database_name, :table_name, :seq, :name, :unique, :origin, :partial
+            )
+        """,
+            indexes_to_insert,
+        )
+
+    await internal_db.execute_write_fn(update_catalog)

datasette/views/row.py

@@ -749,7 +749,9 @@ class RowDeleteView(BaseView):
 
         # Delete table
         def delete_row(conn):
-            sqlite_utils.Database(conn)[resolved.table].delete(resolved.pk_values)
+            sqlite_utils.Database(conn, execute_plugins=False)[resolved.table].delete(
+                resolved.pk_values
+            )
 
         try:
             await resolved.db.execute_write_fn(delete_row, request=request)
@@ -830,7 +832,7 @@ class RowUpdateView(BaseView):
             return Response.error(["Permission denied for alter-table"], 403)
 
         def update_row(conn):
-            sqlite_utils.Database(conn)[resolved.table].update(
+            sqlite_utils.Database(conn, execute_plugins=False)[resolved.table].update(
                 resolved.pk_values, update, alter=alter
             )
 

datasette/views/table.py

@@ -1143,7 +1143,7 @@ class TableInsertView(BaseView):
             row_pk_values_for_later = [tuple(row[pk] for pk in pks) for row in rows]
 
         def insert_or_upsert_rows(conn):
-            table = sqlite_utils.Database(conn)[table_name]
+            table = sqlite_utils.Database(conn, execute_plugins=False)[table_name]
             kwargs = {}
             if upsert:
                 kwargs = {
@@ -1407,7 +1407,7 @@ class TableDropView(BaseView):
 
         # Drop table
         def drop_table(conn):
-            sqlite_utils.Database(conn)[table_name].drop()
+            sqlite_utils.Database(conn, execute_plugins=False)[table_name].drop()
 
         await db.execute_write_fn(drop_table, request=request)
         await self.ds.track_event(

datasette/views/table_create_alter.py

@@ -889,11 +889,13 @@ class TableCreateView(BaseView):
         initial_schema = None
         if table_exists:
             initial_schema = await db.execute_fn(
-                lambda conn: sqlite_utils.Database(conn)[table_name].schema
+                lambda conn: sqlite_utils.Database(conn, execute_plugins=False)[
+                    table_name
+                ].schema
             )
 
         def create_table(conn):
-            db_for_write = sqlite_utils.Database(conn)
+            db_for_write = sqlite_utils.Database(conn, execute_plugins=False)
             table = db_for_write[table_name]
             if rows:
                 table.insert_all(
@@ -1187,7 +1189,9 @@ class TableAlterView(BaseView):
             before_schema = _table_schema_from_conn(conn, table_name)
 
             def apply_operations(operation_conn):
-                db_for_write = sqlite_utils.Database(operation_conn)
+                db_for_write = sqlite_utils.Database(
+                    operation_conn, execute_plugins=False
+                )
                 table = db_for_write[table_name]
                 current_table_name = table_name
 
@@ -1261,62 +1265,62 @@ class TableAlterView(BaseView):
                     elif operation.op == "set_foreign_keys":
                         foreign_keys = [fk.tuple for fk in args.foreign_keys]
 
-                with operation_conn:
-                    for column in add_columns:
-                        not_null_default = None
-                        if column.not_null:
-                            if "default_expr" in column.model_fields_set:
-                                not_null_default = _default_expression_sql(
-                                    column.default_expr
-                                )
-                            else:
-                                not_null_default = _literal_default(
-                                    db_for_write, column.default
-                                )
-                        table.add_column(
-                            column.name,
-                            column.type,
-                            not_null_default=not_null_default,
-                        )
+                # The write task transaction makes these operations atomic
+                for column in add_columns:
+                    not_null_default = None
+                    if column.not_null:
+                        if "default_expr" in column.model_fields_set:
+                            not_null_default = _default_expression_sql(
+                                column.default_expr
+                            )
+                        else:
+                            not_null_default = _literal_default(
+                                db_for_write, column.default
+                            )
+                    table.add_column(
+                        column.name,
+                        column.type,
+                        not_null_default=not_null_default,
+                    )
 
-                    should_transform = any(
-                        (
-                            types,
-                            rename,
-                            drop,
-                            not_null,
-                            defaults,
-                            column_order is not None,
-                            pk is not SQLITE_UTILS_DEFAULT,
-                            add_foreign_keys,
-                            drop_foreign_keys,
-                            foreign_keys is not None,
+                should_transform = any(
+                    (
+                        types,
+                        rename,
+                        drop,
+                        not_null,
+                        defaults,
+                        column_order is not None,
+                        pk is not SQLITE_UTILS_DEFAULT,
+                        add_foreign_keys,
+                        drop_foreign_keys,
+                        foreign_keys is not None,
+                    )
+                )
+                if should_transform:
+                    table.transform(
+                        types=types or None,
+                        rename=rename or None,
+                        drop=drop or None,
+                        pk=pk,
+                        not_null=not_null or None,
+                        defaults=defaults or None,
+                        column_order=column_order,
+                        add_foreign_keys=add_foreign_keys or None,
+                        drop_foreign_keys=drop_foreign_keys or None,
+                        foreign_keys=foreign_keys,
+                    )
+                if (
+                    rename_table_to is not None
+                    and rename_table_to != current_table_name
+                ):
+                    operation_conn.execute(
+                        "alter table {} rename to {}".format(
+                            escape_sqlite(current_table_name),
+                            escape_sqlite(rename_table_to),
                         )
                     )
-                    if should_transform:
-                        table.transform(
-                            types=types or None,
-                            rename=rename or None,
-                            drop=drop or None,
-                            pk=pk,
-                            not_null=not_null or None,
-                            defaults=defaults or None,
-                            column_order=column_order,
-                            add_foreign_keys=add_foreign_keys or None,
-                            drop_foreign_keys=drop_foreign_keys or None,
-                            foreign_keys=foreign_keys,
-                        )
-                    if (
-                        rename_table_to is not None
-                        and rename_table_to != current_table_name
-                    ):
-                        operation_conn.execute(
-                            "alter table {} rename to {}".format(
-                                escape_sqlite(current_table_name),
-                                escape_sqlite(rename_table_to),
-                            )
-                        )
-                        current_table_name = rename_table_to
+                    current_table_name = rename_table_to
 
                 return current_table_name, _table_schema_from_conn(
                     operation_conn, current_table_name

docs/changelog.rst

@@ -4,6 +4,21 @@
 Changelog
 =========
 
+.. _v_unreleased:
+
+Unreleased
+----------
+
+- Write functions run via ``await db.execute_write_fn()`` now execute inside an explicitly opened ``BEGIN IMMEDIATE`` transaction, committed when the function returns or rolled back if it raises. Previously the transaction was only opened implicitly by the first raw data-modifying statement, which meant writes made through sqlite-utils committed independently mid-task - a function that used sqlite-utils and then failed could leave those writes permanently committed. sqlite-utils write methods now nest inside the task transaction as savepoints, so a failing write function rolls back everything it did. Functions run with ``transaction=True`` should no longer manage transactions themselves - use ``transaction=False`` for manual transaction control. (:issue:`2831`)
+- ``await db.execute_write()`` detects statements that SQLite cannot execute inside a transaction - ``VACUUM``, ``ATTACH``, ``DETACH`` and ``PRAGMA`` - and runs them in autocommit mode instead. (:issue:`2831`)
+- ``await db.execute_write_script()`` is now transactional, matching its documentation: if any statement in the script fails, none of its statements are applied. Scripts containing statements that cannot run inside a transaction, or that manage transactions themselves, fall back to the previous ``conn.executescript()`` autocommit behavior. (:issue:`2831`)
+- The JSON write API is now atomic per request: ``/db/-/create`` with initial rows, multi-operation ``/db/table/-/alter`` calls and inserts using ``"return": true`` now either fully apply or roll back entirely if any part fails. Previously a failure part way through could leave earlier writes from the same request permanently committed. (:issue:`2831`)
+- Rebuilding the internal database catalog for a database is now a single atomic write. Previously the rebuild used six separate transactions, so queries against the internal database could observe a database with missing catalog rows while a rebuild was in progress. (:issue:`2831`)
+- sqlite-utils plugins no longer have their ``prepare_connection()`` hooks executed against Datasette's database connections - use Datasette's own :ref:`prepare_connection() <plugin_hook_prepare_connection>` plugin hook to customize connections. ``PRAGMA recursive_triggers=on`` is now applied consistently to every connection Datasette opens - previously it was enabled just on the write connection, as a side effect of the first sqlite-utils based write. (:issue:`2831`)
+- New :ref:`setting_busy_timeout_ms` setting controlling how long SQLite waits for a locked database file before giving up, previously hard-wired to the ``sqlite3`` driver's silent 5 second default. This matters when external processes write to the same database files Datasette is serving. (:issue:`2831`)
+- New :ref:`setting_journal_mode` setting for opting mutable database files into SQLite's `WAL mode <https://www.sqlite.org/wal.html>`__ (or another journal mode), allowing reads and writes to proceed concurrently. Datasette does not change the journal mode of database files by default. (:issue:`2831`)
+- A persistent internal database specified with ``--internal`` now has WAL mode enabled, matching the behavior of the default temporary internal database. (:issue:`2831`)
+
 .. _v1_0_a36:
 
 1.0a36 (2026-07-07)

docs/cli-reference.rst

@@ -251,6 +251,11 @@ These can be passed to ``datasette serve`` using ``datasette serve --setting nam
                                    executing SQLite queries (default=3)
       sql_time_limit_ms            Time limit for a SQL query in milliseconds
                                    (default=1000)
+      busy_timeout_ms              How long SQLite waits for a locked database file
+                                   before giving up (default=5000)
+      journal_mode                 Journal mode to set on mutable database files,
+                                   e.g. wal - leave blank to use each file's
+                                   existing mode (default=)
       default_facet_size           Number of values to return for requested facets
                                    (default=30)
       facet_time_limit_ms          Time limit for calculating a requested facet

docs/internals.rst

@@ -2059,16 +2059,18 @@ 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, with the exception of statements that SQLite does not allow to run inside a transaction: ``VACUUM``, ``ATTACH``, ``DETACH`` and ``PRAGMA``. Those statements are executed in autocommit mode instead.
 
 .. _database_execute_write_script:
 
 await db.execute_write_script(sql, block=True)
 ----------------------------------------------
 
-Like ``execute_write()`` but can be used to send multiple SQL statements in a single string separated by semicolons, using the ``sqlite3`` `conn.executescript() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.executescript>`__ method.
+Like ``execute_write()`` but can be used to send multiple SQL statements in a single string separated by semicolons.
 
-Each call to ``execute_write_script()`` will be executed inside a transaction.
+Each call to ``execute_write_script()`` will be executed inside a transaction - if any statement fails, none of the statements will be applied.
+
+The exception is scripts that include statements which SQLite cannot execute inside a transaction - ``VACUUM``, ``ATTACH``, ``DETACH``, ``PRAGMA`` - or that manage transactions themselves using ``BEGIN``, ``COMMIT``, ``ROLLBACK``, ``SAVEPOINT`` or ``RELEASE``. Those scripts are executed using the ``sqlite3`` `conn.executescript() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.executescript>`__ method instead, where each statement is committed as it executes.
 
 .. _database_execute_write_many:
 
@@ -2151,7 +2153,11 @@ The value returned from ``await database.execute_write_fn(...)`` will be the ret
 
 If your function raises an exception that exception will be propagated up to the ``await`` line.
 
-By default your function will be executed inside a transaction. You can pass ``transaction=False`` to disable this behavior, though if you do that you should be careful to manually apply transactions - ideally using the ``with conn:`` pattern, or you may see ``OperationalError: database table is locked`` errors.
+By default your function will be executed inside a transaction. Datasette executes ``BEGIN IMMEDIATE`` on the write connection before calling your function, then commits the transaction when your function returns - or rolls it back if your function raises an exception. Nothing your function writes will be visible to other connections until that final commit.
+
+Because the transaction is already open when your function is called, write methods from libraries such as `sqlite-utils <https://sqlite-utils.datasette.io/>`__ will nest their work inside it (sqlite-utils uses savepoints) rather than committing independently, so an exception rolls back everything the function did.
+
+Your function should not manage transactions itself when ``transaction=True`` - do not execute ``BEGIN`` or call ``conn.commit()`` or ``conn.rollback()`` on the connection. If you need to manage transactions manually, pass ``transaction=False`` - ideally using the ``with conn:`` pattern, or you may see ``OperationalError: database table is locked`` errors.
 
 If you specify ``block=False`` the method becomes fire-and-forget, queueing your function to be executed and then allowing your code after the call to ``.execute_write_fn()`` to continue running while the underlying thread waits for an opportunity to run your function. A UUID representing the queued task will be returned. Any exceptions in your code will be silently swallowed.
 
@@ -2174,6 +2180,17 @@ Functions run using ``execute_isolated_fn()`` share the same queue as ``execute_
 
 The return value of the function will be returned by this method. Any exceptions raised by the function will be raised out of the ``await`` line as well.
 
+.. _internals_database_transactions:
+
+How Datasette manages transactions
+----------------------------------
+
+Each mutable database has a single write connection, owned by a dedicated write thread that executes queued write tasks one at a time. Each task submitted with ``transaction=True`` (the default) runs inside its own transaction: Datasette executes ``BEGIN IMMEDIATE`` before invoking the task, commits if it completes and rolls back if it raises. One write task equals one transaction, and writes made by a task only become visible to readers when the task completes.
+
+Read connections never open transactions - each read statement executes in autocommit mode with its own implicit SQLite read snapshot.
+
+Datasette's write connections rely on the legacy transaction handling of Python's ``sqlite3`` module (connections opened with ``isolation_level=`` rather than the ``autocommit=`` parameter introduced in Python 3.12). This is a deliberate constraint shared with `sqlite-utils <https://sqlite-utils.datasette.io/>`__, which refuses connections created with ``autocommit=True`` or ``autocommit=False`` - any future migration to the modern transaction API would need to be coordinated across both projects. Plugins should not change the transaction control mode of connections Datasette passes to them.
+
 .. _database_close:
 
 db.close()

docs/settings.rst

@@ -103,6 +103,39 @@ You can optionally set a lower time limit for an individual query using the ``?_
 
 This would set the time limit to 100ms for that specific query. This feature is useful if you are working with databases of unknown size and complexity - a query that might make perfect sense for a smaller table could take too long to execute on a table with millions of rows. By setting custom time limits you can execute queries "optimistically" - e.g. give me an exact count of rows matching this query but only if it takes less than 100ms to calculate.
 
+.. _setting_busy_timeout_ms:
+
+busy_timeout_ms
+~~~~~~~~~~~~~~~
+
+How long SQLite should wait when a database file is locked by another connection or process before giving up with a ``database is locked`` error, in milliseconds. The default is 5000 (five seconds), matching the default used by Python's ``sqlite3`` module.
+
+This mostly matters when other processes write to the same database files that Datasette is serving - a common pattern is a separate script (using `sqlite-utils <https://sqlite-utils.datasette.io/>`__ or similar) that periodically updates a database while Datasette serves it. A larger value makes Datasette more patient with long write transactions from those processes::
+
+    datasette mydatabase.db --setting busy_timeout_ms 10000
+
+.. _setting_journal_mode:
+
+journal_mode
+~~~~~~~~~~~~
+
+`Journal mode <https://www.sqlite.org/pragma.html#pragma_journal_mode>`__ to set on mutable database files. Leave blank (the default) to use whatever journal mode each database file already uses.
+
+Setting this to ``wal`` enables SQLite's `Write-Ahead Logging <https://www.sqlite.org/wal.html>`__ mode, which allows reads and writes to proceed concurrently - in the default rollback journal mode each commit blocks readers, and long reads block the writer::
+
+    datasette mydatabase.db --setting journal_mode wal
+
+When ``wal`` is selected Datasette also sets ``PRAGMA synchronous=NORMAL`` on the write connection, the standard pairing for WAL which reduces the number of ``fsync`` operations without weakening application-level consistency guarantees.
+
+Other allowed values are ``delete``, ``truncate`` and ``persist``.
+
+Things to be aware of before enabling WAL:
+
+- WAL mode is persistent - it is recorded in the database file and stays in effect when other tools open the same file later.
+- SQLite creates ``-wal`` and ``-shm`` files alongside the database file.
+- WAL does not work reliably on network filesystems such as NFS.
+- The directory containing the database must be writable by Datasette.
+
 .. _setting_max_returned_rows:
 
 max_returned_rows

tests/test_api.py

@@ -694,6 +694,8 @@ async def test_settings_json(ds_client):
         "max_insert_rows": 100,
         "max_post_body_bytes": 2 * 1024 * 1024,
         "sql_time_limit_ms": 200,
+        "busy_timeout_ms": 5000,
+        "journal_mode": "",
         "allow_download": True,
         "allow_signed_tokens": True,
         "max_signed_tokens_ttl": 0,

tests/test_api_write.py

@@ -2717,3 +2717,103 @@ async def test_create_using_alter_against_existing_table(
         insert_rows_event = ds_write._tracked_events[1]
         assert insert_rows_event.name == "insert-rows"
         assert insert_rows_event.num_rows == 1
+
+
+@pytest.mark.asyncio
+async def test_create_table_with_failing_rows_is_atomic(ds_write):
+    # https://github.com/simonw/datasette/issues/2831
+    # If inserting the initial rows fails, the create table should
+    # be rolled back as well
+    token = write_token(ds_write)
+    response = await ds_write.client.post(
+        "/data/-/create",
+        json={
+            "table": "atomic_create",
+            "rows": [{"id": 1, "name": "one"}, {"id": 1, "name": "dupe"}],
+            "pk": "id",
+        },
+        headers=_headers(token),
+    )
+    assert response.status_code == 400
+    assert not await ds_write.get_database("data").table_exists("atomic_create")
+
+
+@pytest.mark.asyncio
+async def test_alter_table_with_failing_operation_is_atomic(ds_write):
+    # https://github.com/simonw/datasette/issues/2831
+    # If a later operation fails, earlier operations should be rolled back
+    token = write_token(ds_write)
+    response = await ds_write.client.post(
+        "/data/docs/-/alter",
+        json={
+            "operations": [
+                {"op": "add_column", "args": {"name": "new_col", "type": "text"}},
+                # Fails - column "title" already exists
+                {"op": "add_column", "args": {"name": "title", "type": "text"}},
+            ]
+        },
+        headers=_headers(token),
+    )
+    assert response.status_code == 400
+    columns = await ds_write.get_database("data").table_columns("docs")
+    assert "new_col" not in columns
+
+
+@pytest.mark.asyncio
+async def test_insert_with_return_failing_row_is_atomic(ds_write):
+    # https://github.com/simonw/datasette/issues/2831
+    # Insert with "return": true runs one insert per row - a failure
+    # part way through should roll back the earlier rows
+    token = write_token(ds_write)
+    response = await ds_write.client.post(
+        "/data/docs/-/insert",
+        json={
+            "rows": [
+                {"id": 1, "title": "one"},
+                {"id": 2, "title": "two"},
+                {"id": 2, "title": "dupe"},
+            ],
+            "return": True,
+        },
+        headers=_headers(token),
+    )
+    assert response.status_code == 400
+    count = (
+        await ds_write.get_database("data").execute("select count(*) from docs")
+    ).single_value()
+    assert count == 0
+
+
+@pytest.mark.asyncio
+async def test_write_api_does_not_run_sqlite_utils_plugins(ds_write):
+    # https://github.com/simonw/datasette/issues/2831
+    # sqlite-utils plugins should not have their prepare_connection hooks
+    # executed against Datasette's connections
+    import sqlite_utils.plugins
+    from sqlite_utils import hookimpl
+
+    prepared = []
+
+    class TrackingPlugin:
+        @hookimpl
+        def prepare_connection(self, conn):
+            prepared.append(conn)
+
+    sqlite_utils.plugins.pm.register(TrackingPlugin(), name="datasette-test-tracking")
+    try:
+        token = write_token(ds_write)
+        response = await ds_write.client.post(
+            "/data/docs/-/insert",
+            json={"rows": [{"id": 1, "title": "one"}]},
+            headers=_headers(token),
+        )
+        assert response.status_code == 201
+        response = await ds_write.client.post(
+            "/data/docs/1/-/update",
+            json={"update": {"title": "two"}},
+            headers=_headers(token),
+        )
+        assert response.status_code == 200
+        assert prepared == []
+    finally:
+        sqlite_utils.plugins.pm.unregister(name="datasette-test-tracking")

tests/test_internal_db.py

@@ -334,3 +334,51 @@ async def test_orphan_stale_catalog_child_entries_removed(tmp_path):
     assert response.status_code == 200
 
     ds2.close()
+
+
+@pytest.mark.asyncio
+async def test_populate_schema_tables_is_a_single_write_task(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    # The catalog rebuild should be one atomic write task, so readers of the
+    # internal database never observe a database with deleted-but-not-yet-
+    # reinserted catalog rows
+    from datasette.app import Datasette
+    from datasette.utils.internal_db import populate_schema_tables
+
+    path = str(tmp_path / "data.db")
+    conn = sqlite3.connect(path)
+    conn.execute("create table t (id integer primary key, name text)")
+    conn.execute("create view v as select * from t")
+    conn.close()
+    ds = Datasette([path])
+    await ds.invoke_startup()
+    await ds.refresh_schemas()
+    internal_db = ds.get_internal_database()
+
+    write_fns = []
+    original_execute_write_fn = internal_db.execute_write_fn
+
+    async def counting_execute_write_fn(fn, *args, **kwargs):
+        write_fns.append(fn)
+        return await original_execute_write_fn(fn, *args, **kwargs)
+
+    internal_db.execute_write_fn = counting_execute_write_fn
+    try:
+        await populate_schema_tables(internal_db, ds.get_database("data"))
+    finally:
+        internal_db.execute_write_fn = original_execute_write_fn
+
+    assert len(write_fns) == 1
+    # And the catalog should be correct
+    tables = await internal_db.execute(
+        "select table_name from catalog_tables where database_name = 'data'"
+    )
+    assert [row["table_name"] for row in tables.rows] == ["t"]
+    views = await internal_db.execute(
+        "select view_name from catalog_views where database_name = 'data'"
+    )
+    assert [row["view_name"] for row in views.rows] == ["v"]
+    columns = await internal_db.execute(
+        "select name from catalog_columns where database_name = 'data' and table_name = 't'"
+    )
+    assert {row["name"] for row in columns.rows} == {"id", "name"}

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
 
@@ -111,6 +112,129 @@ async def test_execute_fn_transaction_false():
     await db.execute_write_fn(run, transaction=False)
 
 
+@pytest.mark.asyncio
+async def test_execute_write_fn_wraps_sqlite_utils_writes_in_transaction():
+    # https://github.com/simonw/datasette/issues/2831
+    # sqlite-utils write methods commit their own transactions unless one is
+    # already open - the write thread must open one before running each task
+    # so that a failing task rolls back everything, including those writes.
+    datasette = Datasette(memory=True)
+    db = datasette.add_memory_database("test_txn_sqlite_utils")
+    await db.execute_write("create table t (a integer)")
+
+    def failing_task(conn):
+        sqlite_utils.Database(conn)["t"].insert_all({"a": i} for i in range(5))
+        assert conn.in_transaction
+        raise ValueError("boom")
+
+    with pytest.raises(ValueError):
+        await db.execute_write_fn(failing_task)
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 0
+    # The write connection should be back in autocommit mode
+    assert (
+        await db.execute_write_fn(lambda conn: conn.in_transaction, transaction=False)
+    ) is False
+    # And a transaction=True task should see a transaction already open
+    assert (await db.execute_write_fn(lambda conn: conn.in_transaction)) is True
+
+
+@pytest.mark.asyncio
+async def test_execute_write_statements_disallowed_in_transaction(tmp_path):
+    # VACUUM (and ATTACH/DETACH/PRAGMA) cannot run inside a transaction, so
+    # execute_write() must run them outside one
+    # https://github.com/simonw/datasette/issues/2831
+    path = str(tmp_path / "test.db")
+    setup_conn = sqlite3.connect(path)
+    setup_conn.execute("create table t (a integer)")
+    setup_conn.close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    await db.execute_write("vacuum")
+    await db.execute_write("  -- a comment\n  VACUUM")
+    # But regular DML statements still run inside a transaction
+    from datasette.database import _can_execute_in_transaction
+
+    assert _can_execute_in_transaction("insert into t values (1)")
+    assert _can_execute_in_transaction("with x as (select 1) insert into t select 1")
+    assert not _can_execute_in_transaction("vacuum")
+    assert not _can_execute_in_transaction("/* hi */ PRAGMA optimize")
+
+
+@pytest.mark.asyncio
+async def test_execute_write_fn_sqlite_utils_integrity_error_rolls_back_task():
+    # https://github.com/simonw/datasette/issues/2831
+    datasette = Datasette(memory=True)
+    db = datasette.add_memory_database("test_txn_integrity")
+    await db.execute_write("create table t (id integer primary key)")
+
+    def two_inserts(conn):
+        table = sqlite_utils.Database(conn)["t"]
+        table.insert({"id": 1})
+        table.insert({"id": 1})  # IntegrityError
+
+    with pytest.raises(sqlite3.IntegrityError):
+        await db.execute_write_fn(two_inserts)
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 0
+
+
+@pytest.mark.asyncio
+async def test_execute_write_script_is_transactional():
+    # https://github.com/simonw/datasette/issues/2831
+    # A script that fails part way through should apply none of its statements
+    datasette = Datasette(memory=True)
+    db = datasette.add_memory_database("test_write_script_txn")
+    with pytest.raises(sqlite3.OperationalError):
+        await db.execute_write_script(
+            "create table one (id integer primary key);\n"
+            "insert into one (id) values (1);\n"
+            "insert into no_such_table (id) values (2);"
+        )
+    assert "one" not in await db.table_names()
+
+
+@pytest.mark.asyncio
+async def test_execute_write_script_with_transaction_unsafe_statements(tmp_path):
+    # Scripts containing statements that cannot run inside a transaction
+    # (VACUUM, BEGIN etc) still execute, using the previous autocommit behavior
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    await db.execute_write_script("create table t (id integer primary key);\nvacuum;")
+    assert "t" in await db.table_names()
+    await db.execute_write_script("begin;\ninsert into t (id) values (1);\ncommit;")
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 1
+
+
+@pytest.mark.asyncio
+async def test_execute_write_fn_writes_invisible_to_readers_until_task_ends(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    path = str(tmp_path / "test.db")
+    setup_conn = sqlite3.connect(path)
+    setup_conn.execute("create table t (a integer)")
+    setup_conn.close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+
+    def insert_and_check(conn):
+        sqlite_utils.Database(conn)["t"].insert_all({"a": i} for i in range(5))
+        # A separate read-only connection must not see the rows mid-task
+        reader = sqlite3.connect("file:{}?mode=ro".format(path), uri=True)
+        try:
+            return reader.execute("select count(*) from t").fetchone()[0]
+        finally:
+            reader.close()
+
+    mid_task_count = await db.execute_write_fn(insert_and_check)
+    assert mid_task_count == 0
+    # After the task commits the rows are visible
+    count = (await db.execute("select count(*) from t")).single_value()
+    assert count == 5
+
+
 @pytest.mark.parametrize(
     "tables,exists",
     (
@@ -1178,3 +1302,107 @@ async def test_database_close_is_idempotent(tmpdir):
     # Second call should be a no-op, not raise
     db.close()
     ds._internal_database.close()
+
+
+@pytest.mark.asyncio
+async def test_recursive_triggers_enabled_on_all_connections(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    # Previously recursive_triggers was only enabled on the write connection,
+    # and only as a side effect of the first sqlite-utils based write - so
+    # trigger semantics could differ between connections
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    write_value = await db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA recursive_triggers").fetchone()[0],
+        transaction=False,
+    )
+    read_value = await db.execute_fn(
+        lambda conn: conn.execute("PRAGMA recursive_triggers").fetchone()[0]
+    )
+    assert write_value == 1
+    assert read_value == 1
+
+
+@pytest.mark.asyncio
+async def test_busy_timeout_ms_setting(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    # The SQLite busy timeout should be an explicit, configurable policy
+    # instead of the sqlite3 driver's inherited 5 second default
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path], settings={"busy_timeout_ms": 250})
+    db = datasette.get_database("test")
+    read_value = await db.execute_fn(
+        lambda conn: conn.execute("PRAGMA busy_timeout").fetchone()[0]
+    )
+    write_value = await db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA busy_timeout").fetchone()[0],
+        transaction=False,
+    )
+    assert read_value == 250
+    assert write_value == 250
+
+
+@pytest.mark.asyncio
+async def test_busy_timeout_ms_default(tmp_path):
+    # Default matches the sqlite3 driver's historical 5 second default
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    read_value = await db.execute_fn(
+        lambda conn: conn.execute("PRAGMA busy_timeout").fetchone()[0]
+    )
+    assert read_value == 5000
+
+
+@pytest.mark.asyncio
+async def test_journal_mode_setting_applies_wal(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    # Opt-in WAL support for mutable database files - not the default
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path], settings={"journal_mode": "wal"})
+    db = datasette.get_database("test")
+    mode = await db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA journal_mode").fetchone()[0],
+        transaction=False,
+    )
+    assert mode == "wal"
+    # WAL is paired with synchronous=NORMAL (1) on the write connection
+    synchronous = await db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA synchronous").fetchone()[0],
+        transaction=False,
+    )
+    assert synchronous == 1
+
+
+@pytest.mark.asyncio
+async def test_journal_mode_defaults_to_leaving_files_alone(tmp_path):
+    path = str(tmp_path / "test.db")
+    sqlite3.connect(path).close()
+    datasette = Datasette([path])
+    db = datasette.get_database("test")
+    mode = await db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA journal_mode").fetchone()[0],
+        transaction=False,
+    )
+    assert mode == "delete"
+
+
+@pytest.mark.asyncio
+async def test_persistent_internal_database_gets_wal(tmp_path):
+    # https://github.com/simonw/datasette/issues/2831
+    # The temporary internal database enables WAL - a persistent one passed
+    # via --internal should get the same treatment
+    internal_path = str(tmp_path / "internal.db")
+    datasette = Datasette(memory=True, internal=internal_path)
+    await datasette.invoke_startup()
+    internal_db = datasette.get_internal_database()
+    mode = await internal_db.execute_write_fn(
+        lambda conn: conn.execute("PRAGMA journal_mode").fetchone()[0],
+        transaction=False,
+    )
+    assert mode == "wal"