From 498d3be196435eeb00f553649af3818a3eb2945b Mon Sep 17 00:00:00 2001
From: floor-licker <kingofthecoast78@gmail.com>
Date: Tue, 30 Jun 2026 21:11:47 -0400
Subject: [PATCH] fix(stream): flush websocket pongs

---
 Cargo.lock    |  2 +-
 Cargo.toml    |  2 +-
 src/stream.rs | 56 ++++++++++++++++++++++++++++++++++++++++++++-------
 3 files changed, 51 insertions(+), 9 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 0173731..e7500f3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3220,7 +3220,7 @@ dependencies = [
 
 [[package]]
 name = "polyfill-rs"
-version = "0.4.2"
+version = "0.4.3"
 dependencies = [
  "alloy-primitives",
  "alloy-signer",
diff --git a/Cargo.toml b/Cargo.toml
index 3d10e44..b0d7d2e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "polyfill-rs"
-version = "0.4.2"
+version = "0.4.3"
 edition = "2021"
 authors = ["Julius Tranquilli <jtranqs@gmail.com>"]
 description = "The Fastest Polymarket Client On The Market."
diff --git a/src/stream.rs b/src/stream.rs
index e72dc73..68a3f08 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -7,7 +7,7 @@ use crate::errors::{PolyfillError, Result};
 use crate::types::*;
 use crate::ws_hot_path::{WsBookApplyStats, WsBookUpdateProcessor};
 use chrono::Utc;
-use futures::{SinkExt, Stream, StreamExt};
+use futures::{ready, SinkExt, Stream, StreamExt};
 use parking_lot::Mutex;
 use serde_json::Value;
 use std::collections::VecDeque;
@@ -380,6 +380,36 @@ impl WebSocketStream {
     }
 }
 
+fn poll_send_pong(
+    connection: &mut tokio_tungstenite::WebSocketStream<
+        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+    >,
+    cx: &mut Context<'_>,
+    data: Vec<u8>,
+) -> Poll<Result<()>> {
+    ready!(connection.poll_ready_unpin(cx)).map_err(|e| {
+        PolyfillError::stream(
+            format!("Failed to prepare pong: {}", e),
+            crate::errors::StreamErrorKind::MessageCorrupted,
+        )
+    })?;
+    connection
+        .start_send_unpin(tokio_tungstenite::tungstenite::Message::Pong(data))
+        .map_err(|e| {
+            PolyfillError::stream(
+                format!("Failed to send pong: {}", e),
+                crate::errors::StreamErrorKind::MessageCorrupted,
+            )
+        })?;
+    ready!(connection.poll_flush_unpin(cx)).map_err(|e| {
+        PolyfillError::stream(
+            format!("Failed to flush pong: {}", e),
+            crate::errors::StreamErrorKind::MessageCorrupted,
+        )
+    })?;
+    Poll::Ready(Ok(()))
+}
+
 /// WebSocket stream wrapper that applies `book` updates directly into an [`crate::book::OrderBookManager`].
 ///
 /// This bypasses `StreamMessage` decoding (serde/DOM parsing) for the `book` hot path by using
@@ -484,9 +514,15 @@ impl<'a> Stream for WebSocketBookApplier<'a> {
                         self.stream.connection = None;
                         return Poll::Ready(None);
                     },
-                    tokio_tungstenite::tungstenite::Message::Ping(_) => {
-                        // Best-effort: tokio-tungstenite/tungstenite may handle pings internally.
-                        continue;
+                    tokio_tungstenite::tungstenite::Message::Ping(data) => {
+                        match poll_send_pong(connection, cx, data) {
+                            Poll::Ready(Ok(())) => continue,
+                            Poll::Ready(Err(e)) => {
+                                self.stream.stats.errors += 1;
+                                return Poll::Ready(Some(Err(e)));
+                            },
+                            Poll::Pending => return Poll::Pending,
+                        }
                     },
                     tokio_tungstenite::tungstenite::Message::Pong(_) => continue,
                     tokio_tungstenite::tungstenite::Message::Binary(_) => continue,
@@ -548,9 +584,15 @@ impl Stream for WebSocketStream {
                         self.connection = None;
                         return Poll::Ready(None);
                     },
-                    tokio_tungstenite::tungstenite::Message::Ping(_) => {
-                        // Best-effort: tokio-tungstenite/tungstenite may handle pings internally.
-                        continue;
+                    tokio_tungstenite::tungstenite::Message::Ping(data) => {
+                        match poll_send_pong(connection, cx, data) {
+                            Poll::Ready(Ok(())) => continue,
+                            Poll::Ready(Err(e)) => {
+                                self.stats.errors += 1;
+                                return Poll::Ready(Some(Err(e)));
+                            },
+                            Poll::Pending => return Poll::Pending,
+                        }
                     },
                     tokio_tungstenite::tungstenite::Message::Pong(_) => continue,
                     tokio_tungstenite::tungstenite::Message::Binary(_) => continue,