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()