fix(client-python): retry pipe open() on transient connect-refused

packages/client-python/docs/index.md

@@ -408,6 +408,11 @@ Returned by `await client.pipe(...)`. One streaming upload: **open** -> **write*
 | `__aenter__` | `async def __aenter__(self)`                           | `self`            | Enters context; calls `open()`.                    |
 | `__aexit__`  | `async def __aexit__(self, exc_type, exc_val, exc_tb)` | -                 | Exits context; calls `close()`.                    |
 
+`open()` retries once automatically if it hits a transient "Connect call
+failed" while the pipeline's data listener is still starting up (worst case
+adds ~1.75s); a `PipeException` from `open()` means it kept failing past that
+retry budget.
+
 ---
 
 ## Question

packages/client-python/src/rocketride/mixins/data.py

@@ -60,6 +60,31 @@
 from ..core import DAPClient, PipeException
 from ..types import PIPELINE_RESULT, UPLOAD_RESULT
 
+# A just-registered pipeline's per-pipe data listener can occasionally still be
+# binding when `open()` reaches it - observed under heavy concurrent load (e.g. many
+# pipelines opened at once in CI) as a transient "Connect call failed" on the
+# freshly assigned port. That's indistinguishable from a real "pipeline isn't
+# running" failure to the caller, so give it one short extra retry before
+# surfacing it as an error.
+#
+# The engine already retries this exact connect internally (up to 10 attempts,
+# 150ms apart - see task_engine.py's `_connect_data_client`) before it gives up
+# and reports "Connect call failed" back to us, so a single SDK-level attempt
+# already absorbs ~1.5s of engine-side retrying. Each retry here re-runs that
+# whole engine-side loop from scratch, so keep the attempt count low: at
+# `_PIPE_OPEN_RETRY_ATTEMPTS = 2` (one retry), worst case is two ~1.5s engine
+# cycles plus one short backoff, roughly 3.2s, before `open()` raises.
+_PIPE_OPEN_RETRY_ATTEMPTS = 2
+_PIPE_OPEN_RETRY_BACKOFF_SECONDS = 0.25
+
+
+def _is_transient_pipe_open_error(message: str) -> bool:
+    # Deliberately narrower than "any connection-shaped error": a misconfigured
+    # `remote` node surfaces a plain "Connection refused" for a permanent failure
+    # that this retry can't fix (the engine's inner connect loop doesn't run for
+    # it), so only the exact race-condition signature above is retried.
+    return 'Connect call failed' in message
+
 
 class DataMixin(DAPClient):
     """
@@ -154,12 +179,19 @@ async def open(self) -> 'DataMixin.DataPipe':
             Must be called before writing data. The server assigns a unique
             pipe ID and prepares to receive your data.
 
+            If the only problem is a transient "Connect call failed" while the
+            pipeline's data listener is still starting up, this retries once
+            before giving up - worst case that adds ~1.75s (a short backoff,
+            then the engine's own internal connect retry runs again on the
+            retry). Any other failure raises immediately on the first attempt.
+
             Returns:
                 self: The opened pipe instance for method chaining
 
             Raises:
                 RuntimeError: If the pipe is already opened.
-                PipeException: If the server rejects the open request.
+                PipeException: If the server rejects the open request, or if it
+                    keeps failing with a transient connect error past the retry.
 
             Example:
                 pipe = await client.pipe(token, mimetype="text/plain")
@@ -169,25 +201,34 @@ async def open(self) -> 'DataMixin.DataPipe':
             if self._opened:
                 raise RuntimeError('Pipe already opened')
 
-            request = self._client.build_request(
-                'rrext_process',
-                arguments={
-                    'subcommand': 'open',
-                    'object': self._objinfo,
-                    'mimeType': self._mime_type,
-                    'provider': self._provider,
-                },
-                token=self._token,
-            )
+            response = None
+            for attempt in range(1, _PIPE_OPEN_RETRY_ATTEMPTS + 1):
+                request = self._client.build_request(
+                    'rrext_process',
+                    arguments={
+                        'subcommand': 'open',
+                        'object': self._objinfo,
+                        'mimeType': self._mime_type,
+                        'provider': self._provider,
+                    },
+                    token=self._token,
+                )
 
-            response = await self._client.request(request)
+                response = await self._client.request(request)
+
+                if not self._client.did_fail(response):
+                    break
+
+                message = response.get('message') or ''
+                if attempt < _PIPE_OPEN_RETRY_ATTEMPTS and _is_transient_pipe_open_error(message):
+                    await asyncio.sleep(_PIPE_OPEN_RETRY_BACKOFF_SECONDS * attempt)
+                    continue
 
-            if self._client.did_fail(response):
                 # The server's message stays the message: an application may show
                 # it to an end user. The developer checklist rides along as `hint`
                 # (PipeException.hint), and `code` classifies the failure.
                 response = dict(response)
-                response['message'] = response.get('message') or 'Failed to open a data pipe.'
+                response['message'] = message or 'Failed to open a data pipe.'
                 response['hint'] = (
                     'Common causes:\n'
                     "- Pipeline isn't running (wrong token or task terminated)\n"

packages/client-python/tests/test_pipe_open_retry.py

@@ -0,0 +1,124 @@
+import asyncio
+
+import pytest
+
+from rocketride.core.exceptions import PipeException
+from rocketride.mixins.data import DataMixin, _PIPE_OPEN_RETRY_ATTEMPTS
+
+
+class ScriptedTransport:
+    """Fake transport that answers each `send()` with a scripted response, resolving
+    the request's future the same way a real server reply would via `on_receive`.
+    """
+
+    def __init__(self, results):
+        # results: list of ('ok', body) | ('fail', message) tuples, consumed in send() order.
+        self._results = list(results)
+        self.client = None  # set after construction, once the client exists
+        self.send_count = 0
+
+    def bind(self, **handlers):
+        self.handlers = handlers
+
+    def is_connected(self):
+        return True
+
+    async def send(self, message):
+        self.send_count += 1
+        kind, payload = self._results.pop(0)
+        response = {'type': 'response', 'request_seq': message['seq']}
+        if kind == 'ok':
+            response['success'] = True
+            response['body'] = payload
+        else:
+            response['success'] = False
+            response['message'] = payload
+        await self.client.on_receive(response)
+
+
+def _make_pipe(results):
+    transport = ScriptedTransport(results)
+    client = DataMixin(module='TEST', transport=transport)
+    transport.client = client
+    pipe = DataMixin.DataPipe(client, token='tok', mime_type='text/plain')
+    return pipe, transport
+
+
+def test_open_retries_on_transient_connect_error(monkeypatch):
+    _real_sleep = asyncio.sleep
+    monkeypatch.setattr(asyncio, 'sleep', lambda *_a, **_kw: _real_sleep(0))
+
+    pipe, transport = _make_pipe(
+        [
+            ('fail', "Failed to open a data pipe.\n\nConnect call failed ('127.0.0.1', 40006)"),
+            ('ok', {'pipe_id': 7}),
+        ]
+    )
+
+    async def run_test():
+        await pipe.open()
+        assert transport.send_count == 2
+        assert pipe.pipe_id == 7
+        assert pipe.is_opened
+
+    asyncio.run(run_test())
+
+
+def test_open_gives_up_after_exhausting_retries(monkeypatch):
+    _real_sleep = asyncio.sleep
+    monkeypatch.setattr(asyncio, 'sleep', lambda *_a, **_kw: _real_sleep(0))
+
+    transient_failure = ('fail', "Connect call failed ('127.0.0.1', 40006)")
+    pipe, transport = _make_pipe([transient_failure] * (_PIPE_OPEN_RETRY_ATTEMPTS + 1))
+
+    async def run_test():
+        with pytest.raises(PipeException, match='Connect call failed'):
+            await pipe.open()
+        assert transport.send_count == _PIPE_OPEN_RETRY_ATTEMPTS
+        assert not pipe.is_opened
+
+    asyncio.run(run_test())
+
+
+def test_open_does_not_retry_non_transient_failure():
+    pipe, transport = _make_pipe([('fail', 'No pipeline found for token')])
+
+    async def run_test():
+        with pytest.raises(PipeException, match='No pipeline found'):
+            await pipe.open()
+        assert transport.send_count == 1
+        assert not pipe.is_opened
+
+    asyncio.run(run_test())
+
+
+def test_open_does_not_retry_connection_refused():
+    # "Connection refused" is a distinct, permanent signature (e.g. a
+    # misconfigured `remote` node) that the engine's inner connect-retry loop
+    # never runs for, unlike "Connect call failed" - retrying it would just add
+    # latency to a failure that was never going to succeed.
+    pipe, transport = _make_pipe([('fail', "Connection refused ('127.0.0.1', 40006)")])
+
+    async def run_test():
+        with pytest.raises(PipeException, match='Connection refused'):
+            await pipe.open()
+        assert transport.send_count == 1
+        assert not pipe.is_opened
+
+    asyncio.run(run_test())
+
+
+def test_open_failure_keeps_message_and_carries_hint():
+    pipe, transport = _make_pipe([('fail', 'No pipeline found for token')])
+
+    async def run_test():
+        with pytest.raises(PipeException) as excinfo:
+            await pipe.open()
+        exc = excinfo.value
+        # The server's message is preserved verbatim (fit to show an end user);
+        # the developer checklist rides along separately as `hint`.
+        assert str(exc) == 'No pipeline found for token'
+        assert exc.hint is not None
+        assert 'Common causes' in exc.hint
+
+    asyncio.run(run_test())