Simplify admission control exceptions and relax Cooldown single restriction

Drop CooldownBlocked and DebounceBlocked in favor of raising AdmissionBlocked
directly with reschedule/retry_delay as constructor arguments. Less ceremony,
same behavior.

Cooldown no longer enforces single=True — multiple per-parameter cooldowns on
the same task are independent and work fine. Debounce keeps single=True because
its reschedule mechanism means multiple debounces would loop forever.

Also trims implementation details (Redis key patterns, Lua scripts) from the
user-facing docs.

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

Author: chrisguidry

docs/task-behaviors.md

@@ -423,7 +423,7 @@ async def monitor_concurrency_usage() -> None:
 
 ## Cooldown
 
-Cooldown executes the first submission immediately, then drops duplicates within a window. On entry it atomically sets a Redis key with a TTL (`SET key 1 NX PX window_ms`). If the key already exists the task is silently dropped. The key expires naturally after the window.
+Cooldown executes the first submission immediately, then drops duplicates within a window. If another submission arrives before the window expires, it's silently dropped.
 
 ### Per-Task Cooldown
 
@@ -510,14 +510,6 @@ await docket.add(sync_customer)(customer_id=1001)  # resets 1001's timer
 await docket.add(sync_customer)(customer_id=2002)  # independent window
 ```
 
-### How It Works
-
-Debounce uses two Redis keys per scope — a **winner** key (which task gets to proceed) and a **last_seen** timestamp — managed by an atomic Lua script:
-
-1. **No winner exists** — the task becomes the winner and gets rescheduled for the full settle window.
-2. **Winner returns from reschedule** — if enough time has passed since the last submission, it proceeds. Otherwise it reschedules for the remaining time.
-3. **Non-winner arrives** — updates the last_seen timestamp (resetting the settle timer) and is immediately dropped.
-
 ### Debounce vs. Cooldown
 
 | | Cooldown | Debounce |
@@ -526,6 +518,34 @@ Debounce uses two Redis keys per scope — a **winner** key (which task gets to
 | **Window anchored to** | First execution | Last submission |
 | **Good for** | Deduplicating rapid-fire events | Batching bursts into one action |
 
+### Multiple Cooldowns
+
+You can annotate multiple parameters with `Cooldown` on the same task. Each gets its own independent window scoped to that parameter's value. A task must pass *all* of its cooldown checks to start — if any one blocks, the task is dropped:
+
+```python
+from typing import Annotated
+
+async def sync_data(
+    customer_id: Annotated[int, Cooldown(timedelta(seconds=30))],
+    region: Annotated[str, Cooldown(timedelta(seconds=60))],
+) -> None:
+    await refresh_data(customer_id, region)
+
+# Runs immediately — both windows are clear
+await docket.add(sync_data)(customer_id=1, region="us")
+
+# Blocked — customer_id=1 is still in cooldown
+await docket.add(sync_data)(customer_id=1, region="eu")
+
+# Blocked — region="us" is still in cooldown
+await docket.add(sync_data)(customer_id=2, region="us")
+
+# Runs — both customer_id=2 and region="eu" are clear
+await docket.add(sync_data)(customer_id=2, region="eu")
+```
+
+Only one `Debounce` is allowed per task — its reschedule mechanism requires a single settle window.
+
 ### Combining with Other Controls
 
 Debounce, cooldown, and concurrency limits can all coexist on the same task:

src/docket/dependencies/__init__.py

@@ -19,8 +19,8 @@
     format_duration,
 )
 from ._concurrency import ConcurrencyBlocked, ConcurrencyLimit
-from ._cooldown import Cooldown, CooldownBlocked
-from ._debounce import Debounce, DebounceBlocked
+from ._cooldown import Cooldown
+from ._debounce import Debounce
 from ._cron import Cron
 from ._contextual import (
     CurrentDocket,
@@ -86,9 +86,7 @@
     "ConcurrencyBlocked",
     "ConcurrencyLimit",
     "Cooldown",
-    "CooldownBlocked",
     "Debounce",
-    "DebounceBlocked",
     "Cron",
     "Perpetual",
     "Progress",

src/docket/dependencies/_base.py

@@ -64,12 +64,18 @@ class AdmissionBlocked(Exception):
     ``retry_delay`` overrides the default reschedule delay when set.
     """
 
-    reschedule: bool = True
-    retry_delay: timedelta | None = None
-
-    def __init__(self, execution: Execution, reason: str = "admission control"):
+    def __init__(
+        self,
+        execution: Execution,
+        reason: str = "admission control",
+        *,
+        reschedule: bool = True,
+        retry_delay: timedelta | None = None,
+    ):
         self.execution = execution
         self.reason = reason
+        self.reschedule = reschedule
+        self.retry_delay = retry_delay
         super().__init__(f"Task {execution.key} blocked by {reason}")
 
 

src/docket/dependencies/_cooldown.py

@@ -8,25 +8,10 @@
 
 from datetime import timedelta
 from types import TracebackType
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from ._base import AdmissionBlocked, Dependency, current_docket, current_execution
 
-if TYPE_CHECKING:  # pragma: no cover
-    from ..execution import Execution
-
-
-class CooldownBlocked(AdmissionBlocked):
-    """Raised when a task is blocked by cooldown."""
-
-    reschedule = False
-
-    def __init__(self, execution: Execution, cooldown_key: str, window: timedelta):
-        self.cooldown_key = cooldown_key
-        self.window = window
-        reason = f"cooldown ({window}) on {cooldown_key}"
-        super().__init__(execution, reason=reason)
-
 
 class Cooldown(Dependency["Cooldown"]):
     """Execute first, drop duplicates within window.
@@ -47,8 +32,6 @@ async def process_customer(
         ) -> None: ...
     """
 
-    single: bool = True
-
     def __init__(self, window: timedelta, *, scope: str | None = None) -> None:
         self.window = window
         self.scope = scope
@@ -79,7 +62,11 @@ async def __aenter__(self) -> Cooldown:
             acquired = await redis.set(cooldown_key, 1, nx=True, px=window_ms)
 
         if not acquired:
-            raise CooldownBlocked(execution, cooldown_key, self.window)
+            raise AdmissionBlocked(
+                execution,
+                reason=f"cooldown ({self.window}) on {cooldown_key}",
+                reschedule=False,
+            )
 
         return self
 

src/docket/dependencies/_debounce.py

@@ -10,13 +10,10 @@
 import time
 from datetime import timedelta
 from types import TracebackType
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from ._base import AdmissionBlocked, Dependency, current_docket, current_execution
 
-if TYPE_CHECKING:  # pragma: no cover
-    from ..execution import Execution
-
 # Lua script for atomic debounce logic.
 #
 # KEYS[1] = winner key    (holds the execution key of the chosen task)
@@ -76,26 +73,6 @@
 _ACTION_DROP = 3
 
 
-class DebounceBlocked(AdmissionBlocked):
-    """Raised when a task is blocked by debounce."""
-
-    def __init__(
-        self,
-        execution: Execution,
-        debounce_key: str,
-        settle: timedelta,
-        *,
-        reschedule: bool,
-        retry_delay: timedelta | None = None,
-    ):
-        self.debounce_key = debounce_key
-        self.settle = settle
-        self.reschedule = reschedule  # type: ignore[assignment]
-        self.retry_delay = retry_delay  # type: ignore[assignment]
-        reason = f"debounce ({settle}) on {debounce_key}"
-        super().__init__(execution, reason=reason)
-
-
 class Debounce(Dependency["Debounce"]):
     """Wait for submissions to settle, then fire once.
 
@@ -160,22 +137,17 @@ async def __aenter__(self) -> Debounce:
         if action == _ACTION_PROCEED:
             return self
 
+        reason = f"debounce ({self.settle}) on {base_key}"
+
         if action == _ACTION_RESCHEDULE:
-            raise DebounceBlocked(
+            raise AdmissionBlocked(
                 execution,
-                base_key,
-                self.settle,
-                reschedule=True,
+                reason=reason,
                 retry_delay=timedelta(milliseconds=remaining_ms),
             )
 
         # DROP
-        raise DebounceBlocked(
-            execution,
-            base_key,
-            self.settle,
-            reschedule=False,
-        )
+        raise AdmissionBlocked(execution, reason=reason, reschedule=False)
 
     async def __aexit__(
         self,

tests/test_cooldown.py

@@ -6,8 +6,6 @@
 from datetime import timedelta
 from typing import Annotated
 
-import pytest
-
 from docket import ConcurrencyLimit, Docket, Worker
 from docket.dependencies import Cooldown
 
@@ -71,16 +69,31 @@ async def cooled_task(
     assert results.count(1) == 1
 
 
-async def test_cooldown_single_rejects_two(docket: Docket):
-    """single=True rejects two Cooldown on the same task."""
-    with pytest.raises(ValueError, match="Only one Cooldown"):
+async def test_multiple_cooldowns_on_different_parameters(
+    docket: Docket, worker: Worker
+):
+    """Multiple Cooldown annotations on different parameters are independent."""
+    results: list[tuple[int, str]] = []
+
+    async def task(
+        customer_id: Annotated[int, Cooldown(timedelta(seconds=5))],
+        region: Annotated[str, Cooldown(timedelta(seconds=5))],
+    ):
+        results.append((customer_id, region))
+
+    await docket.add(task)(customer_id=1, region="us")
+    await docket.add(task)(
+        customer_id=1, region="eu"
+    )  # same customer, different region
+    await docket.add(task)(
+        customer_id=2, region="us"
+    )  # different customer, same region
 
-        async def task(
-            a: Annotated[int, Cooldown(timedelta(seconds=1))],
-            b: Annotated[str, Cooldown(timedelta(seconds=2))],
-        ): ...  # pragma: no cover
+    worker.concurrency = 10
+    await worker.run_until_finished()
 
-        await docket.add(task)(a=1, b="x")
+    # First call runs. Second is blocked by customer_id=1. Third is blocked by region="us".
+    assert results == [(1, "us")]
 
 
 async def test_cooldown_coexists_with_concurrency_limit(docket: Docket, worker: Worker):

tests/test_debounce.py

@@ -66,8 +66,8 @@ async def debounced_task(
     assert results.count(1) == 1
 
 
-async def test_debounce_single_rejects_two(docket: Docket):
-    """single=True rejects two Debounce on the same task."""
+async def test_multiple_debounces_rejected(docket: Docket):
+    """Only one Debounce is allowed per task."""
     with pytest.raises(ValueError, match="Only one Debounce"):
 
         async def task(