diff --git a/routstr/wallet.py b/routstr/wallet.py index bd59de91..e3384bfe 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -89,30 +89,82 @@ class _MintRateGuard: asyncio.Semaphore(max_concurrency) if max_concurrency > 0 else None ) self._cooldown_until = 0.0 + self._needs_probe = False + self._probe_lock = asyncio.Lock() def apply_cooldown(self, delay: float) -> None: self._cooldown_until = max( self._cooldown_until, time.monotonic() + max(0.0, delay) ) + self._needs_probe = True def cooldown_remaining(self) -> float: return max(0.0, self._cooldown_until - time.monotonic()) - async def _run_after_cooldown(self, factory: Callable[[], Awaitable[Any]]) -> Any: - wait = self.cooldown_remaining() - if wait > 0: + async def _wait_for_cooldown(self) -> None: + while True: + deadline = self._cooldown_until + wait = max(0.0, deadline - time.monotonic()) + if wait <= 0: + return logger.debug( "Mint rate guard: cooling down", extra={"mint_url": self._mint_url, "wait_seconds": round(wait, 2)}, ) await asyncio.sleep(wait) - return await factory() + if self._cooldown_until <= deadline: + return + + async def _run_probe(self, factory: Callable[[], Awaitable[Any]]) -> Any: + await self._wait_for_cooldown() + logger.warning( + "Mint cooldown ended; sending one probe request", + extra={"event": "mint_cooldown_probe_started", "mint_url": self._mint_url}, + ) + try: + result = await factory() + except Exception as error: + # Keep queued callers behind the probe while _mint_operation applies + # the precise Retry-After/backoff from this failure. + self.apply_cooldown(1.0) + logger.warning( + "Mint cooldown probe failed", + extra={ + "event": "mint_cooldown_probe_failed", + "mint_url": self._mint_url, + "error": str(error), + "error_type": type(error).__name__, + }, + ) + raise + + self._needs_probe = False + self._cooldown_until = 0.0 + logger.warning( + "Mint cooldown probe succeeded; restoring normal concurrency", + extra={ + "event": "mint_cooldown_probe_succeeded", + "mint_url": self._mint_url, + }, + ) + return result async def run(self, factory: Callable[[], Awaitable[Any]]) -> Any: - if self._semaphore is None: - return await self._run_after_cooldown(factory) - async with self._semaphore: - return await self._run_after_cooldown(factory) + while True: + if self._needs_probe or self.cooldown_remaining() > 0: + async with self._probe_lock: + if self.cooldown_remaining() > 0: + self._needs_probe = True + if self._needs_probe: + return await self._run_probe(factory) + continue + + if self._semaphore is None: + return await factory() + async with self._semaphore: + if self._needs_probe: + continue + return await factory() def _mint_cooldown_remaining(mint_url: str) -> float: diff --git a/tests/unit/test_wallet.py b/tests/unit/test_wallet.py index a192eeac..fd3e3011 100644 --- a/tests/unit/test_wallet.py +++ b/tests/unit/test_wallet.py @@ -1511,6 +1511,35 @@ async def test_mint_rate_guard_waits_for_adaptive_cooldown() -> None: operation.assert_awaited_once() +@pytest.mark.asyncio +async def test_mint_rate_guard_allows_one_probe_after_cooldown() -> None: + from routstr.wallet import _MintRateGuard + + guard = _MintRateGuard("http://mint:3338", 4) + guard.apply_cooldown(0) + probe_started = asyncio.Event() + release_probe = asyncio.Event() + calls = 0 + + async def operation() -> int: + nonlocal calls + calls += 1 + if calls == 1: + probe_started.set() + await release_probe.wait() + return calls + + tasks = [asyncio.create_task(guard.run(operation)) for _ in range(5)] + await probe_started.wait() + await asyncio.sleep(0) + assert calls == 1 + + release_probe.set() + await asyncio.gather(*tasks) + assert calls == 5 + assert guard._needs_probe is False + + def test_mint_rate_guard_rebuilds_when_setting_changes() -> None: from routstr.core.settings import settings from routstr.wallet import _MintRateGuard