Address CodeRabbit nitpicks: retry, state alias, test result handling

codecs.py: HTTP schema fetch retries 3 times with exponential backoff,
catching httpx.RequestError and non-200 responses before raising
DecodeError.

connection.py: STATE_CLOSED explicitly assigned from
state.State.STATE_STOPPED with comment clarifying the intentional
alias for is_closed/is_stopped equivalence.

testing.py: process_message now raises AssertionError on
MESSAGE_REQUEUE (unexpected in tests). MESSAGE_DROP returns normally
(valid test outcome). Uses match/case for clarity.

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

Author: gmr

rejected/codecs.py

@@ -262,14 +262,25 @@ async def _load_http_schema(
             )
         if self._http_client is None:
             self._http_client = httpx.AsyncClient(timeout=30)
-        response = await self._http_client.get(uri)
-        if response.status_code != 200:
-            raise DecodeError(
-                f'Failed to fetch Avro schema for {message_type}: '
-                f'HTTP {response.status_code}'
-            )
-        schema: dict[str, typing.Any] = response.json()
-        return schema
+        last_err: Exception | None = None
+        for attempt in range(3):
+            try:
+                response = await self._http_client.get(uri)
+                if response.status_code == 200:
+                    schema: dict[str, typing.Any] = response.json()
+                    return schema
+                last_err = DecodeError(
+                    f'Failed to fetch Avro schema for '
+                    f'{message_type}: HTTP {response.status_code}'
+                )
+            except httpx.RequestError as err:
+                last_err = err
+            if attempt < 2:
+                await asyncio.sleep(0.5 * (2**attempt))
+        raise DecodeError(
+            f'Failed to fetch Avro schema for {message_type} '
+            f'after 3 attempts: {last_err}'
+        ) from last_err
 
 
 # --- Internal helpers (stateless, sync) ---

rejected/connection.py

@@ -19,7 +19,9 @@
 
 class Connection(state.State):
     HB_INTERVAL: typing.ClassVar[int] = 300
-    STATE_CLOSED: typing.ClassVar[int] = 0x08
+    # Intentionally aliased to STATE_STOPPED — is_closed delegates
+    # to is_stopped so both report the same terminal state.
+    STATE_CLOSED: typing.ClassVar[int] = state.State.STATE_STOPPED
     STATES: typing.ClassVar[dict[int, str]] = {
         **state.State.STATES,
         STATE_CLOSED: 'Closed',

rejected/testing.py

@@ -237,16 +237,21 @@ def _capture_log(
 
         await self.consumer.execute(ctx)
 
-        if ctx.result == models.Result.CONSUMER_EXCEPTION:
-            raise exceptions.ConsumerException()
-        elif ctx.result == models.Result.MESSAGE_EXCEPTION:
-            raise exceptions.MessageException()
-        elif ctx.result == models.Result.PROCESSING_EXCEPTION:
-            raise exceptions.ProcessingException()
-        elif ctx.result == models.Result.UNHANDLED_EXCEPTION:
-            if self.exc_info:
-                raise self.exc_info[1]
-            raise AssertionError('UNHANDLED_EXCEPTION')
+        match ctx.result:
+            case models.Result.CONSUMER_EXCEPTION:
+                raise exceptions.ConsumerException()
+            case models.Result.MESSAGE_EXCEPTION:
+                raise exceptions.MessageException()
+            case models.Result.PROCESSING_EXCEPTION:
+                raise exceptions.ProcessingException()
+            case models.Result.UNHANDLED_EXCEPTION:
+                if self.exc_info:
+                    raise self.exc_info[1]
+                raise AssertionError('UNHANDLED_EXCEPTION')
+            case models.Result.MESSAGE_REQUEUE:
+                raise AssertionError(
+                    'Message was requeued — consumer returned MESSAGE_REQUEUE'
+                )
         return ctx.measurement
 
     @staticmethod