diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 3c876fe..8fb321f 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -583,13 +583,21 @@ def _tx(self, op: int, prefix: str) -> dict: return _call(prefix, "arcadedb_n_tx", self._h, op) def begin(self): - self._tx(0, "Failed to begin transaction") + self._begin_state() + + def _begin_state(self) -> dict: + """begin(), and the state right after it: the reply of ``arcadedb_n_tx`` carries ``active``, ``depth`` and ``commits``.""" + return self._tx(0, "Failed to begin transaction") def commit(self): self._tx(1, "Failed to commit transaction") def rollback(self): - self._tx(2, "Failed to rollback transaction") + self._rollback_state() + + def _rollback_state(self) -> dict: + """rollback(), and the state right after it (the reply carries it, as begin's does): the unwind reads it instead of asking.""" + return self._tx(2, "Failed to rollback transaction") def transaction(self) -> TransactionContext: return TransactionContext(self) @@ -605,40 +613,60 @@ def _transaction_commit_mark(self): """The commit count of the current transaction (the witness that a BATCH boundary published part of its work).""" return self._tx(3, "Failed to read the transaction state")["commits"] - def _end_transactions_opened_since(self, mark): + def _unwind_opened_since(self, mark): + """Roll back every level opened since ``mark``, and no other; returns ``(state, first_error)``. + + ``state`` is the transaction state after the last crossing (every rollback reply carries it, so no read sits between two + steps), or None when that crossing failed. A level the caller held at the mark is theirs, as in the JPype ``Database``. + """ was_active, depth_before = mark first_error = None state = self._tx(3, "Failed to read the transaction state") for _ in range(state["depth"] - depth_before): - if not self.is_transaction_active(): + if state is None: + state = self._tx(3, "Failed to check transaction status") + if not state["active"]: break try: - self.rollback() - except Exception as e: - first_error = first_error or e - if not was_active and self.is_transaction_active(): - try: - self.rollback() + state = self._rollback_state() except Exception as e: + state = None first_error = first_error or e + if not was_active: + if state is None: + state = self._tx(3, "Failed to check transaction status") + if state["active"]: + try: + state = self._rollback_state() + except Exception as e: + state = None + first_error = first_error or e + return state, first_error + + def _end_transactions_opened_since(self, mark): + _, first_error = self._unwind_opened_since(mark) if first_error is not None: raise first_error def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): - """``fn`` inside a transaction, retried on NeedRetry conflicts (see the JPype ``Database.run_in_transaction``).""" + """``fn`` inside a transaction, retried on NeedRetry conflicts (see the JPype ``Database.run_in_transaction``). + + Three crossings on the happy path: the mark before begin (what this call owns), begin (its reply is the commit-count + witness) and commit. The mark must stay before begin: a mark read after it would count the level begin pushed as the caller's. + """ duplicated_key_retried = False for attempt in range(retries + 1): mark = self._transaction_stack_mark() commits_at_start = None try: - self.begin() - commits_at_start = self._transaction_commit_mark() + commits_at_start = self._begin_state()["commits"] result = fn() self.commit() return result except BaseException as e: + state = None try: - self._end_transactions_opened_since(mark) + state, _ = self._unwind_opened_since(mark) except Exception: # nosec B110 pass if attempt >= retries or not isinstance(e, Exception): @@ -656,7 +684,11 @@ def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): # durable, and a replay would apply it twice. Only the READ of the witness may fail quietly (no witness: # retry, as before); the raise sits outside that try, or its own except would swallow it. try: - commits_now = self._transaction_commit_mark() + commits_now = ( + state["commits"] + if state is not None + else self._transaction_commit_mark() + ) except ArcadeDBError: commits_now = commits_at_start if commits_now != commits_at_start: diff --git a/tests/conftest.py b/tests/conftest.py index d15d9ed..7669916 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -56,6 +56,8 @@ collect_ignore.append("test_native_point_read.py") # the vector provider the native glue picks by default (ARCADEDB_VECTOR_PROVIDER, #153), observed on the native library collect_ignore.append("test_native_vector_provider.py") + # crossing counts of the transaction entry points (lib.call is the native library's ctypes door) + collect_ignore.append("test_native_tx_crossings.py") # counts the statement texts the native backend hands to the engine (the JPype backend has no such layer) collect_ignore.append("test_native_statement_texts.py") if not os.path.isfile( diff --git a/tests/test_interrupt_hygiene.py b/tests/test_interrupt_hygiene.py index 10e0bae..53e7791 100644 --- a/tests/test_interrupt_hygiene.py +++ b/tests/test_interrupt_hygiene.py @@ -25,9 +25,17 @@ from arcadedb.server import ArcadeDBServer # the wrapper-level checks run against the class the active build's databases are instances of -if arcadedb.__backend__ == "native": +_NATIVE = arcadedb.__backend__ == "native" +if _NATIVE: Database = importlib.import_module("arcadedb._native.database").NativeDatabase +# The methods ``run_in_transaction`` itself calls to open a level and to close one, which is where these tests inject a failure. The +# JPype Database calls begin() and rollback(). The native one calls _begin_state() and _rollback_state(): the same crossings, but +# their reply carries the transaction state, so the call reads none of it again. begin() and rollback() are thin wrappers over them, +# so a patch of begin() never reaches run_in_transaction on the native Database. +BEGIN = "_begin_state" if _NATIVE else "begin" +ROLLBACK = "_rollback_state" if _NATIVE else "rollback" + @pytest.fixture def db(temp_db_path): @@ -61,7 +69,7 @@ class TestRunInTransactionBeginHygiene: def test_interrupt_after_the_transaction_opened_rolls_it_back( self, db, monkeypatch ): - real_begin = Database.begin + real_begin = getattr(Database, BEGIN) def begin_then_interrupt(self): real_begin(self) @@ -69,7 +77,7 @@ def begin_then_interrupt(self): # before begin() hands control back to run_in_transaction raise KeyboardInterrupt - monkeypatch.setattr(Database, "begin", begin_then_interrupt) + monkeypatch.setattr(Database, BEGIN, begin_then_interrupt) with pytest.raises(KeyboardInterrupt): db.run_in_transaction(lambda: None) @@ -100,7 +108,7 @@ def test_a_begin_that_refuses_leaves_the_callers_transaction_alone( def refusing_begin(self): raise ArcadeDBError("Failed to begin transaction: refused") - monkeypatch.setattr(Database, "begin", refusing_begin) + monkeypatch.setattr(Database, BEGIN, refusing_begin) with pytest.raises(ArcadeDBError): db.run_in_transaction(lambda: None) monkeypatch.undo() @@ -119,14 +127,14 @@ def test_interrupt_after_a_nested_begin_pops_only_that_level(self, db, monkeypat db.new_document("R").set("k", "caller").save() assert _depth(db) == 1 - real_begin = Database.begin + real_begin = getattr(Database, BEGIN) def begin_then_interrupt(self): real_begin(self) assert _depth(self) == 2 raise KeyboardInterrupt - monkeypatch.setattr(Database, "begin", begin_then_interrupt) + monkeypatch.setattr(Database, BEGIN, begin_then_interrupt) with pytest.raises(KeyboardInterrupt): db.run_in_transaction(lambda: None) monkeypatch.undo() @@ -184,16 +192,17 @@ def opens_another_level_then_fails(): db.begin() raise ValueError("boom") - real_rollback = Database.rollback + real_rollback = getattr(Database, ROLLBACK) calls = [] def first_rollback_fails_after_popping(self): - real_rollback(self) + state = real_rollback(self) calls.append(1) if len(calls) == 1: raise ArcadeDBError("Failed to rollback transaction: boom") + return state - monkeypatch.setattr(Database, "rollback", first_rollback_fails_after_popping) + monkeypatch.setattr(Database, ROLLBACK, first_rollback_fails_after_popping) with pytest.raises(ValueError): db.run_in_transaction(opens_another_level_then_fails) monkeypatch.undo() @@ -219,6 +228,14 @@ def failing(): assert _count(db) == 1 +if _NATIVE: + # the native begin reply carries the commit mark, so there is no separate read to interrupt (the interrupt after begin above + # interrupts that reply). Removed, not skipped: scripts/check_test_skips.py fails a native run on any skip. + del ( + TestRunInTransactionBeginHygiene.test_interrupt_taking_the_commit_mark_rolls_the_transaction_back + ) + + class TestTransactionContextBeginHygiene: """``with db.transaction():`` calls begin() in __enter__, and an exception out of __enter__ means __exit__ never runs, so __enter__ cleans up (#9970).""" diff --git a/tests/test_native_tx_crossings.py b/tests/test_native_tx_crossings.py new file mode 100644 index 0000000..0a8c04c --- /dev/null +++ b/tests/test_native_tx_crossings.py @@ -0,0 +1,190 @@ +"""The native wheel spends the fewest library crossings a transaction needs. + +Every ``arcadedb_n_tx`` call is one trip into the native library: a ctypes call, a malloc'd JSON reply, ``string_at``, a decode and a +``json.loads``. The reply of begin, commit and rollback already carries the whole transaction state (``active``, ``depth`` and +``commits``), so a caller that needs the state right after one of these calls must not ask for it again with a mark. + +The counts are crossings, not seconds: they are deterministic. The behaviour tests next to them pin what the marks are for (a caller's +transaction is never rolled back by a nested block, a retry rolls the failed attempt back), so a lower count can not buy a different result. + +Needs the native library, so conftest.py does not collect this file on the JPype backend (a skip there would fail +scripts/check_test_skips.py); it runs in the native parity run. +""" + +import arcadedb as arcadedb +import pytest +from arcadedb import ArcadeDBError +from arcadedb._native import lib + +_OPS = {0: "begin", 1: "commit", 2: "rollback", 3: "mark"} + + +class _Crossings: + """Counts every ``lib.call`` made while it is installed; ``tx`` is the sequence of ``arcadedb_n_tx`` operations.""" + + def __init__(self, monkeypatch): + self.calls = [] + real = lib.call + + def counting(name, *args): + self.calls.append(_OPS[args[1]] if name == "arcadedb_n_tx" else name) + return real(name, *args) + + monkeypatch.setattr(lib, "call", counting) + + @property + def tx(self): + return [c for c in self.calls if c in _OPS.values()] + + def reset(self): + self.calls.clear() + + +@pytest.fixture +def db(temp_db_path): + database = arcadedb.create_database(temp_db_path) + database.command("sql", "CREATE DOCUMENT TYPE R") + with ( + database.transaction() + ): # the thread's transaction context exists from here on, as it does in a running application + pass + yield database + database.drop() + + +@pytest.fixture +def crossings(db, monkeypatch): + return _Crossings(monkeypatch) + + +def _count(db): + return db.query("sql", "SELECT count(*) AS c FROM R").first().get("c") + + +def _need_retry(): + error = ArcadeDBError( + "Failed to commit transaction: ConcurrentModificationException" + ) + error.native = {"retry": "need_retry", "retry_ms": 0} + return error + + +# -- crossing counts + + +def test_run_in_transaction_happy_path_is_three_crossings(db, crossings): + # the mark before begin (what this call owns), begin (its reply is the commit-count witness), commit + assert db.run_in_transaction(lambda: 42) == 42 + assert crossings.calls == ["mark", "begin", "commit"] + + +def test_with_transaction_is_three_crossings(db, crossings): + with db.transaction(): + pass + assert crossings.calls == ["mark", "begin", "commit"] + + +def test_with_transaction_that_raises_is_three_crossings(db, crossings): + with pytest.raises(ValueError): + with db.transaction(): + raise ValueError("boom") + assert crossings.calls == ["mark", "begin", "rollback"] + + +def test_run_in_transaction_that_raises_reads_the_state_once(db, crossings): + # mark, begin, then one state read for the unwind: the rollback replies carry the state, so no is-active read and no commit-count read + with pytest.raises(ValueError): + db.run_in_transaction(lambda: (_ for _ in ()).throw(ValueError("boom"))) + assert crossings.calls == ["mark", "begin", "mark", "rollback"] + assert not db.is_transaction_active() + + +def test_run_in_transaction_retry_costs_one_unwind_and_one_attempt(db, crossings): + attempts = [] + + def fn(): + attempts.append(1) + if len(attempts) == 1: + raise _need_retry() + return "done" + + assert db.run_in_transaction(fn, backoff_s=0) == "done" + assert len(attempts) == 2 + assert crossings.calls == [ + *["mark", "begin", "mark", "rollback"], # the refused attempt, unwound + *["mark", "begin", "commit"], # the attempt that commits + ] + + +# -- what the marks are for: the same results at the lower count + + +def test_a_nested_block_that_fails_leaves_the_callers_transaction_open(db): + with db.transaction(): + db.command("sql", "INSERT INTO R SET k = 1") + with pytest.raises(ValueError): + db.run_in_transaction( + lambda: ( + db.command("sql", "INSERT INTO R SET k = 2"), + (_ for _ in ()).throw(ValueError("boom")), + ) + ) + assert db.is_transaction_active() + assert _count(db) == 1 + + +def test_a_nested_block_that_succeeds_commits_with_the_caller(db): + with db.transaction(): + db.run_in_transaction(lambda: db.command("sql", "INSERT INTO R SET k = 1")) + db.command("sql", "INSERT INTO R SET k = 2") + assert _count(db) == 2 + + +def test_a_refused_attempt_is_rolled_back_before_the_retry(db): + attempts = [] + + def fn(): + attempts.append(1) + db.command("sql", "INSERT INTO R SET k = 1") + if len(attempts) == 1: + raise _need_retry() + + db.run_in_transaction(fn, backoff_s=0) + assert len(attempts) == 2 + assert _count(db) == 1 + assert not db.is_transaction_active() + + +def test_a_duplicated_key_is_retried_once(db): + attempts = [] + + def fn(): + attempts.append(1) + error = ArcadeDBError("duplicated key") + error.native = {"retry": "duplicated_key", "retry_ms": 0} + raise error + + with pytest.raises(ArcadeDBError): + db.run_in_transaction(fn, backoff_s=0) + assert len(attempts) == 2 + assert not db.is_transaction_active() + + +def test_an_error_without_a_retry_kind_is_raised_after_one_attempt(db): + attempts = [] + + def fn(): + attempts.append(1) + raise ArcadeDBError("not a conflict") + + with pytest.raises(ArcadeDBError): + db.run_in_transaction(fn, backoff_s=0) + assert attempts == [1] + assert not db.is_transaction_active() + + +def test_begin_commit_rollback_return_nothing(db): + assert db.begin() is None + assert db.rollback() is None + assert db.begin() is None + assert db.commit() is None