fix: harden Lease acquisition disconnect handling

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

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",

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")

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),
 }
 
 

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")