rpc_client: internal timeouts

Add internal timeouts when waiting for `RPC_RSP` and `RPC_DATA_ACK`
packets.

Signed-off-by: Jordan Yates <jordan@embeint.com>

Author: JordanYates

scripts/apn_set.py

@@ -78,6 +78,8 @@ def announce_observed(self, live: Live, infuse_id: int, pkt: readings.announce |
                 hdr, rsp = rpc_client.run_standard_cmd(
                     rpc.kv_write.COMMAND_ID, Auth.DEVICE, params, self.response.vla_from_buffer_copy
                 )
+                if hdr is None:
+                    return
                 if hdr.return_code == 0:
                     assert rsp is not None and hasattr(rsp, "rc")
                     if rsp.rc[0] == 0:

scripts/reboot_count_reset.py

@@ -67,6 +67,8 @@ def announce_observed(self, live: Live, infuse_id: int, pkt: readings.announce |
                 hdr, _ = rpc_client.run_standard_cmd(
                     rpc.kv_write.COMMAND_ID, Auth.DEVICE, params, rpc.kv_write.response.from_buffer_copy
                 )
+                if hdr is None:
+                    return
                 if hdr.return_code == 0:
                     self.updated.append(infuse_id)
 

src/infuse_iot/rpc_client.py

@@ -2,6 +2,7 @@
 
 import ctypes
 import random
+import time
 from collections.abc import Callable
 
 from infuse_iot import rpc
@@ -25,11 +26,15 @@ def __init__(
         rx_cb: Callable[[ClientNotification], None] | None = None,
     ):
         self._request_id = random.randint(0, 2**31 - 1)
+        self._timeout = 10.0
         self._client = client
         self._id = infuse_id
         self._max_payload = max_payload
         self._rx_cb = rx_cb
 
+    def set_timeout(self, timeout: float):
+        self._timeout = timeout
+
     def _finalise_command(
         self, rpc_rsp: PacketReceived, rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure]
     ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
@@ -48,8 +53,9 @@ def _client_recv(self) -> ClientNotification | None:
             self._rx_cb(rsp)
         return rsp
 
-    def _wait_data_ack(self) -> PacketReceived:
-        while True:
+    def _wait_data_ack(self) -> PacketReceived | None:
+        timeout = time.time() + self._timeout
+        while time.time() < timeout:
             rsp = self._client_recv()
             if rsp is None:
                 continue
@@ -66,10 +72,12 @@ def _wait_data_ack(self) -> PacketReceived:
             if data_ack.request_id != self._request_id:
                 continue
             return rsp.epacket
+        return None
 
-    def _wait_rpc_rsp(self) -> PacketReceived:
+    def _wait_rpc_rsp(self) -> PacketReceived | None:
+        timeout = time.time() + self._timeout
         # Wait for responses
-        while True:
+        while time.time() < timeout:
             rsp = self._client_recv()
             if rsp is None:
                 continue
@@ -83,6 +91,7 @@ def _wait_rpc_rsp(self) -> PacketReceived:
             if rsp_header.request_id != self._request_id:
                 continue
             return rsp.epacket
+        return None
 
     def _run_data_send_core(
         self,
@@ -94,7 +103,7 @@ def _run_data_send_core(
         packet_idx: bool,
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         ack_period = 2
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
@@ -112,6 +121,8 @@ def _run_data_send_core(
 
         # Wait for initial ACK
         recv = self._wait_data_ack()
+        if recv is None:
+            return None, None
         if recv.ptype == InfuseType.RPC_RSP:
             return self._finalise_command(recv, rsp_decoder)
 
@@ -133,6 +144,8 @@ def _run_data_send_core(
             # Wait for ACKs at the period
             if ack_cnt == ack_period:
                 recv = self._wait_data_ack()
+                if recv is None:
+                    return None, None
                 if recv.ptype == InfuseType.RPC_RSP:
                     return self._finalise_command(recv, rsp_decoder)
                 ack_cnt = 0
@@ -142,6 +155,9 @@ def _run_data_send_core(
                 progress_cb(chunk_id + 1 if packet_idx else offset)
 
         recv = self._wait_rpc_rsp()
+        if recv is None:
+            return None, None
+
         return self._finalise_command(recv, rsp_decoder)
 
     def run_data_send_cmd(
@@ -152,7 +168,7 @@ def run_data_send_cmd(
         data: bytes,
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         # Maxmimum payload size of interface
         size = self._max_payload - ctypes.sizeof(rpc.DataHeader)
         # Round payload down to multiple of 4 bytes
@@ -170,7 +186,7 @@ def run_data_send_cmd_chunked(
         data: list[bytes],
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         return self._run_data_send_core(cmd_id, auth, params, data, len(data), True, progress_cb, rsp_decoder)
 
     def run_data_recv_cmd(
@@ -225,7 +241,7 @@ def run_data_recv_cmd(
 
     def run_standard_cmd(
         self, cmd_id: int, auth: Auth, params: bytes, rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure]
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
 
@@ -239,4 +255,6 @@ def run_standard_cmd(
         req = GatewayRequestEpacketSend(pkt)
         self._client.send(req)
         recv = self._wait_rpc_rsp()
+        if recv is None:
+            return None, None
         return self._finalise_command(recv, rsp_decoder)

src/infuse_iot/tools/ota_upgrade.py

@@ -153,8 +153,9 @@ def gateway_diff_load(self):
                 None,
                 file_write_basic.response.from_buffer_copy,
             )
-            if hdr.return_code != 0:
-                sys.exit(f"Failed to save diff file to gateway (({errno.strerror(-hdr.return_code)}))")
+            return_code = hdr.return_code if hdr else -1
+            if return_code != 0:
+                sys.exit(f"Failed to save diff file to gateway (({errno.strerror(-return_code)}))")
             print(f"'{self._single_diff}' written to gateway")
 
     def run_file_upload(self, live: Live, mtu: int, source: HopReceived):
@@ -172,7 +173,9 @@ def run_file_upload(self, live: Live, mtu: int, source: HopReceived):
             file_write_basic.response.from_buffer_copy,
         )
 
-        if hdr.return_code == 0:
+        if hdr is None:
+            self._failed += 1
+        elif hdr.return_code == 0:
             self._pending[source.infuse_id] = time.time() + 60
 
     def run_file_copy(self, live: Live, mtu: int, source: HopReceived):
@@ -195,7 +198,9 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived):
             bytes(params),
             bt_file_copy_basic.response.from_buffer_copy,
         )
-        if hdr.return_code == 0:
+        if hdr is None:
+            self._failed += 1
+        elif hdr.return_code == 0:
             self._pending[source.infuse_id] = time.time() + 60
 
     def run(self):

src/infuse_iot/tools/rpc.py

@@ -126,8 +126,9 @@ def run(self):
                         params,
                         decode_fn,
                     )
+                return_code = hdr.return_code if hdr else -1
                 # Handle response
-                self._command.handle_response(hdr.return_code, rsp)
+                self._command.handle_response(return_code, rsp)
 
                 if self._args.conn_log:
                     while True: