fix(gateway): drop WebChat frames for closed sockets

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++) {

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.
  */

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;
 }