From 0fcd385c625482cc5fa77eb39efc79821c69e476 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 23:33:38 +0000 Subject: [PATCH] sqlitedb: recover a poisoned handle inside a write operation by replaying it Closes the last deviation recorded for DID-matlab da83fa3. MATLAB reopens the connection before retrying a READONLY statement, because a poisoned handle (SQLITE_READONLY_DBMOVED) is cleared by reopening and never by waiting. mksqlite autocommits, so there the failed statement is all there is to re-run. In SQLiteDB it is part of a transaction whose earlier statements die with the handle. The four write operations (_do_add_branch, _do_add_doc, _do_remove_doc, _do_delete_branch) now run through a journaling cursor. It records each statement's SQL, parameters, rowcount, an INSERT's lastrowid and every row read. On a READONLY at a statement or at COMMIT, _recover_transaction rolls back the dead transaction, waits on the same backoff schedule, reopens and replays the journal. Each statement is checked against what it did the first time before the failed statement runs or the transaction commits. Only SQL is replayed: files already copied, or deleted with delete_original, are not needed again. A replay that differs (another writer took the next doc_idx, a different file is at the path, a statement fails) is refused, nothing is written, and the original error is raised. The tests reproduce a real DBMOVED by replacing the database file under the open connection. Doing that showed SQLite reports it at the first write of a transaction rather than at COMMIT, so a COMMIT-only recovery would have missed the real case. It also showed that a swap after the first write is not detected at all, which no retry can see. Adding documents is about 4% slower (1,500 documents: 6.1 s -> 6.4 s). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012Dn4gCNCU2jUGdmfPVCAGK --- ..._matlab_python_bridge_implementations.yaml | 53 +-- src/did/implementations/sqlitedb.py | 350 +++++++++++++++++- tests/test_sqlitedb_transient_errors.py | 254 +++++++++++++ 3 files changed, 623 insertions(+), 34 deletions(-) diff --git a/src/did/did_matlab_python_bridge_implementations.yaml b/src/did/did_matlab_python_bridge_implementations.yaml index ef106ae..e4daa18 100644 --- a/src/did/did_matlab_python_bridge_implementations.yaml +++ b/src/did/did_matlab_python_bridge_implementations.yaml @@ -153,27 +153,38 @@ classes: UserWrite), and a temp-file write probe of the directory -- but print the time as ISO-8601 UTC rather than MATLAB's local datestr. - DEVIATION -- where the retry may run. mksqlite autocommits every - statement, so in MATLAB a failed statement is the whole unit of work - and re-running it (even after a reopen) is safe. SQLiteDB groups each - add/remove/branch operation into one sqlite3 transaction. Rolling back - or reopening mid-transaction would silently discard the statements - already pending, and a retried tail would then commit a partial - operation. So do_run_sql_query retries only when the connection had no - uncommitted work before the statement; inside a transaction the error - is reported (with diagnostics) and raised at once, and the caller's - existing rollback runs. - - DEVIATION -- the commit is retried too. Because Python's writes reach - the disk at COMMIT rather than at each statement, that is where a - slow-drive failure lands here. Every commit now goes through - _commit(), which retries a BUSY / READONLY COMMIT on the same schedule - while the transaction is still open. It never reopens -- that would - lose the transaction -- so a genuinely poisoned handle (DBMOVED) is not - recovered at COMMIT: the error is raised and the operation rolled back. - Replaying a whole operation after a reopen would cover that case, but - _do_add_doc also copies ingested files, so it is not obviously - idempotent; not attempted. + DEVIATION -- what is re-run after a reopen. mksqlite autocommits + every statement, so in MATLAB a failed statement is the whole unit of + work and re-running it after a reopen is safe. SQLiteDB groups each + add/remove/branch operation into one sqlite3 transaction, and a + poisoned handle takes the transaction's earlier statements with it. + Since 2026-10-23 (da83fa3's reopen, carried into the transaction + case) the four write operations -- _do_add_branch, + _do_add_doc, _do_remove_doc, _do_delete_branch -- run through a + journaling cursor (_JournalCursor, @_journaled) that records each + statement's SQL, parameters, rowcount, an INSERT's lastrowid and every + row the operation read. On a READONLY inside one -- at a statement or + at COMMIT -- _recover_transaction rolls the dead transaction back, + waits on the same backoff schedule, reopens, replays the journal and + checks each statement against what it did the first time, then runs + the failed statement (or commits). Only SQL is replayed: files the + operation copied, or deleted with delete_original, before the failure + are not touched again, which is why the operation is not simply re-run. + A replay that differs -- another writer took the next doc_idx, a + different file is now at the path, a statement fails -- is refused, + nothing is written, and the original error is raised; it is not + retried. Two things found while porting: SQLite reports DBMOVED at a + transaction's FIRST WRITE, not at COMMIT, so that is where recovery + usually happens; and a file swapped out after that first write is not + detected at all -- COMMIT "succeeds" into the unlinked file -- which no + retry in either language can see. + do_run_sql_query, outside a write operation, keeps its own retry: only + when no transaction is open, reopening first for READONLY; inside an + unjournaled transaction the error is raised at once. + + BUSY at COMMIT is still waited out on the same connection while the + transaction stays open, never replayed: a lock clears with time, and + the transaction is intact. DEVIATION -- classification. Like MATLAB, BUSY vs READONLY is judged from the error message (sqlite3.Error.sqlite_errorcode would be exact diff --git a/src/did/implementations/sqlitedb.py b/src/did/implementations/sqlitedb.py index 7122238..0356aee 100644 --- a/src/did/implementations/sqlitedb.py +++ b/src/did/implementations/sqlitedb.py @@ -1,4 +1,5 @@ import contextlib +import functools import os import re as _re import shutil @@ -41,6 +42,100 @@ def _sql_escape(value): _RETRY_BACKOFFS = (0.1, 0.2, 0.5, 1, 2, 4) +class _ReplayDiverged(Exception): + """A replayed transaction saw a different database than the original.""" + + +class _JournalCursor: + """A cursor that records what each of its statements did. + + Used inside a write operation (see :func:`_journaled`) so that, when the + connection is poisoned part-way through, the transaction can be replayed + on a fresh connection and checked against what it saw the first time -- + see :meth:`SQLiteDB._recover_transaction`. Each statement's entry holds + its SQL and parameters, its rowcount, the lastrowid of an INSERT, and + every row the caller fetched. + """ + + def __init__(self, db, cursor): + self._db = db + self._cursor = cursor + self._entry = None + + def execute(self, sql, params=()): + return self._db._journaled_execute(self, "execute", sql, params) + + def executemany(self, sql, seq_of_params): + # Materialized so the same parameters can be replayed. + return self._db._journaled_execute( + self, "executemany", sql, list(seq_of_params) + ) + + def _seen(self, rows, complete): + if self._entry is not None: + self._entry["rows"].extend(tuple(row) for row in rows) + if complete: + self._entry["complete"] = True + + def fetchone(self): + row = self._cursor.fetchone() + self._seen([] if row is None else [row], row is None) + return row + + def fetchall(self): + rows = self._cursor.fetchall() + self._seen(rows, True) + return rows + + def fetchmany(self, size=None): + rows = self._cursor.fetchmany(self._cursor.arraysize if size is None else size) + self._seen(rows, not rows) + return rows + + def __iter__(self): + return self + + def __next__(self): + row = self.fetchone() + if row is None: + raise StopIteration + return row + + @property + def lastrowid(self): + return self._cursor.lastrowid + + @property + def rowcount(self): + return self._cursor.rowcount + + def close(self): + self._cursor.close() + + +def _journaled(method): + """Record the SQL a write operation runs, so it can be recovered. + + Re-entrant: an operation called from inside another shares its journal. + """ + + @functools.wraps(method) + def wrapper(self, *args, **kwargs): + if self._journal is not None: + return method(self, *args, **kwargs) + self._journal = [] + try: + return method(self, *args, **kwargs) + finally: + self._journal = None + + return wrapper + + +def _is_insert(sql): + return sql.lstrip().upper().startswith(("INSERT", "REPLACE")) + + # Escape character used in LIKE patterns (see _sql_like_escape / ESCAPE clauses). _LIKE_ESCAPE_CHAR = "\\" @@ -128,6 +223,7 @@ def __init__(self, filename): super().__init__(connection=filename) self.dbid = None self._fields_cache = {} # (class, field_name) -> field_idx + self._journal = None # statements of the current write operation self._open_db() # Create FileDir now, as DID-MATLAB's sqlitedb constructor does: @@ -351,20 +447,29 @@ def _fetchall(self, query_str, params=()): return cursor.fetchall() def _commit(self): - """Commit, retrying a transient BUSY / READONLY failure with backoff. + """Commit, recovering a transient BUSY / READONLY failure. In DID-matlab every statement autocommits, so the statement retry in run_sql_noOpen covers the write that reaches the disk. Here writes are grouped into a transaction and reach the disk at COMMIT, so the - commit is retried too. It is never reopened: that would discard the - transaction. If SQLite has already rolled the transaction back there - is nothing left to commit, and the error is raised for the caller. + commit is retried too. A BUSY commit is waited out on the same + connection while the transaction is still open. A READONLY commit + inside a journaled write operation is recovered the way MATLAB + recovers a READONLY statement -- reopen, then re-run -- except that + what is re-run is the whole transaction, replayed and verified by + :meth:`_recover_transaction`, and then committed. """ try: self.dbid.commit() - return except sqlite3.Error as err: last_err = err + else: + self._journal_committed() + return + if self._journal is not None and self._is_readonly_error(last_err): + self._recover_transaction(last_err, commit=True) + self._journal_committed() + return if self._is_retryable_sql_error(last_err): for attempt, backoff in enumerate(_RETRY_BACKOFFS, start=1): if not self.dbid.in_transaction: @@ -379,14 +484,229 @@ def _commit(self): time.sleep(backoff) try: self.dbid.commit() - return except sqlite3.Error as err: last_err = err + else: + self._journal_committed() + return if not self._is_retryable_sql_error(last_err): break self._report_sql_error(last_err) raise last_err + def _rollback(self): + """Roll back, if there is still a connection to roll back. + + After a recovery whose every reopen failed there may be none; the + error that caused that is the one the caller should see. + """ + if self.dbid is not None: + self.dbid.rollback() + + def _journal_cursor(self): + """A cursor for a write operation: recording, when one is journaled.""" + if self._journal is None: + return self.dbid.cursor() + return _JournalCursor(self, self.dbid.cursor()) + + def _journal_committed(self): + """What the journal held is on disk now; a later replay starts here.""" + if self._journal is None: + return + for entry in self._journal: + entry["proxy"]._entry = None + self._journal.clear() + + def _record(self, proxy, method, sql, params): + entry = { + "method": method, + "sql": sql, + "params": params, + "rowcount": proxy._cursor.rowcount, + # Only an INSERT that inserted has a lastrowid of its own: after + # executemany, or an INSERT OR IGNORE that ignored, sqlite3 + # reports a stale value (or None) that no later statement used. + "lastrowid": ( + proxy._cursor.lastrowid + if method == "execute" + and _is_insert(sql) + and proxy._cursor.rowcount > 0 + else None + ), + "rows": [], + "complete": False, + "proxy": proxy, + } + self._journal.append(entry) + proxy._entry = entry + + def _journaled_execute(self, proxy, method, sql, params): + """Run one statement of a journaled write operation. + + A READONLY failure here is the poisoned-handle case (typically + SQLITE_READONLY_DBMOVED), which SQLite reports at the first write of + a transaction. It is recovered by :meth:`_recover_transaction`, which + replays what came before on a fresh connection and then runs this + statement there. + """ + try: + getattr(proxy._cursor, method)(sql, params) + except sqlite3.Error as err: + if not self._is_readonly_error(err): + raise + self._recover_transaction(err, pending=(proxy, method, sql, params)) + return proxy + self._record(proxy, method, sql, params) + return proxy + + def _replay(self, journal): + """Re-run ``journal`` on the current connection, checking each result. + + Every statement must do what it did the first time: the same + rowcount, the same lastrowid for an INSERT (later statements were + built from it), and the same rows for anything the caller read. Any + difference means the database changed in between -- another writer, + or a different file now at the path -- and the decisions the + operation made from those reads no longer hold, so the replay is + refused rather than committed. + """ + cursors = [] + for number, entry in enumerate(journal, start=1): + cursor = self.dbid.cursor() + try: + getattr(cursor, entry["method"])(entry["sql"], entry["params"]) + except sqlite3.Error as err: + if self._is_retryable_sql_error(err): + raise # transient: the next attempt may get through + # It succeeded the first time; failing now (a UNIQUE row + # that is already there, a missing table) is a difference. + raise _ReplayDiverged( + f"statement {number} failed on replay: {err}" + ) from err + if cursor.rowcount != entry["rowcount"]: + raise _ReplayDiverged( + f"statement {number} changed {cursor.rowcount} rows on " + f"replay, {entry['rowcount']} the first time" + ) + if entry["lastrowid"] is not None and ( + cursor.lastrowid != entry["lastrowid"] + ): + raise _ReplayDiverged( + f"statement {number} inserted row {cursor.lastrowid} on " + f"replay, {entry['lastrowid']} the first time" + ) + seen = entry["rows"] + if entry["complete"]: + again = [tuple(row) for row in cursor.fetchall()] + elif seen: + again = [tuple(row) for row in cursor.fetchmany(len(seen))] + else: + again = [] + if again != seen: + raise _ReplayDiverged( + f"statement {number} read different rows on replay" + ) + cursors.append((entry, cursor)) + return cursors + + def _recover_transaction(self, err, pending=None, commit=False): + """Reopen, replay the journaled transaction, and carry on. + + DID-matlab reopens the connection before retrying a READONLY + statement (da83fa3): a poisoned handle -- SQLITE_READONLY_DBMOVED, + SQLite believing the file moved out from under it -- is cleared by + reopening, never by waiting. mksqlite autocommits each statement, so + there the failed statement is all there is to re-run. Here it is part + of a transaction whose earlier statements die with the handle, so all + of them are replayed on the fresh connection and verified (see + :meth:`_replay`) before the operation continues. Only SQL is + replayed: files the operation already copied stay where they are, + which is why the operation itself is not simply run again. + + ``pending`` is the statement that failed, run after the replay; with + ``commit`` the replayed transaction is committed. On the same backoff + schedule as the statement retry; a replay that diverges is never + retried, and the original error is raised. + """ + journal = self._journal + last_err = err + for attempt, backoff in enumerate(_RETRY_BACKOFFS, start=1): + if self.debug: + print( + f"DID:SQLITEDB readonly inside a write operation (attempt " + f"{attempt}/{len(_RETRY_BACKOFFS)}, {backoff:g} s backoff): " + f"reopening and replaying {len(journal)} statement(s)", + file=sys.stderr, + ) + # Let go of the dead transaction first, so its locks do not hold + # other writers up for the length of the backoff. + if self.dbid is not None: + with contextlib.suppress(sqlite3.Error): + self.dbid.rollback() + time.sleep(backoff) + # Suspended while reopening: _open_db commits on its own. + self._journal = None + try: + self._reopen_db() + if self.dbid is None: + continue + try: + cursors = self._replay(journal) + if commit: + self.dbid.commit() + except _ReplayDiverged as diverged: + with contextlib.suppress(sqlite3.Error): + self.dbid.rollback() + print( + "DID:SQLITEDB could not recover the write operation: " + f"{diverged}. Nothing from it was written.", + file=sys.stderr, + ) + self._report_sql_error(err) + raise err from diverged + except sqlite3.Error as replay_err: + with contextlib.suppress(sqlite3.Error): + self.dbid.rollback() + last_err = replay_err + if not self._is_retryable_sql_error(replay_err): + break + continue + finally: + self._journal = journal + + # The caller's cursors now point at the fresh connection, each at + # the result of its last statement. + for entry, cursor in cursors: + entry["proxy"]._cursor = cursor + if pending is not None: + proxy, method, sql, params = pending + proxy._cursor = self.dbid.cursor() + try: + getattr(proxy._cursor, method)(sql, params) + except sqlite3.Error as pending_err: + if not self._is_readonly_error(pending_err): + # An ordinary failure of the statement itself, as it + # would have been the first time: the transaction + # stays open for the caller, just as SQLite leaves it. + raise + last_err = pending_err + continue + self._record(proxy, method, sql, params) + if self.debug: + print( + f"DID:SQLITEDB write operation recovered on attempt " + f"{attempt}/{len(_RETRY_BACKOFFS)} (after reopen)", + file=sys.stderr, + ) + return + if self.dbid is None: + # Every reopen failed. Leave a connection behind if the path has + # come back, so the caller's rollback does not hide this error. + with contextlib.suppress(sqlite3.Error, OSError): + self._open_db() + self._report_sql_error(last_err) + raise last_err + def _reopen_db(self): """Close and reopen the connection, leaving dbid None if that fails.""" with contextlib.suppress(Exception): @@ -498,10 +818,11 @@ def _do_get_branch_ids(self): rows = self.do_run_sql_query("SELECT DISTINCT branch_id FROM branches") return [row["branch_id"] for row in rows] + @_journaled def _do_add_branch(self, branch_id, parent_branch_id): import time - cursor = self.dbid.cursor() + cursor = self._journal_cursor() # Handle empty string parent as NULL if parent_branch_id == "": @@ -657,6 +978,7 @@ def _normalize_loaded_props(props): return props + @_journaled def _do_add_doc( self, document_obj, @@ -676,7 +998,7 @@ def _do_add_doc( ) doc_id = document_obj.id() - cursor = self.dbid.cursor() + cursor = self._journal_cursor() # Validate branch existence BEFORE any INSERT. Previously the docs and # doc_data rows were inserted first and the missing branch was only @@ -788,7 +1110,7 @@ def _do_add_doc( except Exception: # Every failure path rolls back so no partial rows survive into the # next transaction, and the fields cache is restored to match. - self.dbid.rollback() + self._rollback() self._fields_cache = cache_before raise @@ -2551,8 +2873,9 @@ def _remove_cached_files(paths): except OSError: pass + @_journaled def _do_remove_doc(self, document_id, branch_id, **kwargs): - cursor = self.dbid.cursor() + cursor = self._journal_cursor() # Check if branch exists cursor.execute("SELECT 1 FROM branches WHERE branch_id = ?", (branch_id,)) @@ -2590,7 +2913,7 @@ def _do_remove_doc(self, document_id, branch_id, **kwargs): # As in _do_add_doc: a failure part-way through must not leave # the branch_docs delete sitting in an open transaction for the # next commit on this connection to pick up. - self.dbid.rollback() + self._rollback() raise # Only once the rows are committed - see _reclaim_unreferenced_doc. @@ -2603,6 +2926,7 @@ def _do_remove_doc(self, document_id, branch_id, **kwargs): elif on_missing != "ignore": raise ValueError(f"Document id '{document_id}' not found for removal.") + @_journaled def _do_delete_branch(self, branch_id): """Delete a branch, reclaiming any document it was the last to hold. @@ -2613,7 +2937,7 @@ def _do_delete_branch(self, branch_id): holds is left completely alone. Deleting documents from one branch does not delete them from other branches. """ - cursor = self.dbid.cursor() + cursor = self._journal_cursor() cached_paths = [] try: # Note which documents this branch holds BEFORE dropping its rows: @@ -2641,7 +2965,7 @@ def _do_delete_branch(self, branch_id): self._commit() except Exception: - self.dbid.rollback() + self._rollback() raise # Only once the rows are committed - see _reclaim_unreferenced_doc. diff --git a/tests/test_sqlitedb_transient_errors.py b/tests/test_sqlitedb_transient_errors.py index dea3c9e..6cbae46 100644 --- a/tests/test_sqlitedb_transient_errors.py +++ b/tests/test_sqlitedb_transient_errors.py @@ -18,10 +18,12 @@ """ import os +import shutil import sqlite3 import pytest +from did.document import Document from did.implementations import sqlitedb as sqlitedb_module from did.implementations.sqlitedb import SQLiteDB @@ -267,3 +269,255 @@ def refuse(*args, **kwargs): monkeypatch.setattr(sqlitedb_module.tempfile, "NamedTemporaryFile", refuse) diag = db._collect_readonly_diagnostics() assert "db dir writable: NO (Read-only file system)" in diag + + +def _move_file_under_handle(path): + """Replace the database with a byte-identical copy at the same path. + + The open connection still holds the old inode, so its next write fails + with SQLITE_READONLY_DBMOVED ("attempt to write a readonly database") -- + the poisoned handle da83fa3 reopens to clear. No mocking: this is the + real SQLite failure. + """ + shutil.copyfile(path, path + ".moved") + os.replace(path + ".moved", path) + + +def _doc(idhex, value=1): + return Document( + { + "base": {"id": idhex, "datestamp": "2020-01-01"}, + "document_class": {"class_name": "thing"}, + "demoA": {"value": value}, + } + ) + + +def _doc_ids_on_disk(path): + with sqlite3.connect(path) as check: + return {row[0] for row in check.execute("SELECT doc_id FROM docs")} + + +class TestPoisonedHandleRecovery: + """A READONLY inside a write operation: reopen, replay, carry on.""" + + def test_the_failure_is_real(self, db): + db.add_branch("main") + _move_file_under_handle(db.connection) + with pytest.raises(sqlite3.OperationalError, match="readonly"): + db.dbid.execute("INSERT INTO branches VALUES ('x', NULL, 0)") + + def test_add_doc_is_recovered_after_the_file_moved(self, db, sleeps): + db.add_branch("main") + poisoned = db.dbid + _move_file_under_handle(db.connection) + + db.add_docs([_doc("a" * 32)], branch_id="main", validate=False) + + assert sleeps == [0.1] + assert db.dbid is not poisoned + assert "a" * 32 in _doc_ids_on_disk(db.connection) + got = db.get_docs("a" * 32, branch_id="main") + assert got.document_properties["demoA"]["value"] == 1 + + def test_add_branch_is_recovered(self, db, sleeps): + db.add_branch("main") + _move_file_under_handle(db.connection) + + db.add_branch("b2", parent_branch_id="main") + + assert sleeps == [0.1] + assert {"main", "b2"} <= _branch_ids(db.connection) + + def test_remove_doc_is_recovered(self, db, sleeps): + db.add_branch("main") + db.add_docs([_doc("b" * 32)], branch_id="main", validate=False) + _move_file_under_handle(db.connection) + + db.remove_docs(["b" * 32], branch_id="main") + + assert sleeps == [0.1] + assert "b" * 32 not in _doc_ids_on_disk(db.connection) + + def test_the_journal_ends_with_the_operation(self, db, sleeps): + db.add_branch("main") + _move_file_under_handle(db.connection) + db.add_docs([_doc("c" * 32)], branch_id="main", validate=False) + assert db._journal is None + # The next operation replays nothing from the last one. + _move_file_under_handle(db.connection) + db.add_docs([_doc("d" * 32, 2)], branch_id="main", validate=False) + assert {"c" * 32, "d" * 32} <= _doc_ids_on_disk(db.connection) + + def test_a_replay_that_reads_a_different_database_is_refused( + self, db, sleeps, capsys + ): + # The file now at the path is an older copy, without the branch the + # operation already checked for on the poisoned handle. Committing a + # replay against it would write a document into a branch that does + # not exist there, so it is refused and nothing is written. + snapshot = db.connection + ".before" + shutil.copyfile(db.connection, snapshot) + db.add_branch("main") + os.replace(snapshot, db.connection) + + with pytest.raises(sqlite3.OperationalError, match="readonly"): + db.add_docs([_doc("e" * 32)], branch_id="main", validate=False) + + assert sleeps == [0.1] # diverged: not retried + assert "e" * 32 not in _doc_ids_on_disk(db.connection) + assert "read different rows on replay" in capsys.readouterr().err + + def test_without_a_journal_a_readonly_handle_is_not_replayed(self, db, sleeps): + # Outside a write operation, do_run_sql_query's own retry applies. + assert db._journal is None + db.add_branch("main") + _move_file_under_handle(db.connection) + db.do_run_sql_query(INSERT_BRANCH, ("x", None, 0.0)) + db._commit() + assert "x" in _branch_ids(db.connection) + + +class TestRecoveryThatCannotReopen: + def test_the_original_error_reaches_the_caller(self, db, sleeps, monkeypatch): + db.add_branch("main") + db.dbid = _ReadonlyAtCommit(db.dbid) + real_open = SQLiteDB._open_db + + def refuse(self): + raise sqlite3.OperationalError("unable to open database file") + + monkeypatch.setattr(SQLiteDB, "_open_db", refuse) + with pytest.raises(sqlite3.OperationalError): + db.add_docs([_doc("7" * 32)], branch_id="main", validate=False) + assert len(sleeps) == 6 # every attempt tried to reopen + + monkeypatch.setattr(SQLiteDB, "_open_db", real_open) + db._open_db() + assert "7" * 32 not in _doc_ids_on_disk(db.connection) + + +class _ReadonlyAtCommit: + """A connection whose next COMMITs fail READONLY, as a poisoned one does. + + SQLite reports DBMOVED at a transaction's first write, so a READONLY that + reaches COMMIT comes from elsewhere (a hot journal, a remount); it cannot + be produced on demand, so this one is a stand-in around a real handle. + """ + + def __init__(self, conn): + self._conn = conn + + def commit(self): + # Every time: waiting never clears a poisoned handle, only a reopen. + raise sqlite3.OperationalError("attempt to write a readonly database") + + def __getattr__(self, name): + return getattr(self._conn, name) + + +class TestReadonlyCommitRecovery: + def test_a_readonly_commit_is_replayed_on_a_fresh_connection(self, db, sleeps): + db.add_branch("main") + db.dbid = _ReadonlyAtCommit(db.dbid) + + db.add_docs([_doc("f" * 32)], branch_id="main", validate=False) + + assert sleeps == [0.1] + assert isinstance(db.dbid, sqlite3.Connection) # reopened + assert "f" * 32 in _doc_ids_on_disk(db.connection) + + def test_files_ingested_before_the_commit_are_not_needed_again(self, db, sleeps): + # The ingest copies the file into FileDir and, with delete_original, + # deletes the source -- all before COMMIT. Running the operation + # again would fail on the missing source; replaying its SQL does not + # need the file at all. + db.add_branch("main") + source = os.path.join(os.path.dirname(db.connection), "source.bin") + with open(source, "wb") as handle: + handle.write(b"payload") + doc = Document("demoFile", **{"demoFile.value": 1}) + doc.add_file("filename1.ext", "placeholder", ingest=0, delete_original=0) + doc.add_file("filename2.ext", "placeholder", ingest=0, delete_original=0) + _, info, _ = doc.is_in_file_list("filename1.ext") + info["locations"] = [ + { + "location": source, + "location_type": "file", + "uid": "u-local", + "ingest": 1, + "delete_original": 1, + "parameters": "", + } + ] + db.dbid = _ReadonlyAtCommit(db.dbid) + + db.add_docs([doc], branch_id="main", validate=False) + + assert sleeps == [0.1] + assert not os.path.exists(source) + file_obj = db.open_doc(doc.id(), "filename1.ext") + file_obj.fopen() + try: + assert file_obj.fread() == b"payload" + finally: + file_obj.fclose() + + def test_a_busy_commit_is_still_waited_out_not_replayed(self, db, sleeps): + db.add_branch("main") + reader = sqlite3.connect(db.connection, isolation_level=None) + reader.execute("BEGIN") + reader.execute("SELECT * FROM branches").fetchall() + sleeps.on_sleep.append(reader.rollback) + handle = db.dbid + + db.add_docs([_doc("1" * 32)], branch_id="main", validate=False) + + assert db.dbid is handle + assert "1" * 32 in _doc_ids_on_disk(db.connection) + reader.close() + + +class TestReplayRefusesADifferentDatabase: + """A replay that does not do what the original did is never committed.""" + + def test_another_writer_in_between_is_refused(self, db, sleeps, capsys): + # While we wait to reopen, another connection adds a document. Our + # replayed INSERT then gets a different doc_idx, so every doc_data + # row built from the first one would point at the wrong document. + db.add_branch("main") + db.dbid = _ReadonlyAtCommit(db.dbid) + other = SQLiteDB(db.connection) + sleeps.on_sleep.append( + lambda: other.add_docs([_doc("8" * 32)], "main", validate=False) + ) + + with pytest.raises(sqlite3.OperationalError, match="readonly"): + db.add_docs([_doc("9" * 32)], branch_id="main", validate=False) + + on_disk = _doc_ids_on_disk(db.connection) + assert "8" * 32 in on_disk + assert "9" * 32 not in on_disk + assert sleeps == [0.1] # a divergence is not retried + assert "inserted row" in capsys.readouterr().err + other._close_db() + + def test_a_statement_that_fails_on_replay_is_divergence(self, db, sleeps): + # Replaying a journal that is already on disk: its first INSERT now + # hits a UNIQUE row. That is reported as divergence, not as the + # IntegrityError, so the caller still sees the original failure. + db.add_branch("main") + journal = [] + original = db._journal_committed + + def capture(): + journal.extend(dict(entry) for entry in db._journal) + original() + + db._journal_committed = capture + db.add_branch("b2", parent_branch_id="main") + db._journal_committed = original + + with pytest.raises(sqlitedb_module._ReplayDiverged, match="failed on replay"): + db._replay(journal) + db.dbid.rollback()