From d740f97f67434bad5f5687c440e17a3ddff1d6d3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 11 Oct 2026 11:17:58 +0000 Subject: [PATCH] fix(gateway): drop WebChat frames for closed sockets Replies for a disconnected webchat conn_id stayed on the shared outgoing queue. After 64 of those, every live client failed to send until restart. Discard frames on unregister, and do not enqueue when no socket is registered. Co-authored-by: esadrianno --- src/gateway/ws.c | 156 ++++++++++++++++++++++++++++++++--------------- src/gateway/ws.h | 2 + tests/test_ws.c | 54 ++++++++++++++++ 3 files changed, 162 insertions(+), 50 deletions(-) diff --git a/src/gateway/ws.c b/src/gateway/ws.c index c1848f5..d3e4ef0 100644 --- a/src/gateway/ws.c +++ b/src/gateway/ws.c @@ -83,16 +83,92 @@ int ws_register_conn(int conn_id, ws_conn_t wsi) return registered ? 0 : -1; } +static void ws_free_outgoing(ws_msg_t *m) +{ + if (!m) + return; + free(m->text); + free(m); +} + +static ws_msg_t *ws_msg_new(int conn_id, const char *text) +{ + ws_msg_t *m; + + m = malloc(sizeof(*m)); + if (!m) + return NULL; + m->conn_id = conn_id; + m->text = strdup(text); + m->next = NULL; + if (!m->text) { + free(m); + return NULL; + } + return m; +} + +/** Drop every queued frame for @p conn_id. Caller holds g_mutex. */ +static void ws_drop_outgoing_locked(int conn_id) +{ + ws_msg_t *prev = NULL; + ws_msg_t *p = g_outgoing_head; + + while (p) { + ws_msg_t *next = p->next; + if (p->conn_id != conn_id) { + prev = p; + p = next; + continue; + } + if (prev) + prev->next = next; + else + g_outgoing_head = next; + if (g_outgoing_tail == p) + g_outgoing_tail = prev; + g_outgoing_count--; + ws_free_outgoing(p); + p = next; + } +} + +static ws_conn_t ws_find_wsi_locked(int conn_id) +{ + int i; + + for (i = 0; i < MAX_CONNECTIONS; i++) { + if (g_conns[i].wsi && g_conns[i].conn_id == conn_id) + return g_conns[i].wsi; + } + return NULL; +} + +static void ws_enqueue_outgoing_locked(ws_msg_t *m) +{ + if (!g_outgoing_tail) + g_outgoing_head = g_outgoing_tail = m; + else { + g_outgoing_tail->next = m; + g_outgoing_tail = m; + } + g_outgoing_count++; +} + void ws_unregister_conn(int conn_id) { + int i; + pthread_mutex_lock(&g_mutex); - for (int i = 0; i < MAX_CONNECTIONS; i++) { - if (g_conns[i].conn_id == conn_id) { + for (i = 0; i < MAX_CONNECTIONS; i++) { + if (g_conns[i].wsi && g_conns[i].conn_id == conn_id) { g_conns[i].wsi = NULL; g_conns[i].conn_id = 0; break; } } + /* Frames queued before CLOSE are never drained once wsi is gone. */ + ws_drop_outgoing_locked(conn_id); pthread_mutex_unlock(&g_mutex); } @@ -218,38 +294,33 @@ int ws_has_pending_outgoing(int conn_id) int ws_send_to(const char *session_id, const char *text) { - if (!session_id || !ws_text_len_ok(text)) return -1; - int conn_id = parse_conn_id(session_id); - if (conn_id < 0) return -1; - ws_msg_t *m = malloc(sizeof(*m)); - if (!m) return -1; - m->conn_id = conn_id; - m->text = strdup(text); - m->next = NULL; - if (!m->text) { free(m); return -1; } + int conn_id; + ws_msg_t *m; + ws_conn_t wsi; + + if (!session_id || !ws_text_len_ok(text)) + return -1; + conn_id = parse_conn_id(session_id); + if (conn_id < 0) + return -1; + m = ws_msg_new(conn_id, text); + if (!m) + return -1; pthread_mutex_lock(&g_mutex); + wsi = ws_find_wsi_locked(conn_id); + /* No live socket: enqueueing would pin the shared queue until restart. */ + if (!wsi) { + pthread_mutex_unlock(&g_mutex); + ws_free_outgoing(m); + return 0; + } if (g_outgoing_count >= MAX_QUEUE) { pthread_mutex_unlock(&g_mutex); - free(m->text); - free(m); + ws_free_outgoing(m); return -1; } - if (!g_outgoing_tail) - g_outgoing_head = g_outgoing_tail = m; - else { - g_outgoing_tail->next = m; - g_outgoing_tail = m; - } - g_outgoing_count++; - ws_conn_t wsi = NULL; - for (int i = 0; i < MAX_CONNECTIONS; i++) { - if (g_conns[i].conn_id == conn_id) { - wsi = g_conns[i].wsi; - break; - } - } + ws_enqueue_outgoing_locked(m); pthread_mutex_unlock(&g_mutex); - if (!wsi) return 0; lws_callback_on_writable((struct lws *)wsi); return 0; } @@ -263,31 +334,16 @@ void ws_broadcast_text(const char *text) pthread_mutex_lock(&g_mutex); for (i = 0; i < MAX_CONNECTIONS; i++) { ws_msg_t *m; - int cid; - if (!g_conns[i].wsi) continue; - cid = g_conns[i].conn_id; + if (!g_conns[i].wsi) + continue; if (g_outgoing_count >= MAX_QUEUE) break; - m = malloc(sizeof(*m)); - if (!m) break; - m->conn_id = cid; - m->text = strdup(text); - if (!m->text) { - free(m); + m = ws_msg_new(g_conns[i].conn_id, text); + if (!m) break; - } - m->next = NULL; - if (!g_outgoing_tail) - g_outgoing_head = g_outgoing_tail = m; - else { - g_outgoing_tail->next = m; - g_outgoing_tail = m; - } - g_outgoing_count++; - if (wake_count < MAX_CONNECTIONS) { - wake_list[wake_count] = g_conns[i].wsi; - wake_count++; - } + ws_enqueue_outgoing_locked(m); + if (wake_count < MAX_CONNECTIONS) + wake_list[wake_count++] = g_conns[i].wsi; } pthread_mutex_unlock(&g_mutex); for (i = 0; i < wake_count; i++) { diff --git a/src/gateway/ws.h b/src/gateway/ws.h index 9c781e9..dcc8c8d 100644 --- a/src/gateway/ws.h +++ b/src/gateway/ws.h @@ -42,6 +42,8 @@ int ws_register_conn(int conn_id, ws_conn_t wsi); /** * Unregister a WebSocket connection. Called when connection closes. + * Drops outgoing frames still queued for @p conn_id so a closed session + * cannot fill the shared outbound queue. * * @param conn_id Connection ID. */ diff --git a/tests/test_ws.c b/tests/test_ws.c index af53973..97789fd 100644 --- a/tests/test_ws.c +++ b/tests/test_ws.c @@ -280,6 +280,58 @@ static int test_shutdown_stops_pop(void) return 0; } +/** + * A frame queued before CLOSE must leave with the socket. Otherwise it + * sits on the shared outgoing list forever and, with enough disconnects, + * hits MAX_QUEUE so live clients can no longer send. + */ +static int test_unregister_drops_pending_outgoing(void) +{ + char buf[64]; + size_t len; + int i; + + ws_cleanup(); + ASSERT(ws_register_conn(6, (ws_conn_t)(intptr_t)6) == 0); + ASSERT(ws_register_conn(7, (ws_conn_t)(intptr_t)7) == 0); + ASSERT(ws_send_to("webchat:7", "first-7") == 0); + ASSERT(ws_send_to("webchat:6", "for-6") == 0); + ASSERT(ws_send_to("webchat:7", "for-7") == 0); + ws_unregister_conn(6); + ASSERT(ws_has_pending_outgoing(6) == 0); + ASSERT(ws_has_pending_outgoing(7) == 1); + for (i = 0; i < 128; i++) + ASSERT(ws_send_to("webchat:6", "orphan") == 0); + ASSERT(ws_send_to("webchat:7", "still-live") == 0); + ASSERT(ws_dequeue_outgoing(7, buf, sizeof(buf), &len) == 1); + ASSERT(strcmp(buf, "first-7") == 0); + ASSERT(ws_dequeue_outgoing(7, buf, sizeof(buf), &len) == 1); + ASSERT(strcmp(buf, "for-7") == 0); + ASSERT(ws_dequeue_outgoing(7, buf, sizeof(buf), &len) == 1); + ASSERT(strcmp(buf, "still-live") == 0); + ws_cleanup(); + return 0; +} + +/** Replies to a session that was never connected must not fill MAX_QUEUE. */ +static int test_send_to_absent_conn_does_not_fill_queue(void) +{ + char buf[64]; + size_t len; + int i; + + ws_cleanup(); + for (i = 0; i < 128; i++) + ASSERT(ws_send_to("webchat:99", "ghost") == 0); + ASSERT(ws_register_conn(8, (ws_conn_t)(intptr_t)8) == 0); + ASSERT(ws_send_to("webchat:8", "ok") == 0); + ASSERT(ws_dequeue_outgoing(8, buf, sizeof(buf), &len) == 1); + ASSERT(strcmp(buf, "ok") == 0); + ASSERT(ws_has_pending_outgoing(99) == 0); + ws_cleanup(); + return 0; +} + int main(void) { RUN(test_register_conn_full_table()); @@ -297,6 +349,8 @@ int main(void) RUN(test_push_incoming_rejects_empty()); RUN(test_pop_incoming_invalid_args()); RUN(test_shutdown_stops_pop()); + RUN(test_unregister_drops_pending_outgoing()); + RUN(test_send_to_absent_conn_does_not_fill_queue()); printf("test_ws: all tests passed\n"); return 0; }