From 8dbfb4d297c222eaf0f1a8e57027bdb7759572ef Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 01:28:23 +0200 Subject: [PATCH] Exercise supernode relay delivery and preserve ended receive handles --- src/stream/stream_network.c | 16 +++- tests/test_stream_service.c | 147 +++++++++++++++++++++++++++++++++++- 2 files changed, 157 insertions(+), 6 deletions(-) diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 0d28b0d7..893ad97d 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -199,9 +199,17 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin /* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */ static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) { struct UTUN_INSTANCE *inst = net->inst; - if (!etcp_router_has_route(inst, m.descriptor.group_id, peer) - || !etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) { - DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for route/room peer=%llx command=%u", + int route = etcp_router_has_route(inst, m.descriptor.group_id, peer); + if (!route) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for group route peer=%llx group=%llx command=%u", + (unsigned long long)peer, (unsigned long long)m.descriptor.group_id, m.command); return STREAM_WAIT; + } + if (!etcp_router_conn_get(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed channel allocation failed peer=%llx", (unsigned long long)peer); + return STREAM_ERR_MEMORY; + } + if (!etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for queue room peer=%llx command=%u", (unsigned long long)peer, m.command); return STREAM_WAIT; } struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length)); @@ -960,7 +968,7 @@ static void network_tick(void *arg) { && f->info.last_available != UINT64_MAX)) && now >= f->deadline) flow_close(f, f->outgoing ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1); - else if (f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) { + else if (f->outgoing && f->info.state >= STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) { struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .command = STREAM_CMD_RENEW, .subblock = f->subblock}; if (control_put(p, m, NULL, 0) == STREAM_OK) f->renewed = now; diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index d6e9e524..15eb121f 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -54,6 +54,10 @@ static int automatic_drop_relay; static int automatic_drop_tail; static etcp_recv_fn automatic_handlers[4]; static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; +static etcp_recv_fn relay_raw_handlers[3], relay_route_handler; +static unsigned relay_data[3], relay_ends[3], relay_requests_seen; +static int relay_drop_source; +static uint8_t relay_request_wire[STREAM_WIRE_HEADER]; static struct { unsigned data, reinits, responses; int hold_info; @@ -990,18 +994,156 @@ static void test_profile_atomicity(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] profile updates: retained avatar/name/options and atomic size/DB rollback"); } +/* Блок владельца действительно подписан канальным ключом и проходит member_sync, профиль A сохраняется. */ +static void relay_role(unsigned member, int enabled, unsigned version) { + char tags[128]; uint8_t signature[64], body[256]; + snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"supernode\":\"%s\"}", version, enabled ? "yes" : "no"); + int length = member_sync_build_owner_msg(nodes[member]->node_id, tags, body, sizeof(body)); CHECK(length > 0); + CHECK(sc_ed25519_sign(group_private, body, (size_t)length, signature) == SC_OK); + for (unsigned i = 0; i < 4; i++) { + CHECK(member_sync_put(nodes[i], "710002", nodes[member]->node_id, nodes[member]->my_keys.public_key, + nodes[member]->my_ed25519_pubkey, NULL, 0, NULL, 0, NULL, tags, signature, 0, 0, NULL) >= 0); + CHECK(stream_member_supernode(nodes[i], GROUP2, nodes[member]->node_id) == enabled); + } +} + +/* Прямой stream-вход источника блокируем только у зрителей; Merkle/BGP и получение суперузла остаются настоящими. */ +static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { + unsigned index = 0; while (index < 3 && conn->instance != nodes[index]) index++; CHECK(index < 3); + if (relay_drop_source && index != 1 && conn->peer_node_id == nodes[3]->node_id) { + queue_dgram_free(entry); queue_entry_free(entry); return; + } + relay_raw_handlers[index](conn, entry); +} + +/* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */ +static void relay_test_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_RELAY_REQUEST) { + memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++; + } + relay_route_handler(conn, entry); +} + +/* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */ +static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3); + CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2); + if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == sizeof(expected)); relay_data[index]++; } + if (event->type == STREAM_EVENT_END) relay_ends[index]++; +} + +/* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */ +static void test_supernode_relay(void) { + struct TOPO_PEER_REQUEST *connection = NULL; + CHECK(topo_group_peer_open(topo_groups_find(nodes[3]->topo_groups, GROUP2), nodes[0]->node_id, &connection) == 0); + uint64_t deadline = get_time_tb() + 150000; + while (topo_group_peer_phase(connection) != TOPO_PEER_READY && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(topo_group_peer_phase(connection) == TOPO_PEER_READY); + CHECK(member_sync_start(nodes[3], nodes[0]->node_id, "710002", NULL, NULL) == 0); + CHECK(member_sync_start(nodes[0], nodes[3]->node_id, "710002", NULL, NULL) == 0); + relay_role(1, 1, 100); + for (unsigned i = 0; i < 3; i++) { + relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; + CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_test_raw) == 0); + stream_set_event_callback(nodes[i], relay_test_event, NULL); + } + relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; + CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0); + struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE}; + struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0}; + CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); + struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK); + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + run_until(&relay_data[2], 1); test_clock_run_for(ua, 60000); + struct stream_local local; + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen); + stream_close(viewers[2]); viewers[2] = NULL; + test_clock_run_for(ua, 10000); + relay_drop_source = 1; + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + deadline = get_time_tb() + 300000; + while (relay_data[2] < 2 && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(relay_data[2] == 2 && relay_requests_seen && !relay_data[1]); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only); + /* Поздний обычный подписчик также удерживает раздачу, даже без собственного запроса помощи. */ + CHECK(stream_subscribe_open(nodes[0], GROUP2, nodes[3]->node_id, &viewers[0]) == STREAM_OK); + run_until(&relay_data[0], 1); + stream_close(viewers[2]); viewers[2] = NULL; + test_clock_run_for(ua, 80000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK); + /* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */ + CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only); + unsigned before = relay_data[1]; + CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK); + run_until(&relay_data[0], 2); run_until(&relay_data[1], before + 1); + stream_close(viewers[0]); viewers[0] = NULL; + test_clock_run_for(ua, 100000); + CHECK(stream_handle_active(viewers[1])); + CHECK(stream_get_info(viewers[1], &origin) == STREAM_OK && origin.owners == 1); + stream_close(viewers[1]); viewers[1] = NULL; + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + /* Поздний повтор завершённого запроса не подписывает суперузел заново. */ + struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry); + entry->dgram[0] = ETCP_RT_ID_STREAM; memcpy(entry->dgram + 1, relay_request_wire, STREAM_WIRE_HEADER); + entry->len = 1 + STREAM_WIRE_HEADER; + CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0); + test_clock_run_for(ua, 50000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + /* Два relay-only экземпляра не выбирают друг друга, даже если имеют буферы и живые BGP-пути. */ + struct STREAM_HANDLE *helpers[2] = {0}; + relay_role(0, 1, 100); + CHECK(stream_relay_open(nodes[1], &origin.descriptor, &helpers[0]) == STREAM_OK); + CHECK(stream_relay_open(nodes[0], &origin.descriptor, &helpers[1]) == STREAM_OK); + test_clock_run_for(ua, 30000); + for (unsigned i = 0; i < 2; i++) { + struct stream_candidate *candidates = NULL; size_t count = 0; + CHECK(stream_get_candidates(helpers[i], &candidates, &count) == STREAM_OK); + for (size_t c = 0; c < count; c++) CHECK(candidates[c].node_id != nodes[i ? 1 : 0]->node_id); + u_free(candidates); stream_close(helpers[i]); + } + relay_role(0, 0, 101); + stream_close(publisher); publisher = NULL; test_clock_run_for(ua, 10000); + /* Пустой END проходит помощь до READY/DATA; буфер освобождается, хотя источник handle ещё жив. */ + CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); + CHECK(stream_get_info(publisher, &origin) == STREAM_OK); + for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + CHECK(stream_subscribe_manual_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + stream_relay_demand(nodes[2], &origin.descriptor, 1, get_time_tb() / 10); + deadline = get_time_tb() + 100000; + while (stream_local_get(nodes[1], &origin.descriptor, &local) != STREAM_OK && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK); + test_clock_run_for(ua, 20000); CHECK(stream_publish_finish(publisher) == STREAM_OK); + run_until(&relay_ends[2], 1); CHECK(!relay_ends[1]); + test_clock_run_for(ua, 50000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + stream_close(viewers[2]); stream_close(publisher); + relay_drop_source = 0; relay_role(1, 0, 101); + CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0); + for (unsigned i = 0; i < 3; i++) { + CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0); + stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL); + } + topo_group_peer_close(connection); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END"); +} + /* Канонический список исключает неоднозначные IDs, дубли и обрезание значений. */ static void test_member_format(void) { struct stream_member_state state = {.source = {.group_id = GROUP1, .source_node_id = 1, .source_epoch = 2, .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}, .published = 1, .count = 2, - .subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX}}}; - char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[MS_USERINFO_MAX + 1]; + .subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX, 1}}}; + char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[2 * MS_USERINFO_MAX + 64]; CHECK(stream_member_format(&state, source, subscriptions) == STREAM_OK); snprintf(json, sizeof(json), "{\"stream_source\":\"%s\",\"stream_subscriptions\":\"%s\"}", source, subscriptions); struct stream_member_state parsed; CHECK(stream_member_parse(GROUP1, 1, json, &parsed) == STREAM_OK); CHECK(parsed.published && parsed.source.source_epoch == 2 && parsed.count == 2); CHECK(parsed.subscriptions[1].source_node_id == UINT64_MAX && parsed.subscriptions[1].source_epoch == UINT64_MAX); + CHECK(!parsed.subscriptions[0].relay_only && parsed.subscriptions[1].relay_only); const char *bad[] = { "{\"stream_subscriptions\":\"1:2\"}", "{\"stream_subscriptions\":\"0000000000000001:0000000000000002,\"}", @@ -1116,6 +1258,7 @@ int main(void) { test_pressure(); test_global_pressure(); test_automatic(); + test_supernode_relay(); test_pending_down(); /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */