From d7ba23cafd8a1d9a5e08a28f58819901b12010a6 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Tue, 3 Nov 2026 08:21:24 +0900 Subject: [PATCH] Native: run_in_transaction takes the commit-count witness from the begin reply, and the unwind reads no state it already has Port of humemai/arcadedb-embedded-python#522 (branch perf/native-tx-crossings, never merged there) onto current main; the P-03 guard of that PR's base is already here. run_in_transaction made 4 library crossings on the happy path (mark, begin, commit mark, commit) although the begin reply already carries {"active","depth","commits"}: the witness now comes from that reply, 3 crossings (the mark must stay before begin). The unwind after a failed block asked is_transaction_active() before every rollback and once more at the end, and the retry check asked for the commit count again; every rollback reply carries the same state, so the unwind returns the state of its last crossing and the retry check reads the commit count from it (a failed crossing falls back to a read). Failed block 6 -> 4 crossings, refused and retried attempt 11 -> 7. begin() and rollback() stay public and return None; _begin_state() and _rollback_state() are the seams run_in_transaction calls. test_interrupt_hygiene.py patched begin and rollback, which the native run_in_transaction no longer calls, so it patches the seam the active backend calls. test_native_tx_crossings.py counts the arcadedb_n_tx operations and pins the behaviour the marks exist for. Counts only, no speed claim: a timed A/B decides it before merge. Refs #19 Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/database.py | 62 +++++++--- tests/conftest.py | 2 + tests/test_interrupt_hygiene.py | 35 ++++-- tests/test_native_tx_crossings.py | 190 ++++++++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 24 deletions(-) create mode 100644 tests/test_native_tx_crossings.py 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 c5961c2..d9f5d89 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") if not os.path.isfile( os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md") ): 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