From fc4dddfc286c4f17caab7b95c4ae93ad61305264 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 19:02:21 +0100 Subject: [PATCH] Recheck buffered deliveries and align media test membership --- src/routing_layer/etcp_router.c | 6 +++--- tests/Makefile.am | 2 +- tests/test_media_blocks.c | 5 +++-- tests/test_media_delivery_full.c | 16 ++++++++-------- tests/test_media_delivery_sql.c | 1 + tests/test_media_query.c | 3 ++- 6 files changed, 18 insertions(+), 15 deletions(-) diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index b2090fc55..01cfe9a9a 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -1217,13 +1217,13 @@ static int router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn // Собрать доставку из recv_q: последовательно доставлять пакеты по возрастанию rx_seq, пока есть следующий. static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn) { - if (!router_allowed(rconn->inst, rconn->group_id, rconn->remote_node_id, rconn->svc_id)) { - router_close_and_notify(rconn); return; - } struct ETCP_CONN local = {.instance = rconn->inst, .peer_node_id = rconn->inst->node_id}; if (!conn) conn = &local; uint32_t next = rconn->rx_seq; while (1) { + if (!router_allowed(rconn->inst, rconn->group_id, rconn->remote_node_id, rconn->svc_id)) { + router_close_and_notify(rconn); return; + } struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next); if (!entry) break; int (*ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t) = diff --git a/tests/Makefile.am b/tests/Makefile.am index d12a00dd1..3d39b0ed1 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -1,7 +1,7 @@ # Tests Makefile.am for utun - all tests with new ll_queue library EXTRA_DIST = fixtures/reality_openssl_hello.hex data/silero_speech.pcm data/silero_speech.txt test_net.h test_membership.h nat_fixture.h db_sync_fixture.h \ - run_tests.py test_run_tests.py test_service_ids.py test_build_script.py local_transport_ipv6.sh stream_netns/topology.py stream_netns/network.py \ + media_members_fixture.h run_tests.py test_run_tests.py test_service_ids.py test_build_script.py local_transport_ipv6.sh stream_netns/topology.py stream_netns/network.py \ stream_netns/probe.py stream_netns/run.py stream_netns/scenario.py stream_netns/README.md stream_netns/TESTED.md \ stream_netns/test_isolation.py stream_netns/nat_scenario.py stream_netns/stun_server.py stream_netns/stun_scenario.py diff --git a/tests/test_media_blocks.c b/tests/test_media_blocks.c index 3939120cb..b095ec2d1 100644 --- a/tests/test_media_blocks.c +++ b/tests/test_media_blocks.c @@ -120,13 +120,14 @@ static void test_identity_and_bounds(void) { } unchanged(f.client, &bad); } - for (size_t length = 0; length < MEDIA_BLOCK_CHUNK_HDR_SIZE; length++) { + // Нулевой payload зарезервирован router для уведомления о закрытии. + for (size_t length = 1; length < MEDIA_BLOCK_CHUNK_HDR_SIZE; length++) { struct packet bad = good; bad.len = length; unchanged(f.client, &bad); } receive(f.client, &good); unchanged(f.client, &good); REQUIRE(download(f.client)->blocks[0].source->available == 2); struct packet done = done_packet(f.client); - for (size_t length = 0; length < done.len; length++) { + for (size_t length = 1; length < done.len; length++) { struct packet bad = done; bad.len = length; unchanged(f.client, &bad); } for (int variant = 0; variant < 6; variant++) { diff --git a/tests/test_media_delivery_full.c b/tests/test_media_delivery_full.c index 1031b8942..076d92d35 100644 --- a/tests/test_media_delivery_full.c +++ b/tests/test_media_delivery_full.c @@ -271,7 +271,7 @@ static int create_test_file(void) { sc_ed25519_sign(g_inst[0]->my_ed25519_privkey, smsg, soff, result.block_sigs + i * 64); } int rc = media_index_commit(g_inst[0]->topo_sqlite_db, &result, g_nid[0], - g_inst[0]->my_ed25519_privkey, "9223372036854775808", dst_path, media_base); + g_inst[0]->my_ed25519_privkey, CH_ID, dst_path, media_base); if (rc != 0) { media_index_result_free(&result); return -1; } memcpy(g_test_mid, result.media_id, 16); memcpy(g_test_bid0, result.block_ids, 16); @@ -282,9 +282,9 @@ static int create_test_file(void) { /* ── Phase A3: admission control (uses real file after create_test_file) ── */ static void phase_a3_admission(void) { - /* wait for UTUN-group routing to be established (NODEINFO exchange complete) */ + /* Дождаться маршрута CHAT-группы. */ { int a = 0; while (a < 1000) { - struct TOPO_GROUP* g = topo_groups_find(g_inst[1]->topo_groups, TOPO_GROUP_UTUN); + struct TOPO_GROUP* g = topo_groups_find(g_inst[1]->topo_groups, g_group_id); struct ETCP_CONN* c = g ? topo_group_find_conn_for_node(g, g_nid[0]) : NULL; if (c) break; uasync_poll(g_ua, POLL_MS); a++; @@ -294,7 +294,7 @@ static void phase_a3_admission(void) { struct media_pkt_block_req req; memset(&req, 0, sizeof(req)); req.subcmd = MEDIA_SUBCMD_BLOCK_REQ; assert(!media_index_generate_uuid(req.transfer_id)); memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid0, 16); - msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req)); + msend(g_inst[1], g_group_id, g_nid[0], (const uint8_t*)&req, sizeof(req)); int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; } if (g_inst[0]->md.streams_started >= 1) OK(); else FAIL("started=%d", g_inst[0]->md.streams_started); } @@ -304,7 +304,7 @@ static void phase_a3_admission(void) { struct media_pkt_block_req req; memset(&req, 0, sizeof(req)); req.subcmd = MEDIA_SUBCMD_BLOCK_REQ; assert(!media_index_generate_uuid(req.transfer_id)); memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid0, 16); - msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req)); + msend(g_inst[1], g_group_id, g_nid[0], (const uint8_t*)&req, sizeof(req)); } int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; } int started = g_inst[0]->md.streams_started - before; @@ -323,7 +323,7 @@ static void phase_b1_stream_test(void) { memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid1, 16); req.chunk = 1; g_inst[0]->md.stream_completed = 0; - msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req)); + msend(g_inst[1], g_group_id, g_nid[0], (const uint8_t*)&req, sizeof(req)); int a = 0; while (a < 3000 && !g_inst[0]->md.stream_completed) { uasync_poll(g_ua, POLL_MS); a++; } if (g_inst[0]->md.stream_completed) OK(); else FAIL("stream not completed after %dms", 3000 * POLL_MS); @@ -369,7 +369,7 @@ static int create_truncated_file(void) { assert(sc_ed25519_sign(g_inst[0]->my_ed25519_privkey, file_data + i * BS, (size_t)sz, result.block_sigs + i * 64) == SC_OK); } int rc = media_index_commit(g_inst[0]->topo_sqlite_db, &result, g_nid[0], - g_inst[0]->my_ed25519_privkey, "9223372036854775808", dst_path, media_base); + g_inst[0]->my_ed25519_privkey, CH_ID, dst_path, media_base); memcpy(g_t2_mid, result.media_id, 16); for (int i = 0; i < NB; i++) memcpy(g_t2_bids[i], result.block_ids + i * 16, 16); g_truncated = result; @@ -384,7 +384,7 @@ static void phase_b2_truncated_last_block(void) { char destination[1024]; snprintf(destination, sizeof(destination), "%s/media/truncated.bin", g_db_dir[1]); g_download_done = g_download_error = 0; - int started = media_download_start(g_inst[1], TOPO_GROUP_UTUN, &g_truncated, destination, g_db_dir[1], + int started = media_download_start(g_inst[1], g_group_id, &g_truncated, destination, g_db_dir[1], g_nid[0], downloaded, NULL, NULL, NULL); int polls = 0; while (!g_download_done && polls++ < 3000) uasync_poll(g_ua, POLL_MS); diff --git a/tests/test_media_delivery_sql.c b/tests/test_media_delivery_sql.c index 8f853e21b..4fb558bee 100644 --- a/tests/test_media_delivery_sql.c +++ b/tests/test_media_delivery_sql.c @@ -182,6 +182,7 @@ int main(void) { debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO); ua = uasync_create(); REQUIRE(ua); struct UTUN_INSTANCE* a = instance(SERVER); + sql(a->md.db, "UPDATE peers_111 SET adm_tags='{\"supernode\":\"yes\"}' WHERE node_id=256"); uint8_t epoch[16]; memcpy(epoch, a->md.source_epoch, 16); sql(a->md.db, "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status)" " VALUES(zeroblob(16),111,256,0,1,1),(zeroblob(16),222,256,0,1,1);" diff --git a/tests/test_media_query.c b/tests/test_media_query.c index d0e0d589f..5f2776df0 100644 --- a/tests/test_media_query.c +++ b/tests/test_media_query.c @@ -11,7 +11,8 @@ static void test_bounds_and_identity(void) { receive(s, &q); struct packet good = take_packet(); struct media_download snapshot = *download(c); struct media_download_block blocks[2]; memcpy(blocks, download(c)->blocks, sizeof(blocks)); - for (size_t n = 0; n < good.len; n++) { + // Нулевой payload — локальное уведомление router о закрытии, а не QUERY_RESP. + for (size_t n = 1; n < good.len; n++) { bad = good; bad.len = n; receive(c, &bad); REQUIRE(!memcmp(&snapshot, download(c), sizeof(snapshot)) && !memcmp(blocks, download(c)->blocks, sizeof(blocks))); }