From 514b2252219a00d4c1f28a7d998990e39ebe57fb Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 17:29:52 +0900 Subject: [PATCH 1/3] Native: run_in_transaction refuses to replay a block that already committed part of its work (P-03) The guard that refuses a retry after a BATCH boundary or a commit() inside the block sat in a try whose own `except ArcadeDBError: pass` swallowed the `raise` it was written to make, so the block was replayed and its durable part applied twice. Only the read of the commit-count witness may fail quietly; the raise now sits outside that try. The JVM core already refused. The new test uses a real conflict (a second thread commits to a record the block also wrote), so it runs on both wheels: red on the native wheel before this change (20 of 25 rows incremented twice after a BATCH boundary, all 25 after a commit() inside the block), green after, green on the JVM wheel. The contract is added to the run_in_transaction API doc. Co-Authored-By: Claude Sonnet 5.5 --- bindings/python/docs/api/database.md | 2 + .../src/arcadedb_embedded/_native/database.py | 10 +- .../test_run_in_transaction_partial_commit.py | 120 ++++++++++++++++++ 3 files changed, 129 insertions(+), 3 deletions(-) create mode 100644 bindings/python/tests/test_run_in_transaction_partial_commit.py 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..765522b454d 100644 --- a/bindings/python/src/arcadedb_embedded/_native/database.py +++ b/bindings/python/src/arcadedb_embedded/_native/database.py @@ -614,11 +614,15 @@ 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 = 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/test_run_in_transaction_partial_commit.py b/bindings/python/tests/test_run_in_transaction_partial_commit.py new file mode 100644 index 00000000000..bff3c5ed8b2 --- /dev/null +++ b/bindings/python/tests/test_run_in_transaction_partial_commit.py @@ -0,0 +1,120 @@ +""" +``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"): + return ( + db.query("sql", f"SELECT count(*) AS c FROM R WHERE {where}").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 From f93e46a721aa61a70432abb8cb8514a7687514d1 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 18:31:30 +0900 Subject: [PATCH 2/3] test: nosec B608 on the P-03 test's count helper (test-owned predicate text) Co-Authored-By: Claude Opus 5.5 --- .../python/tests/test_run_in_transaction_partial_commit.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/bindings/python/tests/test_run_in_transaction_partial_commit.py b/bindings/python/tests/test_run_in_transaction_partial_commit.py index bff3c5ed8b2..974180f539f 100644 --- a/bindings/python/tests/test_run_in_transaction_partial_commit.py +++ b/bindings/python/tests/test_run_in_transaction_partial_commit.py @@ -32,9 +32,8 @@ def db(temp_db_path): def _count(db, where="true"): - return ( - db.query("sql", f"SELECT count(*) AS c FROM R WHERE {where}").first().get("c") - ) + 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): From aeacaa02a73719e086f610e6870c74f30c624078 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 16:53:46 +0900 Subject: [PATCH 3/3] Native: run_in_transaction takes the commit-count witness from the begin reply, and the unwind reads no state it already has 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 commit-count witness now comes from that reply: 3 crossings (the mark must stay before begin, see the docstring). The unwind after a failed block (_end_transactions_opened_since) 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). A failed block: 6 crossings -> 4; a refused attempt that is retried: 11 -> 7. Tests first: test_native_tx_crossings.py counts lib.call invocations (sequence of arcadedb_n_tx operations) for run_in_transaction, `with db.transaction()` and the exception and retry paths, and pins the behaviour the marks exist for (nested block leaves the caller's transaction open, refused attempt rolled back before the retry, duplicated key retried once). Counts only; no speed claim. The unwind and begin now go through two named seams, _begin_state() and _rollback_state() (begin() and rollback() are thin wrappers over them). test_interrupt_hygiene.py patched Database.begin and Database.rollback, which run_in_transaction no longer calls on the native Database: two tests failed (a refused begin, a failing rollback) and three interrupt tests failed with "DID NOT RAISE KeyboardInterrupt", which conftest.py turns into a silent "NA-JVM" skip. The tests now patch the seam the active backend's run_in_transaction really calls (begin/rollback on the JPype Database, _begin_state/_rollback_state on the native one) and check what they checked before; the one test with no native equivalent (an interrupt between begin and a separate commit-mark read, which the begin reply made one crossing) skips with a stated reason. Based on the P-03 fix (the commit-count witness check raises outside its try), so the retry check here reads the witness from the unwind state and keeps that fix. Co-Authored-By: Claude Sonnet 5.5 --- .../src/arcadedb_embedded/_native/database.py | 62 ++++-- bindings/python/tests/conftest.py | 2 + .../python/tests/test_interrupt_hygiene.py | 31 ++- .../python/tests/test_native_tx_crossings.py | 190 ++++++++++++++++++ 4 files changed, 261 insertions(+), 24 deletions(-) create mode 100644 bindings/python/tests/test_native_tx_crossings.py diff --git a/bindings/python/src/arcadedb_embedded/_native/database.py b/bindings/python/src/arcadedb_embedded/_native/database.py index 765522b454d..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): @@ -618,7 +646,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/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