From 09e86a16f6c00190e259e7494c5ad837dbc543ec Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:55:55 +0200 Subject: [PATCH] Exercise live stream relay, common start, leases and epochs --- tests/test_stream_service.c | 167 ++++++++++++++++++++++++++++++++++++ 1 file changed, 167 insertions(+) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 423ea556..46dd597c 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -11,6 +11,8 @@ #include "../src/chat/chat_core.h" #include "../src/chat/merkle_tree.h" #include "../src/routing_layer/topo_node_sqlite.h" +#include "../src/routing_layer/route_crypto.h" +#include "../src/transport_layer/etcp.h" #include "../src/utun_instance.h" #include "../lib/debug_config.h" #include "../lib/json_flat.h" @@ -30,6 +32,42 @@ static uint8_t group_private[32], group_public[32], group_x[32], expected[1000]; static struct STREAM_HANDLE *reader1, *reader2; static unsigned data_events[4], end_events[4], closed_events[4]; static int close_readers_in_callback; +static unsigned network_data[4], network_end[4], accepts, reverse_requests, dropped_renew; +static uint8_t first_request[STREAM_WIRE_HEADER + 8]; +static etcp_recv_fn saved_raw[4], saved_route; +static int drop_renew; + +/* Протокол проходит настоящие ETCP/router обработчики; перехват только считает команды и имитирует потерю RENEW. */ +static void network_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) { + uint8_t command = entry->dgram[2]; + if (node == 3 && command == STREAM_CMD_SUBSCRIBE && entry->dgram[3] == 0 && !first_request[0]) { + CHECK(entry->len == sizeof(first_request)); memcpy(first_request, entry->dgram, sizeof(first_request)); + } + if (node == 1 && command == STREAM_CMD_ACCEPT) accepts++; + if (node == 3 && command == STREAM_CMD_RENEW && drop_renew) { + dropped_renew++; queue_dgram_free(entry); queue_entry_free(entry); return; + } + } + saved_raw[node](conn, entry); +} + +static void network_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER + && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_REVERSE_REQ) reverse_requests++; + saved_route(conn, entry); +} + +static void on_network(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)arg; unsigned node = inst == nodes[1] ? 1 : 2; + CHECK(event->descriptor.group_id == GROUP2 && event->descriptor.source_node_id == nodes[3]->node_id); + if (event->type == STREAM_EVENT_DATA) { + CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected))); + CHECK(event->first_seq == 1 + network_data[node]); network_data[node]++; + } else if (event->type == STREAM_EVENT_END) network_end[node]++; + else CHECK(event->type == STREAM_EVENT_CLOSED); +} static void hex(const uint8_t *input, char *out) { for (unsigned i = 0; i < 32; i++) snprintf(out + i * 2, 3, "%02x", input[i]); @@ -104,6 +142,134 @@ static void root_hash(unsigned node, uint64_t group, uint8_t hash[32]) { CHECK(merkle_sync_read_hash(nodes[node], MT_DATA_MEMBERS, channel, 0, 0, hash) == 0); } +static void network_write(struct STREAM_HANDLE *publisher, unsigned seq) { + size_t accepted = 0; + CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected)); + uint64_t deadline = get_time_tb() + 100000; + struct stream_info info; + do { + CHECK(stream_get_info(publisher, &info) == STREAM_OK); + if (info.buffer.blocks >= (seq + 1) * 8) return; + test_clock_poll(ua, 100); + } while (get_time_tb() < deadline); + CHECK(0); +} + +/* Источник -> relay -> получатель; данные передаются через raw ETCP, никаких прямых stream_ingest в этой проверке. */ +static void test_network(void) { + for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3); + struct TOPO_PEER_REQUEST *connections[2] = {0}; + struct TOPO_GROUP *root = topo_groups_find(nodes[3]->topo_groups, GROUP2); + for (unsigned i = 1; i < 3; i++) CHECK(topo_group_peer_open(root, nodes[i]->node_id, &connections[i - 1]) == 0); + uint64_t deadline = get_time_tb() + 150000; + while (get_time_tb() < deadline + && (topo_group_peer_phase(connections[0]) != TOPO_PEER_READY || topo_group_peer_phase(connections[1]) != TOPO_PEER_READY)) + test_clock_poll(ua, 100); + for (unsigned i = 1; i < 3; i++) { + CHECK(topo_group_peer_phase(connections[i - 1]) == TOPO_PEER_READY); + CHECK(member_sync_start(nodes[3], nodes[i]->node_id, "710002", NULL, NULL) == 0); + CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0); + } + struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0}; + CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK); + struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + /* Подписанный END отдельно отвергает подмену подписи, индекса и группы. */ + uint8_t end[STREAM_END_SIZE]; uint64_t last; + CHECK(stream_end_sign(&origin.descriptor, UINT64_MAX - 1, nodes[3]->my_ed25519_privkey, end) == STREAM_OK); + CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_OK && last == UINT64_MAX - 1); + end[105] ^= 1; CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_ERR_SIGNATURE); + end[105] ^= 1; end[41] ^= 1; + CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_ERR_SIGNATURE); + for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + for (unsigned i = 1; i < 3; i++) { + CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); + stream_set_event_callback(nodes[i], on_network, NULL); + } + network_write(publisher, 0); network_write(publisher, 1); + for (unsigned i = 1; i < 4; i++) { + saved_raw[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(saved_raw[i] != NULL); + CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, network_raw) == 0); + } + saved_route = nodes[3]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; CHECK(saved_route != NULL); + CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, network_routed) == 0); + struct STREAM_RECEIVE *up[8] = {0}, *down[8] = {0}; + for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, i, UINT64_MAX, &up[i]) == STREAM_OK); + run_until(&network_data[1], 1); + for (unsigned i = 0; i < 8; i++) { + struct stream_receive_info info; CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK); + if (!info.blocks) { deadline = get_time_tb() + 100000; + do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK); } + while (!info.blocks && get_time_tb() < deadline); + } + CHECK(info.blocks == 1 && info.start == 1 && info.next == 2 && info.handle); + } + /* Node2 видит подписку relay через обычный Merkle-обмен, а не копирование из его SQLite. */ + deadline = get_time_tb() + 150000; + int ready = 0; + while (!ready && get_time_tb() < deadline) { + struct stream_member_state member; + ready = stream_member_load(nodes[2], GROUP2, nodes[1]->node_id, &member, NULL) == STREAM_OK + && stream_member_has_subscription(&member, nodes[3]->node_id, origin.descriptor.source_epoch); + if (!ready) test_clock_poll(ua, 100); + } + CHECK(ready); + for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscribers[2], nodes[1]->node_id, i, UINT64_MAX, &down[i]) == STREAM_OK); + run_until(&network_data[2], 1); + struct stream_local retained; CHECK(stream_local_get(nodes[2], &origin.descriptor, &retained) == STREAM_OK); + struct stream_path path; CHECK(stream_buffer_path(retained.buffer, 1, 0, &path) == STREAM_OK); + CHECK(path.count == 3 && path.nodes[0] == nodes[3]->node_id && path.nodes[1] == nodes[1]->node_id && path.nodes[2] == nodes[2]->node_id); + /* Замкнуть сохранённую цепочку обратно в node1 нельзя, даже при актуальной Merkle-подписке. */ + struct STREAM_RECEIVE *loop = NULL; + CHECK(stream_receive_open(subscribers[1], nodes[2]->node_id, 0, 1, &loop) == STREAM_OK); + struct stream_receive_info loop_info; + deadline = get_time_tb() + 100000; + do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(loop, &loop_info) == STREAM_OK); } + while (loop_info.state != STREAM_RECEIVE_CLOSED && get_time_tb() < deadline); + CHECK(loop_info.state == STREAM_RECEIVE_CLOSED && loop_info.reason == STREAM_REASON_LOOP); stream_receive_close(loop); + /* Повтор SUBSCRIBE сохраняет handle и текущий курсор, не отправляет старый DATA заново. */ + CHECK(first_request[0]); unsigned before = accepts; + struct ETCP_CONN *conn = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(conn && conn->links_up); + while (conn->send_input_q->count) test_clock_poll(ua, 100); + struct ll_entry *duplicate = ll_alloc_lldgram(sizeof(first_request)); CHECK(duplicate != NULL); + memcpy(duplicate->dgram, first_request, sizeof(first_request)); duplicate->len = sizeof(first_request); CHECK(etcp_send(conn, duplicate) == 0); + run_until(&accepts, before + 1); + struct stream_receive_info info; CHECK(stream_receive_get_info(up[0], &info) == STREAM_OK && info.blocks == 1 && info.next == 2); + /* Одна и та же routed reverse-попытка не меняет сессию/действующие handles. */ + struct stream_message reverse; CHECK(stream_wire_parse(first_request, sizeof(first_request), &reverse) == STREAM_OK); + reverse.command = STREAM_CMD_REVERSE_REQ; reverse.receiver_epoch = reverse.handle = 0; + for (unsigned i = 0; i < 2; i++) { + struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry != NULL); + entry->dgram[0] = ETCP_RT_ID_STREAM; stream_wire_header(entry->dgram + 1, &reverse); entry->len = 1 + STREAM_WIRE_HEADER; + CHECK(etcp_route_send(nodes[1], GROUP2, nodes[3]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0); + } + run_until(&reverse_requests, 2); + network_write(publisher, 2); run_until(&network_data[1], 2); run_until(&network_data[2], 2); + /* Снять два лишних проверочных субпотока: шесть оставшихся восстанавливают новые пачки. */ + for (unsigned i = 6; i < 8; i++) { stream_receive_close(up[i]); up[i] = NULL; stream_receive_close(down[i]); down[i] = NULL; } + network_write(publisher, 3); run_until(&network_data[1], 3); run_until(&network_data[2], 3); + CHECK(stream_publish_finish(publisher) == STREAM_OK); + run_until(&network_end[1], 1); run_until(&network_end[2], 1); + /* RENEW каждые 2с удерживает даже окончившуюся выдачу; затем имитируем отсутствие продлений. */ + test_clock_run_for(ua, 71000); + CHECK(stream_receive_get_info(up[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_ENDED); + drop_renew = 1; test_clock_run_for(ua, 71000); drop_renew = 0; + CHECK(dropped_renew && stream_receive_get_info(up[0], &info) == STREAM_OK + && info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_LEASE); + CHECK(stream_handle_active(reader1) && stream_handle_active(reader2)); + /* REINIT не обязан разрушать ETCP_CONN, но старые stream handles становятся недействительны. */ + conn = instance_find_conn(nodes[2], nodes[1]->node_id); CHECK(conn != NULL); + etcp_conn_reinit(conn, "stream network test"); + CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED + && info.reason == STREAM_REASON_TRANSPORT); + for (unsigned i = 0; i < 8; i++) { stream_receive_close(up[i]); stream_receive_close(down[i]); } + for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, saved_raw[i]) == 0); + CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, saved_route) == 0); + stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL); + for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher); + for (unsigned i = 0; i < 2; i++) topo_group_peer_close(connections[i]); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live raw ETCP chain: common index, FEC, retained paths/LOOP, duplicate/reverse, END, renew/lease, REINIT"); +} + /* Ошибка размера/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]; @@ -247,6 +413,7 @@ int main(void) { for (size_t i = 0; i < count; i++) CHECK(candidates[i].node_id != nodes[3]->node_id); u_free(candidates); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] Merkle subscriptions and BGP gating; mobile source allowed, mobile relay excluded"); + test_network(); /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */ struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);