Rewrap async cursor COMMIT leadership-loss as AmbiguousCommitError

The sync cursor and async Connection.commit() both re-raise a leadership-lost
OperationalError as AmbiguousCommitError for an explicit COMMIT (the write is
in-doubt once the entry was submitted). The async cursor's DML path lacked this,
so await cursor.execute("COMMIT") on a leadership-lost code surfaced a plain
OperationalError, diverging from both sibling paths. Mirror the sync cursor:
remap on an explicit COMMIT with a leadership-lost code; a not-leader rejection
is a clean failure and stays OperationalError.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -17,6 +17,7 @@
     _convert_params_async,
     _convert_rows_async,
     _ExecuteManyAccumulator,
+    _is_commit_statement,
     _is_dml_rowcount_meaningful,
     _is_dml_with_returning,
     _is_insert_or_replace,
@@ -29,9 +30,12 @@
     _validate_executemany_seq_shape,
 )
 from dqlitedbapi.exceptions import (
+    AMBIGUOUS_COMMIT_CODES,
+    AmbiguousCommitError,
     DataError,
     InterfaceError,
     NotSupportedError,
+    OperationalError,
     ProgrammingError,
 )
 from dqlitedbapi.types import UNKNOWN as _UNKNOWN_TYPE
@@ -328,7 +332,21 @@ async def _execute_unlocked(
             # -1 for ALL PRAGMA, so match that rather than len.
             self._rowcount = -1 if _is_pragma(operation) else len(rows)
         else:
-            last_id, affected = await _call_client(conn.execute(operation, params))
+            try:
+                last_id, affected = await _call_client(conn.execute(operation, params))
+            except OperationalError as e:
+                # Mirror Connection.commit() and the sync cursor: an explicit COMMIT that
+                # loses leadership after the entry was submitted leaves the write in doubt.
+                # A not-leader rejection is a clean pre-apply failure and stays OperationalError.
+                if _is_commit_statement(operation) and e.code in AMBIGUOUS_COMMIT_CODES:
+                    raise AmbiguousCommitError(
+                        "ambiguous commit: leadership lost during COMMIT; "
+                        "the write may or may not have been persisted. "
+                        f"Original: {e}",
+                        code=e.code,
+                        raw_message=getattr(e, "raw_message", None),
+                    ) from e
+                raise
             # Post-await close-race guard (see query branch). _lastrowid is
             # PRESERVED: stdlib persists it across a close-during-DML boundary.
             if self._closed:

tests/test_commit_leader_flip_ambiguous_commit_error.py

@@ -13,14 +13,15 @@
 import asyncio
 import os
 import weakref
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 import dqliteclient.exceptions as _client_exc
 import dqlitedbapi
 from dqlitedbapi import Connection
 from dqlitedbapi.aio.connection import AsyncConnection
+from dqlitedbapi.aio.cursor import AsyncCursor
 from dqlitedbapi.exceptions import (
     AMBIGUOUS_COMMIT_CODES,
     AmbiguousCommitError,
@@ -166,6 +167,75 @@ def test_cursor_insert_leadership_lost_is_not_remapped(code: int) -> None:
         conn._closed = True
 
 
+def _make_async_cursor(code: int) -> AsyncCursor:
+    """An AsyncCursor whose underlying connection.execute raises a client leader error."""
+    cur = AsyncCursor.__new__(AsyncCursor)
+    cur._closed = False
+    cur._description = None
+    cur._rows = []
+    cur._rowcount = -1
+    cur._lastrowid = None
+    cur._row_index = 0
+    cur._arraysize = 1
+    cur.messages = []
+    cur._completed_iterations = 0
+    cur._executing_task = None
+
+    inner = MagicMock()
+
+    async def _raise(_sql: str, _params: object) -> None:
+        raise _client_exc.OperationalError("leader", code)
+
+    inner.execute = _raise
+
+    conn = MagicMock()
+    conn._max_total_rows = None
+
+    async def _ensure() -> object:
+        return inner
+
+    conn._ensure_connection = _ensure
+    cur._connection = conn
+    return cur
+
+
+@pytest.mark.parametrize("code", _IN_DOUBT_CODES)
+async def test_async_cursor_commit_leadership_lost_is_ambiguous(code: int) -> None:
+    cur = _make_async_cursor(code)
+    with (
+        patch.object(AsyncCursor, "_check_closed", lambda self: None),
+        pytest.raises(AmbiguousCommitError) as ei,
+    ):
+        await cur._execute_unlocked("COMMIT", None)
+    assert isinstance(ei.value, OperationalError)
+    assert ei.value.code == code
+
+
+@pytest.mark.parametrize("code", _NOT_LEADER_CODES)
+async def test_async_cursor_commit_not_leader_is_plain_operational(code: int) -> None:
+    cur = _make_async_cursor(code)
+    with (
+        patch.object(AsyncCursor, "_check_closed", lambda self: None),
+        pytest.raises(OperationalError) as ei,
+    ):
+        await cur._execute_unlocked("COMMIT", None)
+    assert not isinstance(ei.value, AmbiguousCommitError)
+    assert ei.value.code == code
+
+
+@pytest.mark.parametrize("code", _IN_DOUBT_CODES)
+async def test_async_cursor_insert_leadership_lost_is_not_remapped(code: int) -> None:
+    """The remap is scoped to explicit COMMIT; a bare DML leader-loss stays Operational."""
+    cur = _make_async_cursor(code)
+    with (
+        patch.object(AsyncCursor, "_check_closed", lambda self: None),
+        pytest.raises(OperationalError) as ei,
+    ):
+        await cur._execute_unlocked("INSERT INTO t VALUES (1)", None)
+    assert not isinstance(ei.value, AmbiguousCommitError)
+    assert ei.value.code == code
+
+
 def test_ambiguous_commit_error_is_exported_at_package_level() -> None:
     assert dqlitedbapi.AmbiguousCommitError is AmbiguousCommitError
     assert "AmbiguousCommitError" in dqlitedbapi.__all__