fix(errors): add typed exceptions and capture trailing metadata

Add TimeoutError (DEADLINE_EXCEEDED), ResourceExhaustedError,
CancelledError, and UnimplementedError to the gRPC status code map.

DecreeError now captures trailing_metadata from the RpcError and
exposes a retry_after timedelta parsed from google.rpc.RetryInfo
packed in the grpc-status-details-bin trailer.

Closes #54

Co-Authored-By: Claude <noreply@anthropic.com>

Author: zeevdr

sdk/src/opendecree/__init__.py

@@ -14,14 +14,18 @@
 from opendecree.client import ConfigClient
 from opendecree.errors import (
     AlreadyExistsError,
+    CancelledError,
     ChecksumMismatchError,
     DecreeError,
     IncompatibleServerError,
     InvalidArgumentError,
     LockedError,
     NotFoundError,
     PermissionDeniedError,
+    ResourceExhaustedError,
+    TimeoutError,
     TypeMismatchError,
+    UnimplementedError,
     UnavailableError,
 )
 from opendecree.types import Change, ConfigValue, FieldUpdate, ServerVersion
@@ -35,6 +39,7 @@
     "AsyncConfigClient",
     "AsyncConfigWatcher",
     "AsyncWatchedField",
+    "CancelledError",
     "Change",
     "ChecksumMismatchError",
     "ConfigClient",
@@ -47,9 +52,12 @@
     "LockedError",
     "NotFoundError",
     "PermissionDeniedError",
+    "ResourceExhaustedError",
     "RetryConfig",
     "ServerVersion",
+    "TimeoutError",
     "TypeMismatchError",
+    "UnimplementedError",
     "UnavailableError",
     "WatchedField",
     "__version__",

sdk/src/opendecree/errors.py

@@ -5,21 +5,28 @@
 
 from __future__ import annotations
 
+import datetime
+
 import grpc
+import grpc.aio
+from google.rpc import error_details_pb2, status_pb2
 
 
 class DecreeError(Exception):
     """Base exception for all OpenDecree SDK errors."""
 
-    def __init__(self, message: str, code: grpc.StatusCode | None = None) -> None:
-        """Create a DecreeError.
-
-        Args:
-            message: Human-readable error description.
-            code: The gRPC status code that caused this error, if any.
-        """
+    def __init__(
+        self,
+        message: str,
+        code: grpc.StatusCode | None = None,
+        *,
+        trailing_metadata: grpc.aio.Metadata | None = None,
+        retry_after: datetime.timedelta | None = None,
+    ) -> None:
         super().__init__(message)
         self.code = code
+        self.trailing_metadata = trailing_metadata
+        self.retry_after = retry_after
 
 
 class NotFoundError(DecreeError):
@@ -58,6 +65,22 @@ class TypeMismatchError(DecreeError):
     """Raised when a typed getter receives a value of the wrong type."""
 
 
+class TimeoutError(DecreeError):
+    """Raised when the operation deadline was exceeded."""
+
+
+class ResourceExhaustedError(DecreeError):
+    """Raised when a resource quota or rate limit is exceeded."""
+
+
+class CancelledError(DecreeError):
+    """Raised when the operation was cancelled."""
+
+
+class UnimplementedError(DecreeError):
+    """Raised when the server has not implemented the operation."""
+
+
 _STATUS_MAP: dict[grpc.StatusCode, type[DecreeError]] = {
     grpc.StatusCode.NOT_FOUND: NotFoundError,
     grpc.StatusCode.ALREADY_EXISTS: AlreadyExistsError,
@@ -67,9 +90,27 @@ class TypeMismatchError(DecreeError):
     grpc.StatusCode.PERMISSION_DENIED: PermissionDeniedError,
     grpc.StatusCode.UNAUTHENTICATED: PermissionDeniedError,
     grpc.StatusCode.UNAVAILABLE: UnavailableError,
+    grpc.StatusCode.DEADLINE_EXCEEDED: TimeoutError,
+    grpc.StatusCode.RESOURCE_EXHAUSTED: ResourceExhaustedError,
+    grpc.StatusCode.CANCELLED: CancelledError,
+    grpc.StatusCode.UNIMPLEMENTED: UnimplementedError,
 }
 
 
+def _parse_retry_after(metadata: grpc.aio.Metadata) -> datetime.timedelta | None:
+    for key, value in metadata:
+        if key != "grpc-status-details-bin" or not isinstance(value, bytes):
+            continue
+        rpc_status = status_pb2.Status()
+        rpc_status.ParseFromString(value)
+        for detail in rpc_status.details:
+            retry_info = error_details_pb2.RetryInfo()
+            if detail.Unpack(retry_info):
+                d = retry_info.retry_delay
+                return datetime.timedelta(seconds=d.seconds, microseconds=d.nanos // 1000)
+    return None
+
+
 def map_grpc_error(err: grpc.RpcError) -> DecreeError:
     """Convert a gRPC ``RpcError`` to a typed ``DecreeError``.
 
@@ -78,5 +119,7 @@ def map_grpc_error(err: grpc.RpcError) -> DecreeError:
     """
     code = err.code()
     details = err.details()
+    trailing: grpc.aio.Metadata | None = getattr(err, "trailing_metadata", lambda: None)()
+    retry_after = _parse_retry_after(trailing) if trailing else None
     exc_class = _STATUS_MAP.get(code, DecreeError)
-    return exc_class(details or str(err), code)
+    return exc_class(details or str(err), code, trailing_metadata=trailing, retry_after=retry_after)

sdk/tests/conftest.py

@@ -13,7 +13,12 @@ class FakeRpcError(grpc.aio.AioRpcError):
     sync and async error handling.
     """
 
-    def __init__(self, code: grpc.StatusCode, details: str = "test") -> None:
+    def __init__(
+        self,
+        code: grpc.StatusCode,
+        details: str = "test",
+        trailing_metadata: grpc.aio.Metadata | None = None,
+    ) -> None:
         # AioRpcError.__init__ expects (code, initial_metadata, trailing_metadata,
         # details, debug_error_string).
-        super().__init__(code, None, None, details, None)  # type: ignore[arg-type]
+        super().__init__(code, None, trailing_metadata, details, None)  # type: ignore[arg-type]

sdk/tests/test_errors.py

@@ -1,20 +1,39 @@
 """Tests for error mapping."""
 
+import datetime
+
 import grpc
+import grpc.aio
+from google.protobuf import any_pb2, duration_pb2
+from google.rpc import error_details_pb2, status_pb2
 
 from opendecree.errors import (
     AlreadyExistsError,
+    CancelledError,
     ChecksumMismatchError,
     DecreeError,
     LockedError,
     NotFoundError,
     PermissionDeniedError,
+    ResourceExhaustedError,
+    TimeoutError,
+    UnimplementedError,
     UnavailableError,
     map_grpc_error,
 )
 from tests.conftest import FakeRpcError
 
 
+def _make_retry_metadata(seconds: int, nanos: int = 0) -> grpc.aio.Metadata:
+    retry_info = error_details_pb2.RetryInfo(
+        retry_delay=duration_pb2.Duration(seconds=seconds, nanos=nanos)
+    )
+    detail = any_pb2.Any()
+    detail.Pack(retry_info)
+    rpc_status = status_pb2.Status(details=[detail])
+    return grpc.aio.Metadata(("grpc-status-details-bin", rpc_status.SerializeToString()))
+
+
 def test_not_found():
     err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND, "gone"))
     assert isinstance(err, NotFoundError)
@@ -61,3 +80,55 @@ def test_unknown_code_falls_back():
 def test_empty_details():
     err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND, ""))
     assert isinstance(err, NotFoundError)
+
+
+def test_deadline_exceeded_maps_to_timeout():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.DEADLINE_EXCEEDED))
+    assert isinstance(err, TimeoutError)
+    assert err.code == grpc.StatusCode.DEADLINE_EXCEEDED
+
+
+def test_resource_exhausted():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.RESOURCE_EXHAUSTED))
+    assert isinstance(err, ResourceExhaustedError)
+
+
+def test_cancelled():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.CANCELLED))
+    assert isinstance(err, CancelledError)
+
+
+def test_unimplemented():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNIMPLEMENTED))
+    assert isinstance(err, UnimplementedError)
+
+
+def test_trailing_metadata_captured():
+    meta = grpc.aio.Metadata(("x-custom", "value"))
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNAVAILABLE, trailing_metadata=meta))
+    assert err.trailing_metadata is not None
+    assert dict(err.trailing_metadata).get("x-custom") == "value"
+
+
+def test_no_trailing_metadata():
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.NOT_FOUND))
+    assert err.trailing_metadata is None
+    assert err.retry_after is None
+
+
+def test_retry_info_parsed():
+    meta = _make_retry_metadata(seconds=5, nanos=500_000_000)
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.RESOURCE_EXHAUSTED, trailing_metadata=meta))
+    assert err.retry_after == datetime.timedelta(seconds=5, microseconds=500_000)
+
+
+def test_retry_info_nanos_precision():
+    meta = _make_retry_metadata(seconds=0, nanos=250_000_000)
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.DEADLINE_EXCEEDED, trailing_metadata=meta))
+    assert err.retry_after == datetime.timedelta(microseconds=250_000)
+
+
+def test_no_retry_info_in_metadata():
+    meta = grpc.aio.Metadata(("x-other", "val"))
+    err = map_grpc_error(FakeRpcError(grpc.StatusCode.UNAVAILABLE, trailing_metadata=meta))
+    assert err.retry_after is None