fix(api-server): ignore duplicate task-api responses in RmfService (#1089)

RMF's request/response channel is a pub/sub pseudo-service, so more than one node can answer the same request_id (the RmfService docstring already notes clients must drop duplicated response ids). _handle_response dropped only *unknown* ids; for a known id it called fut.set_result() unconditionally.

A second response for an already-resolved future raises asyncio.InvalidStateError inside the subscription callback, which runs in the rclpy.spin() thread (ros.py) and terminates it. After that the api-server node stops processing every subscription, so all task-api calls time out with HTTP 500 "rmf service timed out" even though the request was delivered and executed.

Drop duplicate/late responses (fut.done()) so the spin thread survives, and add a regression test.

Signed-off-by: levonti <levonti@ileo.cloud>
Co-authored-by: levonti <levonti@ileo.cloud>

Author: levonti

packages/api-server/api_server/rmf_io/rmf_service.py

@@ -94,6 +94,18 @@ def _handle_response(self, msg: ApiResponse):
                 f"Received response for unknown request id: {msg.request_id}"
             )
             return
+        if fut.done():
+            # RMF's request/response is a pub/sub pseudo-service, so more than
+            # one node may answer the same request_id (see the class docstring:
+            # clients must drop duplicated response ids). Calling set_result on
+            # an already-resolved future raises asyncio.InvalidStateError inside
+            # this subscription callback, which runs in the rclpy.spin() thread
+            # (see ros.py) and would terminate it -- after which the node stops
+            # processing every subscription and all later calls time out.
+            logging.warning(
+                f"Ignoring duplicate response for request id: {msg.request_id}"
+            )
+            return
         fut.set_result(msg.json_msg)
 
 

packages/api-server/api_server/rmf_io/test_rmf_service.py

@@ -122,3 +122,26 @@ async def run():
             self.assertListEqual(["hello", "world"], list(results))
 
         self.loop.run_until_complete(run())
+
+    def test_duplicate_response_is_ignored(self):
+        # Regression: RMF's request/response is a pub/sub pseudo-service, so
+        # more than one node may answer the same request_id. A duplicate
+        # ApiResponse for an already-resolved future used to raise
+        # asyncio.InvalidStateError inside the subscription callback, which runs
+        # in the rclpy spin thread and would terminate it -- silently breaking
+        # every later task-api call. The handler must drop the duplicate.
+        req_id = self._create_node_id()
+        fut: asyncio.Future = self.loop.create_future()
+        self.rmf_service._requests[req_id] = fut
+        try:
+            self.rmf_service._handle_response(
+                ApiResponse(request_id=req_id, json_msg="first")
+            )
+            # second, duplicate delivery of the same request_id must not raise
+            self.rmf_service._handle_response(
+                ApiResponse(request_id=req_id, json_msg="second")
+            )
+        finally:
+            self.rmf_service._requests.pop(req_id, None)
+        self.assertTrue(fut.done())
+        self.assertEqual("first", fut.result())