Fix _ping_pong WS send for ConnectionResetError (OPS-4640)

CHANGELOG.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## [3.8.2] - 2026-06-13
+
+- [Fixed] `_ping_pong` now catches transient websocket send errors (ConnectionResetError, BrokenPipeError, etc.) at WARNING level and sets `wss_conn_halted` to trigger reconnection, matching the behavior already present in `_sub_routing_loop` for recv. Non-transient errors continue to log at ERROR. (OPS-4640)
+
 ## [3.8.1] - 2026-06-11
 
 - [Fixed] Stale `httpx.AsyncClient` instances are now best-effort `aclose()`d instead of being dropped to the garbage collector. Previously three paths leaked the client's transports: `_get_async_client` when it detected a closed event loop, `async_execute`'s "Event loop is closed" retry, and `_close()`/`__del__`. Orphaned `_SelectorTransport.__del__` socket finalizers then ran during cyclic-GC sweeps on arbitrary threads, contributing to false TMPRL1101 deadlock detections in Temporal workers (see valiot/python-tooling#151). A new private `_drop_async_client()` helper awaits `aclose()` (swallowing failures when the original loop is gone) and is used by `_get_async_client`, the `async_execute` retry, and `async_cleanup`; `_close()` now schedules `aclose()` on the running loop via `call_soon_threadsafe` when one exists, falling back to GC only when no loop is available.

pygqlc/GraphQLClient.py

@@ -781,10 +781,13 @@ def _ping_pong(self):
                     # No need to log normal ping operations
                 except Exception as e:
                     if not self.closing:
-                        log(
-                            LogLevel.ERROR,
-                            "error trying to send ping, WSS Pipe is broken",
-                        )
+                        if isinstance(e, TRANSIENT_WS_ERRORS):
+                            log(
+                                LogLevel.WARNING,
+                                "WSS connection reset or closed by peer",
+                            )
+                        else:
+                            log(LogLevel.ERROR, "Some error trying to send ping")
                         self.wss_conn_halted = True
 
     def _registerSub(self, _id=None):

pygqlc/__version__.py

@@ -1 +1 @@
-__version__ = "3.8.1"
+__version__ = "3.8.2"

tests/pygqlc/gql_client/test_subscribe.py

@@ -299,3 +299,75 @@ def test_new_conn_returns_false_when_no_environment():
     assert result is False
     assert any(level == LogLevel.ERROR for level, _ in records)
     Singleton._instances.pop(GraphQLClient, None)
+
+
+def test_ping_pong_connection_reset_logged_as_warning(routing_client):
+    """OPS-4640: ConnectionResetError from _ping_pong send is transient — log at WARNING
+    (not ERROR+traceback) and set wss_conn_halted so router will reconnect."""
+    gql = routing_client
+    gql.closing = False
+    gql.wss_conn_halted = False
+    gql.pingIntervalTime = -1
+
+    # Time sequence so first check after init yields a large delta > interval:
+    # init pingTimer gets 0; the current_time read gets 100 → delta 100 > -1.
+    time_values = iter([0, 100])
+
+    def _send_then_raise(_data):
+        raise ConnectionResetError(104, "Connection reset by peer")
+
+    gql._conn.send.side_effect = _send_then_raise
+
+    with (
+        patch("pygqlc.GraphQLClient.time.sleep", return_value=None),
+        patch("pygqlc.GraphQLClient.time.time", side_effect=lambda: next(time_values)),
+        _capture_logs() as records,
+    ):
+        thread = threading.Thread(target=gql._ping_pong, daemon=True)
+        thread.start()
+        # Let the thread attempt one send (which raises) and set halted; then unblock exit.
+        time.sleep(0.01)
+        gql.closing = True
+        thread.join(1.0)
+        assert not thread.is_alive(), "ping loop did not terminate"
+
+    assert gql.wss_conn_halted is True
+    assert any(
+        level == LogLevel.WARNING and "reset or closed by peer" in msg
+        for level, msg in records
+    )
+    assert not any(level == LogLevel.ERROR for level, msg in records)
+
+
+def test_ping_pong_unexpected_error_logged_as_error(routing_client):
+    """A non-transient error (not in TRANSIENT_WS_ERRORS) during ping send must surface
+    at ERROR level and halt for reconnection."""
+    gql = routing_client
+    gql.closing = False
+    gql.wss_conn_halted = False
+    gql.pingIntervalTime = -1
+
+    time_values = iter([0, 100])
+
+    def _send_then_raise(_data):
+        raise ValueError("unexpected ping send failure")
+
+    gql._conn.send.side_effect = _send_then_raise
+
+    with (
+        patch("pygqlc.GraphQLClient.time.sleep", return_value=None),
+        patch("pygqlc.GraphQLClient.time.time", side_effect=lambda: next(time_values)),
+        _capture_logs() as records,
+    ):
+        thread = threading.Thread(target=gql._ping_pong, daemon=True)
+        thread.start()
+        time.sleep(0.01)
+        gql.closing = True
+        thread.join(1.0)
+        assert not thread.is_alive(), "ping loop did not terminate"
+
+    assert gql.wss_conn_halted is True
+    assert any(
+        level == LogLevel.ERROR and "Some error trying to send ping" in msg
+        for level, msg in records
+    )