Address review: drop the fragments and the extra timeout branch

Nothing rate-limit has been released, so 12.feature.rst already covers the
base class and the two #25 fragments go.

The exhausted-budget case stops being handled separately. One message again,
printing the real remaining rather than clamping a negative one to 0.000s, so
an acquire() that overran still shows up in it. _BudgetEater goes too, its
scenario folded into a parametrised case on _ElapsedAcquire that pins the
negative budget, so re-adding the clamp still fails a test.

asyncio.ensure_future becomes asyncio.create_task in the two places called
out; the two that predate this branch are left alone. parametrize takes a
tuple. The _fake_request docstring goes back to what it said, since 3.15
will date the replacement.

The prose around the Redis sketch and release() was longer than what it had
to say. Per-domain keying keeps the fact the deferred fix turns on -- that
clone() takes no arguments, so it cannot learn the host.

Author: rodrigobnogueira

CHANGES/25.breaking.rst

@@ -26,10 +26,9 @@ class RateLimiter(ABC):
     logic lives in :meth:`wait`, shared by every algorithm, so reserving a
     slot may perform I/O of its own -- against Redis or a database, say.
 
-    An :meth:`acquire` implementation must be cancellation-safe. If it is
-    cancelled or raises before returning, it is responsible for ensuring
-    that no reservation is left behind. Once it returns, :meth:`wait` owns
-    the reservation and calls :meth:`release` if the slot cannot be used.
+    Until :meth:`acquire` returns, cleaning up a half-made reservation is
+    its own responsibility; once it returns, :meth:`wait` owns the slot and
+    calls :meth:`release` if it cannot be used.
 
     An async method that contains no suspension point still runs atomically
     when awaited directly. :class:`TokenBucket` relies on that property to
@@ -40,11 +39,9 @@ class RateLimiter(ABC):
     async def acquire(self) -> float:
         """Reserve a slot and return the delay to sleep before sending.
 
-        The delay must be a non-negative, finite number of seconds.
-        :meth:`wait` takes it on trust: a NaN compares false against both
-        the budget and zero, so the request would go out unthrottled.
-
-        If cancellation or another exception prevents this method from
+        The delay must be non-negative, finite seconds; :meth:`wait` takes that
+        on trust, and a NaN would send the request through unthrottled. If
+        cancellation or another exception prevents this method from
         returning, it must not leave a reservation behind.
         """
 
@@ -64,40 +61,31 @@ def release(self) -> None:
         cancelled while sleeping. The default is a no-op for algorithms
         that have nothing to return.
 
-        Runs from an ``except asyncio.CancelledError`` block, so it must
-        not await: a second cancellation, or the loop shutting down, would
-        truncate it part-way and lose the slot for good. A limiter that has
-        to reach its backend to hand a slot back can schedule that round
-        trip as a task from here.
-
-        It must not raise, either. :meth:`wait` calls it while unwinding, so
-        an exception here would replace the :exc:`asyncio.TimeoutError` the
-        caller is owed -- or the :exc:`asyncio.CancelledError`, leaving a
-        cancelled request reporting an ordinary failure.
+        Must neither await nor raise, since one of those calls is from an
+        ``except asyncio.CancelledError`` block: awaiting there can be
+        truncated part-way, and raising would replace the exception the
+        caller is owed. A limiter that has to reach its backend to hand a
+        slot back can schedule that round trip as a task from here.
         """
 
     async def wait(self, timeout: float | None = None) -> None:
         """Reserve a slot and wait until the request may be sent.
 
-        Time spent in :meth:`acquire` is charged against *timeout* once it
-        returns, but is not bounded by it, so an implementation that can
-        hang needs a deadline of its own. When the delay left to serve
-        exceeds what is left of the budget, the slot is handed back and
-        :exc:`asyncio.TimeoutError` is raised without sleeping.
+        Time in :meth:`acquire` is charged against *timeout* once it
+        returns, though not bounded by it, so an implementation that can
+        hang needs its own deadline. When the delay exceeds what is left,
+        the slot is handed back and :exc:`asyncio.TimeoutError` raised
+        without sleeping.
         """
         started = time.monotonic()
         delay = await self.acquire()
 
         if timeout is not None:
-            elapsed = time.monotonic() - started
-            remaining = timeout - elapsed
+            # Goes negative when acquiring alone outlasted the timeout; the
+            # message reports it as such rather than clamping it to zero.
+            remaining = timeout - (time.monotonic() - started)
             if delay > remaining:
                 self.release()
-                if remaining <= 0.0:
-                    raise asyncio.TimeoutError(
-                        f"reserving a rate-limit slot took {elapsed:.3f}s, "
-                        f"exhausting the {timeout:.3f}s timeout"
-                    )
                 raise asyncio.TimeoutError(
                     f"rate limiter would delay the request {delay:.3f}s, "
                     f"beyond the {remaining:.3f}s remaining timeout"

CHANGES/25.bugfix.rst

@@ -103,29 +103,22 @@ Rate limiting
            def clone(self):
                return RedisLimiter(self._redis, self._key, self._script)
 
-   Three shortcuts in that sketch matter. Its ``clone()`` reuses one key, so
-   under ``per_domain=True`` every host would draw on a single shared limit --
-   and since ``clone()`` is called as a no-argument factory it never learns the
-   host, so a shared backend cannot key itself per host that way. Give each host
-   its own limiter, built with its own key, rather than using ``per_domain=True``.
-   It leaves ``release()`` at the default no-op, so a slot reserved just before
-   a timeout bail is not handed back. And a cancellation between the script
-   running and its reply arriving leaves a reservation nobody holds. The last
-   two are what an expiry on the reservation is for: give every slot one, and
-   an abandoned slot lapses on its own.
+   The sketch keeps one key across clones, and ``clone()`` takes no arguments
+   so it cannot learn the host; give each host its own limiter rather than
+   using ``per_domain=True``. It also leaves ``release()`` at the default
+   no-op, and a cancelled round trip can leave a reservation nobody holds;
+   an expiry on each reservation covers both.
 
    ``wait(timeout=None)`` is supplied by the base class. It charges async
-   acquisition against *timeout* once ``acquire()`` returns -- it does not
-   bound the call itself, so an implementation that can hang needs a deadline
-   of its own -- then fails fast when the delay left to serve exceeds what is
-   left of the budget, sleeps otherwise, and calls ``release()`` if a
-   successfully acquired slot cannot be used. ``release()`` is synchronous
-   and defaults to a no-op for algorithms that have nothing to return. It stays
-   synchronous because ``wait()`` also calls it from a cancellation handler,
-   where an awaiting implementation can be truncated part-way and lose the slot
-   for good; a limiter that has to reach its backend to hand one back can
-   schedule that round trip as a task. It must not raise either, since
-   ``wait()`` calls it while unwinding.
+   acquisition against *timeout* once ``acquire()`` returns -- without bounding
+   the call itself, so an implementation that can hang needs its own deadline --
+   then fails fast when the delay exceeds what is left, sleeps otherwise, and
+   calls ``release()`` if an acquired slot cannot be used. ``release()`` defaults
+   to a no-op for algorithms with nothing to return, and stays synchronous: one
+   of those calls is from a cancellation handler, where awaiting can be truncated
+   part-way and lose the slot for good, and raising would replace the exception
+   the caller is owed. A limiter that has to reach its backend to hand a slot
+   back can schedule that round trip as a task.
 
    ``acquire()`` must be cancellation-safe: if cancellation or another
    exception prevents it from returning, it must leave no reservation behind.

aiohttp_client_middlewares/rate_limit.py

@@ -35,9 +35,8 @@ def _fake_request(
     """A stand-in ``ClientRequest`` exposing just ``.url`` and ``.timeout``.
 
     The timeout is set here, while the mock is still untyped, rather than on
-    the returned value: the supported aiohttp releases have no
-    ``ClientRequest.timeout`` at all, and 3.15 adds it read-only, so assigning
-    it through the ``ClientRequest`` annotation type-checks on neither.
+    the returned value: ``ClientRequest.timeout`` is a read-only property, so
+    assigning it through the ``ClientRequest`` annotation does not type-check.
     """
     req = mock.create_autospec(ClientRequest, instance=True)
     req.url = URL(f"http://{host}")
@@ -428,28 +427,45 @@ async def handler(req: ClientRequest) -> ClientResponse:
 
 
 class _ElapsedAcquire(RateLimiter):
-    """Limiter that advances a fake clock while reserving a slot."""
+    """Limiter that spends fake time reserving a slot, then asks for a delay."""
 
-    def __init__(self, clock: _FakeClock) -> None:
+    def __init__(self, clock: _FakeClock, spend: float, delay: float) -> None:
         self._clock = clock
+        self._spend = spend
+        self._delay = delay
         self.releases = 0
 
     async def acquire(self) -> float:
-        self._clock.advance(0.075)
-        return 0.05
+        self._clock.advance(self._spend)
+        return self._delay
 
     def release(self) -> None:
         self.releases += 1
 
     def clone(self) -> "_ElapsedAcquire":
-        return _ElapsedAcquire(self._clock)
+        return _ElapsedAcquire(self._clock, self._spend, self._delay)
 
 
-async def test_acquisition_time_reduces_remaining_timeout(clock: _FakeClock) -> None:
-    """The delay is checked against the budget left after acquisition."""
-    limiter = _ElapsedAcquire(clock)
+@pytest.mark.parametrize(
+    ("spend", "delay", "budget_left"),
+    (
+        (0.075, 0.05, r"0\.025s"),  # some budget survived acquisition
+        (0.5, 0.0, r"-0\.400s"),  # acquiring alone outlasted the whole timeout
+    ),
+)
+async def test_acquisition_time_reduces_remaining_timeout(
+    clock: _FakeClock, spend: float, delay: float, budget_left: str
+) -> None:
+    """The delay is checked against the budget left after acquisition.
 
-    with pytest.raises(asyncio.TimeoutError, match="remaining timeout"):
+    The overrun case reports a negative budget rather than clamping it: a
+    0.000s delay said to exceed a 0.000s timeout explains nothing.
+    """
+    limiter = _ElapsedAcquire(clock, spend, delay)
+
+    with pytest.raises(
+        asyncio.TimeoutError, match=f"beyond the {budget_left} remaining"
+    ):
         await limiter.wait(timeout=0.1)
 
     assert limiter.releases == 1
@@ -480,7 +496,7 @@ def clone(self) -> "_CancelledAcquire":
 async def test_acquisition_cancellation_does_not_call_release() -> None:
     """Cleanup before acquire returns belongs to the acquire implementation."""
     limiter = _CancelledAcquire()
-    task = asyncio.ensure_future(limiter.wait(timeout=60.0))
+    task = asyncio.create_task(limiter.wait(timeout=60.0))
     await asyncio.sleep(0)
 
     await _cancel_and_join(task)
@@ -489,40 +505,6 @@ async def test_acquisition_cancellation_does_not_call_release() -> None:
     assert limiter.releases == 0
 
 
-class _BudgetEater(RateLimiter):
-    """Limiter whose reservation alone outlasts the caller's whole timeout."""
-
-    def __init__(self, clock: _FakeClock) -> None:
-        self._clock = clock
-        self.releases = 0
-
-    async def acquire(self) -> float:
-        self._clock.advance(0.5)
-        return 0.0  # no delay left to serve -- the budget went on acquiring
-
-    def release(self) -> None:
-        self.releases += 1
-
-    def clone(self) -> "_BudgetEater":
-        return _BudgetEater(self._clock)
-
-
-async def test_acquisition_alone_can_exhaust_the_timeout(clock: _FakeClock) -> None:
-    """A zero delay still fails once acquiring has spent the whole budget.
-
-    The message has to name acquisition as the cause: reporting a 0.000s delay
-    as exceeding a 0.000s budget would be self-contradictory.
-    """
-    limiter = _BudgetEater(clock)
-
-    with pytest.raises(
-        asyncio.TimeoutError, match=r"took 0\.500s, exhausting the 0\.100s"
-    ):
-        await limiter.wait(timeout=0.1)
-
-    assert limiter.releases == 1
-
-
 async def test_non_positive_total_timeout_means_no_deadline() -> None:
     """aiohttp arms no deadline for total <= 0, so the limiter must not either.
 
@@ -566,7 +548,7 @@ async def test_token_bucket_acquire_never_yields_to_the_loop(clock: _FakeClock)
     async def competitor() -> None:
         ran.append("competitor")
 
-    task = asyncio.ensure_future(competitor())
+    task = asyncio.create_task(competitor())
     await bucket.acquire()
     assert ran == [], "acquire() yielded, so callers are no longer ordered"
 
@@ -594,7 +576,7 @@ def clone(self) -> "_RecordsReleases":
         return _RecordsReleases(self._delay, self._error)
 
 
-@pytest.mark.parametrize("delay", [0.0, 0.001])
+@pytest.mark.parametrize("delay", (0.0, 0.001))
 async def test_wait_keeps_a_slot_it_actually_used(delay: float) -> None:
     """A slot the caller goes on to use must never be handed back.