Apply control TCP backpressure and stream node snapshots by ID

src/control_server.c

@@ -91,9 +91,12 @@ static void accept_callback(socket_t fd, void* arg);
 static void client_read_callback(socket_t fd, void* arg);
 static void client_write_callback(socket_t fd, void* arg);
 static void client_except_callback(socket_t fd, void* arg);
-static void handle_client_data(struct control_server* server, struct control_client* client);
+static int handle_client_data(struct control_server* server, struct control_client* client);
 static void close_client(struct control_server* server, struct control_client* client);
 static void control_client_idle_cb(void* arg);
+static void control_output_ready(struct ll_queue* queue, void* arg);
+static int control_wait_output(struct control_client* client);
+static int control_client_arm_timer(struct control_client* client, int tx);
 
 static void send_conn_list(struct control_server* server, struct control_client* client, uint8_t seq_id);
 static void send_socket_list(struct control_server* server, struct control_client* client, uint8_t seq_id);
@@ -430,6 +433,9 @@ static void accept_callback(socket_t fd, void* arg) {
     }
     client->subscribed_nodes = 0;
     client->write_registered = 0;
+    queue_set_size_limit(client->output_queue, 1);
+    queue_set_threshold(client->output_queue, 0, 0);
+    queue_set_waiter_defer(client->output_queue, 1);
 
     /* Register with uasync */
     client->socket_id = uasync_add_socket_t(server->ua, client_fd,
@@ -490,6 +496,7 @@ static void accept_callback(socket_t fd, void* arg) {
 static void client_read_callback(socket_t fd, void* arg) {
     struct control_client* client = (struct control_client*)arg;
     struct control_server* server = client->server;
+    if (client->read_paused) return;
 
     if (client->recv_len >= ETCPMON_MAX_MSG_SIZE) {
         DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control recv buffer overflow (pre-check)");
@@ -586,57 +593,87 @@ static void client_read_callback(socket_t fd, void* arg) {
     }
 }
 
+/* Один таймер: ожидание команд в покое, отсутствие положительного send во время TX. */
+static int control_client_arm_timer(struct control_client* client, int tx) {
+    if (client->idle_timer) uasync_cancel_timeout(client->server->ua, client->idle_timer);
+    client->timer_is_tx = tx != 0;
+    client->idle_timer = uasync_set_timeout(client->server->ua, 300000, client, control_client_idle_cb, tx ? "ctrl_tx" : "ctrl_idle");
+    if (!client->idle_timer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control timer failed fd=%d mode=%s", (int)client->fd, tx ? "TX" : "idle");
+        client->failed = 1;
+        return -1;
+    }
+    return 0;
+}
+
+/* Останавливает входных producers до пустой очереди. Waiter всегда отложен, повторных регистраций нет. */
+static int control_wait_output(struct control_client* client) {
+    if (!client->read_paused) {
+        if (uasync_set_socket_read(client->server->ua, client->socket_id, 0) < 0) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to pause control reads fd=%d", (int)client->fd);
+            client->failed = 1;
+            return -1;
+        }
+        client->read_paused = 1;
+        DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control reads paused fd=%d queued=%zu snapshot=%zu/%zu dirty=%zu",
+                    (int)client->fd, queue_total_bytes(client->output_queue), client->snapshot_index,
+                    client->snapshot_count, client->dirty_count);
+    }
+    if (!client->timer_is_tx && control_client_arm_timer(client, 1) < 0) return -1;
+    if (!client->output_waiter.internal && !client->output_waiter.call_soon_id &&
+        queue_waiter_wait(client->output_queue, &client->output_waiter, control_output_ready, client) < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to wait for control output fd=%d", (int)client->fd);
+        client->failed = 1;
+        return -1;
+    }
+    return 0;
+}
+
 static void client_write_callback(socket_t fd, void* arg) {
-    struct control_client* client = (struct control_client*)arg;
+    struct control_client* client = arg;
     struct control_server* server = client->server;
-    struct ll_queue* q = client->output_queue;
-
-    while (q->head) {
-        struct ll_entry* entry = q->head;
-
+    struct ll_queue* queue = client->output_queue;
+    struct ll_entry* entry = queue->head;
+    if (client->failed) { close_client(server, client); return; }
+    if (entry) {
 #ifdef _WIN32
         int sent = send(fd, (const char*)entry->dgram + client->output_offset, entry->len - client->output_offset, 0);
         if (sent == SOCKET_ERROR) {
-            int err = WSAGetLastError();
-            if (err != WSAEWOULDBLOCK) {
-                DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Client send error: %d, closing", err);
-                close_client(server, client);
-                return;
-            }
-            return;
-        }
+            int error = WSAGetLastError();
+            if (error == WSAEWOULDBLOCK) return;
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control send failed fd=%d error=%d", (int)fd, error);
 #else
         ssize_t sent = send(fd, entry->dgram + client->output_offset, entry->len - client->output_offset, 0);
         if (sent < 0) {
-            if (errno != EAGAIN && errno != EWOULDBLOCK) {
-                DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Client send error: %s, closing", strerror(errno));
-                close_client(server, client);
-                return;
-            }
+            if (errno == EAGAIN || errno == EWOULDBLOCK) return;
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control send failed fd=%d: %s", (int)fd, strerror(errno));
+#endif
+            close_client(server, client);
             return;
         }
-#endif
-        if (sent == 0) {
-            DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client send made no progress fd=%d, closing", (int)fd);
+        if (!sent) {
+            DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control send made no progress fd=%d", (int)fd);
             close_client(server, client);
             return;
         }
         client->output_offset += (uint16_t)sent;
         if (client->output_offset < entry->len) {
-            DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control partial send fd=%d offset=%u size=%u queued_bytes=%zu",
-                        (int)fd, client->output_offset, entry->len, queue_total_bytes(q));
+            DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control partial send fd=%d offset=%u/%u",
+                        (int)fd, client->output_offset, entry->len);
+            if (control_client_arm_timer(client, 1) < 0) close_client(server, client);
             return;
         }
-
-        struct ll_entry* e = queue_data_get(q);
-        queue_dgram_free(e); queue_entry_free(e);
+        entry = queue_data_get(queue);
+        queue_dgram_free(entry); queue_entry_free(entry);
         client->output_offset = 0;
     }
-
-    uasync_set_socket_write(server->ua, client->socket_id, 0);
+    if (uasync_set_socket_write(server->ua, client->socket_id, 0) < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to disable control writes fd=%d", (int)fd);
+        close_client(server, client);
+        return;
+    }
     client->write_registered = 0;
-    if (q->count > 128) DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "ctrl_out: large output queue count=%d fd=%d", q->count, (int)client->fd);
-    (void)fd;
+    if (control_client_arm_timer(client, client->snapshot_active || client->dirty_count) < 0) close_client(server, client);
 }
 
 static void client_except_callback(socket_t fd, void* arg) {
@@ -652,33 +689,48 @@ static void client_except_callback(socket_t fd, void* arg) {
 
 /* Принимает владение буфером, включая ветки ошибки постановки в очередь. */
 static int control_client_send(struct control_server* server, struct control_client* client, uint8_t* dgram, uint16_t len) {
-    (void)server;
-    if (!client || !client->output_queue || !dgram || len == 0 || len > ETCPMON_MAX_MSG_SIZE) {
-        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output message client=%p size=%u", (void*)client, len);
+    if (!client || !client->output_queue || !dgram || !len || len > ETCPMON_MAX_MSG_SIZE) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output client=%p size=%u", (void*)client, len);
+        if (client) client->failed = 1;
         u_free(dgram);
         return -1;
     }
+    if (client->output_queue->head || client->failed) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control producer violated TX backpressure fd=%d queued=%zu incoming=%u",
+                    (int)client->fd, queue_total_bytes(client->output_queue), len);
+        client->failed = 1; u_free(dgram);
+        return -1;
+    }
     struct ll_entry* entry = queue_entry_new(0);
     if (!entry) {
-        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry size=%u", len);
-        u_free(dgram);
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate control output entry fd=%d size=%u", (int)client->fd, len);
+        client->failed = 1; u_free(dgram);
+        return -1;
+    }
+    entry->dgram = dgram; entry->len = len;
+    if (queue_data_put(client->output_queue, entry) < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to enqueue control output fd=%d size=%u", (int)client->fd, len);
+        client->failed = 1;
         return -1;
     }
-    entry->dgram = dgram;
-    entry->len = len;
-    queue_data_put(client->output_queue, entry);
     if (!client->write_registered) {
-        uasync_set_socket_write(client->server->ua, client->socket_id, 1);
+        if (uasync_set_socket_write(server->ua, client->socket_id, 1) < 0) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to enable control writes fd=%d", (int)client->fd);
+            client->failed = 1;
+            return -1;
+        }
         client->write_registered = 1;
     }
-    return 0;
+    return control_wait_output(client);
 }
 
 static void control_client_idle_cb(void* arg) {
-    struct control_client* client = (struct control_client*)arg;
+    struct control_client* client = arg;
     if (!client || !client->server) return;
     client->idle_timer = NULL;
-    DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client idle timeout fd=%d, closing", client->fd);
+    DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control %s timeout fd=%d queued_bytes=%zu snapshot=%zu/%zu dirty=%zu",
+               client->timer_is_tx ? "TX progress" : "idle", (int)client->fd, queue_total_bytes(client->output_queue),
+               client->snapshot_index, client->snapshot_count, client->dirty_count);
     close_client(client->server, client);
 }
 
@@ -690,6 +742,8 @@ static void close_client(struct control_server* server, struct control_client* c
                 client->output_offset, client->idle_timer != NULL);
     
     if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); client->idle_timer = NULL; }
+    if (client->output_queue) queue_waiter_cancel(client->output_queue, &client->output_waiter);
+    u_free(client->snapshot_ids); client->snapshot_ids = NULL;
 
     /* Log disconnection */
     if (server->log_file) {
@@ -742,19 +796,24 @@ static void close_client(struct control_server* server, struct control_client* c
  * Message Handling
  * ============================================================================ */
 
-static void handle_client_data(struct control_server* server, struct control_client* client) {
+static int handle_client_data(struct control_server* server, struct control_client* client) {
+    if (client->failed) { close_client(server, client); return -1; }
+    if (client->output_queue->head || client->snapshot_active || client->dirty_count) {
+        if (control_wait_output(client) < 0) { close_client(server, client); return -1; }
+        return 0;
+    }
     while (client->recv_len >= sizeof(struct etcpmon_msg_header)) {
         struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer;
         
         if (hdr->size == 0 || hdr->size > ETCPMON_MAX_MSG_SIZE) {
             DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid message size from client: %u", hdr->size);
             close_client(server, client);
-            return;
+            return -1;
         }
         if (hdr->type < ETCPMON_CMD_LIST_CONN || hdr->type > ETCPMON_CMD_SUBSCRIBE_NODES) {
             DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid command type from client: 0x%02X", hdr->type);
             close_client(server, client);
-            return;
+            return -1;
         }
         
         /* Validate header */
@@ -766,7 +825,7 @@ static void handle_client_data(struct control_server* server, struct control_cli
             }
             DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid message header from client");
             close_client(server, client);
-            return;
+            return -1;
         }
         
         /* Check if full message received */
@@ -779,6 +838,14 @@ static void handle_client_data(struct control_server* server, struct control_cli
         uint16_t payload_size = hdr->size - sizeof(struct etcpmon_msg_header);
         
         uint8_t req_seq = hdr->seq_id;
+        if ((hdr->type == ETCPMON_CMD_SELECT_CONN && payload_size != sizeof(struct etcpmon_cmd_select)) ||
+            (hdr->type != ETCPMON_CMD_SELECT_CONN && hdr->type != ETCPMON_CMD_ACTION &&
+             hdr->type != ETCPMON_CMD_SET_DEBUG_CONFIG && payload_size)) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid command payload fd=%d type=0x%02X size=%u",
+                        (int)client->fd, hdr->type, payload_size);
+            close_client(server, client);
+            return -1;
+        }
         
         if (server->log_file) {
             fprintf(server->log_file, "%llu: [LOG] Received command type=0x%02X seq=%d size=%d\n",
@@ -805,7 +872,7 @@ static void handle_client_data(struct control_server* server, struct control_cli
                 } else {
                     DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Bad SELECT_CONN payload size %u", payload_size);
                     close_client(server, client);
-                    return;
+                    return -1;
                 }
                 break;
 
@@ -825,7 +892,7 @@ static void handle_client_data(struct control_server* server, struct control_cli
             case ETCPMON_CMD_ACTION:
                 if (handle_action(server, client, payload, payload_size, req_seq) < 0) {
                     close_client(server, client);
-                    return;
+                    return -1;
                 }
                 break;
 
@@ -838,7 +905,7 @@ static void handle_client_data(struct control_server* server, struct control_cli
                     DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Bad SET_DEBUG_CONFIG payload size=%u categories=%d",
                                 payload_size, payload_size >= 2 ? payload[1] : -1);
                     close_client(server, client);
-                    return;
+                    return -1;
                 }
                 uint8_t global_lvl = payload[0];
                 uint8_t cat_count = payload[1];
@@ -858,13 +925,6 @@ static void handle_client_data(struct control_server* server, struct control_cli
             }
 
             case ETCPMON_CMD_SUBSCRIBE_NODES:
-                client->subscribed_nodes = 1;
-                if (server->log_file) {
-                    fprintf(server->log_file, "%llu: [LOG] Client subscribed to node updates\n",
-                           (unsigned long long)get_timestamp_ms());
-                    fflush(server->log_file);
-                }
-                DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Client subscribed to node updates");
                 control_server_send_node_list(server, client, req_seq);
                 break;
 
@@ -875,22 +935,15 @@ static void handle_client_data(struct control_server* server, struct control_cli
                     fflush(server->log_file);
                 }
                 close_client(server, client);
-                return;
+                return -1;
 
             default:
                 DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Unknown command from client: 0x%02X", hdr->type);
                 close_client(server, client);
-                return;
+                return -1;
         }
         
-        /* Reset idle timer on each valid command */
-        if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); }
-        client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle");
-        if (!client->idle_timer) {
-            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to reset control client idle timer fd=%d", (int)client->fd);
-            close_client(server, client);
-            return;
-        }
+        if (client->failed) { close_client(server, client); return -1; }
 
         /* Remove processed message from buffer */
         uint16_t msg_size = hdr->size;
@@ -899,7 +952,13 @@ static void handle_client_data(struct control_server* server, struct control_cli
                     client->recv_len - msg_size);
         }
         client->recv_len -= msg_size;
+        if (client->output_queue->head || client->snapshot_active || client->dirty_count) {
+            if (control_wait_output(client) < 0) { close_client(server, client); return -1; }
+            return 0;
+        }
+        if (control_client_arm_timer(client, 0) < 0) { close_client(server, client); return -1; }
     }
+    return 0;
 }
 
 /* ============================================================================
@@ -973,6 +1032,7 @@ static void send_conn_list(struct control_server* server, struct control_client*
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
     if (!buffer) {
         DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate connection list buffer");
+        client->failed = 1;
         return;
     }
     
@@ -1012,7 +1072,11 @@ static void send_socket_list(struct control_server* server, struct control_clien
     if (!instance || !instance->etcp_sockets) {
         uint16_t rsp_size = ETCPMON_SOCKET_LIST_SIZE(0);
         uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
-        if (!buffer) return;
+        if (!buffer) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate empty socket list fd=%d", (int)client->fd);
+            client->failed = 1;
+            return;
+        }
 
         struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
         etcpmon_build_header(hdr, sizeof(struct etcpmon_rsp_socket_list),
@@ -1044,6 +1108,7 @@ static void send_socket_list(struct control_server* server, struct control_clien
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
     if (!buffer) {
         DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate socket list buffer");
+        client->failed = 1;
         return;
     }
 
@@ -1135,6 +1200,7 @@ static void send_metrics(struct control_server* server, struct control_client* c
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
     if (!buffer) {
         DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate metrics buffer");
+        client->failed = 1;
         return;
     }
     
@@ -1387,7 +1453,11 @@ static void send_error(struct control_client* client, uint8_t error_code, const
     uint16_t rsp_size = ETCPMON_ERROR_SIZE(msg_len);
 
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
-    if (!buffer) return;
+    if (!buffer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate control error fd=%d seq=%u", (int)client->fd, seq_id);
+        client->failed = 1;
+        return;
+    }
 
     struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
     etcpmon_build_header(hdr, sizeof(struct etcpmon_rsp_error) + msg_len + 1,
@@ -1428,7 +1498,11 @@ static void send_debug_config(struct control_server* server, struct control_clie
     uint16_t payload_size = 2 + names_len + cat_count;
     uint16_t rsp_size = sizeof(struct etcpmon_msg_header) + payload_size;
     uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
-    if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate debug config buffer"); return; }
+    if (!buffer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate debug config buffer");
+        client->failed = 1;
+        return;
+    }
 
     struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
     etcpmon_build_header(hdr, payload_size, ETCPMON_RSP_DEBUG_CONFIG, seq_id);
@@ -1468,16 +1542,29 @@ static struct ETCP_CONN* find_connection_by_peer_id(struct UTUN_INSTANCE* instan
 static void send_single_node_info(struct control_server* server, struct control_client* client, struct TOPO_GROUP_NODE* nq) {
     if (!server || !nq) return;
     struct TOPO_NODE* ni = topo_node_registry_find(server->instance->topo_groups, nq->node_id);
-    if (!ni) return;
-    uint8_t buf[8192];
+    if (!ni) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot has no registry entry fd=%d node=%016llx", (int)client->fd,
+                    (unsigned long long)nq->node_id);
+        client->failed = 1;
+        return;
+    }
+    uint8_t buf[TOPO_NODE_WIRE_MAX_SIZE + 2];
     int ser_len = topo_node_serialize(ni, nq, TOPO_GROUP_UTUN, TOPO_FLAG_SEND_SUBNETS, buf + 2, sizeof(buf) - 2, 0);
-    if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to serialize node info"); return; }
+    if (ser_len < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to serialize node info node=%016llx", (unsigned long long)nq->node_id);
+        client->failed = 1;
+        return;
+    }
     buf[0] = CTRL_BGP_CMD; buf[1] = CTRL_BGP_SUBCMD;
     size_t data_size = 2 + (size_t)ser_len;
 
     size_t np_size = sizeof(struct etcpmon_msg_header) + data_size;
     uint8_t* buffer = (uint8_t*)u_malloc(np_size);
-    if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node info buffer"); return; }
+    if (!buffer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node info buffer");
+        client->failed = 1;
+        return;
+    }
 
     struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
     etcpmon_build_header(hdr, (uint16_t)data_size, ETCPMON_RSP_NODE_INFO, 0);
@@ -1488,61 +1575,148 @@ static void send_single_node_info(struct control_server* server, struct control_
 
 static void control_server_send_node_list(struct control_server* server, struct control_client* client, uint8_t seq_id) {
     struct TOPO_GROUP* group = topo_groups_get_default(server->instance->topo_groups);
-    if (!group) { send_error(client, ETCPMON_ERR_SERVER_BUSY, "BGP not available", seq_id); return; }
-
-    /* Send local node first */
-    if (group->local_node) send_single_node_info(server, client, group->local_node);
-
-    /* Send all known remote nodes */
-    struct ll_entry* e = group->nodes ? group->nodes->head : NULL;
-    while (e) {
-        struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
-        send_single_node_info(server, client, nq);
-        e = e->next;
+    if (!group) {
+        DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Node subscription unavailable fd=%d seq=%u", (int)client->fd, seq_id);
+        send_error(client, ETCPMON_ERR_SERVER_BUSY, "BGP not available", seq_id);
+        return;
     }
+    size_t count = (group->nodes ? queue_entry_count(group->nodes) : 0) + (group->local_node != NULL);
+    if (count > UINT32_MAX / sizeof(uint64_t)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot too large fd=%d nodes=%zu", (int)client->fd, count);
+        client->failed = 1;
+        return;
+    }
+    uint64_t* ids = count ? u_malloc(count * sizeof(*ids)) : NULL;
+    if (count && !ids) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot allocation failed fd=%d nodes=%zu", (int)client->fd, count);
+        client->failed = 1;
+        return;
+    }
+    size_t used = 0;
+    if (group->local_node) ids[used++] = group->local_node->node_id;
+    for (struct ll_entry* e = group->nodes ? group->nodes->head : NULL; e; e = e->next) {
+        if (used >= count) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot count mismatch fd=%d count=%zu", (int)client->fd, count);
+            u_free(ids); client->failed = 1;
+            return;
+        }
+        ids[used++] = ((struct TOPO_GROUP_NODE*)e)->node_id;
+    }
+    u_free(client->snapshot_ids);
+    client->snapshot_ids = ids; client->snapshot_count = used; client->snapshot_index = 0;
+    client->snapshot_active = 1; client->snapshot_seq = seq_id; client->subscribed_nodes = 1;
+    DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Node snapshot started fd=%d seq=%u nodes=%zu id_bytes=%zu",
+               (int)client->fd, seq_id, used, used * sizeof(*ids));
+}
 
-    /* Send end-of-list marker */
-    uint8_t* buffer = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header));
-    if (!buffer) return;
-    struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
-    etcpmon_build_header(hdr, 0, ETCPMON_RSP_NODE_INFO_END, seq_id);
-    control_client_send(server, client, buffer, sizeof(struct etcpmon_msg_header));
+/* Продолжение producer после опустошения TX; за один вызов создаётся максимум один кадр. */
+static void control_output_ready(struct ll_queue* queue, void* arg) {
+    struct control_client* client = arg;
+    struct control_server* server = client->server;
+    if (client->failed) { close_client(server, client); return; }
+    if (queue->head) {
+        if (control_wait_output(client) < 0) close_client(server, client);
+        return;
+    }
+    struct TOPO_GROUP* group = topo_groups_get_default(server->instance->topo_groups);
+    if ((client->snapshot_active || client->dirty_count) && !group) {
+        DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Subscribed UTUN group disappeared fd=%d", (int)client->fd);
+        close_client(server, client);
+        return;
+    }
+    while (client->snapshot_active && client->snapshot_index < client->snapshot_count && !queue->head && !client->failed) {
+        uint64_t id = client->snapshot_ids[client->snapshot_index++];
+        struct TOPO_GROUP_NODE* node = group->local_node && group->local_node->node_id == id ?
+                                      group->local_node : topo_node_find_by_id(group, id);
+        if (node) send_single_node_info(server, client, node);
+        else DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Snapshot skips removed node fd=%d node=%016llx", (int)client->fd,
+                         (unsigned long long)id);
+    }
+    if (client->snapshot_active && client->snapshot_index == client->snapshot_count && !queue->head && !client->failed) {
+        uint8_t* buffer = u_malloc(sizeof(struct etcpmon_msg_header));
+        if (!buffer) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node list end fd=%d", (int)client->fd);
+            client->failed = 1;
+        } else {
+            etcpmon_build_header((void*)buffer, 0, ETCPMON_RSP_NODE_INFO_END, client->snapshot_seq);
+            DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Node snapshot queued fd=%d seq=%u nodes=%zu",
+                       (int)client->fd, client->snapshot_seq, client->snapshot_count);
+            client->snapshot_active = 0;
+            u_free(client->snapshot_ids); client->snapshot_ids = NULL;
+            client->snapshot_count = 0; client->snapshot_index = 0;
+            control_client_send(server, client, buffer, sizeof(struct etcpmon_msg_header));
+        }
+    }
+    while (!client->snapshot_active && client->dirty_count && !queue->head && !client->failed) {
+        uint64_t id = client->dirty_ids[0];
+        int removed = client->dirty_removed[0];
+        client->dirty_count--;
+        memmove(client->dirty_ids, client->dirty_ids + 1, client->dirty_count * sizeof(*client->dirty_ids));
+        memmove(client->dirty_removed, client->dirty_removed + 1, client->dirty_count * sizeof(*client->dirty_removed));
+        struct TOPO_GROUP_NODE* node = group->local_node && group->local_node->node_id == id ?
+                                      group->local_node : topo_node_find_by_id(group, id);
+        if (!removed && node) send_single_node_info(server, client, node);
+        else {
+            size_t size = sizeof(struct etcpmon_msg_header) + sizeof(id);
+            uint8_t* buffer = u_malloc(size);
+            if (!buffer) {
+                DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node removal fd=%d", (int)client->fd);
+                client->failed = 1;
+                break;
+            }
+            etcpmon_build_header((void*)buffer, sizeof(id), ETCPMON_RSP_NODE_REMOVED, 0);
+            memcpy(buffer + sizeof(struct etcpmon_msg_header), &id, sizeof(id));
+            control_client_send(server, client, buffer, (uint16_t)size);
+        }
+    }
+    if (client->failed) { close_client(server, client); return; }
+    if (queue->head) return;
+    if (handle_client_data(server, client) < 0) return;
+    if (queue->head || client->snapshot_active || client->dirty_count) return;
+    if (client->read_paused) {
+        if (uasync_set_socket_read(server->ua, client->socket_id, 1) < 0) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to resume control reads fd=%d", (int)client->fd);
+            close_client(server, client);
+            return;
+        }
+        client->read_paused = 0;
+        DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control reads resumed fd=%d buffered=%u", (int)client->fd, client->recv_len);
+    }
+    if (client->timer_is_tx && control_client_arm_timer(client, 0) < 0) close_client(server, client);
+}
 
-    DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Sent node list to client (seq=%d)", seq_id);
+/* Сохраняет только последнее событие узла. Ограниченный набор не растёт от повторов одного ID. */
+static int control_mark_node(struct control_client* client, uint64_t id, int removed) {
+    size_t index = 0;
+    while (index < client->dirty_count && client->dirty_ids[index] != id) index++;
+    if (index == CONTROL_NODE_DIRTY_MAX) {
+        DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control subscriber overwhelmed fd=%d dirty_nodes=%zu queued_bytes=%zu",
+                   (int)client->fd, client->dirty_count, queue_total_bytes(client->output_queue));
+        return -1;
+    }
+    if (index == client->dirty_count) client->dirty_count++;
+    client->dirty_ids[index] = id; client->dirty_removed[index] = removed != 0;
+    return control_wait_output(client);
 }
 
 void control_server_notify_node_change(struct control_server* server, struct TOPO_GROUP_NODE* node) {
     if (!server || !node) return;
     struct control_client* client = server->clients;
     while (client) {
-        if (client->subscribed_nodes) send_single_node_info(server, client, node);
-        client = client->next;
+        struct control_client* next = client->next;
+        if (client->subscribed_nodes && control_mark_node(client, node->node_id, 0) < 0) close_client(server, client);
+        client = next;
     }
 }
 
 void control_server_notify_node_removed(struct control_server* server, uint64_t node_id) {
     if (!server) return;
-
-    uint8_t* buffer = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header) + sizeof(uint64_t));
-    if (!buffer) return;
-
-    struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
-    etcpmon_build_header(hdr, sizeof(uint64_t), ETCPMON_RSP_NODE_REMOVED, 0);
-    *(uint64_t*)(buffer + sizeof(*hdr)) = node_id;
-
     struct control_client* client = server->clients;
     while (client) {
-        if (client->subscribed_nodes) {
-            /* Every client needs its own copy since ownership transfers to the queue */
-            uint8_t* copy = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header) + sizeof(uint64_t));
-            if (copy) {
-                memcpy(copy, buffer, sizeof(struct etcpmon_msg_header) + sizeof(uint64_t));
-                control_client_send(server, client, copy, sizeof(struct etcpmon_msg_header) + sizeof(uint64_t));
-            }
-        }
-        client = client->next;
+        struct control_client* next = client->next;
+        if (client->subscribed_nodes && control_mark_node(client, node_id, 1) < 0) close_client(server, client);
+        client = next;
     }
-    u_free(buffer);
 }
 
 /* ============================================================================

src/control_server.h

@@ -2,7 +2,9 @@
  * control_server.h - Control Socket Server for ETCP Monitoring
  * 
  * TCP-мониторинг ETCP: init → обработка команд в uasync-потоке → shutdown.
- * Метрики отправляются по запросу; доступ — по IPv4 control_allow, idle timeout — 30 секунд.
+ * Метрики отправляются по запросу; доступ — по IPv4 [control] allow, idle timeout — 30 секунд без команд.
+ * Шифрования/аутентификации протокола нет. Общий wire-контракт — etcpmon_protocol.h.
+ * Client владеет своим сокетом, таймером и очередью TX; очередь сейчас без лимита/backpressure.
  */
 
 #ifndef CONTROL_SERVER_H
@@ -10,6 +12,7 @@
 
 #include "../tools/etcpmon/etcpmon_protocol.h"
 #include "../lib/socket_compat.h"
+#include "../lib/ll_queue.h"
 #include <stdint.h>
 #include <stdio.h>
 
@@ -26,6 +29,8 @@ struct ll_queue;
 /* Forward declaration */
 struct control_server;
 
+#define CONTROL_NODE_DIRTY_MAX 256
+
 /* Client connection state */
 struct control_client {
     struct control_client* next;        /* Linked list */
@@ -40,11 +45,23 @@ struct control_client {
     /* Selected connection */
     uint64_t selected_peer_id;          /* 0 = none selected */
 
-    struct ll_queue* output_queue;      /* Output queue for async sends */
+    struct ll_queue* output_queue;      /* Владеет TX-буферами; сейчас без ограничения объёма/числа сообщений */
     uint16_t output_offset;             /* Отправленные байты головы очереди; dgram/len не меняются */
     void* idle_timer;                  /* Client idle timeout */
     uint8_t subscribed_nodes;           /* 1 = subscribed to node updates */
     uint8_t write_registered;           /* 1 = write events registered in uasync */
+    uint8_t read_paused;                /* Чтение/разбор ждут пустой TX-очереди */
+    uint8_t failed;                     /* Ошибка producer; владелец callback закроет client после возврата */
+    uint8_t timer_is_tx;                /* Таймер контролирует прогресс TX, иначе отсутствие команд */
+    struct queue_waiter_handle output_waiter; /* Отложенное продолжение; отменить до освобождения client */
+    uint64_t* snapshot_ids;             /* ID начального списка; никаких заимствованных TOPO_GROUP_NODE* */
+    size_t snapshot_count;
+    size_t snapshot_index;              /* Следующий ID для сериализации */
+    uint8_t snapshot_active;            /* 1 до постановки NODE_INFO_END */
+    uint8_t snapshot_seq;               /* seq запроса для NODE_INFO_END */
+    uint64_t dirty_ids[CONTROL_NODE_DIRTY_MAX]; /* Последние изменения объединяются по ID */
+    uint8_t dirty_removed[CONTROL_NODE_DIRTY_MAX]; /* Последнее событие: 1 = удаление, 0 = обновление */
+    size_t dirty_count;                 /* При переполнении подписчик отключается, события не теряются молча */
 };
 
 /* Control server state */
@@ -79,10 +96,11 @@ struct control_server {
  *   ua      - uasync context for event handling
  *   instance - Parent utun instance
  *   bind_addr - Address to bind (from config control_sock)
- *   max_clients - Maximum concurrent client connections
+ *   max_clients - Maximum concurrent client connections; 0 selects 8
  * 
  * Returns:
- *   0 on success, -1 on error
+ *   0 on success, -1 on error. Server copies bind_addr; caller retains ua/instance/server.
+ *   With valid arguments, shutdown also cleans resources left by a failed initialization.
  */
 int control_server_init(struct control_server* server,
                         struct UASYNC* ua,
@@ -100,7 +118,8 @@ int control_server_init(struct control_server* server,
 void control_server_shutdown(struct control_server* server);
 
 /* Разобрать накопленные команды всех клиентов в uasync-потоке.
- * Чтение сокета также запускает разбор; метрики отправляются только по CMD_GET_METRICS.
+ * Чтение сокета само запускает разбор; отдельный периодический вызов не требуется.
+ * Функция лишь повторно разбирает уже принятые данные, не читает сокеты и не рассылает метрики по таймеру.
  * 
  * Parameters:
  *   server - Server state
@@ -117,7 +136,11 @@ void control_server_process_updates(struct control_server* server);
  */
 uint32_t control_server_get_client_count(const struct control_server* server);
 
+/* В uasync-потоке поставить подписанным клиентам снимок node (заимствован на время вызова).
+ * NODE_INFO содержит 01 04 + topo_node_serialize для UTUN с подсетями; seq=0.
+ * Неизвестный в registry узел не отправляется. Буферы ответов принадлежат очередям клиентов. */
 void control_server_notify_node_change(struct control_server* server, struct TOPO_GROUP_NODE* node);
+/* В uasync-потоке поставить подписанным клиентам NODE_REMOVED с node_id и seq=0. */
 void control_server_notify_node_removed(struct control_server* server, uint64_t node_id);
 
 #ifdef __cplusplus

src/routing_layer/topo_group.c

@@ -1200,9 +1200,11 @@ static int topo_group_routes_changed(struct TOPO_GROUP* group, struct TOPO_GROUP
     topo_fire_nodeinfo_cbk(group->instance, group, node);
     if (!node->paths) {
         event = TOPO_NODE_EVENT_REMOVE;
-        if (group->instance->control_srv) control_server_notify_node_removed(group->instance->control_srv, id);
+        if (group->group_id == TOPO_GROUP_UTUN && group->instance->control_srv)
+            control_server_notify_node_removed(group->instance->control_srv, id);
         topo_nodeq_remove_node(group, node);
-    } else if (group->instance->control_srv) control_server_notify_node_change(group->instance->control_srv, node);
+    } else if (group->group_id == TOPO_GROUP_UTUN && group->instance->control_srv)
+        control_server_notify_node_change(group->instance->control_srv, node);
     for (struct topo_node_cbk_entry* c = group->node_cbks; c; c = c->next) c->fn(group, id, event, c->arg);
     topo_group_publish_route(group, id);
     topo_recovery_changed(group);