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

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

A just-registered pipeline's per-pipe data listener can occasionally
still be binding when open() reaches it - under heavy concurrent load
(e.g. many pipelines opened at once) this surfaces as a transient
ECONNREFUSED ("Connect call failed") that is indistinguishable from a
real "pipeline isn't running" failure to the caller.

Retry a few times with a short backoff when the failure message
matches this specific transient pattern; any other failure still
raises immediately on the first attempt, so genuine errors (bad
token, wrong MIME type, terminated pipeline) are not masked or
delayed.

Surfaced by CI flakiness on unrelated PRs: the same
[Errno 111] Connect call failed error hit two different, unrelated
tests (nodes/test/guardrails/test_lane_forward_once.py and
nodes/test/test_dynamic.py) on consecutive runs, each on a different
ephemeral port - consistent with this race rather than a test bug.

* fix(client-python): address review — narrow retry, recalibrate, rebase fix

Nihal's review on #2117:

- Rebase onto develop needed a deliberate resolution: #2127 (merged after
  this branch was cut) rewrote the same open() failure hunk to split the
  server message from a `hint` field and add `code`. Keep that split; only
  the retry loop wraps it now.
- The predicate also matched "Connection refused", which a misconfigured
  `remote` node raises for a permanent failure the engine's inner connect
  retry never runs for — retrying it only added latency. Narrow to the
  actual race-condition signature, "Connect call failed".
- The retry budget undersold itself: the engine already retries this same
  connect internally (10x, 150ms apart, ~1.5s worst case) before it reports
  failure, so each SDK-level attempt re-runs that whole loop. Drop from 3
  attempts to 2 (one retry) and document the ~3.2s worst case instead of
  silently multiplying it further.

Also pins the rebase resolution with tests for the narrowed predicate and
the message/hint split, and updates the co-located SDK doc's retry note.

* fix(client-python): correct the documented pipe-open retry latency

CodeRabbit on #2117: the retry's added delay is the 0.25s SDK backoff plus
the engine's own ~1.5s internal connect-retry cycle, so ~1.75s - not the
~1.5s the docstring and doc claimed. Also reworded the doc's PipeException
description to "past that retry budget" rather than "genuinely could not be
opened" (the earlier module comment already gets this right).

* fix(client-python): guard pipe-open retry predicate against a non-string message

CodeRabbit on #2117: response.get('message') or '' passes through a truthy
non-string value (e.g. a malformed server response with an int message)
unchanged, so _is_transient_pipe_open_error()'s `'Connect call failed' in
message` raises TypeError before PipeException can even be raised. Coerce to
str right after extraction so the retry classification and the exception it
may raise both stay safe.

* fix(client-python): don't let a falsey non-string message fall back to generic

CodeRabbit on #2117: response.get('message') or '' runs before the
just-added isinstance check, so a real-but-falsey message like 0 or False
was already flattened to '' by the time it got there - PipeException then
showed the generic "Failed to open a data pipe." instead of the server's
actual value. Check for None explicitly instead of using `or`.

---------

Co-authored-by: madhumithachandrasekaran31 <madhumitha2110549@ssn.edu.in>

Author: madhumitha-chandrasekaran-1

packages/client-python/docs/index.md

@@ -463,6 +463,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,43 @@ 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')
+                if message is None:
+                    message = ''
+                elif not isinstance(message, str):
+                    # A well-behaved server always sends a string, but don't let a
+                    # malformed one (e.g. a bare int/bool) blow up the `in` check
+                    # below or the PipeException raised past the retry. `or ''`
+                    # here would also turn falsey-but-real values like `0`/`False`
+                    # into an empty message, so check for absence explicitly.
+                    message = str(message)
+                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,152 @@
+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_does_not_retry_non_string_failure_message():
+    # A malformed response with a non-string `message` (e.g. a bare int) must
+    # not blow up the `in` check that classifies transient errors.
+    pipe, transport = _make_pipe([('fail', 404)])
+
+    async def run_test():
+        with pytest.raises(PipeException, match='404'):
+            await pipe.open()
+        assert transport.send_count == 1
+        assert not pipe.is_opened
+
+    asyncio.run(run_test())
+
+
+def test_open_preserves_falsey_non_string_failure_message():
+    # `0`/`False` are real messages, not "no message" - they must survive
+    # normalization instead of being swallowed into the generic fallback.
+    pipe, transport = _make_pipe([('fail', 0)])
+
+    async def run_test():
+        with pytest.raises(PipeException, match='0'):
+            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())