fix(api): bail early on module gcode parse errors (#20194)

We were waiting for multiple acks from certain commands, but if those
commands fail at the module's gcode parser stage (i.e. because the
module isn't updated and doesn't handle that gcode) then we'll never get
a second ack and we'll just wait a really long time. This is ultimately
fine in that it doesn't fail the protocol, but it makes anything that
uses modules that have this problem take a really really long time.

## testing
- [x] on a robot with a module that doesn't handle M411, check that the
module polls don't take 60 seconds to complete

Closes RABR-843

Author: sfoster1

api/src/opentrons/drivers/asyncio/communication/serial_connection.py

@@ -12,6 +12,8 @@
     ErrorResponse,
     BaseErrorCode,
     DefaultErrorCodes,
+    UnhandledGcode,
+    GCodeCacheFull,
 )
 from .async_serial import AsyncSerial
 
@@ -555,6 +557,23 @@ async def _consume_responses(
                 # A read timeout, end
                 yield "empty-unknown", data
 
+    def _raise_on_parser_error(self, data: str, response: bytes) -> None:
+        """Raise an exception if this response contains an error from the gcode parser on the module.
+
+        This has to be treated specially because multiack commands won't get multiple acks if the command
+        fails at the parse stage. The errors handled here should be kept in sync with the module gcode
+        parse code.
+        """
+        try:
+            str_response = self.process_raw_response(
+                command=data, response=response.replace(self._ack, b"").decode()
+            )
+            self.raise_on_error(response=str_response, request=data)
+        except (UnhandledGcode, GCodeCacheFull):
+            raise
+        except Exception:
+            pass
+
     async def _send_one_retry(self, data: str, acks: int) -> list[str]:
         data_encode = data.encode("utf-8")
         log.debug(f"{self._name}: Write -> {data_encode!r}")
@@ -567,8 +586,10 @@ async def _send_one_retry(self, data: str, acks: int) -> list[str]:
         async for response_type, response in self._consume_responses(acks):
             if response_type == "error":
                 async_errors.append(response)
+                self._raise_on_parser_error(data, response)
             elif response_type == "response":
                 command_acks.append(response)
+                self._raise_on_parser_error(data, response)
             else:
                 break
 

api/tests/opentrons/drivers/asyncio/communication/test_serial_connection.py

@@ -338,7 +338,7 @@ async def test_send_data_multiple_ack_some_errors(
     """It should return all acks."""
     successful_response = "M411"
     data = "M411"
-    error_response = "ERR003:test"
+    error_response = "ERR007:test"
     serial_successful_response = f" {successful_response}  {ack}"
     encoded_successful_response = serial_successful_response.encode()
     serial_error_response = f" {error_response}  {ack}"
@@ -424,3 +424,15 @@ class CustomDefaultErrorCodes(BaseErrorCode):
     assert error.value.command == "G28"
     assert error.value.response == "ERR999:test"
     assert error.value.port == "test_port"
+
+
+async def test_send_data_multiple_raises_unhandled(
+    mock_serial_port: AsyncMock, async_subject: AsyncResponseSerialConnection, ack: str
+) -> None:
+    """It shouldn't wait for both acks before raising an unhandled gcode"""
+    mock_serial_port.read_until.side_effect = [
+        f"ERR003:unhandled gcode {ack}".encode(),
+    ]
+    with pytest.raises(UnhandledGcode):
+        await async_subject._send_data_multiack(data="M411", retries=0, acks=3)
+    mock_serial_port.read_until.assert_called_once()