From 9a596e65e1cce1071cd308697614d7d3723ef18f Mon Sep 17 00:00:00 2001 From: wxj Date: Wed, 1 Jul 2026 12:53:33 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20WS=20frame=20parser=20=E2=80=94=20skip?= =?UTF-8?q?=20oversized=20payload=20to=20keep=20sync?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ws_client.c: When payload exceeds buffer, consume remaining bytes before returning error to prevent parser desynchronization. - market_feed.c: Increase receive buffer 4K→64K, support both standard (symbol/timestamp) and compact (s/ts) JSON field names. - ws_client.c: Disable per-frame debug print in release mode. Co-Authored-By: Claude --- app/market_feed.c | 10 ++++++---- src/ws_client.c | 12 ++++++++++-- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/app/market_feed.c b/app/market_feed.c index e3143a2..8d9a6fa 100644 --- a/app/market_feed.c +++ b/app/market_feed.c @@ -351,7 +351,7 @@ int main(int argc, char *argv[]) rte_ring_sp_enqueue_burst(ring->in, (void **)rx, nb_rx, NULL); /* Receive WebSocket frame */ - uint8_t buf[4096]; + uint8_t buf[65536]; int opcode = 0, buflen = sizeof(buf); int ws_ret = ws_recv_frame(sockfd, &opcode, buf, &buflen); @@ -360,11 +360,13 @@ int main(int argc, char *argv[]) buf[buflen] = '\0'; uint64_t now_us = rte_get_timer_cycles() * 1000000 / tsc_hz; - /* Parse symbol and timestamp from JSON */ + /* Parse symbol and timestamp from JSON (supports compact & full keys) */ char sym[32] = "unknown"; uint64_t ts_us = 0; - json_get_str((char *)buf, "symbol", sym, sizeof(sym)); - json_get_int((char *)buf, "timestamp", &ts_us); + if (json_get_str((char *)buf, "symbol", sym, sizeof(sym)) < 0) + json_get_str((char *)buf, "s", sym, sizeof(sym)); /* compact format */ + if (json_get_int((char *)buf, "timestamp", &ts_us) < 0) + json_get_int((char *)buf, "ts", &ts_us); /* compact format */ int idx = stats_find_or_add(sym); if (idx >= 0 && ts_us > 0 && now_us > ts_us) diff --git a/src/ws_client.c b/src/ws_client.c index b168fae..cec56c4 100755 --- a/src/ws_client.c +++ b/src/ws_client.c @@ -314,7 +314,7 @@ int ws_recv_frame(int sockfd, int *opcode, uint8_t *buf, int *buflen) static int frame_count = 0; frame_count++; - if (frame_count <= 3 || *opcode != 0x01) + if (0) /* debug disabled — enable for parser troubleshooting */ printf("[WS DEBUG #%d] byte0=0x%02x byte1=0x%02x opcode=%d fin=%d masked=%d plen=%lu\n", frame_count, byte0, byte1, *opcode, fin, masked, payload_len); @@ -386,8 +386,16 @@ int ws_recv_frame(int sockfd, int *opcode, uint8_t *buf, int *buflen) /* Read payload */ if (payload_len > (uint64_t)(*buflen)) { - printf("[WS] WARNING: payload %lu > buffer %d\n", + printf("[WS] WARNING: payload %lu > buffer %d, skipping\n", payload_len, *buflen); + /* Consume (skip) oversized payload to keep parser in sync */ + uint64_t remain = payload_len; + uint8_t skip_buf[2048]; + while (remain > 0) { + int chunk = (int)(remain < sizeof(skip_buf) ? remain : sizeof(skip_buf)); + if (ws_read(sockfd, skip_buf, chunk) <= 0) break; + remain -= chunk; + } return -1; }