Native: run_in_transaction refuses to replay a block that already committed part of its work (P-03)

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

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

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