fix: retry transient connection failures for API queries

src/hardcover_sync/api.py

@@ -8,10 +8,13 @@
 import json
 import logging
 import math
+import re
+import socket
 import time
 from datetime import date
+from http.client import IncompleteRead
 from typing import Any
-from urllib.error import HTTPError
+from urllib.error import HTTPError, URLError
 from urllib.request import HTTPRedirectHandler, Request, build_opener
 
 from . import queries  # noqa: E402
@@ -32,7 +35,7 @@
 API_URL = "https://api.hardcover.app/v1/graphql"
 DEFAULT_TIMEOUT = 30  # seconds
 _MAX_ERROR_MESSAGE_LENGTH = 500
-_MAX_RATE_LIMIT_RETRIES = 2
+_MAX_RETRIES = 2
 _MAX_RETRY_DELAY_SECONDS = 10.0
 
 logger = logging.getLogger(__name__)
@@ -142,8 +145,16 @@ def _extract_data(cls, payload: object, status: int | None = None) -> dict[str,
         return data
 
     @staticmethod
-    def _retry_delay(error: HTTPError, retry_number: int) -> float:
-        retry_after = error.headers.get("Retry-After") if error.headers else None
+    def _is_transient_connection_error(error: Exception) -> bool:
+        # urllib wraps connection errors, but response/read errors can be unwrapped.
+        reason = error.reason if isinstance(error, URLError) else error
+        if isinstance(reason, (TimeoutError, ConnectionError, IncompleteRead)):
+            return True
+        return isinstance(reason, socket.gaierror) and reason.errno == socket.EAI_AGAIN
+
+    @staticmethod
+    def _retry_delay(error: HTTPError | None, retry_number: int) -> float:
+        retry_after = error.headers.get("Retry-After") if error and error.headers else None
         if retry_after is not None:
             try:
                 delay = float(retry_after)
@@ -163,8 +174,11 @@ def execute(
             payload["variables"] = variables
 
         request_body = json.dumps(payload).encode("utf-8")
+        # Only recognizable queries can be replayed after an ambiguous network failure.
+        # We send no operationName, so the server rejects multi-operation documents.
+        read_only = re.match(r"^\s*(?:query\b|\{)", query) is not None
         response_body = b""
-        for retry_number in range(_MAX_RATE_LIMIT_RETRIES + 1):
+        for retry_number in range(_MAX_RETRIES + 1):
             # Request objects may be mutated by urllib handlers, so rebuild on retry.
             request = Request(  # noqa: S310 - API_URL is a fixed HTTPS endpoint
                 API_URL,
@@ -185,7 +199,7 @@ def execute(
             except HTTPError as error:
                 if error.code < 400:
                     raise error from None
-                if error.code == 429 and retry_number < _MAX_RATE_LIMIT_RETRIES:
+                if error.code == 429 and retry_number < _MAX_RETRIES:
                     delay = self._retry_delay(error, retry_number)
                     error.close()
                     time.sleep(delay)
@@ -198,6 +212,22 @@ def execute(
                 except (json.JSONDecodeError, UnicodeDecodeError):
                     raise error from None
                 return self._extract_data(error_payload, status=error.code)
+            except (OSError, IncompleteRead) as error:
+                if (
+                    not read_only
+                    or not self._is_transient_connection_error(error)
+                    or retry_number >= _MAX_RETRIES
+                ):
+                    raise
+                delay = self._retry_delay(None, retry_number)
+                logger.warning(
+                    "Transient %s during read-only API query; retry %s/%s in %.1fs",
+                    type(error).__name__,
+                    retry_number + 1,
+                    _MAX_RETRIES,
+                    delay,
+                )
+                time.sleep(delay)
 
         try:
             response_payload = json.loads(response_body)

test/test_http_client.py

@@ -2,14 +2,19 @@
 
 import io
 import json
-from urllib.error import HTTPError
+import socket
+import ssl
+from http.client import IncompleteRead, RemoteDisconnected
+from urllib.error import HTTPError, URLError
 from unittest.mock import call, patch
 
 import pytest
 
 from hardcover_sync.api import (
     API_URL,
     GraphQLResponseError,
+    HardcoverAPI,
+    HardcoverAPIError,
     HardcoverHTTPClient,
     _NoRedirectHandler,
 )
@@ -33,7 +38,8 @@ def http_error(status, body, headers=None):
 
 
 class TestHardcoverHTTPClient:
-    def test_execute_posts_graphql_request(self):
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_posts_graphql_request(self, sleep):
         client, opener = make_client()
         opener.open.return_value = json_response({"data": {"me": {"id": 1}}})
 
@@ -52,6 +58,8 @@ def test_execute_posts_graphql_request(self):
             "variables": {"limit": 10},
         }
         assert opener.open.call_args.kwargs == {"timeout": 12}
+        assert opener.open.call_count == 1
+        sleep.assert_not_called()
 
     def test_execute_omits_variables_when_not_provided(self):
         client, opener = make_client()
@@ -71,15 +79,158 @@ def test_execute_raises_graphql_error(self):
         with pytest.raises(GraphQLResponseError, match="first error; second error"):
             client.execute("query Broken { broken }")
 
+    @pytest.mark.parametrize(
+        "error",
+        [
+            URLError(TimeoutError("timed out")),
+            TimeoutError("timed out"),
+            URLError(ConnectionRefusedError("connection refused")),
+            ConnectionResetError("connection reset"),
+            RemoteDisconnected("remote closed connection"),
+            IncompleteRead(b"partial response", 100),
+            URLError(socket.gaierror(socket.EAI_AGAIN, "temporary DNS failure")),
+            socket.gaierror(socket.EAI_AGAIN, "temporary DNS failure"),
+        ],
+    )
+    @pytest.mark.parametrize("stage", ["open", "read"])
     @patch("hardcover_sync.api.time.sleep")
-    def test_execute_retries_http_rate_limit(self, sleep):
+    def test_execute_retries_transient_query_failure(self, sleep, error, stage, caplog):
+        client, opener = make_client()
+        failed_response = io.BytesIO()
+        opener.open.side_effect = [
+            error if stage == "open" else failed_response,
+            json_response({"data": {"me": {"id": 1}}}),
+        ]
+
+        with patch.object(failed_response, "read", side_effect=error):
+            result = client.execute("\nquery Me { me { id } }", {"limit": 10})
+
+        assert result == {"me": {"id": 1}}
+        assert opener.open.call_count == 2
+        assert sleep.call_args_list == [call(1.0)]
+        if stage == "read":
+            assert failed_response.closed
+        requests = [args.args[0] for args in opener.open.call_args_list]
+        assert requests[0] is not requests[1]
+        assert requests[0].data == requests[1].data
+        assert requests[0].headers == requests[1].headers
+        assert all(args.kwargs == {"timeout": 12} for args in opener.open.call_args_list)
+        assert "retry 1/2" in caplog.text
+        assert "test-token" not in caplog.text
+
+    @pytest.mark.parametrize("query", ["query{ me { id } }", "\n{ me { id } }"])
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_retries_anonymous_queries(self, sleep, query):
+        client, opener = make_client()
+        opener.open.side_effect = [TimeoutError(), json_response({"data": {"me": None}})]
+
+        assert client.execute(query) == {"me": None}
+        assert opener.open.call_count == 2
+        sleep.assert_called_once_with(1.0)
+
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_preserves_final_connection_error_after_retries(self, sleep):
+        client, opener = make_client()
+        errors = [URLError(TimeoutError(f"timeout {attempt}")) for attempt in range(3)]
+        opener.open.side_effect = errors
+
+        with pytest.raises(URLError) as exc_info:
+            client.execute("query Me { me { id } }")
+
+        assert exc_info.value is errors[-1]
+        assert opener.open.call_count == 3
+        assert sleep.call_args_list == [call(1.0), call(2.0)]
+
+    @pytest.mark.parametrize(
+        "query",
+        [
+            "\nmutation Update { update_user_book(id: 1, object: {}) { id } }",
+            "# comment\nmutation { update_user_book(id: 1, object: {}) { id } }",
+            "subscription { user_books { id } }",
+            "queryNotAnOperation { me { id } }",
+            "# unrecognized query prefix\nquery Me { me { id } }",
+        ],
+    )
+    @pytest.mark.parametrize("stage", ["open", "read"])
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_does_not_replay_mutations_or_unknown_operations(self, sleep, query, stage):
+        client, opener = make_client()
+        error = URLError(TimeoutError("timed out"))
+        failed_response = io.BytesIO()
+        opener.open.side_effect = [error if stage == "open" else failed_response]
+
+        with patch.object(failed_response, "read", side_effect=error):
+            with pytest.raises(URLError) as exc_info:
+                client.execute(query)
+
+        assert exc_info.value is error
+        assert opener.open.call_count == 1
+        sleep.assert_not_called()
+        if stage == "read":
+            assert failed_response.closed
+
+    @pytest.mark.parametrize(
+        "error",
+        [
+            ssl.SSLCertVerificationError("invalid certificate"),
+            URLError(ssl.SSLCertVerificationError("invalid certificate")),
+            ssl.SSLError("TLS failure"),
+            socket.gaierror(socket.EAI_NONAME, "unknown host"),
+            URLError(socket.gaierror(socket.EAI_NONAME, "unknown host")),
+            URLError("unknown error"),
+            OSError("not a transient error"),
+        ],
+    )
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_does_not_retry_permanent_connection_errors(self, sleep, error):
+        client, opener = make_client()
+        opener.open.side_effect = error
+
+        with pytest.raises(type(error)) as exc_info:
+            client.execute("query Me { me { id } }")
+
+        assert exc_info.value is error
+        assert opener.open.call_count == 1
+        sleep.assert_not_called()
+
+    @pytest.mark.parametrize("recover", [True, False])
+    @pytest.mark.parametrize("rate_limit_first", [True, False])
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_shares_retry_budget_with_rate_limits(self, sleep, recover, rate_limit_first):
+        client, opener = make_client()
+        rate_limit = http_error(429, b"rate limited")
+        timeout = URLError(TimeoutError("timed out"))
+        errors = [rate_limit, timeout] if rate_limit_first else [timeout, rate_limit]
+        final_error = URLError(TimeoutError("still timed out"))
+        opener.open.side_effect = [
+            *errors,
+            json_response({"data": {"me": None}}) if recover else final_error,
+        ]
+
+        if recover:
+            assert client.execute("query Me { me { id } }") == {"me": None}
+        else:
+            with pytest.raises(URLError) as exc_info:
+                client.execute("query Me { me { id } }")
+            assert exc_info.value is final_error
+
+        assert opener.open.call_count == 3
+        assert sleep.call_args_list == [call(1.0), call(2.0)]
+        assert rate_limit.closed
+
+    @pytest.mark.parametrize(
+        "query",
+        ["query Me { me { id } }", "mutation { update_user_book(id: 1, object: {}) { id } }"],
+    )
+    @patch("hardcover_sync.api.time.sleep")
+    def test_execute_retries_http_rate_limit(self, sleep, query):
         client, opener = make_client()
         opener.open.side_effect = [
             http_error(429, b"rate limited", {"Retry-After": "3"}),
             json_response({"data": {"me": {"id": 1}}}),
         ]
 
-        result = client.execute("query Me { me { id } }")
+        result = client.execute(query)
 
         assert result == {"me": {"id": 1}}
         assert opener.open.call_count == 2
@@ -223,6 +374,38 @@ def test_error_messages_are_truncated(self):
         assert len(str(error)) == 500
 
 
+class TestConnectivityThroughAPI:
+    @patch("hardcover_sync.api.time.sleep")
+    def test_get_me_recovers_from_connection_timeout(self, sleep):
+        with patch("hardcover_sync.api.build_opener") as build_opener:
+            api = HardcoverAPI("test-token")  # noqa: S106
+            build_opener.return_value.open.side_effect = [
+                URLError(TimeoutError("timed out")),
+                json_response({"data": {"me": [{"id": 1, "username": "reader"}]}}),
+            ]
+
+            user = api.get_me()
+
+        assert user.id == 1
+        assert user.username == "reader"
+        assert build_opener.return_value.open.call_count == 2
+        sleep.assert_called_once_with(1.0)
+
+    @patch("hardcover_sync.api.time.sleep")
+    def test_get_me_preserves_final_timeout_as_cause(self, sleep):
+        errors = [URLError(TimeoutError(f"timeout {attempt}")) for attempt in range(3)]
+        with patch("hardcover_sync.api.build_opener") as build_opener:
+            api = HardcoverAPI("test-token")  # noqa: S106
+            build_opener.return_value.open.side_effect = errors
+
+            with pytest.raises(HardcoverAPIError, match="timeout 2") as exc_info:
+                api.get_me()
+
+        assert exc_info.value.__cause__ is errors[-1]
+        assert build_opener.return_value.open.call_count == 3
+        assert sleep.call_args_list == [call(1.0), call(2.0)]
+
+
 class TestNoRedirectHandler:
     def test_redirects_are_refused(self):
         handler = _NoRedirectHandler()