diff --git a/bindings/python/docs/api/database.md b/bindings/python/docs/api/database.md index 683a0036b94..cf744c29647 100644 --- a/bindings/python/docs/api/database.md +++ b/bindings/python/docs/api/database.md @@ -364,6 +364,8 @@ and `fn` is re-executed, with linear backoff. The `with db.transaction():` conte manager cannot retry (a `with` block can't be re-entered), so use this for contended multi-threaded writes. +A block that already made part of its work durable is never re-run: a statement with a `BATCH n` boundary, or a `commit()` inside `fn`, publishes rows that the rollback cannot take back, so a replay would apply them twice. When such a block hits a retryable conflict, the conflict is raised as it is (the same `ArcadeDBError` the last attempt failed with), after the rollback of what was still open. The JVM and the native wheel behave the same. + Any other way out of `fn` rolls back too, and then propagates: a plain bug (`TypeError`, `KeyError`) and `KeyboardInterrupt` / `SystemExit` included, matching the Java side's `catch (final Throwable e)`. Nothing leaves an open transaction behind for diff --git a/bindings/python/src/arcadedb_embedded/_native/database.py b/bindings/python/src/arcadedb_embedded/_native/database.py index 68fe220b894..7123524bae9 100644 --- a/bindings/python/src/arcadedb_embedded/_native/database.py +++ b/bindings/python/src/arcadedb_embedded/_native/database.py @@ -545,13 +545,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) @@ -567,40 +575,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): @@ -614,11 +642,19 @@ def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): raise duplicated_key_retried = True if commits_at_start is not None: + # #7916: the rollback above cannot take back what a BATCH boundary or a commit() inside the block already made + # 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: - if self._transaction_commit_mark() != commits_at_start: - raise + commits_now = ( + state["commits"] + if state is not None + else self._transaction_commit_mark() + ) except ArcadeDBError: - pass + commits_now = commits_at_start + if commits_now != commits_at_start: + raise _time.sleep( max(backoff_s * (attempt + 1), reply.get("retry_ms", 0) / 1000.0) ) diff --git a/bindings/python/tests/conftest.py b/bindings/python/tests/conftest.py index 77820a36136..52f0b81c0d9 100644 --- a/bindings/python/tests/conftest.py +++ b/bindings/python/tests/conftest.py @@ -29,6 +29,8 @@ collect_ignore.append("test_native_udf.py") # the native library's start-up output (#477) can only be checked with the native library loaded collect_ignore.append("test_native_startup_runtime.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.isdir(os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs")): # the upstream pull request branch has no docs/ directory collect_ignore.append("test_docs_examples.py") diff --git a/bindings/python/tests/test_interrupt_hygiene.py b/bindings/python/tests/test_interrupt_hygiene.py index 71ca7ef86b3..099c3a9c2e5 100644 --- a/bindings/python/tests/test_interrupt_hygiene.py +++ b/bindings/python/tests/test_interrupt_hygiene.py @@ -25,11 +25,19 @@ from arcadedb_embedded.server import ArcadeDBServer # the wrapper-level checks run against the class the active backend's databases are instances of -if os.environ.get("ARCADEDB_BACKEND", "jpype").lower() == "native": +_NATIVE = os.environ.get("ARCADEDB_BACKEND", "jpype").lower() == "native" +if _NATIVE: Database = importlib.import_module( "arcadedb_embedded._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): @@ -63,7 +71,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) @@ -71,7 +79,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) @@ -79,6 +87,10 @@ def begin_then_interrupt(self): monkeypatch.undo() assert db.is_transaction_active() is False + @pytest.mark.skipif( + _NATIVE, + reason="the native begin reply carries the commit mark: no separate read to interrupt (the test above interrupts that reply)", + ) def test_interrupt_taking_the_commit_mark_rolls_the_transaction_back( self, db, monkeypatch ): @@ -102,7 +114,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() @@ -121,14 +133,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() @@ -186,16 +198,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() diff --git a/bindings/python/tests/test_native_tx_crossings.py b/bindings/python/tests/test_native_tx_crossings.py new file mode 100644 index 00000000000..749a0c0f04b --- /dev/null +++ b/bindings/python/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_embedded as arcadedb +import pytest +from arcadedb_embedded import ArcadeDBError +from arcadedb_embedded._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 diff --git a/bindings/python/tests/test_run_in_transaction_partial_commit.py b/bindings/python/tests/test_run_in_transaction_partial_commit.py new file mode 100644 index 00000000000..974180f539f --- /dev/null +++ b/bindings/python/tests/test_run_in_transaction_partial_commit.py @@ -0,0 +1,119 @@ +""" +``Database.run_in_transaction`` never replays a block that already committed part of its work (#7916), on both wheels. + +A statement with a ``BATCH n`` boundary, or a ``commit()`` inside the block, makes part of the block's work durable. When the block +then hits a retryable conflict, the rollback cannot take that part back, and a replay would apply it twice and report clean success. +The conflict has to be raised instead. The conflict here is a real one (a second thread commits to a record the block also wrote), so +the same test runs against the JPype wheel and the native wheel, and it needs neither a Java class nor a fabricated error. + +The native wheel's guard once sat in a ``try`` whose own ``except ArcadeDBError: pass`` swallowed the ``raise`` it was written to +make, so a native block that had committed part of its work was replayed. +""" + +import threading + +import arcadedb_embedded as arcadedb +import pytest +from arcadedb_embedded import ArcadeDBError + + +@pytest.fixture +def db(temp_db_path): + database = arcadedb.create_database(temp_db_path) + database.command("sql", "CREATE DOCUMENT TYPE R") + database.command("sql", "CREATE PROPERTY R.k INTEGER") + database.command("sql", "CREATE PROPERTY R.v INTEGER") + database.command("sql", "CREATE DOCUMENT TYPE C") + database.command("sql", "CREATE PROPERTY C.n INTEGER") + with database.transaction(): + database.command("sql", "INSERT INTO C SET n = 0") + yield database + database.drop() + + +def _count(db, where="true"): + sql = f"SELECT count(*) AS c FROM R WHERE {where}" # nosec B608 - test-owned predicate text, no user input + return db.query("sql", sql).first().get("c") + + +def _commit_from_another_thread(db): + """Commit a write to the record the block also wrote, so the block's own commit meets a ConcurrentModificationException.""" + errors = [] + + def other(): + try: + with db.transaction(): + db.command("sql", "UPDATE C SET n = n + 100") + except BaseException as e: # reported on the calling thread + errors.append(e) + + thread = threading.Thread(target=other) + thread.start() + thread.join() + assert errors == [] + + +def _commit_inside_the_block(db): + """Make the rows written so far durable with an explicit commit, then carry on in a new transaction.""" + db.command("sql", "UPDATE R SET v = v + 1") + db.commit() + db.begin() + + +def _batch_boundary(db): + """The rows the BATCH boundaries committed (20 of 25) are durable; the remaining 5 sit in the open transaction.""" + db.command("sql", "UPDATE R SET v = v + 1 BATCH 10") + + +@pytest.mark.parametrize( + "publish, durable_v1", + [ + pytest.param(_batch_boundary, 20, id="batch-boundary"), + pytest.param(_commit_inside_the_block, 25, id="commit-inside-the-block"), + ], +) +def test_a_block_that_committed_part_of_its_work_is_not_replayed( + db, publish, durable_v1 +): + with db.transaction(): + for _ in range(25): + db.command("sql", "INSERT INTO R SET k = 0, v = 0") + attempts = [] + + def fn(): + attempts.append(1) + publish(db) + db.command("sql", "UPDATE C SET n = n + 1") + if len(attempts) == 1: + _commit_from_another_thread(db) + return "done" + + with pytest.raises(ArcadeDBError, match="ConcurrentModification"): + db.run_in_transaction(fn, retries=5, backoff_s=0) + + assert ( + len(attempts) == 1 + ), "the block was replayed after it had committed part of its work" + # a replay would have raised every durable row to v = 2 and reported success + assert _count(db, "v = 2") == 0 + assert _count(db, "v = 1") == durable_v1 + assert not db.is_transaction_active() + + +def test_a_block_that_committed_nothing_is_still_retried(db): + """The guard must not turn every conflict into a refusal: a block with no durable part is replayed, once, and succeeds.""" + attempts = [] + + def fn(): + attempts.append(1) + db.command("sql", "INSERT INTO R SET k = 1, v = 0") + db.command("sql", "UPDATE C SET n = n + 1") + if len(attempts) == 1: + _commit_from_another_thread(db) + return "done" + + assert db.run_in_transaction(fn, retries=5, backoff_s=0) == "done" + assert len(attempts) == 2 + assert _count(db, "k = 1") == 1 + # the first attempt was rolled back; the other thread added 100 and the second attempt 1 + assert db.query("sql", "SELECT n FROM C").first().get("n") == 101