Recheck buffered deliveries and align media test membership

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) =

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
 

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++) {

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);

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);"

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)));
     }