diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a8111b..9632f8e 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. - `read_file` reads that do not fit the result buffer now fail with a JSON error instead of returning a clipped prefix, so a later `write_file` cannot persist a silently truncated file. - 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. 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..18737b2 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. */ @@ -71,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. 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; }