refactor:

Replace as_dict with dataclasses.asdict;
Remove user-level retry orchestration from broker/hub/protocol

Author: alexted

taskiq/brokers/nng/broker.py

@@ -83,9 +83,6 @@ def __init__(
         heartbeat_interval: float = 5.0,
         lease_timeout: float = 20.0,
         capacity: int = 1,
-        max_retries: int = 0,
-        retry_backoff: float = 1.0,
-        retry_jitter: float = 0.0,
         recv_timeout_ms: int = 5_000,
         send_timeout_ms: int = 5_000,
     ) -> None:
@@ -102,9 +99,6 @@ def __init__(
         :param heartbeat_interval: seconds between heartbeat messages to hub.
         :param lease_timeout: seconds a dispatched task lease remains valid.
         :param capacity: max concurrent tasks this worker will accept.
-        :param max_retries: default max retries for submitted tasks.
-        :param retry_backoff: base seconds for exponential backoff.
-        :param retry_jitter: jitter multiplier added to backoff (0 = no jitter).
         :param recv_timeout_ms: Req0 recv timeout in milliseconds.
         :param send_timeout_ms: Req0 send timeout in milliseconds.
         """
@@ -123,9 +117,6 @@ def __init__(
         self.heartbeat_interval = heartbeat_interval
         self.lease_timeout = lease_timeout
         self.capacity = capacity
-        self.max_retries = max_retries
-        self.retry_backoff = retry_backoff
-        self.retry_jitter = retry_jitter
         self.recv_timeout_ms = recv_timeout_ms
         self.send_timeout_ms = send_timeout_ms
 
@@ -249,16 +240,6 @@ async def kick(self, message: BrokerMessage) -> None:
             "payload_b64": base64.b64encode(message.message).decode("ascii"),
             "labels": message.labels,
             "lease_id": "",  # hub assigns the real lease_id at dispatch time
-            "attempts": int(message.labels.get("attempts", 0)),
-            "max_retries": int(
-                message.labels.get("max_retries", self.max_retries),
-            ),
-            "retry_backoff": float(
-                message.labels.get("retry_backoff", self.retry_backoff),
-            ),
-            "retry_jitter": float(
-                message.labels.get("retry_jitter", self.retry_jitter),
-            ),
             "priority": int(message.labels.get("priority", 0)),
             "created_at": time.time(),
         }

taskiq/brokers/nng/hub.py

@@ -300,7 +300,7 @@ async def _dispatch_once(self) -> bool:
                 task_name=row["task_name"],
                 labels=row["labels"],
                 priority=int(row["priority"]),
-                attempts=int(row["attempts"]),
+                attempts=int(row.get("attempts", 0)),
             )
             worker = self.store.choose_worker(
                 self._routing,
@@ -334,10 +334,6 @@ async def _dispatch_once(self) -> bool:
                 payload_b64=base64.b64encode(row["payload"]).decode("ascii"),
                 labels=row["labels"],
                 lease_id=lease_id,
-                attempts=int(row["attempts"]) + 1,
-                max_retries=int(row["max_retries"]),
-                retry_backoff=float(row["retry_backoff"]),
-                retry_jitter=float(row["retry_jitter"]),
                 priority=int(row["priority"]),
                 created_at=float(row["created_at"]),
             )

taskiq/brokers/nng/protocol.py

@@ -57,17 +57,18 @@ class TaskEnvelope:
     ``lease_id`` is the UUID assigned by the hub at dispatch time.
     Workers must echo it back in the ACK so the hub can validate
     that the ack is not stale (e.g. after lease expiry and requeue).
+
+    User-level retry policy is the responsibility of the
+    :class:`~taskiq.middlewares.SmartRetryMiddleware` (or any compatible
+    middleware) and travels in :attr:`labels`; the envelope itself carries
+    no retry knobs.
     """
 
     task_id: str
     task_name: str
     payload_b64: str
     labels: dict[str, Any] = field(default_factory=dict)
     lease_id: str = ""
-    attempts: int = 0
-    max_retries: int = 0
-    retry_backoff: float = 1.0
-    retry_jitter: float = 0.0
     priority: int = 0
     created_at: float = 0.0
 
@@ -157,3 +158,4 @@ def to_dict(self) -> dict[str, Any]:
         d = asdict(self)
         d["status"] = str(self.status)
         return d
+

taskiq/brokers/nng/storage.py

@@ -5,7 +5,7 @@
 import inspect
 import random
 import time
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
 
 if TYPE_CHECKING:
@@ -19,7 +19,11 @@ class StoreConfig:
     path: str = ""  # kept for API compat; not used
     max_pending: int = 10_000
     lease_timeout: float = 30.0
-    backoff_base: float = 1.0
+    # Hub-internal cap on delivery retries (lease expiry / worker death).
+    # Has nothing to do with user-level retry policy, which is handled by
+    # taskiq retry middlewares.  Set to 0 to disable redelivery entirely.
+    max_delivery_attempts: int = 5
+    delivery_backoff: float = 1.0
     backoff_cap: float = 60.0
 
 
@@ -34,10 +38,9 @@ class _Task:
     payload: bytes
     labels: dict[str, Any]
     state: str  # ready / leased / done / failed
+    # Internal delivery-attempt counter (incremented on each dispatch).
+    # NOT related to user-level retry policy — that lives in middlewares.
     attempts: int = 0
-    max_retries: int = 0
-    retry_backoff: float = 1.0
-    retry_jitter: float = 0.0
     priority: int = 0
     created_at: float = field(default_factory=time.time)
     updated_at: float = field(default_factory=time.time)
@@ -49,25 +52,7 @@ class _Task:
 
     def as_dict(self) -> dict[str, Any]:
         """Return a dict view of this task record."""
-        return {
-            "task_id": self.task_id,
-            "task_name": self.task_name,
-            "payload": self.payload,
-            "labels": self.labels,
-            "state": self.state,
-            "attempts": self.attempts,
-            "max_retries": self.max_retries,
-            "retry_backoff": self.retry_backoff,
-            "retry_jitter": self.retry_jitter,
-            "priority": self.priority,
-            "created_at": self.created_at,
-            "updated_at": self.updated_at,
-            "next_run_at": self.next_run_at,
-            "lease_id": self.lease_id,
-            "leased_worker_id": self.leased_worker_id,
-            "lease_until": self.lease_until,
-            "last_error": self.last_error,
-        }
+        return asdict(self)
 
     def as_status_dict(self) -> dict[str, Any]:
         """Return a JSON-safe dict (no raw bytes) for control-plane status responses."""
@@ -91,18 +76,7 @@ class _Worker:
 
     def as_dict(self) -> dict[str, Any]:
         """Return a dict view of this worker record."""
-        return {
-            "worker_id": self.worker_id,
-            "task_addr": self.task_addr,
-            "capacity": self.capacity,
-            "inflight": self.inflight,
-            "last_seen": self.last_seen,
-            "heartbeat_interval": self.heartbeat_interval,
-            "lease_timeout": self.lease_timeout,
-            "draining": self.draining,
-            "status": self.status,
-            "version": self.version,
-        }
+        return asdict(self)
 
 
 # ── task context ─────────────────────────────────────────────────────────────
@@ -325,16 +299,21 @@ def __init__(self, config: StoreConfig) -> None:
 
     # ── helpers ───────────────────────────────────────────────────────────────
 
-    def _backoff(self, attempts: int, backoff_base: float) -> float:
-        return min(self.config.backoff_cap, backoff_base * (2 ** max(0, attempts - 1)))
+    def _backoff(self, attempts: int) -> float:
+        return min(
+            self.config.backoff_cap,
+            self.config.delivery_backoff * (2 ** max(0, attempts - 1)),
+        )
 
     def _requeue_or_fail(self, task: _Task, worker_id: str, error: str) -> bool:
         now = time.time()
-        if task.attempts > task.max_retries:
+        # Hub-internal delivery cap.  User-level retries are handled by
+        # retry middlewares, which re-kick the task with updated labels.
+        if task.attempts > self.config.max_delivery_attempts:
             task.state = "failed"
         else:
             task.state = "ready"
-            task.next_run_at = now + self._backoff(task.attempts, task.retry_backoff)
+            task.next_run_at = now + self._backoff(task.attempts)
         task.last_error = error
         task.lease_id = None
         task.leased_worker_id = None
@@ -367,9 +346,6 @@ def submit(self, envelope: TaskEnvelope) -> None:
             payload=envelope.payload,
             labels=envelope.labels,
             state="ready",
-            max_retries=envelope.max_retries,
-            retry_backoff=envelope.retry_backoff,
-            retry_jitter=envelope.retry_jitter,
             priority=envelope.priority,
             created_at=envelope.created_at or now,
             updated_at=now,

tests/brokers/test_nng_broker.py

@@ -65,10 +65,6 @@ def _envelope(**kwargs: object) -> TaskEnvelope:
         "payload_b64": "dGVzdA==",
         "labels": {},
         "lease_id": "",
-        "attempts": 0,
-        "max_retries": 0,
-        "retry_backoff": 1.0,
-        "retry_jitter": 0.0,
         "priority": 0,
         "created_at": time.time(),
     }
@@ -208,10 +204,6 @@ async def submit(self, **labels: object) -> str:
             "payload_b64": "dGVzdA==",
             "labels": {},
             "lease_id": "",
-            "attempts": 0,
-            "max_retries": labels.pop("max_retries", 0),
-            "retry_backoff": labels.pop("retry_backoff", 1.0),
-            "retry_jitter": 0.0,
             "priority": labels.pop("priority", 0),
             "created_at": time.time(),
         }
@@ -321,32 +313,46 @@ def test_late_ack_after_requeue_ignored(store: InMemoryStore) -> None:
     assert not store.ack(env.task_id, w.worker_id, "L2")
 
 
-def test_nack_requeues_with_backoff(store: InMemoryStore) -> None:
-    env = _envelope(max_retries=2, retry_backoff=1.0)
-    store.submit(env)
+def test_nack_requeues_with_backoff(db_path: str) -> None:
+    """Hub-level delivery retry: nack within delivery cap requeues with backoff."""
+    s = InMemoryStore(
+        StoreConfig(
+            path=db_path, max_pending=50, lease_timeout=5.0,
+            max_delivery_attempts=2, delivery_backoff=1.0,
+        ),
+    )
+    env = _envelope()
+    s.submit(env)
     w = _worker_state()
-    store.register_worker(w)
-    store.mark_leased(env.task_id, w.worker_id, "L3", time.time() + 60)
-    assert store.nack(env.task_id, w.worker_id, "L3", "boom")
-    task = store.get_task(env.task_id)
+    s.register_worker(w)
+    s.mark_leased(env.task_id, w.worker_id, "L3", time.time() + 60)
+    assert s.nack(env.task_id, w.worker_id, "L3", "boom")
+    task = s.get_task(env.task_id)
     assert task["state"] == "ready"
     assert float(task["next_run_at"]) > time.time()
 
 
-def test_nack_exceeds_retries_fails(store: InMemoryStore) -> None:
-    env = _envelope(max_retries=0)
-    store.submit(env)
+def test_nack_exceeds_delivery_cap_fails(db_path: str) -> None:
+    """Delivery cap of 0 → first nack fails the task immediately."""
+    s = InMemoryStore(
+        StoreConfig(
+            path=db_path, max_pending=50, lease_timeout=5.0,
+            max_delivery_attempts=0,
+        ),
+    )
+    env = _envelope()
+    s.submit(env)
     w = _worker_state()
-    store.register_worker(w)
-    store.mark_leased(env.task_id, w.worker_id, "L4", time.time() + 60)
-    store.nack(env.task_id, w.worker_id, "L4", "error")
-    assert store.get_task(env.task_id)["state"] == "failed"
+    s.register_worker(w)
+    s.mark_leased(env.task_id, w.worker_id, "L4", time.time() + 60)
+    s.nack(env.task_id, w.worker_id, "L4", "error")
+    assert s.get_task(env.task_id)["state"] == "failed"
 
 
 def test_dead_worker_tasks_requeued(store: InMemoryStore) -> None:
     w = _worker_state()
     store.register_worker(w)
-    env = _envelope(max_retries=3)
+    env = _envelope()
     store.submit(env)
     store.mark_leased(env.task_id, w.worker_id, "L5", time.time() + 60)
     store._workers[w.worker_id].last_seen = 0  # simulate missed heartbeats
@@ -456,7 +462,7 @@ async def test_worker_crash_before_ack_task_requeued(
     client = FakeClient(ctrl_addr)
     try:
         await w1.register()
-        tid = await client.submit(max_retries=3)
+        tid = await client.submit()
         env1 = await w1.recv_task(timeout=3.0)
         assert env1.task_id == tid
         w1.close()  # simulate crash without acking
@@ -493,7 +499,7 @@ async def test_late_ack_after_requeue_rejected(
     client = FakeClient(ctrl_addr)
     try:
         await w1.register()
-        tid = await client.submit(max_retries=3)
+        tid = await client.submit()
         env1 = await w1.recv_task(timeout=3.0)
         await asyncio.sleep(3.5)  # let lease expire
 
@@ -670,10 +676,6 @@ async def test_backpressure_hub_rejects_when_full(
             "payload_b64": "dGVzdA==",
             "labels": {},
             "lease_id": "",
-            "attempts": 0,
-            "max_retries": 0,
-            "retry_backoff": 1.0,
-            "retry_jitter": 0.0,
             "priority": 0,
             "created_at": time.time(),
         }