diff --git a/src/donkeyq/client.py b/src/donkeyq/client.py index 1b4293d..73b2287 100644 --- a/src/donkeyq/client.py +++ b/src/donkeyq/client.py @@ -23,10 +23,10 @@ async def handle(msg): import socket from collections.abc import Callable from dataclasses import dataclass -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast import redis.asyncio as aioredis -from redis.exceptions import ResponseError +from redis.exceptions import RedisError, ResponseError from . import keys from .connection import make_client @@ -37,6 +37,9 @@ async def handle(msg): from .scripts import register_redrive from .serde import Schema, resolve_schema, to_jsonable +if TYPE_CHECKING: + from pydantic import BaseModel + logger = logging.getLogger(__name__) # Stable for this process, unique across processes/hosts — so workers sharing a @@ -200,7 +203,7 @@ def _register( on_dlq: OnDLQ | None = None, block_ms: int = 2000, count: int = 10, - schema: Any = None, + schema: type[BaseModel] | Schema | None = None, start_from: str = "$", dedupe_ttl_s: float | None = None, ) -> _Subscription: @@ -410,8 +413,18 @@ async def _register_in_registry(self, sub: _Subscription) -> None: ) try: await self._require_client().sadd(self._registry_key, record) - except Exception: - logger.warning("failed to write registry entry for %s/%s", sub.stream, sub.group) + except RedisError: + # The registry is best-effort metadata (cross-stream enumeration for a + # future dashboard), not on the delivery path — a Redis hiccup here must + # not fail activation. Log with the cause so it's diagnosable; anything + # that isn't a RedisError is a real bug and is left to propagate. + logger.warning( + "failed to write registry entry for %s/%s; dashboard enumeration may " + "miss this subscription (delivery is unaffected)", + sub.stream, + sub.group, + exc_info=True, + ) # -- publish ----------------------------------------------------------- @@ -422,7 +435,7 @@ async def publish( *, headers: dict[str, Any] | None = None, maxlen: int | None = None, - schema: Any = None, + schema: type[BaseModel] | Schema | None = None, ) -> str: """Publish ``body`` to ``stream``. Returns the new entry id. diff --git a/src/donkeyq/reclaimer.py b/src/donkeyq/reclaimer.py index 47d70fb..6daf71e 100644 --- a/src/donkeyq/reclaimer.py +++ b/src/donkeyq/reclaimer.py @@ -309,28 +309,37 @@ async def _reclaim_entry( ) if action in ("dlq", "poison_dlq"): logger.warning( - "message %s routed to DLQ %s (%s, count=%s)", + "message %s on %s (group %s) routed to DLQ %s (%s, count=%s)", msg_id, + config.stream, + config.group, dlq, action, new_count, ) await self._invoke_on_dlq(config, body_for_cb, original_id, new_count) elif action == "poison_drop": - logger.error("message %s has an unreadable envelope and no DLQ; dropped", msg_id) + logger.error( + "message %s on %s (group %s) has an unreadable envelope and no DLQ; dropped", + msg_id, + config.stream, + config.group, + ) elif action == "gone": logger.warning( - "message %s on %s was trimmed from the stream while still pending; " - "PEL cleared and the message is LOST — a retry/stream MAXLEN cap is " - "set below the in-flight backlog", + "message %s on %s (group %s) was trimmed from the stream while still " + "pending; PEL cleared and the message is LOST — a retry/stream MAXLEN " + "cap is set below the in-flight backlog", msg_id, config.stream, + config.group, ) elif action == "retry": logger.debug( - "reclaimed message %s on %s → retry stream %s (count=%s)", + "reclaimed message %s on %s (group %s) → retry stream %s (count=%s)", msg_id, config.stream, + config.group, config.retry_stream, new_count, )