fix(connection): only retry idempotent queries after mid-send transport loss

OWNd/connection/command_session.py

@@ -209,6 +209,10 @@ async def _locked_send(
         # gateway may have executed it even if its acknowledgement was lost.
         # Status requests are idempotent and can be retried safely after a
         # transport reset.
+        is_idempotent_query = is_status_request and (
+            getattr(msg_obj, "_message_type", None)
+            in ("STATUS_REQUEST", "DIMENSION_REQUEST")
+        )
         max_attempts = 2
 
         for attempt in range(1, max_attempts + 1):
@@ -304,7 +308,7 @@ async def _locked_send(
                 # state yet: a retry that succeeds is an internal recycle and
                 # must not flap the consumer (offline, then online again).
                 await self._close_streams()
-                if attempt < max_attempts and (is_status_request or not written):
+                if attempt < max_attempts and (is_idempotent_query or not written):
                     self._logger.debug(
                         "%s Command session connection lost (%s), retrying once...",
                         self._log_id,

tests/test_connection.py

@@ -336,6 +336,55 @@ async def reconnect() -> dict:
     assert retry_writer.written == [b"*#1*0##"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "frame",
+    [
+        "*1*1*11##",
+        "*#4*000*0*0215##",
+        "*#4*000*#0*0215##",
+    ],
+)
+async def test_written_non_query_with_status_request_flag_is_not_replayed_after_lost_ack(
+    frame: str,
+) -> None:
+    session, writer = make_session(OWNCommandSession)
+    assert isinstance(session, OWNCommandSession)
+    assert session._stream_reader is not None
+    session._stream_reader.feed_eof()
+
+    with patch.object(session, "connect", new_callable=AsyncMock) as connect:
+        result = await session.send(frame, is_status_request=True)
+
+    assert result is None
+    assert writer.written == [frame.encode()]
+    connect.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_disconnected_dimension_request_retries_once() -> None:
+    session, original_writer = make_session(OWNCommandSession)
+    assert isinstance(session, OWNCommandSession)
+    assert session._stream_reader is not None
+    session._stream_reader.feed_eof()
+
+    retry_writer = FakeWriter()
+
+    async def reconnect() -> dict[str, object]:
+        session._stream_reader = asyncio.StreamReader()
+        session._stream_reader.feed_eof()
+        session._stream_writer = retry_writer  # type: ignore[assignment]
+        return {"Success": True, "Message": None}
+
+    with patch.object(session, "connect", side_effect=reconnect) as connect:
+        result = await session.send("*#4*000*0##", is_status_request=True)
+
+    assert result is None
+    connect.assert_awaited_once()
+    assert original_writer.written == [b"*#4*000*0##"]
+    assert retry_writer.written == [b"*#4*000*0##"]
+
+
 @pytest.mark.asyncio
 async def test_cancelled_command_closes_stream() -> None:
     session, writer = make_session(OWNCommandSession)