This commit is contained in:
evgeny committed 2026-10-21 20:26:26 +02:00
commit e3f2feb480
6 files changed
+360 -40

No files matched your search

+6
View File
@@ -34,6 +34,12 @@
Временный regression-проверочный стенд и лог: /tmp/multiview-review/test_stream_view.cpp,
/tmp/multiview-review/lifecycle.log. Исходники реализации при проверке не менялись.
[ ] **Проверить ожидание запросов ключей в test_encrypted_attachment** — 21.10.2026:
До rebase исправлений control_server: check.sh вне sandbox 165/166; отдельный повтор также
phase=5/heartbeats=12001, ожидание двух pending incoming-запросов после перезапуска автора.
Логи: /tmp/control-server-full-check.log, /tmp/control-server-attachment-recheck.log;
артефакты общего прогона: /tmp/utun_encrypted_attachment_6JMO0c. Код вложений в этой работе не менялся.
[ ] **Разобрать нестабильную загрузку storage-узлов в test_media_delivery_chat** — 19.10.2026:
В том же общем прогоне storage st1/st2 не собрали файл, n2 исчерпал попытки блока 0 (15/19 внутренних проверок).
Отдельный повтор прошёл 19/19. Причина не установлена; сравнить маршруты и готовность storage под общей нагрузкой.
+82 -31
View File
@@ -377,7 +377,6 @@ static void accept_callback(socket_t fd, void* arg) {
#endif
return;
}
DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Accept...");
/* Check max clients */
if (server->client_count >= server->max_clients) {
@@ -429,7 +428,6 @@ static void accept_callback(socket_t fd, void* arg) {
}
client->subscribed_nodes = 0;
client->write_registered = 0;
client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle");
/* Register with uasync */
client->socket_id = uasync_add_socket_t(server->ua, client_fd,
@@ -446,6 +444,7 @@ static void accept_callback(socket_t fd, void* arg) {
fflush(server->log_file);
}
DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to register client socket with uasync");
queue_free(client->output_queue);
u_free(client);
#ifdef _WIN32
closesocket(client_fd);
@@ -460,6 +459,13 @@ static void accept_callback(socket_t fd, void* arg) {
server->clients = client;
server->client_count++;
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 arm control client idle timer fd=%d", (int)client_fd);
close_client(server, client);
return;
}
const char* addr_str;
if (client_addr.ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)&client_addr;
@@ -587,7 +593,7 @@ static void client_write_callback(socket_t fd, void* arg) {
struct ll_entry* entry = q->head;
#ifdef _WIN32
int sent = send(fd, (const char*)entry->dgram, entry->len, 0);
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) {
@@ -598,7 +604,7 @@ static void client_write_callback(socket_t fd, void* arg) {
return;
}
#else
ssize_t sent = send(fd, entry->dgram, entry->len, 0);
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));
@@ -608,10 +614,21 @@ static void client_write_callback(socket_t fd, void* arg) {
return;
}
#endif
if ((size_t)sent < entry->len) { entry->dgram += sent; entry->len -= (uint16_t)sent; q->total_bytes -= (size_t)sent; return; }
if (sent == 0) {
DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client send made no progress fd=%d, closing", (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));
return;
}
struct ll_entry* e = queue_data_get(q);
queue_dgram_free(e); queue_entry_free(e);
client->output_offset = 0;
}
uasync_set_socket_write(server->ua, client->socket_id, 0);
@@ -631,14 +648,22 @@ static void client_except_callback(socket_t fd, void* arg) {
(void)fd;
}
/* Принимает владение буфером, включая ветки ошибки постановки в очередь. */
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) return -1;
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);
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"); return -1; }
if (!entry) {
DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry size=%u", len);
u_free(dgram);
return -1;
}
entry->dgram = dgram;
entry->len = len;
entry->int_len = len;
queue_data_put(client->output_queue, entry);
if (!client->write_registered) {
uasync_set_socket_write(client->server->ua, client->socket_id, 1);
@@ -657,6 +682,10 @@ static void control_client_idle_cb(void* arg) {
static void close_client(struct control_server* server, struct control_client* client) {
if (!server || !client) return;
DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Closing control client fd=%d queued_packets=%d queued_bytes=%zu output_offset=%u timer=%d",
(int)client->fd, queue_entry_count(client->output_queue), queue_total_bytes(client->output_queue),
client->output_offset, client->idle_timer != NULL);
if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); client->idle_timer = NULL; }
@@ -680,7 +709,15 @@ static void close_client(struct control_server* server, struct control_client* c
}
#endif
if (client->output_queue) { queue_free(client->output_queue); client->output_queue = NULL; }
if (client->output_queue) {
struct ll_entry* entry;
while ((entry = queue_data_get(client->output_queue))) {
queue_dgram_free(entry);
queue_entry_free(entry);
}
queue_free(client->output_queue);
client->output_queue = NULL;
}
/* Remove from list */
struct control_client** curr = &server->clients;
@@ -822,25 +859,29 @@ static void handle_client_data(struct control_server* server, struct control_cli
send_debug_config(server, client, req_seq);
break;
case ETCPMON_CMD_SET_DEBUG_CONFIG:
if (payload_size >= 2) {
uint8_t global_lvl = payload[0];
uint8_t cat_count = payload[1];
if (cat_count > ETCPMON_MAX_DEBUG_CATEGORIES) cat_count = ETCPMON_MAX_DEBUG_CATEGORIES;
if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl);
const uint8_t* levels = payload + 2;
for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) {
if (levels[i] <= DEBUG_LEVEL_TRACE)
debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]);
}
DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count);
if (server->log_file) {
fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n",
(unsigned long long)get_timestamp_ms(), global_lvl, cat_count);
fflush(server->log_file);
}
case ETCPMON_CMD_SET_DEBUG_CONFIG: {
if (payload_size < 2 || payload[1] > ETCPMON_MAX_DEBUG_CATEGORIES || payload_size != 2u + payload[1]) {
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;
}
uint8_t global_lvl = payload[0];
uint8_t cat_count = payload[1];
if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl);
const uint8_t* levels = payload + 2;
for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) {
if (levels[i] <= DEBUG_LEVEL_TRACE)
debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]);
}
DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count);
if (server->log_file) {
fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n",
(unsigned long long)get_timestamp_ms(), global_lvl, cat_count);
fflush(server->log_file);
}
break;
}
case ETCPMON_CMD_SUBSCRIBE_NODES:
client->subscribed_nodes = 1;
@@ -871,6 +912,11 @@ static void handle_client_data(struct control_server* server, struct control_cli
/* 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;
}
/* Remove processed message from buffer */
uint16_t msg_size = hdr->size;
@@ -957,12 +1003,18 @@ static void send_socket_list(struct control_server* server, struct control_clien
uint8_t count = 0;
struct ETCP_SOCKET* sock = instance->etcp_sockets;
while (sock && count < ETCPMON_MAX_CONNECTIONS) {
if (!sock->probe_only) count++;
if (!sock->probe_only) {
if (ETCPMON_SOCKET_LIST_SIZE(count + 1u) > ETCPMON_MAX_MSG_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Socket list exceeds protocol limit: count>%u next_size=%zu max_size=%u",
count, ETCPMON_SOCKET_LIST_SIZE(count + 1u), ETCPMON_MAX_MSG_SIZE);
send_error(client, ETCPMON_ERR_SERVER_BUSY, "Socket list exceeds protocol limit", seq_id);
return;
}
count++;
}
sock = sock->next;
}
if (count > ETCPMON_MAX_CONNECTIONS) count = ETCPMON_MAX_CONNECTIONS;
uint16_t rsp_size = ETCPMON_SOCKET_LIST_SIZE(count);
uint8_t* buffer = (uint8_t*)u_malloc(rsp_size);
if (!buffer) {
@@ -1241,6 +1293,7 @@ static void send_metrics(struct control_server* server, struct control_client* c
struct etcpmon_link_metrics* link_info = (struct etcpmon_link_metrics*)(buffer + sizeof(*hdr) + sizeof(*rsp));
link = conn->links;
for (uint8_t i = 0; i < links_count && link; i++) {
memset(&link_info[i], 0, sizeof(link_info[i]));
link_info[i].local_link_id = link->local_link_id;
link_info[i].status = link->link_status;
link_info[i].encrypt_errors = (uint32_t)link->encrypt_errors;
@@ -1278,8 +1331,6 @@ static void send_metrics(struct control_server* server, struct control_client* c
uint32_t max_bw = link->bbr->bw_hi[0] > link->bbr->bw_hi[1] ? link->bbr->bw_hi[0] : link->bbr->bw_hi[1];
link_info[i].bbr_bw_hi = max_bw ? (uint32_t)((uint64_t)max_bw * link->mtu * 1000000ULL / 16777216ULL) : 0;
link_info[i].bbr_bw_lo = link->bbr->bw_lo != ~0U ? (uint32_t)((uint64_t)link->bbr->bw_lo * link->mtu * 1000000ULL / 16777216ULL) : 0;
} else {
memset(&link_info[i].bbr_mode, 0, 11 * 4 + 2 + 2); // zero out all BBR fields
}
link = link->next;
}
+5 -5
View File
@@ -1,7 +1,8 @@
/*
* control_server.h - Control Socket Server for ETCP Monitoring
*
* Provides TCP control interface for ETCP connection monitoring
* TCP-мониторинг ETCP: init → обработка команд в uasync-потоке → shutdown.
* Метрики отправляются по запросу; доступ — по IPv4 control_allow, idle timeout — 30 секунд.
*/
#ifndef CONTROL_SERVER_H
@@ -40,6 +41,7 @@ struct control_client {
uint64_t selected_peer_id; /* 0 = none selected */
struct ll_queue* output_queue; /* Output queue for async sends */
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 */
@@ -97,10 +99,8 @@ int control_server_init(struct control_server* server,
*/
void control_server_shutdown(struct control_server* server);
/* Process periodic updates
*
* Should be called periodically (e.g., from main loop) to send
* metrics updates to clients that have selected connections
/* Разобрать накопленные команды всех клиентов в uasync-потоке.
* Чтение сокета также запускает разбор; метрики отправляются только по CMD_GET_METRICS.
*
* Parameters:
* server - Server state
+10 -4
View File
@@ -10,9 +10,15 @@ TCP-сервер для мониторинга ETCP-соединений. При
4. При остановке вызвать `control_server_shutdown()`.
**Нюансы:**
- Доступ по IP контролируется через `control_allow` в конфиге (белый список); без правил доступ запрещён.
- Клиенты отключаются по idle-таймауту (300 секунд без команд).
- Лог пишется в `control_server.log` (полные дампы RX/TX).
- Доступ разрешён только IPv4-клиентам по `control_allow` в конфиге; без правил доступ запрещён. IPv6-клиенты отклоняются.
- Клиенты отключаются по idle-таймауту (30 секунд без команд). Каждая обработанная команда перезапускает таймер.
- Лог пишется в `control_server.log`: дампы содержат первые 256 байт каждого чтения RX и некоторых ответов TX.
- `control_server_process_updates()` разбирает команды; метрики отправляются только по `CMD_GET_METRICS`.
- Очередь отправки владеет исходными буферами до полной отправки или отключения; частичная отправка хранит отдельное смещение.
- Если список сокетов превышает предел сообщения (32 768 байт; сейчас помещается 230 записей),
возвращается `RSP_ERROR` с кодом `SERVER_BUSY`. Список не усекается; сокеты `probe_only` не учитываются.
- `SET_DEBUG_CONFIG` требует ровно `2 + category_count` байт, не более 64 категорий.
Несоответствие длины закрывает соединение без применения настроек.
## 3. API
@@ -42,7 +48,7 @@ TCP-сервер для мониторинга ETCP-соединений. При
| `CMD_SELECT_CONN` | — | Выбрать соединение по `peer_node_id` для метрик. |
| `CMD_GET_METRICS` | `RSP_METRICS` | Полные метрики: ETCP, TUN, router, все каналы с BBR. |
| `CMD_LIST_SOCKETS` | `RSP_SOCKET_LIST` | Список локальных сокетов с NAT-типом и NAT-адресом. |
| `CMD_ACTION` | `RSP_ACTION_RESULT` | `"nat"` — запрос NAT-проверки; `"nodes"` — дамп BGP-узлов текстом. |
| `CMD_ACTION` | Для `"nodes"` — `RSP_ACTION_RESULT`; для `"nat"` — без ответа | `"nat"` — запрос NAT-проверки; `"nodes"` — дамп BGP-узлов текстом. |
| `CMD_GET_DEBUG_CONFIG` | `RSP_DEBUG_CONFIG` | Текущие уровни отладки (глобальный + по категориям). |
| `CMD_SET_DEBUG_CONFIG` | — | Установить уровни отладки. |
| `CMD_SUBSCRIBE_NODES` | `RSP_NODE_INFO` (список) + `RSP_NODE_INFO_END` | Подписка на изменения узлов + текущий список. |
+5
View File
@@ -310,6 +310,11 @@ test_config_updater_LDFLAGS = -Wl,--wrap=fopen -Wl,--wrap=fdopen -Wl,--wrap=frea
-Wl,--wrap=fwrite -Wl,--wrap=fflush -Wl,--wrap=fclose -Wl,--wrap=fsync -Wl,--wrap=fchmod -Wl,--wrap=rename -Wl,--wrap=open \
-Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl -Wl,--wrap=u_realloc_impl -Wl,--wrap=u_strdup_impl
check_PROGRAMS += test_control_server
test_control_server_SOURCES = test_control_server.c
test_control_server_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_control_server_LDFLAGS = -Wl,--wrap=accept -Wl,--wrap=send -Wl,--wrap=uasync_add_socket_t \
-Wl,--wrap=uasync_set_timeout -Wl,--wrap=u_malloc_impl
async_regression_sources = async_regression.c async_regression.h
async_regression_ldflags = -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl -Wl,--wrap=u_realloc_impl \
-Wl,--wrap=gettimeofday -Wl,--wrap=epoll_wait -Wl,--wrap=epoll_ctl -Wl,--wrap=eventfd -Wl,--wrap=epoll_create1
+252
View File
@@ -0,0 +1,252 @@
/* Регрессии control_server: настоящие очереди/таймеры, управляемые короткие send
* и ошибки регистрации. Включение реализации даёт доступ к socket callbacks. */
#include "../src/control_server.c"
#include <assert.h>
static socket_t pending_accept = -1, capture_fd = -1;
static int fail_registration, fail_timer, fail_entry, send_mode;
static size_t send_chunk, wire_size;
static uint8_t wire[ETCPMON_MAX_MSG_SIZE];
socket_t __real_accept(socket_t, struct sockaddr*, socklen_t*);
socket_t __wrap_accept(socket_t fd, struct sockaddr* addr, socklen_t* len) {
if (pending_accept < 0) return __real_accept(fd, addr, len);
struct sockaddr_in ip = {.sin_family = AF_INET};
ip.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
assert(*len >= sizeof(ip)); memcpy(addr, &ip, sizeof(ip)); *len = sizeof(ip);
socket_t result = pending_accept; pending_accept = -1;
return result;
}
void* __real_uasync_add_socket_t(struct UASYNC*, socket_t, socket_t_callback_t, socket_t_callback_t,
socket_t_callback_t, const char*, void*);
void* __wrap_uasync_add_socket_t(struct UASYNC* ua, socket_t fd, socket_t_callback_t read, socket_t_callback_t write,
socket_t_callback_t error, const char* name, void* arg) {
if (fail_registration && !strcmp(name, "control_client")) { fail_registration = 0; return NULL; }
return __real_uasync_add_socket_t(ua, fd, read, write, error, name, arg);
}
void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* ua, int tb, void* arg, timeout_callback_t callback, const char* name) {
if (fail_timer && !strcmp(name, "ctrl_idle")) { fail_timer = 0; return NULL; }
return __real_uasync_set_timeout(ua, tb, arg, callback, name);
}
void* __real_u_malloc_impl(uint32_t, const char*);
void* __wrap_u_malloc_impl(uint32_t size, const char* location) {
if (fail_entry && strstr(location, "ll_queue.c:")) { fail_entry = 0; return NULL; }
return __real_u_malloc_impl(size, location);
}
ssize_t __real_send(socket_t, const void*, size_t, int);
ssize_t __wrap_send(socket_t fd, const void* data, size_t len, int flags) {
if (fd != capture_fd) return __real_send(fd, data, len, flags);
if (send_mode == 1) { errno = EAGAIN; return -1; }
if (send_mode == 2) { errno = EPIPE; return -1; }
if (send_mode == 3) return 0;
size_t take = len < send_chunk ? len : send_chunk;
assert(take > 0 && wire_size + take <= sizeof(wire));
memcpy(wire + wire_size, data, take); wire_size += take;
return (ssize_t)take;
}
struct fixture {
struct UTUN_INSTANCE inst;
struct control_server server;
struct CFG_CONTROL_ALLOW allow;
socket_t peer;
size_t allocations;
};
static void fixture_init(struct fixture* f) {
memset(f, 0, sizeof(*f));
f->allocations = u_get_allocated_count();
f->inst.ua = uasync_create(); assert(f->inst.ua);
f->inst.config = u_calloc(1, sizeof(*f->inst.config)); assert(f->inst.config);
f->inst.config->global.control_allows = &f->allow;
f->inst.config->global.control_allow_count = 1;
f->inst.config->global.control_allows[0].network = INADDR_LOOPBACK;
f->inst.config->global.control_allows[0].netmask = UINT32_MAX;
f->server.ua = f->inst.ua; f->server.instance = &f->inst;
f->server.max_clients = 8; f->server.listen_fd = -1; f->peer = -1;
}
static struct control_client* fixture_accept(struct fixture* f) {
socket_t pair[2]; assert(socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == 0);
assert(f->peer == -1); f->peer = pair[1]; pending_accept = pair[0];
accept_callback(-1, &f->server);
return f->server.clients;
}
static void fixture_destroy(struct fixture* f) {
capture_fd = -1; send_mode = 0;
control_server_shutdown(&f->server);
if (f->peer >= 0) close(f->peer);
TimeoutEntry pending;
assert(timeout_heap_peek(f->inst.ua->timeout_heap, &pending) != 0);
assert(timeout_heap_get_size(f->inst.ua->timeout_heap) == 0);
assert(f->inst.ua->timer_alloc_count == f->inst.ua->timer_free_count);
uasync_destroy(f->inst.ua, 0); u_free(f->inst.config);
assert(u_get_allocated_count() == f->allocations);
}
static void enqueue(struct fixture* f, struct control_client* client, const char* text) {
uint16_t len = (uint16_t)strlen(text);
uint8_t* buffer = u_malloc(len); assert(buffer); memcpy(buffer, text, len);
assert(control_client_send(&f->server, client, buffer, len) == 0);
}
/* Проверяем wire-байты и учёт памяти после нескольких partial send и EAGAIN. */
static void test_partial_send(void) {
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
capture_fd = client->fd; send_chunk = 3; wire_size = 0;
enqueue(&f, client, "0123456789"); enqueue(&f, client, "abcdefgh");
uint8_t* base = client->output_queue->head->dgram;
client_write_callback(client->fd, client);
assert(client->output_offset == 3 && client->output_queue->head->dgram == base);
assert(queue_total_bytes(client->output_queue) == 18);
send_mode = 1; client_write_callback(client->fd, client);
assert(client->output_offset == 3 && wire_size == 3);
send_mode = 0;
while (client->output_queue->head) client_write_callback(client->fd, client);
assert(wire_size == 18 && !memcmp(wire, "0123456789abcdefgh", 18));
assert(!client->output_offset && !client->write_registered && !queue_total_bytes(client->output_queue));
fixture_destroy(&f);
puts("PASS partial send: exact wire bytes, EAGAIN, stable buffer and queue accounting");
}
static void test_cleanup(void) {
for (int mode = 0; mode <= 3; mode++) {
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
enqueue(&f, client, "first message"); enqueue(&f, client, "second message");
capture_fd = client->fd; send_chunk = 2; wire_size = 0;
client_write_callback(client->fd, client); assert(client->output_offset == 2);
if (mode >= 2) { send_mode = mode; client_write_callback(client->fd, client); }
else if (mode == 1) {
uasync_cancel_timeout(f.inst.ua, client->idle_timer);
control_client_idle_cb(client);
} else close_client(&f.server, client);
assert(!f.server.clients && !f.server.client_count);
fixture_destroy(&f);
}
for (int fault = 0; fault < 2; fault++) {
struct fixture f; fixture_init(&f);
fail_registration = fault == 0; fail_timer = fault == 1;
assert(!fixture_accept(&f));
assert(!f.server.client_count && !fail_registration && !fail_timer);
fixture_destroy(&f);
}
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
uint8_t* buffer = u_malloc(8); assert(buffer);
size_t before = u_get_allocated_count(); fail_entry = 1;
assert(control_client_send(&f.server, client, buffer, 8) == -1 && !fail_entry);
assert(u_get_allocated_count() == before - 1 && !client->output_queue->head);
fixture_destroy(&f);
puts("PASS cleanup: pending/partial sends, idle, send failure, registration/timer failure, allocation failure; no leaks");
}
/* Байты после объявленного SET_DEBUG_CONFIG не должны менять настройки. */
static void test_debug_payload(void) {
for (int variant = 0; variant < 6; variant++) {
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
debug_config_t saved = g_debug_config;
struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer;
uint8_t* payload = client->recv_buffer + sizeof(*hdr);
memset(payload, DEBUG_LEVEL_TRACE, 100);
payload[0] = DEBUG_LEVEL_ERROR; payload[1] = 3;
uint16_t len = variant == 0 ? 1 : variant == 1 ? 2 : variant == 2 ? 4 : variant == 3 ? 6 : 5;
if (variant == 4) { payload[1] = ETCPMON_MAX_DEBUG_CATEGORIES + 1; len = 2 + payload[1]; }
etcpmon_build_header(hdr, len, ETCPMON_CMD_SET_DEBUG_CONFIG, 17);
client->recv_len = hdr->size;
if (variant == 5) {
client->recv_len--; handle_client_data(&f.server, client);
assert(!memcmp(&saved, &g_debug_config, sizeof(saved)) && f.server.clients == client);
client->recv_len++; handle_client_data(&f.server, client);
assert(g_debug_config.level == DEBUG_LEVEL_ERROR);
for (int i = 1; i <= 3; i++) assert(g_debug_config.category_levels[i] == DEBUG_LEVEL_TRACE);
} else {
handle_client_data(&f.server, client);
assert(!f.server.clients && !memcmp(&saved, &g_debug_config, sizeof(saved)));
}
g_debug_config = saved;
fixture_destroy(&f);
}
puts("PASS SET_DEBUG_CONFIG: short, truncated, extra and excessive payloads rejected; fragmented valid command applied");
}
static void test_metrics(void) {
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
struct ll_queue empty = {0};
struct ETCP_CONN conn = {.peer_node_id = 42, .input_queue = &empty, .input_send_q = &empty,
.input_wait_ack = &empty, .ack_q = &empty, .recv_q = &empty, .output_queue = &empty};
struct ETCP_LINK links[2] = {{.local_link_id = 7}, {.local_link_id = 8}};
links[0].next = &links[1]; conn.links = links;
f.inst.connections = queue_new(f.inst.ua, 16, 0, 8, "test connections"); assert(f.inst.connections);
struct ll_entry* entry = queue_entry_new(sizeof(struct conn_queue_entry)); assert(entry);
struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
ce->peer_node_id = conn.peer_node_id; ce->conn = &conn;
assert(queue_data_put_with_index(f.inst.connections, entry) == 0);
client->selected_peer_id = conn.peer_node_id;
for (int count = 1; count <= 2; count++) {
links[0].next = count == 1 ? NULL : &links[1];
send_metrics(&f.server, client, 23);
struct ll_entry* response = queue_data_get(client->output_queue); assert(response);
u_check(response->dgram, "metrics bounds", LOCATION);
struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)response->dgram;
assert(hdr->type == ETCPMON_RSP_METRICS && hdr->seq_id == 23 && hdr->size == ETCPMON_METRICS_SIZE(count));
struct etcpmon_link_metrics* metrics = (void*)(response->dgram + sizeof(*hdr) + sizeof(struct etcpmon_rsp_metrics));
for (int i = 0; i < count; i++) {
assert(metrics[i].local_link_id == 7 + i);
const uint8_t* start = &metrics[i].bbr_mode;
for (size_t j = 0; j < sizeof(*metrics) - offsetof(struct etcpmon_link_metrics, bbr_mode); j++) assert(start[j] == 0);
}
queue_dgram_free(response); queue_entry_free(response);
}
queue_entry_free(queue_data_get(f.inst.connections)); queue_free(f.inst.connections);
fixture_destroy(&f);
puts("PASS metrics: one/two links without BBR, zeroed BBR fields and intact allocation guards");
}
static void test_socket_limit(void) {
struct fixture f; fixture_init(&f);
struct control_client* client = fixture_accept(&f); assert(client);
size_t limit = (ETCPMON_MAX_MSG_SIZE - ETCPMON_SOCKET_LIST_SIZE(0)) / sizeof(struct etcpmon_socket_info);
struct ETCP_SOCKET* sockets = u_calloc(251, sizeof(*sockets)); assert(sockets);
for (size_t i = 0; i < 251; i++) {
snprintf(sockets[i].name, sizeof(sockets[i].name), "socket-%zu", i);
sockets[i].interface_addr.ss_family = AF_INET;
if (i < 250) sockets[i].next = &sockets[i + 1];
}
f.inst.etcp_sockets = sockets;
const size_t counts[] = {0, limit, limit + 1, 250};
for (size_t n = 0; n < sizeof(counts) / sizeof(counts[0]); n++) {
size_t count = counts[n];
for (size_t i = 0; i < 251; i++) sockets[i].probe_only = i >= count;
send_socket_list(&f.server, client, 31);
struct ll_entry* response = queue_data_get(client->output_queue); assert(response);
struct etcpmon_msg_header* hdr = (void*)response->dgram;
assert(etcpmon_validate_header(hdr) == 0 && hdr->size == response->len && hdr->seq_id == 31);
if (count <= limit) {
assert(hdr->type == ETCPMON_RSP_SOCKET_LIST && hdr->size == ETCPMON_SOCKET_LIST_SIZE(count));
assert(response->dgram[sizeof(*hdr)] == count);
} else {
assert(hdr->type == ETCPMON_RSP_ERROR && response->dgram[sizeof(*hdr)] == ETCPMON_ERR_SERVER_BUSY);
}
queue_dgram_free(response); queue_entry_free(response);
}
u_free(sockets); f.inst.etcp_sockets = NULL;
fixture_destroy(&f);
puts("PASS socket list: zero, maximum fitting, first excessive and 250 sockets; probe-only sockets excluded");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
debug_set_category_level(DEBUG_CATEGORY_CONTROL, DEBUG_LEVEL_DEBUG);
test_partial_send(); test_cleanup(); test_debug_payload(); test_metrics(); test_socket_limit();
return 0;
}