diff --git a/src/stream/stream.c b/src/stream/stream.c index d05a10ff..b257a846 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -549,7 +549,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0); } u_free(h); s->holds--; - if (!s->holds && !s->delivery) session_free(s); + if (s->closing && !s->holds && !s->delivery) session_free(s); return result; } s->holds--; diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 9af76c09..28ceb0ba 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -748,8 +748,6 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t struct stream_message m; struct stream_local local; if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return; if (m.command >= STREAM_CMD_RELAY_REQUEST) { - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay dispatcher self=%llx sender=%llx command=%u routed=%d bytes=%zu", - (unsigned long long)net->inst->node_id, (unsigned long long)sender, m.command, routed, length); if (routed) relay_dispatch(net, sender, &m); else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u", (unsigned long long)sender, m.command); @@ -910,11 +908,6 @@ static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */ static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL; - if (entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER - && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] >= STREAM_CMD_RELAY_REQUEST) - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay routed entry self=%llx net=%p bytes=%u flags=%u cmd=%u", - (unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), (void *)net, entry->len, - entry->dgram[ROUTER_SVC_FLAGS_OFF], entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2]); if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER && (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) { uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index df2a75c7..e6cd28bb 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1018,10 +1018,6 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test routed callback self=%llx bytes=%u payload_command=%u flags=%u", - (unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), entry->len, - entry->len > ROUTER_SVC_PAYLOAD_OFF + 2 ? entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] : 0, - entry->len > ROUTER_SVC_FLAGS_OFF ? entry->dgram[ROUTER_SVC_FLAGS_OFF] : 0); 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++; @@ -1037,12 +1033,18 @@ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_eve if (event->type == STREAM_EVENT_END) relay_ends[index]++; } +/* Повторяем управление через штатный подписанный router, сохраняя nonce/эпохи исходной попытки. */ +static void relay_test_control(const struct stream_message *message) { + struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry); + entry->dgram[0] = ETCP_RT_ID_STREAM; stream_wire_header(entry->dgram + 1, message); + 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); +} + /* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */ static void test_supernode_relay(void) { int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE]; debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO); - int diagnostic_level = g_debug_config.category_levels[DEBUG_CATEGORY_DEBUG]; - debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG); struct TOPO_PEER_REQUEST *connections[3] = {0}; uint64_t deadline = get_time_tb() + 150000; for (unsigned i = 0; i < 3; i++) { @@ -1059,8 +1061,6 @@ static void test_supernode_relay(void) { stream_set_event_callback(nodes[i], relay_test_event, NULL); } relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test original routed handler=%p network=%p", (void *)relay_route_handler, - (void *)stream_network_get(nodes[1])); 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}; @@ -1099,12 +1099,18 @@ static void test_supernode_relay(void) { 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); + struct stream_message replay; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &replay) == STREAM_OK); + CHECK(replay.receiver_epoch); relay_test_control(&replay); test_clock_run_for(ua, 50000); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + /* Даже после удаления tombstone старое подтверждение поколения не может снова выделить owner. */ + test_clock_run_for(ua, 700000); relay_test_control(&replay); test_clock_run_for(ua, 20000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + replay.handle ^= UINT64_C(0x123456789abcdef0); replay.sender_epoch ^= UINT64_C(0x9876543210fedcba); + replay.receiver_epoch = 0; replay.command = STREAM_CMD_RELAY_CANCEL; relay_test_control(&replay); + replay.command = STREAM_CMD_RELAY_REQUEST; relay_test_control(&replay); + test_clock_run_for(ua, 20000); + 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); @@ -1119,6 +1125,22 @@ static void test_supernode_relay(void) { } relay_role(0, 0, 101); stream_close(publisher); publisher = NULL; test_clock_run_for(ua, 10000); + /* READY без SUBSCRIBE истекает; публичный reader сохраняется, внутренний приём останавливается. */ + 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_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK); + 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_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id) && get_time_tb() < deadline) + test_clock_poll(ua, 100); + CHECK(stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id)); + test_clock_run_for(ua, 100000); + CHECK(!stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id)); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && stream_handle_active(viewers[2])); + stream_close(viewers[2]); viewers[2] = NULL; 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); @@ -1141,7 +1163,6 @@ static void test_supernode_relay(void) { } for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level); - debug_set_category_level(DEBUG_CATEGORY_DEBUG, diagnostic_level); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END"); } @@ -1202,6 +1223,8 @@ int main(int argc, char **argv) { nodes[i] = utun_instance_create_from_str(ua, config); CHECK(nodes[i] != NULL); snprintf(nodes[i]->name, sizeof(nodes[i]->name), "node%u", i); CHECK(utun_core_start(nodes[i]) == 0 && chat_service_start(nodes[i]) == 0 && nodes[i]->stream); + CHECK(nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM] + && nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM] != nodes[i]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES]); for (unsigned g = 0; g < 2; g++) { char id[32]; uint64_t group = g ? GROUP2 : GROUP1; snprintf(id, sizeof(id), "%llu", (unsigned long long)group); uint8_t signature[64] = {0};