Fix BrokenPipeError crash in _ping_pong keep-alive (OPS-3523)

CHANGELOG.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## [3.8.2] - 2026-06-12
+
+- [Fixed] `_ping_pong` background thread no longer dies with unhandled BrokenPipeError (or ConnectionResetError etc.) when the WebSocket becomes invalid; transient send errors now log at WARNING (matching the `_sub_routing_loop` behavior from OPS-3485), set `wss_conn_halted`, and let the keep-alive thread survive so reconnection can restore pings. Added `if not self._conn` guard + regression test modeled on prior websocket tests. (OPS-3523)
+
 ## [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.
@@ -11,6 +15,10 @@
 - [Fixed] `addEnvironment` no longer wipes an existing environment's `wss`/`url`/`headers`/`post_timeout`/`ipv4_only` when re-registering the same environment name without those arguments. Re-registration now MERGES — omitted arguments keep their previous values. Previously a second `addEnvironment("prod", url=..., headers=...)` on the shared (Singleton) client — e.g. a library configuring its own environment — silently reset `wss` to `None`, which crashed the WSS reconnect loop in `_new_conn` with `TypeError: argument of type 'NoneType' is not a container` and produced endless "Failed connecting to None" errors. This was the root cause of the production incident. (OPS-3496)
 - [Fixed] `_new_conn` now guards against an unset/unregistered environment or a missing `wss`, logging a clear ERROR and returning `False` instead of raising `AttributeError`/`TypeError` and killing the subscription router thread. Defense-in-depth alongside the `addEnvironment` fix. (OPS-3496)
 
+## [3.8.1] - 2026-05-28
+
+- [Fixed] `_ping_pong` background thread no longer dies with unhandled BrokenPipeError (or ConnectionResetError etc.) when the WebSocket becomes invalid; transient send errors now log at WARNING (matching the `_sub_routing_loop` behavior from OPS-3485), set `wss_conn_halted`, and let the keep-alive thread survive so reconnection can restore pings. Added `if not self._conn` guard + regression test modeled on prior websocket tests. (OPS-3523)
+
 ## [3.7.2] - 2026-05-28
 
 - [Changed] Repository-wide `ruff format` pass — formatting only, no behavior change.

pygqlc/GraphQLClient.py

@@ -775,16 +775,26 @@ def _ping_pong(self):
             current_time = time.time()
             if (current_time - self.pingTimer) > self.pingIntervalTime:
                 self.pingTimer = current_time
+                if not self._conn:
+                    if not self.closing:
+                        self.wss_conn_halted = True
+                    continue
                 try:
                     self._conn.send(PING_JSON)
                     ping_count += 1
                     # 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 while sending ping",
+                            )
+                        else:
+                            log(
+                                LogLevel.ERROR,
+                                "error trying to send ping, WSS Pipe is broken",
+                            )
                         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,49 @@ 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_transient_error_logged_as_warning():
+    """OPS-3523: BrokenPipeError (transient) from _conn.send(PING_JSON) in _ping_pong
+    must log at WARNING (not ERROR), set wss_conn_halted, and must not kill the ping thread.
+    Mirrors the recv handling and regression tests added for OPS-3485."""
+    Singleton._instances.pop(GraphQLClient, None)
+    gql = GraphQLClient()
+    gql.addEnvironment("ping-test", url="http://ex", wss="ws://ex", default=True)
+    conn = MagicMock()
+    conn.settimeout.return_value = None
+    conn.close.return_value = None
+    gql._conn = conn
+    gql.pingIntervalTime = 0.0
+    gql._conn.send.side_effect = BrokenPipeError(32, "Broken pipe")
+
+    call_times = [0.0, 100.0]
+    with (
+        patch("time.sleep"),
+        patch("time.time", side_effect=call_times),
+        _capture_logs() as records,
+    ):
+        thread = threading.Thread(target=gql._ping_pong, daemon=True)
+        thread.start()
+        for _ in range(1_000_000):
+            if gql._conn.send.call_count > 0:
+                break
+        assert gql._conn.send.call_count > 0, "ping send was never attempted"
+
+        assert gql.wss_conn_halted is True
+
+        gql.closing = True
+        thread.join(timeout=2.0)
+        assert not thread.is_alive(), (
+            "ping_pong thread did not exit cleanly after error"
+        )
+
+    assert any(
+        level == LogLevel.WARNING and "WSS connection reset or closed by peer" in msg
+        for level, msg in records
+    )
+    assert not any(
+        level == LogLevel.ERROR and "error trying to send ping" in msg
+        for level, msg in records
+    )
+    Singleton._instances.pop(GraphQLClient, None)