Native: run_in_transaction takes the commit-count witness from the begin reply (4 to 3 crossings)

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:

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(

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,15 +69,15 @@ 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)
             # the transaction is open on the Java side; the interrupt lands
             # 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)."""