reference parity

Author: pavel-kirienko

src/pycyphal/_publisher.py

@@ -8,7 +8,7 @@
 from ._api import DeliveryError, Instant, LivenessError, Priority, SendError
 from ._api import Publisher, Topic, ResponseStream, Response
 from ._header import MsgBeHeader, MsgRelHeader, RspBeHeader, RspRelHeader
-from ._node import ACK_BASELINE_DEFAULT_TIMEOUT, NodeImpl, SESSION_LIFETIME, TopicImpl
+from ._node import ACK_BASELINE_DEFAULT_TIMEOUT, NodeImpl, PublishTracker, SESSION_LIFETIME, TopicImpl
 from ._transport import TransportArrival
 
 _logger = logging.getLogger(__name__)
@@ -93,33 +93,15 @@ async def __call__(
             raise SendError("Publisher closed")
 
         tag = self._topic.next_tag()
-        lage = self._topic.lage(Instant.now().s)
-
-        hdr: MsgBeHeader | MsgRelHeader
-        if reliable:
-            hdr = MsgRelHeader(
-                topic_log_age=lage,
-                topic_evictions=self._topic.evictions,
-                topic_hash=self._topic.hash,
-                tag=tag,
-            )
-        else:
-            hdr = MsgBeHeader(
-                topic_log_age=lage,
-                topic_evictions=self._topic.evictions,
-                topic_hash=self._topic.hash,
-                tag=tag,
-            )
-
-        data = hdr.serialize() + bytes(message)
-        writer = self._topic.ensure_writer()
+        payload = bytes(message)
 
         if not reliable:
-            await writer(deadline, self._priority, data)
+            writer = self._topic.ensure_writer()
+            await writer(deadline, self._priority, self._serialize_message(tag, payload, reliable=False))
             _logger.debug("Published BE tag=%d topic='%s'", tag, self._topic.name)
             return
 
-        await self._reliable_publish(deadline, tag, data)
+        await self._reliable_publish(deadline, tag, payload)
 
     async def request(
         self,
@@ -131,15 +113,7 @@ async def request(
             raise SendError("Publisher closed")
 
         tag = self._topic.next_tag()
-        lage = self._topic.lage(Instant.now().s)
-
-        hdr = MsgRelHeader(
-            topic_log_age=lage,
-            topic_evictions=self._topic.evictions,
-            topic_hash=self._topic.hash,
-            tag=tag,
-        )
-        data = hdr.serialize() + bytes(message)
+        payload = bytes(message)
 
         # Create response stream before publishing so it's ready to receive.
         stream = ResponseStreamImpl(
@@ -150,81 +124,160 @@ async def request(
         )
         self._topic.request_futures[tag] = stream
 
-        task = self._node.loop.create_task(self._request_publish(delivery_deadline, tag, data, stream))
+        tracker = self._prepare_reliable_publish_tracker(tag, delivery_deadline.ns, payload)
+        try:
+            initial_window = await self._reliable_publish_start(delivery_deadline, tag, payload, tracker)
+        except BaseException:
+            self._topic.request_futures.pop(tag, None)
+            self._release_reliable_publish_tracker(tag, tracker)
+            raise
+
+        task = self._node.loop.create_task(
+            self._request_publish(delivery_deadline, tag, payload, stream, tracker, initial_window)
+        )
         stream.set_publish_task(task)
         return stream
 
     async def _request_publish(
         self,
         deadline: Instant,
         tag: int,
-        data: bytes,
+        payload: bytes,
         stream: ResponseStreamImpl,
+        tracker: PublishTracker,
+        initial_window: tuple[int, bool],
     ) -> None:
         try:
-            await self._reliable_publish(deadline, tag, data)
+            await self._reliable_publish_continue(deadline, tag, payload, tracker, initial_window)
         except asyncio.CancelledError:
             raise
         except BaseException as ex:
             stream.on_publish_error(ex)
+        finally:
+            self._release_reliable_publish_tracker(tag, tracker)
 
     @staticmethod
     def _ack_is_last_attempt(current_ack_deadline_ns: int, current_ack_timeout: float, total_deadline_ns: int) -> bool:
         next_ack_timeout_ns = round(current_ack_timeout * 2 * 1e9)
         remaining_budget_ns = total_deadline_ns - current_ack_deadline_ns
         return remaining_budget_ns < next_ack_timeout_ns
 
-    async def _reliable_publish(self, deadline: Instant, tag: int, data: bytes) -> None:
-        writer = self._topic.ensure_writer()
-        tracker = self._node.prepare_publish_tracker(self._topic, tag, deadline.ns, data)
+    def _serialize_message(self, tag: int, payload: bytes, *, reliable: bool) -> bytes:
+        lage = self._topic.lage(Instant.now().s)
+        hdr = (MsgRelHeader if reliable else MsgBeHeader)(
+            topic_log_age=lage,
+            topic_evictions=self._topic.evictions,
+            topic_hash=self._topic.hash,
+            tag=tag,
+        )
+        return hdr.serialize() + payload
+
+    @staticmethod
+    def _reliable_publish_window(deadline_ns: int, ack_timeout: float) -> tuple[int, bool] | None:
+        now_ns = Instant.now().ns
+        if now_ns >= deadline_ns:
+            return None
+        ack_deadline_ns = min(deadline_ns, now_ns + round(ack_timeout * 1e9))
+        return ack_deadline_ns, PublisherImpl._ack_is_last_attempt(ack_deadline_ns, ack_timeout, deadline_ns)
+
+    def _prepare_reliable_publish_tracker(self, tag: int, deadline_ns: int, payload: bytes) -> PublishTracker:
+        tracker = self._node.prepare_publish_tracker(self._topic, tag, deadline_ns, payload)
         tracker.ack_timeout = self.ack_timeout
         self._topic.publish_futures[tag] = tracker
-        first_attempt = True
-        try:
-            while True:
-                if tracker.acknowledged and not tracker.remaining:
-                    _logger.debug("Reliable publish ACKed tag=%d topic='%s'", tag, self._topic.name)
-                    return
-                now_ns = Instant.now().ns
-                if now_ns >= deadline.ns:
-                    break
-                ack_deadline_ns = min(deadline.ns, now_ns + round(tracker.ack_timeout * 1e9))
-                last_attempt = self._ack_is_last_attempt(ack_deadline_ns, tracker.ack_timeout, deadline.ns)
-                tracker.ack_event.clear()
-                try:
-                    tx_deadline = Instant(ns=ack_deadline_ns)
-                    if (not first_attempt) and (len(tracker.remaining) == 1):
-                        remote_id = next(iter(tracker.remaining))
-                        await self._node.transport.unicast(tx_deadline, self._priority, remote_id, data)
-                    else:
-                        await writer(tx_deadline, self._priority, data)
-                except (SendError, OSError):
-                    tracker.compromised = True
-
-                if tracker.acknowledged and not tracker.remaining:
-                    _logger.debug("Reliable publish ACKed tag=%d topic='%s'", tag, self._topic.name)
-                    return
-
-                wait_until_ns = deadline.ns if last_attempt else ack_deadline_ns
-                wait_timeout = max(0.0, (wait_until_ns - Instant.now().ns) * 1e-9)
-                if wait_timeout > 0:
-                    try:
-                        await asyncio.wait_for(tracker.ack_event.wait(), timeout=wait_timeout)
-                    except asyncio.TimeoutError:
-                        pass
-
-                if tracker.acknowledged and not tracker.remaining:
-                    _logger.debug("Reliable publish ACKed tag=%d topic='%s'", tag, self._topic.name)
-                    return
-                if last_attempt:
-                    break
-                first_attempt = False
-                tracker.ack_timeout *= 2
+        return tracker
+
+    def _release_reliable_publish_tracker(self, tag: int, tracker: PublishTracker) -> None:
+        self._topic.publish_futures.pop(tag, None)
+        self._node.publish_tracker_release(self._topic, tracker)
+
+    async def _send_reliable_publish(
+        self,
+        deadline: Instant,
+        tag: int,
+        payload: bytes,
+        tracker: PublishTracker,
+        *,
+        first_attempt: bool,
+    ) -> None:
+        data = self._serialize_message(tag, payload, reliable=True)
+        if (not first_attempt) and (len(tracker.remaining) == 1):
+            remote_id = next(iter(tracker.remaining))
+            await self._node.transport.unicast(deadline, self._priority, remote_id, data)
+        else:
+            writer = self._topic.ensure_writer()
+            await writer(deadline, self._priority, data)
 
+    async def _reliable_publish_start(
+        self,
+        deadline: Instant,
+        tag: int,
+        payload: bytes,
+        tracker: PublishTracker,
+    ) -> tuple[int, bool]:
+        initial_window = self._reliable_publish_window(deadline.ns, tracker.ack_timeout)
+        if initial_window is None:
             raise DeliveryError("Reliable publish not acknowledged before deadline")
+        ack_deadline_ns, _ = initial_window
+        tracker.ack_event.clear()
+        try:
+            await self._send_reliable_publish(Instant(ns=ack_deadline_ns), tag, payload, tracker, first_attempt=True)
+        except SendError:
+            tracker.compromised = True
+            raise
+        except OSError as ex:
+            tracker.compromised = True
+            raise SendError("Reliable publish initial send failed") from ex
+        return initial_window
+
+    async def _reliable_publish_continue(
+        self,
+        deadline: Instant,
+        tag: int,
+        payload: bytes,
+        tracker: PublishTracker,
+        initial_window: tuple[int, bool],
+    ) -> None:
+        ack_deadline_ns, last_attempt = initial_window
+        while True:
+            if tracker.acknowledged and not tracker.remaining:
+                _logger.debug("Reliable publish ACKed tag=%d topic='%s'", tag, self._topic.name)
+                return
+
+            wait_until_ns = deadline.ns if last_attempt else ack_deadline_ns
+            wait_timeout = max(0.0, (wait_until_ns - Instant.now().ns) * 1e-9)
+            if wait_timeout > 0:
+                try:
+                    await asyncio.wait_for(tracker.ack_event.wait(), timeout=wait_timeout)
+                except asyncio.TimeoutError:
+                    pass
+
+            if tracker.acknowledged and not tracker.remaining:
+                _logger.debug("Reliable publish ACKed tag=%d topic='%s'", tag, self._topic.name)
+                return
+            if last_attempt:
+                break
+            tracker.ack_timeout *= 2
+            next_window = self._reliable_publish_window(deadline.ns, tracker.ack_timeout)
+            if next_window is None:
+                break
+            ack_deadline_ns, last_attempt = next_window
+            tracker.ack_event.clear()
+            try:
+                await self._send_reliable_publish(
+                    Instant(ns=ack_deadline_ns), tag, payload, tracker, first_attempt=False
+                )
+            except (SendError, OSError):
+                tracker.compromised = True
+
+        raise DeliveryError("Reliable publish not acknowledged before deadline")
+
+    async def _reliable_publish(self, deadline: Instant, tag: int, payload: bytes) -> None:
+        tracker = self._prepare_reliable_publish_tracker(tag, deadline.ns, payload)
+        try:
+            initial_window = await self._reliable_publish_start(deadline, tag, payload, tracker)
+            await self._reliable_publish_continue(deadline, tag, payload, tracker, initial_window)
         finally:
-            self._topic.publish_futures.pop(tag, None)
-            self._node.publish_tracker_release(self._topic, tracker)
+            self._release_reliable_publish_tracker(tag, tracker)
 
     def close(self) -> None:
         if self.closed:

src/pycyphal/_subscriber.py

@@ -324,25 +324,25 @@ async def __call__(
 
         ack_timeout = ACK_BASELINE_DEFAULT_TIMEOUT * (1 << int(self._priority))
         try:
+            initial_window = _ack_window(deadline.ns, ack_timeout)
+            if initial_window is None:
+                raise DeliveryError("Reliable response not acknowledged before deadline")
+
+            ack_deadline_ns, last_attempt = initial_window
+            tracker.ack_event.clear()
+            try:
+                await self._node.transport.unicast(Instant(ns=ack_deadline_ns), self._priority, self._remote_id, data)
+            except SendError:
+                raise
+            except OSError as ex:
+                raise SendError("Reliable response initial send failed") from ex
+
             while True:
                 if tracker.done:
                     if tracker.nacked:
                         raise NackError("Response NACK'd by remote")
                     return
 
-                now_ns = Instant.now().ns
-                if now_ns >= deadline.ns:
-                    break
-                ack_deadline_ns = min(deadline.ns, now_ns + round(ack_timeout * 1e9))
-                last_attempt = _ack_is_last_attempt(ack_deadline_ns, ack_timeout, deadline.ns)
-                tracker.ack_event.clear()
-                try:
-                    await self._node.transport.unicast(
-                        Instant(ns=ack_deadline_ns), self._priority, self._remote_id, data
-                    )
-                except (SendError, OSError):
-                    pass
-
                 wait_until_ns = deadline.ns if last_attempt else ack_deadline_ns
                 wait_time = max(0.0, (wait_until_ns - Instant.now().ns) * 1e-9)
                 try:
@@ -358,6 +358,17 @@ async def __call__(
                 if last_attempt:
                     break
                 ack_timeout *= 2
+                next_window = _ack_window(deadline.ns, ack_timeout)
+                if next_window is None:
+                    break
+                ack_deadline_ns, last_attempt = next_window
+                tracker.ack_event.clear()
+                try:
+                    await self._node.transport.unicast(
+                        Instant(ns=ack_deadline_ns), self._priority, self._remote_id, data
+                    )
+                except (SendError, OSError):
+                    pass
 
             if not tracker.done:
                 raise DeliveryError("Reliable response not acknowledged before deadline")
@@ -396,3 +407,11 @@ def _ack_is_last_attempt(current_ack_deadline_ns: int, current_ack_timeout: floa
     next_ack_timeout_ns = round(current_ack_timeout * 2 * 1e9)
     remaining_budget_ns = total_deadline_ns - current_ack_deadline_ns
     return remaining_budget_ns < next_ack_timeout_ns
+
+
+def _ack_window(deadline_ns: int, ack_timeout: float) -> tuple[int, bool] | None:
+    now_ns = Instant.now().ns
+    if now_ns >= deadline_ns:
+        return None
+    ack_deadline_ns = min(deadline_ns, now_ns + round(ack_timeout * 1e9))
+    return ack_deadline_ns, _ack_is_last_attempt(ack_deadline_ns, ack_timeout, deadline_ns)

tests/test_parity_coverage.py

@@ -586,7 +586,7 @@ async def test_subscriber_wraparound_drop_and_head_of_line_rearm() -> None:
     node.close()
 
 
-async def test_breadcrumb_reliable_ack_path_survives_transient_send_error() -> None:
+async def test_breadcrumb_reliable_initial_send_failure_raises_send_error() -> None:
     net = MockNetwork()
     tr = MockTransport(node_id=1, network=net)
     node = new_node(tr, home="n1")
@@ -600,7 +600,6 @@ async def test_breadcrumb_reliable_ack_path_survives_transient_send_error() -> N
         initial_priority=pycyphal.Priority.NOMINAL,
     )
 
-    original_unicast = tr.unicast
     call_count = 0
 
     async def flaky_unicast(
@@ -611,26 +610,24 @@ async def flaky_unicast(
     ) -> None:
         nonlocal call_count
         call_count += 1
-        if call_count == 1:
-            raise OSError("synthetic failure")
-        await original_unicast(deadline, priority, remote_id, message)
+        del deadline, priority, remote_id, message
+        raise OSError("synthetic failure")
 
     tr.unicast = flaky_unicast  # type: ignore[assignment]
 
-    task = asyncio.create_task(bc(pycyphal.Instant.now() + 0.2, b"response", reliable=True))
-    await asyncio.sleep(0.02)
-    tag = next(iter(node.respond_futures.values())).tag
+    with pytest.raises(pycyphal.SendError):
+        await bc(pycyphal.Instant.now() + 0.2, b"response", reliable=True)
+
     node.on_unicast_arrival(
         TransportArrival(
             timestamp=pycyphal.Instant.now(),
             priority=pycyphal.Priority.NOMINAL,
             remote_id=42,
-            message=RspAckHeader(tag=tag, seqno=0, topic_hash=topic.hash, message_tag=123).serialize(),
+            message=RspAckHeader(tag=0, seqno=0, topic_hash=topic.hash, message_tag=123).serialize(),
         )
     )
-    await task
 
-    assert call_count >= 1
+    assert call_count == 1
     assert node.respond_futures == {}
     node.close()