From fd6155aa4a0d71a1ec2ecccd584d9f6f9e989a0a Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 18:17:02 +0200 Subject: [PATCH] Exercise media QUERY bounds, paging, failures and peer reuse --- tests/test_media_download_timeout.c | 27 ++- tests/test_media_query.c | 308 ++++++++++++++++++++++++++++ 2 files changed, 327 insertions(+), 8 deletions(-) create mode 100644 tests/test_media_query.c diff --git a/tests/test_media_download_timeout.c b/tests/test_media_download_timeout.c index c54481d2..ac6f4778 100644 --- a/tests/test_media_download_timeout.c +++ b/tests/test_media_download_timeout.c @@ -17,11 +17,18 @@ #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../src/utun_instance.h" +#include "../lib/platform_compat.h" #include #include #include #include -#include + +/* Транспорт принимает пакет; недостижимый получатель не отвечает. */ +int __wrap_etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t dst, + struct ll_entry* entry, int force, int mode) { + (void)inst; (void)group; (void)dst; (void)force; (void)mode; + queue_dgram_free(entry); queue_entry_free(entry); return 0; +} static int G_PASSED = 0, G_FAILED = 0, G_TOTAL = 0; #define TEST(n) do { G_TOTAL++; printf(" %-58s", n); fflush(stdout); } while(0) @@ -33,7 +40,7 @@ static int g_done = 0, g_err = 0; static void make_uuid(uint8_t b[16]) { for (int i = 0; i < 16; i++) b[i] = (uint8_t)(rand() & 0xFF); } -static void done_cb(void* a, int e) { (void)a; g_done = 1; g_err = e; printf(" [done_cb err=%d]\n", e); } +static void done_cb(void* a, int e) { (void)a; g_done++; g_err = e; printf(" [done_cb err=%d]\n", e); } static struct UTUN_INSTANCE* make_minimal(void) { struct UTUN_INSTANCE* inst = u_calloc(1, sizeof(*inst)); @@ -49,6 +56,7 @@ static struct UTUN_INSTANCE* make_minimal(void) { memset(inst->my_ed25519_pubkey, 0xDD, 32); inst->md.inst = inst; inst->md.db = db; inst->md.self_node_id = inst->node_id; inst->md.initialized = 1; inst->md.dl_stall_timeout_tb = 1000; /* 100ms */ + inst->md.query_timeout_tb = 1000; inst->md.dl_max_attempts = 3; return inst; } @@ -103,13 +111,15 @@ int main(void) { /* имитируем QUERY_RESP: автор сообщает, что держит блок */ struct media_pkt_query_resp_entry e; memset(&e, 0, sizeof(e)); - e.node_id = 0xBADF00D00000002ULL; + e.num_holders = 1; e.holders[0] = 0xBADF00D00000002ULL; memcpy(e.block_id, r.block_ids, 16); - uint8_t pkt[3 + sizeof(e)]; - pkt[0] = MEDIA_SUBCMD_QUERY_RESP; - uint16_t ne = 1; memcpy(pkt + 1, &ne, 2); - memcpy(pkt + 3, &e, sizeof(e)); - media_download_handle_query_resp(inst, pkt, sizeof(pkt)); + uint8_t pkt[MEDIA_QUERY_RESP_HDR_SIZE + sizeof(e)] = {0}; + struct media_download* dl = (struct media_download*)inst->md.downloads->head->data; + struct media_pkt_query_resp* response = (struct media_pkt_query_resp*)pkt; + response->subcmd = MEDIA_SUBCMD_QUERY_RESP; response->num_blocks = 1; + memcpy(response->media_id, r.media_id, 16); memcpy(response->request_id, dl->query_id, 16); + memcpy(pkt + MEDIA_QUERY_RESP_HDR_SIZE, &e, sizeof(e)); + media_download_handle_query_resp(inst, dl->group_id, dl->query_node, pkt, sizeof(pkt)); /* держатель «завис» — чанков нет, watchdog сделает фейловер → попытки → fail */ if (poll_until_done(5000) && g_err != 0) OK(); @@ -117,6 +127,7 @@ int main(void) { } media_index_result_free(&r); + if (inst->md.downloads) queue_free(inst->md.downloads); sqlite3_close(inst->topo_sqlite_db); u_free(inst); } diff --git a/tests/test_media_query.c b/tests/test_media_query.c new file mode 100644 index 00000000..b87ed536 --- /dev/null +++ b/tests/test_media_query.c @@ -0,0 +1,308 @@ +/* Регрессии QUERY: реальные обработчики сервиса и SQLite, управляемая доставка/ошибки транспорта. + * Стенд сохраняет владение router: пакет потребляется даже при ошибке. Ответы доставляются отдельно, + * поэтому можно проверить задержки, дубликаты, чужие ответы и состояние между пачками. */ +#include "test_utils.h" +#include "../src/utun_instance.h" +#include "../src/chat/chat_core_priv.h" +#include "../src/media_delivery/media_index.h" +#include "../src/media_delivery/media_download.h" +#include "../src/routing_layer/etcp_router.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "../lib/platform_compat.h" +#include +#include +#include +#ifdef __SANITIZE_ADDRESS__ +#include +#endif + +#define GROUP 111 +#define SERVER 0x100 +#define CLIENT 0x200 +#define MAX_WIRE 2048 +#define REQUIRE(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "QUERY test line=%d: %s", __LINE__, #x); exit(1); } } while (0) + +struct packet { uint64_t from, to, group; size_t len; uint8_t data[MAX_WIRE]; }; +static struct packet packets[128]; +static int packet_count, query_count, request_count, done_count, done_err, fail_send, fail_rng; +static struct UASYNC* ua; +static char directory[256]; +static void* injected; +static uint8_t* poison_tail; + +int __real_RAND_bytes(unsigned char* out, int count); +int __wrap_RAND_bytes(unsigned char* out, int count) { + if (fail_rng) return 0; + return __real_RAND_bytes(out, count); +} + +int __wrap_etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t dst, + struct ll_entry* entry, int force, int mode) { + (void)force; (void)mode; + int cmd = entry->dgram[1], rc = cmd == fail_send ? -1 : 0; + if (cmd == MEDIA_SUBCMD_BLOCK_REQ) request_count++; + if (cmd == MEDIA_SUBCMD_QUERY) query_count++; + if (!rc && (cmd == MEDIA_SUBCMD_QUERY || cmd == MEDIA_SUBCMD_QUERY_RESP)) { + REQUIRE(packet_count < 128 && entry->len - 1 <= MAX_WIRE); + struct packet* p = &packets[packet_count++]; + p->from = inst->node_id; p->to = dst; p->group = group; p->len = entry->len - 1; + memcpy(p->data, entry->dgram + 1, p->len); + } + queue_dgram_free(entry); queue_entry_free(entry); return rc; +} + +/* u_malloc округляет выделение: ASAN защищает именно логическую границу входного payload. */ +static void release_injected(void* data) { +#ifdef __SANITIZE_ADDRESS__ + if (data == injected) __asan_unpoison_memory_region(poison_tail, 64); +#endif + injected = NULL; poison_tail = NULL; u_free(data); +} + +static void receive(struct UTUN_INSTANCE* inst, const struct packet* p) { + struct ll_entry* entry = queue_entry_new(0); REQUIRE(entry); + size_t bytes = ROUTER_SVC_PAYLOAD_OFF + p->len; + entry->dgram = u_calloc(1, bytes + 64); REQUIRE(entry->dgram); + entry->len = (uint16_t)bytes; entry->dgram_free_fn = release_injected; + entry->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; + memcpy(entry->dgram + ROUTER_SVC_SRC_OFF, &p->from, 8); + memcpy(entry->dgram + ROUTER_SVC_GROUP_OFF, &p->group, 8); + memcpy(entry->dgram + ROUTER_SVC_PAYLOAD_OFF, p->data, p->len); + injected = entry->dgram; poison_tail = entry->dgram + bytes; +#ifdef __SANITIZE_ADDRESS__ + __asan_poison_memory_region(poison_tail, 64); +#endif + struct ETCP_CONN conn = {.instance = inst}; + inst->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY](&conn, entry); + REQUIRE(!injected); +} + +static struct packet take_packet(void) { + REQUIRE(packet_count > 0); + struct packet result = packets[0]; + memmove(packets, packets + 1, (size_t)--packet_count * sizeof(*packets)); + return result; +} + +static struct UTUN_INSTANCE* instance(uint64_t node) { + struct UTUN_INSTANCE* inst = u_calloc(1, sizeof(*inst)); REQUIRE(inst); + inst->ua = ua; inst->node_id = node; + inst->chat_core = u_calloc(1, sizeof(struct chat_core_ctx)); REQUIRE(inst->chat_core); + REQUIRE(sqlite3_open(":memory:", &inst->topo_sqlite_db) == SQLITE_OK); + REQUIRE(!media_index_init(inst->topo_sqlite_db) && !media_delivery_init(inst)); + return inst; +} + +static void destroy(struct UTUN_INSTANCE* inst) { + media_delivery_destroy(inst); sqlite3_close(inst->topo_sqlite_db); + u_free(inst->chat_core); u_free(inst); uasync_poll(ua, 0); +} + +static void completed(void* arg, int err) { (void)arg; done_count++; done_err = err; } + +static struct media_index_result result(int count) { + struct media_index_result r = {.file_size = count, .block_size = 1, .num_blocks = count}; + memset(r.media_id, 0xAA, 16); + r.block_ids = u_calloc(count, 16); r.block_sigs = u_calloc(count, 64); REQUIRE(r.block_ids && r.block_sigs); + for (int i = 0; i < count; i++) { uint32_t id = (uint32_t)i + 1; memcpy(r.block_ids + i * 16, &id, 4); } + return r; +} + +static struct media_download* download(struct UTUN_INSTANCE* inst) { + return inst->md.downloads && inst->md.downloads->head ? (struct media_download*)inst->md.downloads->head->data : NULL; +} + +static void start(struct UTUN_INSTANCE* inst, struct media_index_result* r) { + char path[512]; snprintf(path, sizeof(path), "%s/result.bin", directory); + REQUIRE(!media_download_start(inst, GROUP, r, path, directory, SERVER, completed, NULL, NULL, NULL)); +} + +static void seed(struct UTUN_INSTANCE* inst, const struct media_index_result* r, int count, int holders, int distinct) { + sqlite3_stmt* st = NULL; + REQUIRE(sqlite3_prepare_v2(inst->topo_sqlite_db, + "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,111,?,?,1,1)", -1, &st, NULL) == SQLITE_OK); + for (int i = 0; i < count; i++) for (int h = 0; h < holders; h++) { + REQUIRE(sqlite3_bind_blob(st, 1, r->block_ids + i * 16, 16, SQLITE_STATIC) == SQLITE_OK); + REQUIRE(sqlite3_bind_int64(st, 2, 1000 + (distinct ? i % distinct : h)) == SQLITE_OK); + REQUIRE(sqlite3_bind_int(st, 3, i) == SQLITE_OK && sqlite3_step(st) == SQLITE_DONE); + REQUIRE(sqlite3_reset(st) == SQLITE_OK); + } + REQUIRE(sqlite3_finalize(st) == SQLITE_OK); +} + +static void reset(void) { + packet_count = query_count = request_count = done_count = done_err = fail_send = fail_rng = 0; +} + +static struct packet answer(struct UTUN_INSTANCE* server) { + struct packet query = take_packet(); REQUIRE(query.data[0] == MEDIA_SUBCMD_QUERY); + receive(server, &query); + struct packet response = take_packet(); REQUIRE(response.data[0] == MEDIA_SUBCMD_QUERY_RESP); + return response; +} + +static void test_bounds_and_identity(void) { + reset(); struct UTUN_INSTANCE* s = instance(SERVER); struct UTUN_INSTANCE* c = instance(CLIENT); + struct media_index_result r = result(2); seed(s, &r, 2, 2, 0); start(c, &r); + struct packet q = packets[0]; packet_count = 0; + for (size_t n = 0; n < q.len; n++) { struct packet bad = q; bad.len = n; receive(s, &bad); REQUIRE(!packet_count); } + struct packet bad = q; ((struct media_pkt_query*)bad.data)->num_blocks = MEDIA_QUERY_MAX_BLOCKS + 1; + receive(s, &bad); REQUIRE(!packet_count); + bad = q; bad.len++; receive(s, &bad); REQUIRE(!packet_count); + 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++) { + bad = good; bad.len = n; receive(c, &bad); + REQUIRE(!memcmp(&snapshot, download(c), sizeof(snapshot)) && !memcmp(blocks, download(c)->blocks, sizeof(blocks))); + } + for (int mutation = 0; mutation < 12; mutation++) { + bad = good; + struct media_pkt_query_resp* header = (struct media_pkt_query_resp*)bad.data; + struct media_pkt_query_resp_entry* e = (struct media_pkt_query_resp_entry*)(bad.data + MEDIA_QUERY_RESP_HDR_SIZE); + switch (mutation) { + case 0: bad.from++; break; + case 1: bad.group++; break; + case 2: header->media_id[0]++; break; + case 3: header->request_id[0]++; break; + case 4: header->num_blocks = 1; bad.len -= sizeof(*e); break; + case 5: e[1].block_id[0]++; break; // Ошибка последней записи не меняет первую. + case 6: e[1].holders[1] = e[1].holders[0]; break; + case 7: e[1].holders[0] = CLIENT; break; + case 8: e[1].num_holders = 5; break; + case 9: header->status = MEDIA_QUERY_DB_ERROR; break; + case 10: e[1].holders[2] = 1; break; + case 11: bad.len++; break; + } + receive(c, &bad); + REQUIRE(!memcmp(&snapshot, download(c), sizeof(snapshot)) && !memcmp(blocks, download(c)->blocks, sizeof(blocks))); + } + receive(c, &good); REQUIRE(download(c)->inflight == 2 && download(c)->blocks[1].num_holders == 2); + snapshot = *download(c); memcpy(blocks, download(c)->blocks, sizeof(blocks)); + receive(c, &good); REQUIRE(!memcmp(&snapshot, download(c), sizeof(snapshot)) && !memcmp(blocks, download(c)->blocks, sizeof(blocks))); + destroy(c); destroy(s); media_index_result_free(&r); REQUIRE(done_count == 1 && done_err == -2); + puts("QUERY bounds, atomic validation, identity and duplicate: PASS"); +} + +static void test_paging(void) { + reset(); struct UTUN_INSTANCE* s = instance(SERVER); struct UTUN_INSTANCE* c = instance(CLIENT); + struct media_index_result r = result(65); seed(s, &r, 65, 6, 0); start(c, &r); + int covered = 0, replies = 0; + while (packet_count) { + struct packet p = answer(s); + struct media_pkt_query_resp* header = (struct media_pkt_query_resp*)p.data; + REQUIRE(header->status == MEDIA_QUERY_OK && p.len <= 1604 && header->num_blocks > 0); + covered += header->num_blocks; replies++; + receive(c, &p); + } + REQUIRE(covered == 65 && replies == 3 && query_count == 3 && !done_count); + struct media_download* dl = download(c); REQUIRE(dl && dl->inflight == 65); + for (int i = 0; i < 65; i++) REQUIRE(dl->blocks[i].num_holders == 4 && dl->blocks[i].state == MD_BLK_REQ); + destroy(c); destroy(s); media_index_result_free(&r); REQUIRE(done_count == 1); + puts("65 blocks, 260 holder records, complete bounded batches: PASS"); +} + +/* Нулевые результаты ищутся дальше, в то время как найденные блоки уже скачиваются. */ +static void test_partial_and_timeout(void) { + reset(); struct UTUN_INSTANCE* s = instance(SERVER); struct UTUN_INSTANCE* c = instance(CLIENT); + struct media_index_result r = result(2); seed(s, &r, 1, 1, 0); start(c, &r); + struct media_download* dl = download(c); + dl->super_nodes[1] = SERVER + 1; dl->super_nodes[2] = SERVER + 2; dl->super_count = 3; + struct packet p = answer(s); receive(c, &p); + REQUIRE(dl->blocks[0].state == MD_BLK_REQ && dl->blocks[1].num_holders == 0); + REQUIRE(dl->query_node == SERVER + 1 && dl->query_count == 1 && dl->query_blocks[0] == 1); + REQUIRE(dl->announce_node == SERVER && dl->inflight == 1 && !dl->blocks[1].author_tried); + uint64_t deadline = dl->query_deadline_tb; + dl->blocks[0].last_progress_tb = deadline - 1; dl->last_activity_tb = deadline - 1; + REQUIRE(!md_dl_check_stall(dl, deadline)); + REQUIRE(dl->query_node == SERVER + 2 && dl->query_deadline_tb > deadline && dl->inflight == 1); + REQUIRE(dl->blocks[0].attempts == 0 && dl->announce_node == SERVER); + destroy(c); destroy(s); media_index_result_free(&r); REQUIRE(done_count == 1); + puts("partial results and QUERY timeout during another block's progress: PASS"); +} + +static void deliver_block(struct UTUN_INSTANCE* c, struct media_download* dl, int bi) { + struct packet p = {.from = 1000, .to = CLIENT, .group = GROUP, .len = MEDIA_BLOCK_CHUNK_HDR_SIZE + 1}; + struct media_pkt_block_chunk* chunk = (struct media_pkt_block_chunk*)p.data; + chunk->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; chunk->chunk = bi; chunk->data_len = 1; + memcpy(chunk->media_id, dl->media_id, 16); memcpy(chunk->block_id, dl->blocks[bi].block_id, 16); + p.data[MEDIA_BLOCK_CHUNK_HDR_SIZE] = (uint8_t)bi; receive(c, &p); + memset(p.data, 0, sizeof(p.data)); p.len = MEDIA_BLOCK_DONE_SIZE; + struct media_pkt_block_done* done = (struct media_pkt_block_done*)p.data; + done->subcmd = MEDIA_SUBCMD_BLOCK_DONE; done->chunk = bi; done->total_size = 1; + memcpy(done->media_id, dl->media_id, 16); memcpy(done->block_id, dl->blocks[bi].block_id, 16); + receive(c, &p); +} + +static void test_peer_reuse_and_completion(void) { + reset(); struct UTUN_INSTANCE* s = instance(SERVER); struct UTUN_INSTANCE* c = instance(CLIENT); + struct media_index_result r = result(65); seed(s, &r, 65, 1, 13); start(c, &r); + while (packet_count) { struct packet p = answer(s); receive(c, &p); } + REQUIRE(download(c)->num_peers == MD_MAX_PEERS && download(c)->blocks[10].wait_reason == MD_WAIT_SLOT); + int received = 0; + while (download(c)) { + struct media_download* dl = download(c); int bi = -1, inflight = 0; + REQUIRE(dl->num_peers <= MD_MAX_PEERS && !done_count); + for (int i = 0; i < dl->num_blocks; i++) { + if (dl->blocks[i].state == MD_BLK_REQ) { bi = i; inflight++; } + REQUIRE(dl->blocks[i].attempts == 0); + } + REQUIRE(bi >= 0 && inflight == dl->inflight && received < 65); + deliver_block(c, dl, bi); received++; + } + REQUIRE(received == 65 && done_count == 1 && !done_err); + char path[512]; snprintf(path, sizeof(path), "%s/result.bin", directory); + FILE* f = fopen(path, "rb"); REQUIRE(f); + for (int i = 0; i < 65; i++) REQUIRE(fgetc(f) == i); + REQUIRE(fgetc(f) == EOF && !fclose(f)); + sqlite3_stmt* st = NULL; + REQUIRE(sqlite3_prepare_v2(c->topo_sqlite_db, "SELECT COUNT(*) FROM media_files", -1, &st, NULL) == SQLITE_OK); + REQUIRE(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 0) == 65); sqlite3_finalize(st); + destroy(c); destroy(s); media_index_result_free(&r); REQUIRE(done_count == 1); remove(path); + puts("65 blocks from 13 holders, peer slot reuse, file contents and registration: PASS"); +} + +static int denied_reads; +static int deny_second_block(void* arg, int action, const char* table, const char* column, const char* db, const char* trigger) { + (void)arg; (void)db; (void)trigger; + if (action == SQLITE_READ && table && column && !strcmp(table, "block_availability") && !strcmp(column, "block_uuid")) + if (++denied_reads == 2) return SQLITE_DENY; + return SQLITE_OK; +} + +static void test_failures_and_restart(void) { + reset(); struct UTUN_INSTANCE* s = instance(SERVER); struct UTUN_INSTANCE* c = instance(CLIENT); + struct media_index_result r = result(2); seed(s, &r, 2, 2, 0); start(c, &r); + denied_reads = 0; sqlite3_set_authorizer(s->topo_sqlite_db, deny_second_block, NULL); + struct packet error = answer(s); + struct media_pkt_query_resp* header = (struct media_pkt_query_resp*)error.data; + REQUIRE(denied_reads == 2 && header->status == MEDIA_QUERY_DB_ERROR && !header->num_blocks && error.len == MEDIA_QUERY_RESP_HDR_SIZE); + sqlite3_set_authorizer(s->topo_sqlite_db, NULL, NULL); + struct media_download* dl = download(c); + REQUIRE(!media_download_cancel(c, dl->media_id, NULL, 0) && done_count == 1); + start(c, &r); dl = download(c); struct media_download snapshot = *dl; + receive(c, &error); REQUIRE(!memcmp(&snapshot, dl, sizeof(snapshot)) && done_count == 1); + fail_send = MEDIA_SUBCMD_QUERY_RESP; + struct packet q = take_packet(); receive(s, &q); REQUIRE(!packet_count); fail_send = 0; + REQUIRE(!media_download_cancel(c, dl->media_id, NULL, 0) && done_count == 2); + fail_rng = 1; start(c, &r); fail_rng = 0; + REQUIRE(!download(c) && done_count == 3 && done_err == -1 && !packet_count); + fail_send = MEDIA_SUBCMD_QUERY; start(c, &r); fail_send = 0; + dl = download(c); REQUIRE(dl && !dl->query_deadline_tb && dl->inflight == 2); + destroy(c); destroy(s); media_index_result_free(&r); REQUIRE(done_count == 4); + puts("SQL error after partial lookup, cancel/restart, stale response, RNG and send ownership: PASS"); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG); + ua = uasync_create(); REQUIRE(ua); + snprintf(directory, sizeof(directory), "/tmp/utun-query-XXXXXX"); REQUIRE(!test_mkdtemp(directory)); + test_bounds_and_identity(); test_paging(); test_partial_and_timeout(); + test_peer_reuse_and_completion(); test_failures_and_restart(); + char path[512]; snprintf(path, sizeof(path), "%s/result.bin", directory); remove(path); + for (int i = 0; i < 65; i++) { snprintf(path, sizeof(path), "%s/result.bin.chunk_%d", directory, i); remove(path); } + REQUIRE(!rmdir(directory)); uasync_destroy(ua, 0); + puts("media QUERY regressions: PASS"); return 0; +}