Check live QUERY deadlines and deferred CM failover

tests/test_media_delivery_download.c

@@ -20,6 +20,7 @@
 #include "../lib/u_async.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
+#include "../lib/platform_compat.h"
 #include "../transport_layer/secure_channel.h"
 #include "../routing_layer/topo_node_sqlite.h"
 #include "../routing_layer/conn_mgr.h"
@@ -30,7 +31,6 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <unistd.h>
 #include <sys/stat.h>
 
 /* Доставка принята транспортом, ответа нет. Владение entry совпадает с реальным router во всех исходах. */
@@ -149,6 +149,7 @@ static void free_test_dl(struct UTUN_INSTANCE* inst, struct media_download* dl)
     if (inst->md.downloads) {
         struct ll_entry* e = queue_find_data_by_index(inst->md.downloads, dl->media_id);
         if (e) { queue_remove_data(inst->md.downloads, e); queue_entry_free(e); }
+        if (!inst->md.downloads->head) { queue_free(inst->md.downloads); inst->md.downloads = NULL; }
     }
 }
 
@@ -426,7 +427,7 @@ static void test_relay_full(void) {
 /* ── 10. пустой QUERY_RESP ── */
 
 static void test_query_resp_empty(void) {
-    TEST("QUERY_RESP empty → no crash"); {
+    TEST("QUERY_RESP empty result → author fallback"); {
         sqlite3* db = NULL; sqlite3_open(":memory:", &db);
         struct UTUN_INSTANCE* inst = make_minimal_instance(db);
         struct media_download* dl = make_test_dl(db, inst, 1);
@@ -525,10 +526,29 @@ static void test_cm_handle_lifetime(void) {
         && dl->blocks[1].state == MD_BLK_CONN && dl->blocks[0].holder_idx == 0 && dl->blocks[1].holder_idx == 0;
     md_dl_conn_cb(h, 0x1111, dl->group_id, CONN_EVENT_TIMEOUT, dl);
     int closed = h->closed && !h->deliver_up_token && !dl->peers[0].cm_handle;
-    if (!h->closed) conn_mgr_close(h);
+    uasync_poll(g_ua, 0);
+    int processed = !dl->pump_timer && dl->blocks[0].state == MD_BLK_REQ && dl->blocks[1].state == MD_BLK_REQ;
     free_test_dl(inst, dl); sqlite3_close(db); u_free(inst);
     uasync_poll(g_ua, 0);
-    if (retained && closed && !late) OK(); else FAIL("retained=%d closed=%d late=%d", retained, closed, late);
+    if (retained && closed && processed && !late) OK();
+    else FAIL("retained=%d closed=%d processed=%d late=%d", retained, closed, processed, late);
+}
+
+/* Вызов pump из другого события не должен обходить срок OVERLOADED. */
+static void test_overloaded_deadline(void) {
+    TEST("pump preserves OVERLOADED retry deadline");
+    sqlite3* db = NULL; sqlite3_open(":memory:", &db);
+    struct UTUN_INSTANCE* inst = make_minimal_instance(db);
+    struct media_download* dl = make_test_dl(db, inst, 1);
+    add_holder(dl, 0, 0x1111); dl->blocks[0].holder_idx = 0; dl->blocks[0].state = MD_BLK_REQ; dl->inflight = 1;
+    struct media_pkt_block_overloaded ov = {.subcmd = MEDIA_SUBCMD_BLOCK_OVERLOADED, .retry_after_ms = 1000};
+    memcpy(ov.media_id, dl->media_id, 16); memcpy(ov.block_id, dl->blocks[0].block_id, 16);
+    media_download_handle_overloaded(inst, (const uint8_t*)&ov, sizeof(ov));
+    int rc = md_dl_check_stall(dl, get_time_tb());
+    int ok = !rc && dl->blocks[0].state == MD_BLK_WAIT && dl->blocks[0].wait_reason == MD_WAIT_RETRY
+        && dl->blocks[0].retry_at_tb > get_time_tb() && !dl->blocks[0].attempts && !dl->inflight;
+    free_test_dl(inst, dl); sqlite3_close(db); u_free(inst);
+    if (ok) OK(); else FAIL("retry deadline was bypassed");
 }
 
 int main(void) {
@@ -548,6 +568,7 @@ int main(void) {
     test_query_resp_assign();
     test_dedup();
     test_cm_handle_lifetime();
+    test_overloaded_deadline();
 
     printf("\n%d/%d passed, %d failed\n", g_passed, g_total, g_failed);
     test_cleanup();

tests/test_media_query.c

@@ -264,6 +264,34 @@ static void test_peer_reuse_and_completion(void) {
     puts("65 blocks from 13 holders, peer slot reuse, file contents and registration: PASS");
 }
 
+/* Размер файла больше прежнего предела одного QUERY; в транспорт всё равно уходит только 32 ID. */
+static void test_large_file_and_live_deadline(void) {
+    reset(); struct UTUN_INSTANCE* c = instance(CLIENT);
+    struct media_index_result large = result(4100); start(c, &large);
+    REQUIRE(download(c) && download(c)->num_blocks == 4100 && download(c)->query_count == 32);
+    REQUIRE(packet_count == 1 && packets[0].len == MEDIA_QUERY_HDR_SIZE + 32 * 16);
+    destroy(c); media_index_result_free(&large); REQUIRE(done_count == 1);
+
+    reset(); struct UTUN_INSTANCE* s = instance(SERVER); c = instance(CLIENT);
+    c->md.query_timeout_tb = 1000; c->md.dl_stall_timeout_tb = 10000;
+    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_count = 2;
+    struct packet p = answer(s); receive(c, &p);
+    REQUIRE(dl->query_node == SERVER + 1 && dl->query_deadline_tb);
+    struct packet chunk = {.from = 1000, .to = CLIENT, .group = GROUP, .len = MEDIA_BLOCK_CHUNK_HDR_SIZE + 1};
+    struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)chunk.data;
+    ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; ch->data_len = 1;
+    memcpy(ch->media_id, r.media_id, 16); memcpy(ch->block_id, r.block_ids, 16);
+    uint64_t stop = get_time_tb() + 5000;
+    while (dl->query_deadline_tb && get_time_tb() < stop) {
+        receive(c, &chunk); uasync_poll(ua, 5); REQUIRE(download(c) == dl && !done_count);
+    }
+    REQUIRE(!dl->query_deadline_tb && dl->blocks[0].state == MD_BLK_RECV && !dl->blocks[0].attempts);
+    REQUIRE(dl->blocks[1].state == MD_BLK_REQ && dl->blocks[1].holders[0] == SERVER);
+    destroy(c); destroy(s); media_index_result_free(&r); uasync_poll(ua, 20); REQUIRE(done_count == 1);
+    puts("4100-block start and real QUERY watchdog during incoming chunks: PASS");
+}
+
 static int denied_reads, started_selects;
 static int trace_query(unsigned event, void* arg, void* statement, void* sql) {
     (void)arg; (void)sql;
@@ -310,7 +338,7 @@ int main(void) {
     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();
+    test_peer_reuse_and_completion(); test_large_file_and_live_deadline(); 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);