From f11592be3e3390bdde69c51a2b58f7d5897d551d Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 08:37:14 +0200 Subject: [PATCH] Exercise stream overload under real ETCP backpressure --- tests/test_stream_service.c | 109 ++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index e77ed6d7..99315a47 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -39,6 +39,38 @@ static uint64_t network_next[4] = {0, 1, 1, 0}; static uint8_t first_request[STREAM_WIRE_HEADER + 8]; static etcp_recv_fn saved_raw[4], saved_route; static int drop_renew, gap_batch; +static unsigned pressure_data, overload_seen, selects_seen; +static int ignore_select; +static uint8_t pressure_payload[7080], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER]; +static size_t stale_select_length; +static etcp_recv_fn pressure_handlers[4]; + +static void pressure_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { + unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4); + if (entry->len >= STREAM_WIRE_HEADER) { + if (entry->dgram[2] == STREAM_CMD_OVERLOAD) overload_seen++; + if (entry->dgram[2] == STREAM_CMD_OVERLOAD_SELECT) { + selects_seen++; CHECK(entry->len <= sizeof(stale_select)); + memcpy(stale_select, entry->dgram, entry->len); stale_select_length = entry->len; + if (ignore_select) { queue_dgram_free(entry); queue_entry_free(entry); return; } + } + } + pressure_handlers[node](conn, entry); +} + +static void on_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)arg; CHECK(inst == nodes[1] && event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2); + if (event->type == STREAM_EVENT_DATA) { + CHECK(event->length == sizeof(pressure_payload) && !memcmp(event->data, pressure_payload, sizeof(pressure_payload))); pressure_data++; + } else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END); +} + +static struct stream_peer_load_info source_load(void) { + struct stream_peer_load_info info; CHECK(stream_peer_get_load(nodes[3], nodes[1]->node_id, &info) == STREAM_OK); + CHECK(info.load.arrivals.bytes == info.load.sent.bytes + info.load.removed.bytes + info.load.queued.bytes); + CHECK(info.load.arrivals.packets == info.load.sent.packets + info.load.removed.packets + info.load.queued.packets); + return info; +} /* Протокол проходит настоящие ETCP/router обработчики; перехват только считает команды и имитирует потерю RENEW. */ static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { @@ -328,6 +360,82 @@ static void test_network(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live ETCP chain: common index/restart, FEC, paths/LOOP, GAP, duplicate/reverse, END, leases, REINIT"); } +/* Настоящий send_input_q намеренно держим непустым: короткий глитч не режет, устойчивый рост вызывает OVERLOAD. */ +static void test_pressure(void) { + memset(pressure_payload, 42, sizeof(pressure_payload)); + struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL; + CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK); + struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0); + CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK); + stream_set_event_callback(nodes[1], on_pressure, NULL); + for (unsigned i = 1; i <= 3; i += 2) { + pressure_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(pressure_handlers[i] != NULL); + CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_raw) == 0); + } + size_t accepted = 0; + CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK && accepted == sizeof(pressure_payload)); + struct STREAM_RECEIVE *receives[8] = {0}; + for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK); + run_until(&pressure_data, 1); + uint64_t deadline = get_time_tb() + 100000; + while (get_time_tb() < deadline) { + int ready = 1; + for (unsigned i = 0; i < 8; i++) { + struct stream_receive_info info; CHECK(stream_receive_get_info(receives[i], &info) == STREAM_OK); + if (!info.blocks) ready = 0; + CHECK(info.rtt_valid && info.supplier_metrics_valid); + } + if (ready) break; test_clock_poll(ua, 100); + } + struct ETCP_CONN *conn = instance_find_conn(nodes[3], nodes[1]->node_id); CHECK(conn && conn->links_up); + struct ll_queue *queue = conn->send_input_q; + while (queue->count) test_clock_poll(ua, 100); + queue_callback_fn callback = queue->callback; void *callback_arg = queue->callback_arg; + CHECK(callback != NULL); queue_set_callback(queue, NULL, NULL); + for (unsigned i = 0; i < 4; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK); + test_clock_run_for(ua, 3000); + struct stream_peer_load_info info = source_load(); CHECK(info.load.queued.bytes && !info.request_id && !overload_seen); + queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue); + run_until(&pressure_data, 5); + test_clock_run_for(ua, 1000); info = source_load(); CHECK(!info.load.queued.bytes && !info.request_id); + /* Тот же транспорт остаётся UP. Теперь держим очередь дольше подтверждения и возвращаем управление получателю. */ + while (queue->count) test_clock_poll(ua, 100); + queue_set_callback(queue, NULL, NULL); + for (unsigned i = 0; i < 24; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK); + deadline = get_time_tb() + 15000; + do { test_clock_poll(ua, 100); info = source_load(); } while (!info.request_active && get_time_tb() < deadline); + CHECK(info.request_active && info.required_count > 1 && info.required_rate && !info.released_count); + uint64_t request = info.request_id, expires = info.deadline_ms; + test_clock_run_for(ua, 3000); info = source_load(); CHECK(info.request_id == request && info.deadline_ms == expires); + ignore_select = 1; queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue); + deadline = get_time_tb() + 100000; + while (!selects_seen && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(overload_seen && selects_seen && stale_select_length); + /* Повторы SELECT намеренно не доходят: source исполняет достаточный остаток на исходном deadline. */ + while (get_time_tb() / 10 < expires + 100) test_clock_poll(ua, 100); + info = source_load(); CHECK(!info.request_active && info.released_count > 1 && info.released_rate >= info.required_rate); + unsigned closed = 0; + for (unsigned i = 0; i < 8; i++) { + struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK); + if (receive.state == STREAM_RECEIVE_CLOSED) { CHECK(receive.reason == STREAM_REASON_OVERLOAD); closed++; } + } + CHECK(closed == info.released_count); + /* Поздний выбор старого id не закрывает дополнительную выдачу. */ + ignore_select = 0; + struct ETCP_CONN *back = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(back != NULL); + while (back->send_input_q->count) test_clock_poll(ua, 100); + struct ll_entry *late = ll_alloc_lldgram((uint16_t)stale_select_length); CHECK(late != NULL); + memcpy(late->dgram, stale_select, stale_select_length); late->len = (uint16_t)stale_select_length; CHECK(etcp_send(back, late) == 0); + test_clock_run_for(ua, 1000); CHECK(source_load().released_count == closed && conn->links_up); + for (unsigned i = 0; i < 8; i++) stream_receive_close(receives[i]); + stream_close(subscriber); stream_close(publisher); test_clock_run_for(ua, 1000); + for (unsigned i = 1; i <= 3; i += 2) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_handlers[i]) == 0); + stream_set_event_callback(nodes[1], on_stream, NULL); + CHECK(stream_handle_active(reader1) && stream_handle_active(reader2)); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: short glitch, multi-flow frozen rates, fixed 2s deadline, fallback, late SELECT and shared ETCP"); +} + /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */ static void test_profile_atomicity(void) { char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1]; @@ -472,6 +580,7 @@ int main(void) { u_free(candidates); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] Merkle subscriptions and BGP gating; mobile source allowed, mobile relay excluded"); test_network(); + test_pressure(); /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */ struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);