diff --git a/src/inflowpay/x402/_payment.py b/src/inflowpay/x402/_payment.py index 180876a..42f0d66 100644 --- a/src/inflowpay/x402/_payment.py +++ b/src/inflowpay/x402/_payment.py @@ -106,8 +106,10 @@ async def await_payload( completion = self._completion self._waiters += 1 try: - # A cancelled waiter must not cancel another caller's wait for this same payment. - result = await asyncio.shield(completion) + # Keep cancellation local to this waiter without shield() logging late + # hook errors on Python 3.14; _observe_completion already observes them. + await asyncio.wait((completion,)) + result = completion.result() return EncodedPayment( result.encoded_payload, result.payment_payload.model_copy(deep=True), diff --git a/src/inflowpay/x402/buyer.py b/src/inflowpay/x402/buyer.py index 6b1ec21..67e22fb 100644 --- a/src/inflowpay/x402/buyer.py +++ b/src/inflowpay/x402/buyer.py @@ -131,7 +131,11 @@ async def get_supported(self, *, refresh: bool = False) -> SupportedResponse: if self._refresh is None: self._refresh = asyncio.create_task(self._refresh_supported()) self._refresh.add_done_callback(self._finish_refresh) - await asyncio.shield(self._refresh) + # shield() logs late failures after cancellation on Python 3.14, + # even though _finish_refresh observes them. Keep the shared request independent. + task = self._refresh + await asyncio.wait((task,)) + task.result() return self._supported.model_copy(deep=True) async def _refresh_supported(self) -> None: diff --git a/src/inflowpay/x402/seller.py b/src/inflowpay/x402/seller.py index 29c41c7..48782fd 100644 --- a/src/inflowpay/x402/seller.py +++ b/src/inflowpay/x402/seller.py @@ -33,7 +33,11 @@ async def get(self, refresh: bool = False) -> _T: if self.task is None: self.task = asyncio.create_task(self._fetch()) self.task.add_done_callback(self._finish) - await asyncio.shield(self.task) + # shield() logs late failures after cancellation on Python 3.14, + # even though _finish observes them. Keep the shared request independent. + task = self.task + await asyncio.wait((task,)) + task.result() assert self.value is not None return self.value.model_copy(deep=True) diff --git a/tests/test_x402_buyer.py b/tests/test_x402_buyer.py index 5e02b13..1b360a4 100644 --- a/tests/test_x402_buyer.py +++ b/tests/test_x402_buyer.py @@ -796,6 +796,106 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: await task +@pytest.mark.parametrize("remaining_waiter", [False, True]) +async def test_cancelled_refresh_waiter_does_not_log_late_failure(remaining_waiter: bool) -> None: + class BlockingPlatform(Platform): + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("x402-supported") and self.requests: + self.poll_started.set() + await self.release.wait() + return await super().handle_async_request(request) + + platform = BlockingPlatform() + observed: list[dict[str, Any]] = [] + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + loop.set_exception_handler(lambda _, event: observed.append(event)) + try: + async with await buyer(platform) as client: + first = asyncio.create_task(client.get_supported(refresh=True)) + await platform.poll_started.wait() + shared = client._refresh + assert shared is not None + second = ( + asyncio.create_task(client.get_supported(refresh=True)) + if remaining_waiter + else None + ) + await asyncio.sleep(0) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + assert not shared.done() + platform.supported = httpx.Response(503) + platform.release.set() + await asyncio.gather(shared, return_exceptions=True) + if second is not None: + with pytest.raises(InflowApiError) as failure: + await second + assert failure.value is shared.exception() + await asyncio.sleep(0) + assert observed == [] + assert client._refresh is None + assert client.supports(REQUIREMENT) + assert len(platform.requests) == 2 + platform.supported = SUPPORTED + await client.get_supported(refresh=True) + assert len(platform.requests) == 3 + finally: + loop.set_exception_handler(previous) + + +@pytest.mark.parametrize("remaining_waiter", [False, True]) +async def test_cancelled_payment_waiter_preserves_late_hook_error(remaining_waiter: bool) -> None: + platform = Platform() + started, release = asyncio.Event(), asyncio.Event() + failure = RuntimeError("application hook failed") + calls = 0 + observed: list[dict[str, Any]] = [] + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + loop.set_exception_handler(lambda _, event: observed.append(event)) + try: + async with await buyer(platform) as client: + + async def after(_: PaymentCreatedContext) -> None: + nonlocal calls + calls += 1 + started.set() + await release.wait() + raise failure + + client.on_after_payment_creation(after) + payment = await client.prepare(REQUIREMENT, RESOURCE) + first = asyncio.create_task(payment.await_payload()) + await started.wait() + completion = payment._completion + assert completion is not None + second = asyncio.create_task(payment.await_payload()) if remaining_waiter else None + await asyncio.sleep(0) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + assert not completion.done() + release.set() + await asyncio.gather(completion, return_exceptions=True) + if second is not None: + with pytest.raises(RuntimeError) as caught: + await second + assert caught.value is failure + await asyncio.sleep(0) + assert observed == [] + for _ in range(2): + with pytest.raises(RuntimeError) as caught: + await payment.await_payload() + assert caught.value is failure + assert calls == 1 + assert payment._completion is completion + assert not any(r.url.path.endswith("/cancel") for r in platform.requests) + finally: + loop.set_exception_handler(previous) + + async def test_expired_capabilities_refresh() -> None: platform = Platform() async with await buyer(platform) as client: diff --git a/tests/test_x402_seller.py b/tests/test_x402_seller.py index 8d0905b..9c26015 100644 --- a/tests/test_x402_seller.py +++ b/tests/test_x402_seller.py @@ -264,6 +264,48 @@ async def test_refresh_concurrency_cancellation_and_failure() -> None: assert platform.closed +@pytest.mark.parametrize("operation", ["config", "supported"]) +@pytest.mark.parametrize("remaining_waiter", [False, True]) +async def test_cancelled_refresh_waiter_does_not_log_late_failure( + operation: str, remaining_waiter: bool +) -> None: + platform = Platform() + observed: list[dict[str, Any]] = [] + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + loop.set_exception_handler(lambda _, event: observed.append(event)) + try: + async with await Seller.create(platform.options()) as seller: + platform.block = operation + refresh = seller.config if operation == "config" else seller.get_supported + first = asyncio.create_task(refresh(refresh=True)) + await platform.started.wait() + cache = seller._config if operation == "config" else seller._supported + shared = cache.task + assert shared is not None + second = asyncio.create_task(refresh(refresh=True)) if remaining_waiter else None + await asyncio.sleep(0) + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + assert not shared.done() + setattr(platform, operation, httpx.Response(503)) + platform.release.set() + await asyncio.gather(shared, return_exceptions=True) + if second is not None: + with pytest.raises(InflowApiError) as failure: + await second + assert failure.value is shared.exception() + await asyncio.sleep(0) + assert observed == [] + assert cache.task is None + assert sum(r.url.path.endswith(operation) for r in platform.requests) == 2 + await refresh() + assert len(platform.requests) == 3 + finally: + loop.set_exception_handler(previous) + + async def test_fixed_facilitator_snapshot_and_recreation() -> None: platform = Platform() async with await Facilitator.create(platform.options()) as facilitator: