From 14bacdc9d50a57f71ef1de474e157111c1f5ad2c Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Tue, 25 Aug 2026 10:15:44 -0400 Subject: [PATCH] fix: harden lease disconnect handling --- README.md | 4 +++ src/fitz_py/__init__.py | 2 ++ src/fitz_py/domains/lease.py | 35 +++++++++++++++---- src/fitz_py/errors.py | 4 +++ tests/unit/test_contracts.py | 65 ++++++++++++++++++++++++++++++++++++ 5 files changed, 104 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 77b38b6..e0cdcb2 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,10 @@ queue, stale-handle, and domain failures have stable string codes and structured cancellation is preserved. Requests cancelled after transmission leave a FIFO tombstone so a late reply cannot corrupt the next same-type request. +Schedule backend unavailability and broker saturation use the distinct coded +error `ERR_SCHEDULE_BACKEND_ERROR` (`7010`). It is retryable subject to the +operation's replay safety; it is never reported as a cron or parse error. + ## Verification ```bash diff --git a/src/fitz_py/__init__.py b/src/fitz_py/__init__.py index 0515d9e..30f5ac4 100644 --- a/src/fitz_py/__init__.py +++ b/src/fitz_py/__init__.py @@ -46,6 +46,7 @@ Worker, ) from fitz_py.errors import ( + ERR_SCHEDULE_BACKEND_ERROR, AuthenticationError, CodecError, DomainError, @@ -87,6 +88,7 @@ ) __all__ = ( + "ERR_SCHEDULE_BACKEND_ERROR", "AsyncSubscription", "AuthenticationError", "Availability", diff --git a/src/fitz_py/domains/lease.py b/src/fitz_py/domains/lease.py index b861a4c..2a3c20e 100644 --- a/src/fitz_py/domains/lease.py +++ b/src/fitz_py/domains/lease.py @@ -173,11 +173,19 @@ async def acquire( resolved_owner_id = self._owner_id if owner_id is None else owner_id if not resolved_owner_id: raise ValueError("owner_id must not be empty") - queued = asyncio.get_running_loop().create_future() + queued: asyncio.Future[bytes] | None = None + if wait > 0: + queued = asyncio.get_running_loop().create_future() + # A disconnect can fail this internal future while the primary + # ACQUIRE request is still pending. Observe that failure from the + # moment it can be published; awaiting the same future later still + # preserves the public acquisition error. + queued.add_done_callback(_observe_acquire_future) await self._acquire_lock.acquire() release_lock = True try: - self._queued.append(queued) + if queued is not None: + self._queued.append(queued) writer = BufferWriter() writer.write_route(route) writer.write_route(resolved_owner_id) @@ -186,19 +194,21 @@ async def acquire( payload = await self.request_frame(MSG_LEASE_ACQUIRE, writer.build()) response_type, token = self._decode_acquire(payload) if response_type not in {2, 3}: - self._queued.remove(queued) + if queued is not None: + self._queued.remove(queued) else: - response_type, token = self._decode_acquire(await asyncio.shield(queued)) + queued_response = _require_queued_acquire_future(queued) + response_type, token = self._decode_acquire(await asyncio.shield(queued_response)) _validate_final_acquire(response_type) except asyncio.CancelledError: - if queued in self._queued: + if queued is not None and queued in self._queued: task = asyncio.create_task(self._drain_cancelled_acquire(queued)) self._cancelled_acquires.add(task) task.add_done_callback(self._cancelled_acquire_completed) release_lock = False raise except BaseException: - if queued in self._queued: + if queued is not None and queued in self._queued: with contextlib.suppress(ValueError): self._queued.remove(queued) raise @@ -356,6 +366,19 @@ def _validate_final_acquire(response_type: int) -> None: raise LeaseError("ACQUIRE returned a second queued response", "INVALID_RESPONSE") +def _observe_acquire_future(future: asyncio.Future[bytes]) -> None: + if not future.cancelled(): + future.exception() + + +def _require_queued_acquire_future( + future: asyncio.Future[bytes] | None, +) -> asyncio.Future[bytes]: + if future is None: + raise LeaseError("ACQUIRE queued without a positive wait", "INVALID_RESPONSE") + return future + + def _duration(value: object, name: str, *, positive: bool = False) -> None: if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{name} must be an integer number of seconds") diff --git a/src/fitz_py/errors.py b/src/fitz_py/errors.py index e1544c7..8202698 100644 --- a/src/fitz_py/errors.py +++ b/src/fitz_py/errors.py @@ -137,6 +137,9 @@ class ScheduleError(DomainError): prefix = "SCHEDULE" +ERR_SCHEDULE_BACKEND_ERROR = 7010 + + _RETRYABLE = { ("KV", 1004), ("KV", 1009), @@ -148,6 +151,7 @@ class ScheduleError(DomainError): ("RPC", 6002), ("RPC", 6003), ("RPC", 6004), + ("SCHEDULE", ERR_SCHEDULE_BACKEND_ERROR), } diff --git a/tests/unit/test_contracts.py b/tests/unit/test_contracts.py index 47027e0..cd7bfc8 100644 --- a/tests/unit/test_contracts.py +++ b/tests/unit/test_contracts.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import gc import json from pathlib import Path @@ -16,6 +17,7 @@ from fitz_py.domains.schedule import DeliveryMode, ScheduleClient from fitz_py.domains.stream import StreamClient, StreamSession, _assert_stream_pattern from fitz_py.errors import ( + ERR_SCHEDULE_BACKEND_ERROR, FitzConnectionError, FitzTimeoutError, FitzTransportError, @@ -27,6 +29,7 @@ ScheduleError, StreamError, SubscriptionBackpressureError, + is_retryable, ) from fitz_py.multiplexer import Multiplexer from fitz_py.protocol.buffer import BufferReader, BufferWriter @@ -72,6 +75,13 @@ from fitz_py.types import ClientConfig, ConcurrencyLimits +def test_schedule_backend_error_code_is_distinct_and_retryable() -> None: + error = ScheduleError("backend busy", "BACKEND_ERROR", ERR_SCHEDULE_BACKEND_ERROR) + + assert ERR_SCHEDULE_BACKEND_ERROR == 7010 + assert is_retryable(error) + + class FakeConnection: def __init__(self, responses: dict[int, bytes] | None = None) -> None: self.config = ClientConfig(url="tcp://localhost:1") @@ -111,6 +121,18 @@ async def run_with_retry(self, operation, *, replay_safe): return await operation() +class PendingLeaseConnection(FakeConnection): + def __init__(self) -> None: + super().__init__() + self.release_request = asyncio.Event() + + async def request(self, message_type: int, payload: bytes) -> bytes: + self.sent.append((message_type, payload)) + self.request_started.set() + await self.release_request.wait() + raise FitzConnectionError("transport disconnected") + + def test_clean_break_public_surface() -> None: assert hasattr(fitz_py, "ClientConfig") assert hasattr(fitz_py, "FitzLogger") @@ -414,6 +436,19 @@ async def test_kv_error_response_decodes_domain_code_and_message() -> None: assert raised.value.domain_code == 1006 +@pytest.mark.asyncio +async def test_schedule_backend_error_response_preserves_distinct_code() -> None: + connection = FakeConnection( + {MSG_SCHEDULE_LIST: _coded_error(ERR_SCHEDULE_BACKEND_ERROR, "backend busy")} + ) + + with pytest.raises(ScheduleError, match="backend busy") as raised: + await ScheduleClient(connection).list_schedules() + + assert raised.value.domain_code == 7010 + assert is_retryable(raised.value) + + @pytest.mark.asyncio async def test_domain_success_decoders_reject_malformed_flags_and_lengths() -> None: transaction = KVTransaction(FakeConnection({MSG_KV_GET: b"\0\2"}), "kv://r/a/items", 7) @@ -578,6 +613,36 @@ async def test_queued_lease_wait_is_unblocked_by_disconnect() -> None: await pending +@pytest.mark.asyncio +@pytest.mark.parametrize("wait", [0, 30]) +async def test_disconnect_during_primary_acquire_has_no_unretrieved_internal_future( + wait: int, +) -> None: + connection = PendingLeaseConnection() + client = LeaseClient(connection) + loop = asyncio.get_running_loop() + contexts: list[dict[str, object]] = [] + previous_handler = loop.get_exception_handler() + loop.set_exception_handler(lambda _loop, context: contexts.append(context)) + + try: + pending = asyncio.create_task(client.acquire("lease://r/a/first", ttl=30, wait=wait)) + await asyncio.wait_for(connection.request_started.wait(), 1) + + client._disconnect() + connection.release_request.set() + + with pytest.raises(FitzConnectionError, match="transport disconnected"): + await pending + del pending + gc.collect() + await asyncio.sleep(0) + assert contexts == [] + assert not client._queued + finally: + loop.set_exception_handler(previous_handler) + + @pytest.mark.asyncio async def test_stream_operations_decode_plain_error_envelopes() -> None: error = _plain_error("stream conflict")