feat: make FailedWrite, ConflictedWrite exceptions

Rather than throwing the lower-level errors from datapath, we'll now use
one of these two higher-level errors.

Author: h4l

src/denokv/_kv_writes.py

@@ -35,7 +35,9 @@
 from denokv._pycompat.typing import Any
 from denokv._pycompat.typing import ClassVar
 from denokv._pycompat.typing import Container
+from denokv._pycompat.typing import Final
 from denokv._pycompat.typing import Generic
+from denokv._pycompat.typing import Iterable
 from denokv._pycompat.typing import Mapping
 from denokv._pycompat.typing import MutableSequence
 from denokv._pycompat.typing import Never
@@ -60,6 +62,7 @@
 from denokv.datapath import AnyKvKey
 from denokv.datapath import CheckFailure
 from denokv.datapath import pack_key
+from denokv.errors import DenoKvError
 from denokv.kv_keys import KvKey
 from denokv.result import AnyFailure
 from denokv.result import AnySuccess
@@ -1330,65 +1333,126 @@ def _enqueue(self, enqueue: Enqueue, /) -> Self:
         return self
 
 
-@dataclass(init=False, unsafe_hash=True, **slots_if310())
-class ConflictedWrite(FrozenAfterInitDataclass, AnyFailure):
+EMPTY_MAP: Final[Mapping[Any, Any]] = MappingProxyType({})
+
+
+# TODO: Support capturing retries in the FailedWrite/CommittedWrite?
+@dataclass(init=False, unsafe_hash=True)
+class FailedWrite(FrozenAfterInitDataclass, AnyFailure, DenoKvError):
     if TYPE_CHECKING:
 
         def _AnyFailure_marker(self, no_call: Never) -> Never: ...
 
-    ok: Literal[False]
-    conflicts: Mapping[AnyKvKey, CheckRepresentation]
-    versionstamp: None
-    checks: Sequence[CheckRepresentation]
-    mutations: Sequence[MutationRepresentation]
-    enqueues: Sequence[EnqueueRepresentation]
-    endpoint: EndpointInfo
+    checks: Final[Sequence[CheckRepresentation]] = field()
+    failed_checks: Final[Sequence[int]] = field()
+    mutations: Final[Sequence[MutationRepresentation]] = field()
+    enqueues: Final[Sequence[EnqueueRepresentation]] = field()
+    endpoint: Final[EndpointInfo] = field()
+    ok: Final[Literal[False]] = False  # noqa: PYI064
+    versionstamp: Final[None] = None
 
     def __init__(
         self,
-        failed_checks: Sequence[int],
-        checks: Sequence[CheckRepresentation],
-        mutations: Sequence[MutationRepresentation],
-        enqueues: Sequence[EnqueueRepresentation],
+        checks: Iterable[CheckRepresentation],
+        mutations: Iterable[MutationRepresentation],
+        enqueues: Iterable[EnqueueRepresentation],
         endpoint: EndpointInfo,
+        *,
+        cause: BaseException | None = None,
     ) -> None:
-        self.ok = False
-        try:
-            self.conflicts = MappingProxyType(
-                {checks[i].key: checks[i] for i in failed_checks}
-            )
-        except IndexError as e:
-            raise ValueError("failed_checks contains out-of-bounds index") from e
-        self.versionstamp = None
-        self.checks = tuple(checks)
-        self.mutations = tuple(mutations)
-        self.enqueues = tuple(enqueues)
-        self.endpoint = endpoint
+        super(FailedWrite, self).__init__()
+        self.checks = tuple(checks)  # type: ignore[misc] # Cannot assign to final
+        # Allow subclass to initialise failed_checks
+        if not hasattr(self, "failed_checks"):
+            self.failed_checks = tuple()  # type: ignore[misc] # Cannot assign to final
+        self.mutations = tuple(mutations)  # type: ignore[misc] # Cannot assign to final
+        self.enqueues = tuple(enqueues)  # type: ignore[misc] # Cannot assign to final
+        self.endpoint = endpoint  # type: ignore[misc] # Cannot assign to final
+        self.__cause__ = cause
+
+    @property
+    def conflicts(self) -> Mapping[AnyKvKey, CheckRepresentation]:
+        checks = self.checks
+        return {checks[i].key: checks[i] for i in self.failed_checks}
+
+    def _get_cause_description(self) -> str:
+        if self.__cause__:
+            return type(self.__cause__).__name__
+        return "unspecified cause"
+
+    @property
+    def message(self) -> str:
+        # TODO: after xxx attempts?
+        return (
+            f"to {str(self.endpoint.url)!r} "
+            f"due to {self._get_cause_description()}, "
+            f"with {len(self.checks)} checks, "
+            f"{len(self.mutations)} mutations, "
+            f"{len(self.enqueues)} enqueues"
+        )
+
+    def __str__(self) -> str:
+        return f"Write failed {self.message}"
 
     def __repr__(self) -> str:
+        return f"<{type(self).__name__} {self.message}>"
+
+
+def _normalise_failed_checks(
+    failed_checks: Iterable[int], check_count: int
+) -> tuple[int, ...]:
+    failed_checks = tuple(sorted(failed_checks))
+    if failed_checks and (failed_checks[0] < 0 or failed_checks[-1] >= check_count):
+        raise ValueError("failed_checks contains out-of-bounds index")
+    return failed_checks
+
+
+class ConflictedWrite(FailedWrite):
+    def __init__(
+        self,
+        failed_checks: Iterable[int],
+        checks: Iterable[CheckRepresentation],
+        mutations: Iterable[MutationRepresentation],
+        enqueues: Iterable[EnqueueRepresentation],
+        endpoint: EndpointInfo,
+        *,
+        cause: BaseException | None = None,
+    ) -> None:
+        _checks = tuple(checks)
+        self.failed_checks = _normalise_failed_checks(  # type: ignore[misc] # Cannot assign to final attribute "failed_checks"
+            failed_checks,
+            check_count=len(_checks),
+        )
+        super(ConflictedWrite, self).__init__(
+            _checks, mutations, enqueues, endpoint, cause=cause
+        )
+
+    @property
+    def message(self) -> str:
         return (
-            f"<{type(self).__name__} "
             f"NOT APPLIED to {str(self.endpoint.url)!r} with "
             f"{len(self.conflicts)}/{len(self.checks)} checks CONFLICTING, "
             f"{len(self.mutations)} mutations, "
             f"{len(self.enqueues)} enqueues"
-            f">"
         )
 
+    def __str__(self) -> str:
+        return f"Write {self.message}"
+
 
 @dataclass(init=False, unsafe_hash=True, **slots_if310())
 class CommittedWrite(FrozenAfterInitDataclass, AnySuccess):
     if TYPE_CHECKING:
 
         def _AnySuccess_marker(self, no_call: Never) -> Never: ...
 
-    ok: Literal[True]
-    conflicts: Mapping[KvKey, CheckRepresentation]  # empty
-    versionstamp: VersionStamp
-    checks: Sequence[CheckRepresentation]
-    mutations: Sequence[MutationRepresentation]
-    enqueues: Sequence[EnqueueRepresentation]
-    endpoint: EndpointInfo
+    ok: Final[Literal[True]]  # noqa: PYI064
+    conflicts: Final[Mapping[KvKey, CheckRepresentation]]  # empty
+    versionstamp: Final[VersionStamp]
+    checks: Final[Sequence[CheckRepresentation]]
+    mutations: Final[Sequence[MutationRepresentation]]
+    enqueues: Final[Sequence[EnqueueRepresentation]]
+    endpoint: Final[EndpointInfo]
 
     def __init__(
         self,
@@ -1399,23 +1463,28 @@ def __init__(
         endpoint: EndpointInfo,
     ) -> None:
         self.ok = True
-        self.conflicts = MappingProxyType({})
+        self.conflicts = EMPTY_MAP
         self.versionstamp = versionstamp
         self.checks = tuple(checks)
         self.mutations = tuple(mutations)
         self.enqueues = tuple(enqueues)
         self.endpoint = endpoint
 
-    def __repr__(self) -> str:
+    @property
+    def _message(self) -> str:
         return (
-            f"<{type(self).__name__} "
             f"version 0x{self.versionstamp} to {str(self.endpoint.url)!r} with "
             f"{len(self.checks)} checks, "
             f"{len(self.mutations)} mutations, "
             f"{len(self.enqueues)} enqueues"
-            f">"
         )
 
+    def __str__(self) -> str:
+        return f"Write committed {self._message}"
+
+    def __repr__(self) -> str:
+        return f"<{type(self).__name__} {self._message}>"
+
 
 CompletedWrite: TypeAlias = Union[CommittedWrite, ConflictedWrite]
 
@@ -2078,4 +2147,6 @@ def _evaluate_backoff_schedule(self) -> Sequence[int]:
         return [int(delay * 1000) for delay in delay_seconds]
 
 
-WriteOperation: TypeAlias = Union[Check, Set, Sum, Min, Max, Delete, Enqueue]
+WriteOperation: TypeAlias = Union[
+    CheckRepresentation, MutationRepresentation, EnqueueRepresentation
+]

test/test__kv_writes__CommittedWrite.py

@@ -61,7 +61,10 @@ def test_str_repr() -> None:
         endpoint=EP,
     )
     assert (
-        str(instance) == "<CommittedWrite version 0x00000000000000000001 "
+        str(instance) == "Write committed version 0x00000000000000000001 "
+        "to 'https://example.com/' with 1 checks, 1 mutations, 1 enqueues"
+    )
+    assert (
+        repr(instance) == "<CommittedWrite version 0x00000000000000000001 "
         "to 'https://example.com/' with 1 checks, 1 mutations, 1 enqueues>"
     )
-    assert str(instance) == repr(instance)

test/test__kv_writes__ConflictedWrite.py

@@ -1,14 +1,17 @@
+import traceback
 from datetime import datetime
 
 import pytest
 from yarl import URL
 
+from denokv import _datapath_pb2 as datapath_pb2
 from denokv._kv_writes import Check
 from denokv._kv_writes import ConflictedWrite
 from denokv._kv_writes import Enqueue
 from denokv._kv_writes import Set
 from denokv.auth import ConsistencyLevel
 from denokv.auth import EndpointInfo
+from denokv.datapath import CheckFailure
 from denokv.kv_keys import KvKey
 from denokv.result import is_err
 
@@ -18,17 +21,32 @@
 
 @pytest.fixture
 def instance() -> ConflictedWrite:
+    pb_checks = [
+        datapath_pb2.Check(key=bytes(KvKey("a")), versionstamp=None),
+        datapath_pb2.Check(key=bytes(KvKey("b")), versionstamp=None),
+        datapath_pb2.Check(key=bytes(KvKey("c")), versionstamp=None),
+    ]
     checks = [
         Check.for_key_not_set(KvKey("a")),
         Check.for_key_not_set(KvKey("b")),
         Check.for_key_not_set(KvKey("c")),
     ]
+    failed_checks = [0, 2]
+
+    cause = CheckFailure(
+        "Not all checks required by the Atomic Write passed",
+        all_checks=pb_checks,
+        failed_check_indexes=failed_checks,
+        endpoint=EP,
+    )
+
     return ConflictedWrite(
-        failed_checks=[0, 2],
-        checks=list(checks),
+        failed_checks=failed_checks,
+        checks=checks,
         mutations=[Set(KvKey("a"), 42)],
         enqueues=[Enqueue("Hi")],
         endpoint=EP,
+        cause=cause,
     )
 
 
@@ -53,13 +71,9 @@ def test_constructor(instance: ConflictedWrite) -> None:
     assert instance.enqueues == (Enqueue("Hi"),)
     assert instance.endpoint is EP
 
-    assert dict(instance.conflicts) == {KvKey("a"): checks[0], KvKey("c"): checks[2]}
+    assert instance.conflicts == {KvKey("a"): checks[0], KvKey("c"): checks[2]}
     assert instance.conflicts[KvKey("a")] is checks[0]
 
-    # conflicts is immutable
-    with pytest.raises(TypeError):
-        del instance.conflicts[KvKey("a")]  # type: ignore[attr-defined]
-
     with pytest.raises(ValueError, match=r"failed_checks contains out-of-bounds index"):
         ConflictedWrite(
             failed_checks=[0, 10],
@@ -70,13 +84,51 @@ def test_constructor(instance: ConflictedWrite) -> None:
         )
 
 
+def test_changes_to_conflicts_do_not_persist(instance: ConflictedWrite) -> None:
+    assert isinstance(instance.conflicts, dict)
+    # Changes to conflicts do not persist
+    assert KvKey("a") in instance.conflicts
+    del instance.conflicts[KvKey("a")]
+    assert KvKey("a") in instance.conflicts
+
+
 def test_is_AnyFailure(instance: ConflictedWrite) -> None:
     assert is_err(instance)
 
 
-def test_str_repr(instance: ConflictedWrite) -> None:
+@pytest.mark.parametrize("with_cause", [True, False])
+def test_str(instance: ConflictedWrite, with_cause: bool) -> None:
+    assert instance.__cause__
+    if not with_cause:
+        instance.__cause__ = None
     assert (
-        str(instance) == "<ConflictedWrite NOT APPLIED to 'https://example.com/' "
+        str(instance) == "Write NOT APPLIED to 'https://example.com/' "
+        "with 2/3 checks CONFLICTING, 1 mutations, 1 enqueues"
+    )
+
+
+@pytest.mark.parametrize("with_cause", [True, False])
+def test_repr(instance: ConflictedWrite, with_cause: bool) -> None:
+    assert instance.__cause__
+    if not with_cause:
+        instance.__cause__ = None
+
+    assert (
+        repr(instance) == "<ConflictedWrite NOT APPLIED to 'https://example.com/' "
         "with 2/3 checks CONFLICTING, 1 mutations, 1 enqueues>"
     )
-    assert str(instance) == repr(instance)
+
+
+@pytest.mark.parametrize("with_cause", [True, False])
+def test_traceback_presentation(instance: ConflictedWrite, with_cause: bool) -> None:
+    assert instance.__cause__
+    if not with_cause:
+        instance.__cause__ = None
+
+    assert "\n".join(
+        traceback.format_exception_only(type(instance), instance)
+    ).strip() == (
+        "denokv._kv_writes.ConflictedWrite: "
+        "Write NOT APPLIED to 'https://example.com/' "
+        "with 2/3 checks CONFLICTING, 1 mutations, 1 enqueues"
+    )