From 2651781066b496d4d066f79666de233c864ae3ed Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Wed, 5 Aug 2026 14:18:32 -0700 Subject: [PATCH] python: retry connection resets on idempotent GET requests A connection reset mid-poll (e.g. status polling during compilation) was raised immediately as FelderaCommunicationError with no retry, turning a transient network blip into a hard client failure. GET is idempotent, so retry ConnectionError there; leave POST/PUT/PATCH/DELETE unretried since a lost response may hide an already-applied write. Signed-off-by: Ben Pfaff --- python/feldera/rest/_httprequests.py | 21 ++++++++++++--- python/tests/unit/test_httprequests_retry.py | 28 +++++++++++++++++--- 2 files changed, 42 insertions(+), 7 deletions(-) diff --git a/python/feldera/rest/_httprequests.py b/python/feldera/rest/_httprequests.py index 8c9353404df..077c055558f 100644 --- a/python/feldera/rest/_httprequests.py +++ b/python/feldera/rest/_httprequests.py @@ -111,10 +111,18 @@ def _check_cluster_health(self) -> bool: logging.error("Health check failed: %s", e) return False - def _is_retryable(self, exc: BaseException) -> bool: - """Define which exceptions are worth retrying.""" + def _is_retryable(self, exc: BaseException, idempotent: bool) -> bool: + """Define which exceptions are worth retrying. + + `idempotent` gates `ConnectionError` (e.g. a connection reset mid-poll): + safe to retry for GET, where a lost response can't have caused a + server-side side effect, but not for POST/PUT/PATCH/DELETE, where the + original request may already have been applied. + """ if isinstance(exc, requests.exceptions.Timeout): return True + if idempotent and isinstance(exc, requests.exceptions.ConnectionError): + return True if isinstance(exc, FelderaAPIError): return exc.status_code in self.config.retry_config.retryable_status_codes return False @@ -204,6 +212,10 @@ def send_request( Retry policy: - Status codes in `retry_config.retryable_status_codes` (default 408, 429, 502, 503, 504) and connection/read timeouts retry. + - For GET, a `ConnectionError` (e.g. connection reset mid-request) + also retries, since a lost response can't have caused a + server-side side effect. Not retried for POST/PUT/PATCH/DELETE, + where the original request may already have been applied. - 502 probes `/cluster_healthz` to distinguish a spurious gateway error (cluster healthy → retry immediately) from a real outage (cluster unhealthy → wait `unhealthy_backoff` seconds before @@ -213,6 +225,7 @@ def send_request( (capped at `max_backoff`). - All other errors are raised immediately. """ + is_idempotent = http_method is requests.get request_path = self.config.url + "/" + self.config.version + path # Serialize the body once, not per retry. None / bytes / `serialize=False` @@ -235,7 +248,9 @@ def send_request( cfg = self.config.retry_config retryer = Retrying( - retry=retry_if_exception(self._is_retryable), + retry=retry_if_exception( + lambda exc: self._is_retryable(exc, is_idempotent) + ), wait=self._custom_wait, stop=stop_after_attempt(cfg.max_retries + 1), reraise=True, diff --git a/python/tests/unit/test_httprequests_retry.py b/python/tests/unit/test_httprequests_retry.py index c28cadb75a8..0b7008bf42c 100644 --- a/python/tests/unit/test_httprequests_retry.py +++ b/python/tests/unit/test_httprequests_retry.py @@ -226,13 +226,33 @@ def test_timeout_exhausts_raises_timeout_error(self): client.get("/foo") assert m.call_count == 2 - def test_connection_error_wrapped(self): + def test_get_connection_error_then_success(self): + # GET is idempotent — a connection reset mid-poll is safe to retry. client = _make_client() - # ConnectionError isn't retryable — the first raise should propagate - # and be wrapped as FelderaCommunicationError. - with patch_requests("get", [requests.exceptions.ConnectionError("down")]): + with patch_requests( + "get", + [requests.exceptions.ConnectionError("reset"), _make_response(200, b"{}")], + ) as m: + client.get("/foo") + assert m.call_count == 2 + + def test_get_connection_error_exhausts_raises_wrapped(self): + client = _make_client(_fast_retry(max_retries=1)) + with patch_requests( + "get", [requests.exceptions.ConnectionError("down")] * 2 + ) as m: with pytest.raises(FelderaCommunicationError): client.get("/foo") + assert m.call_count == 2 + + def test_post_connection_error_is_not_retried(self): + # POST isn't idempotent — a lost response may hide an applied write, + # so retrying could resubmit it. The first raise must propagate. + client = _make_client() + with patch_requests("post", [requests.exceptions.ConnectionError("down")]) as m: + with pytest.raises(FelderaCommunicationError): + client.post("/foo") + assert m.call_count == 1 def test_no_retries_when_max_retries_zero(self): client = _make_client(_fast_retry(max_retries=0))