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..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