From b97a28dad17908fa92c3a2acee40962021ee0c0d Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Sereno Date: Thu, 15 Oct 2026 23:55:22 -0300 Subject: [PATCH 1/3] 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. Supersedes #115. --- 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; } From e741b7c3e7a6716a85f400f34c38c1543d41b6a2 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Sereno Date: Sat, 17 Oct 2026 02:22:48 -0300 Subject: [PATCH 2/3] docs(changelog): record the closed-socket WebChat frames fix WebChat stalling every live client until restart after ~64 abandoned replies is an operator-visible availability fix, so give it a Fixed bullet under Unreleased alongside the other gateway/WebChat entries. --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ce4ee7..88cd152 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ All notable changes to ShellClaw are documented here. Format follows [Keep a Cha ## [Unreleased] ### Fixed +- WebChat `ws_send_to` drops a send whose session has no live `wsi` (still returning 0) and `ws_unregister_conn` drops frames still queued for the closing session, so ~64 abandoned replies cannot pin the shared outgoing queue at `MAX_QUEUE` (64) and stall every live client until restart. - Cron re-offer waits the full timeout when a due id could not be stored in the 16-slot table, and an ack frees that slot so the next due job can be delivered and then backed off. - Landlock device grants stay `/dev/null`, `/dev/zero`, `/dev/urandom`, and `/dev/tty` without `IOCTL_DEV`. A failed `landlock_add_rule` fails the ruleset closed. - Dashboard config save and JSON patch validation use a unique temp plus `fsync`, and the live gateway config pointer is published once under `agent_lock`. From d9e7f45ee5655b4a51248c6b359544c0fb1a6089 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Sereno Date: Sat, 17 Oct 2026 02:22:52 -0300 Subject: [PATCH 3/3] docs(gateway): document ws_send_to drop on closed sockets ws_send_to intentionally frees the frame and returns 0 when the target socket is absent, unregistered, or closed (the no-live-wsi path), but the header only promised '0 on success'. State the drop in the contract so webchat_send callers can rely on the actual behavior. --- src/gateway/ws.h | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/gateway/ws.h b/src/gateway/ws.h index dcc8c8d..18737b2 100644 --- a/src/gateway/ws.h +++ b/src/gateway/ws.h @@ -73,6 +73,9 @@ int ws_pop_incoming(char *session_id_out, size_t session_size, /** * Send text to a WebSocket connection by session_id. * Message is queued and sent from the lws thread. Session ID format: "webchat:". + * Frames targeted at an absent, unregistered, or closed socket (no live wsi) are + * dropped instead of queued and 0 is still returned, so a dead session cannot + * pin the shared outgoing queue. * * @param session_id Recipient session ID. * @param text Text to send.