Fix RateLimit sorted set member uniqueness for Perpetual tasks

Using `execution.key` as the sorted set member meant Perpetual tasks
(which reuse the same key via `replace()`) only ever had one entry —
ZADD overwrites the score instead of adding a new member, so ZCARD
stays at 1 and the rate limit never fires.

The member is now `{execution.key}:{now_ms}`, so each execution
attempt gets its own entry. Also cleans up phantom slots in
`__aexit__` when a later dependency (like ConcurrencyLimit) blocks —
the task never ran but would otherwise consume a rate-limit slot.

Caught during PR #356 review.

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

Author: chrisguidry

src/docket/dependencies/_ratelimit.py

@@ -2,7 +2,8 @@
 
 Caps how many times a task (or a per-parameter scope) can execute within a
 sliding window.  Uses a Redis sorted set as a sliding window log: members are
-execution keys, scores are millisecond timestamps.
+``{execution_key}:{now_ms}`` strings (unique per attempt), scores are
+millisecond timestamps.
 """
 
 from __future__ import annotations
@@ -17,7 +18,7 @@
 # Lua script for atomic sliding-window rate limit check.
 #
 # KEYS[1] = sorted set key (one per scope)
-# ARGV[1] = execution key (member)
+# ARGV[1] = member (execution key + timestamp, unique per attempt)
 # ARGV[2] = current time in milliseconds
 # ARGV[3] = window size in milliseconds
 # ARGV[4] = max allowed count (limit)
@@ -99,6 +100,8 @@ def __init__(
         self.scope = scope
         self._argument_name: str | None = None
         self._argument_value: Any = None
+        self._ratelimit_key: str | None = None
+        self._member: str | None = None
 
     def bind_to_parameter(self, name: str, value: Any) -> RateLimit:
         bound = RateLimit(self.limit, per=self.per, drop=self.drop, scope=self.scope)
@@ -121,18 +124,21 @@ async def __aenter__(self) -> RateLimit:
         window_ms = int(self.per.total_seconds() * 1000)
         now_ms = int(time.time() * 1000)
         ttl_ms = window_ms * 2
+        member = f"{execution.key}:{now_ms}"
 
         async with docket.redis() as redis:
             script = redis.register_script(_RATELIMIT_LUA)
             result: list[int] = await script(
                 keys=[ratelimit_key],
-                args=[execution.key, now_ms, window_ms, self.limit, ttl_ms],
+                args=[member, now_ms, window_ms, self.limit, ttl_ms],
             )
 
         action = result[0]
         retry_after_ms = result[1]
 
         if action == _ACTION_PROCEED:
+            self._ratelimit_key = ratelimit_key
+            self._member = member
             return self
 
         reason = f"rate limit ({self.limit}/{self.per}) on {ratelimit_key}"
@@ -152,4 +158,9 @@ async def __aexit__(
         exc_value: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        pass
+        if exc_type is not None and self._member is not None:
+            if issubclass(exc_type, AdmissionBlocked):
+                assert self._ratelimit_key is not None
+                docket = current_docket.get()
+                async with docket.redis() as redis:
+                    await redis.zrem(self._ratelimit_key, self._member)

tests/test_ratelimit.py

@@ -6,7 +6,7 @@
 from datetime import timedelta
 from typing import Annotated
 
-from docket import ConcurrencyLimit, Docket, Worker
+from docket import ConcurrencyLimit, Docket, Perpetual, Worker
 from docket.dependencies import RateLimit
 
 
@@ -169,3 +169,71 @@ async def rated_task(
             )
         ]
         assert ratelimit_keys == []
+
+
+async def test_rate_limit_slot_kept_on_task_failure(docket: Docket, worker: Worker):
+    """A failed task still counts against the rate limit."""
+    results: list[str] = []
+
+    async def failing_task(
+        rate: RateLimit = RateLimit(2, per=timedelta(seconds=5), drop=True),
+    ):
+        results.append("attempted")
+        if len(results) == 1:
+            raise RuntimeError("boom")
+
+    await docket.add(failing_task)()
+    await docket.add(failing_task)()
+    await docket.add(failing_task)()
+
+    await worker.run_until_finished()
+
+    assert len(results) == 2
+
+
+async def test_perpetual_task_counts_each_execution(docket: Docket, worker: Worker):
+    """Perpetual re-executions each count against the rate limit."""
+    results: list[str] = []
+
+    async def perpetual_rated(
+        perpetual: Perpetual = Perpetual(every=timedelta(milliseconds=10)),
+        rate: RateLimit = RateLimit(2, per=timedelta(seconds=5), drop=True),
+    ):
+        results.append("executed")
+
+    execution = await docket.add(perpetual_rated)()
+    await worker.run_at_most({execution.key: 4})
+
+    assert len(results) == 2
+
+
+async def test_rate_limit_slot_freed_when_another_dep_blocks(
+    docket: Docket, worker: Worker
+):
+    """RateLimit slot is freed if a later dependency blocks the task.
+
+    RateLimit (default-param, resolved first) proceeds and records a slot,
+    then ConcurrencyLimit (annotation, resolved second) blocks. Without
+    __aexit__ cleanup the phantom slot stays in the sorted set.
+    """
+    results: list[str] = []
+
+    async def blocker(
+        customer_id: Annotated[int, ConcurrencyLimit(1)],
+        rate: RateLimit = RateLimit(2, per=timedelta(seconds=5), drop=True),
+    ):
+        results.append(f"executed_{customer_id}")
+        if len(results) == 1:
+            await asyncio.sleep(0.3)
+
+    # Task 1 grabs both the rate-limit slot and the concurrency slot.
+    # Task 2 passes rate-limit (2/2) but is concurrency-blocked, then
+    # rescheduled.  Without cleanup, the phantom slot means task 2's
+    # retry hits 2/2 and gets dropped.
+    await docket.add(blocker)(customer_id=1)
+    await docket.add(blocker)(customer_id=1)
+
+    worker.concurrency = 2
+    await worker.run_until_finished()
+
+    assert len(results) == 2