`Debounce` dependency

[chrisguidry]

Differently from a concurrency or a rate limit, sometime we just need to debounce rapidly-fired off tasks and totally skip them. I think it could look something like this:

async def send_alert(debounce = Debounce(cooldown=timedelta(seconds=30))):
    ...

Perhaps like concurrency limiting, though, it should have some intra-key awareness, so:

async def send_alert(
    customer_id: UUID,
    debounce = Debounce("customer_id", cooldown=timedelta(seconds=30))
):
    ...

But here's where I wish we had explored Annotated for this in Concurrency and maybe we can do that here:

async def send_alert(
    customer_id: Annotated[UUID, Debounce(cooldown=timedelta(seconds=30))]
):
    ...

Regardless of how we define it, the behavior would be tied in with the worker's admission control to skip this work (and to log that we are skipping it). Not sure if this needs new support in the worker or if we can do it with the hooks we have, but it should be a fairly lightweight addition and would support other future hypothetical dependencies.

Inspired by some work @zangell44 is doing on Prefect Cloud.

[chrisguidry]

One critical aspect is that this is kind of a weird hybrid lock/mutex/semaphore thingy during the first task's run. There needs to be a mutex where the first task is the only one allowed to try; then if it succeeds then we promote that mutex to the full cooldown period, but if it fails that mutex needs to be cleared so that another task can try. This also has an interesting interaction with Retry, so maybe that behavior is just opted-in by using Retry?

[chrisguidry]

Design decision: splitting debounce into two named classes with distinct semantics:

Debounce (leading edge) — "don't start this if one was recently started." Sets the Redis key on entry. Blocks rapid duplicate events. Key is set immediately before the task runs.

Cooldown (trailing edge) — "don't start this if one recently succeeded." Sets the Redis key on successful exit only. Failed tasks don't set the key, so the next attempt can proceed.

Both use a window parameter for the time duration. Both work as annotations and bare defaults:

# Annotation style — infers argument name from position
customer_id: Annotated[int, Debounce(timedelta(minutes=5))]
customer_id: Annotated[int, Cooldown(timedelta(minutes=5))]

# Bare default style — per-task scope
debounce: Debounce = Debounce(timedelta(minutes=5))

Both raise AdmissionBlocked subclasses when blocked, so the worker's existing reschedule flow handles them.