diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index f72fd740..da452b4a 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -393,7 +393,7 @@ static void md_dl_pump_cb(void* arg) { /* Только суперузлы запрошенной CHAT-группы; автор всегда имеет отдельное последнее место. */ static void md_dl_collect_supernodes(struct UTUN_INSTANCE* inst, struct media_download* dl) { - dl->super_count = 0; dl->super_current = 0; + dl->super_count = 0; struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, dl->group_id) : NULL; if (g && g->group_type == TOPO_GROUP_TYPE_CHAT && g->channel_id[0] && inst->topo_sqlite_db) { char sql[256]; @@ -446,7 +446,7 @@ static int md_dl_query_pump(struct media_download* dl, uint64_t now_tb) { if (media_index_generate_uuid(dl->query_id)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: QUERY UUID failed", MDL_ID); return -1; } - dl->super_current = source; dl->query_node = dl->super_nodes[source]; + dl->query_node = dl->super_nodes[source]; uint8_t packet[MEDIA_QUERY_HDR_SIZE + MEDIA_QUERY_MAX_BLOCKS * 16] = {0}; struct media_pkt_query* q = (struct media_pkt_query*)packet; q->subcmd = MEDIA_SUBCMD_QUERY; q->num_blocks = (uint16_t)dl->query_count; @@ -556,8 +556,7 @@ void media_download_handle_chunk(struct UTUN_INSTANCE* inst, uint32_t off = ch->offset; uint32_t end = off + (uint32_t)wlen; if (end > b->bytes_received) b->bytes_received = end; - dl->last_activity_tb = get_time_tb(); - b->last_progress_tb = dl->last_activity_tb; + b->last_progress_tb = get_time_tb(); if (b->state == MD_BLK_REQ || b->state == MD_BLK_CONN) b->state = MD_BLK_RECV; /* rate-limited progress: log at block start, every 1MB, and when block data is complete */ @@ -641,7 +640,6 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst, struct media_download_block* b = &dl->blocks[bi]; if (b->state == MD_BLK_DONE) return; - dl->last_activity_tb = get_time_tb(); /* verify block data: Ed25519 signature from author, or size-only fallback */ int sig_ok = 0; @@ -914,7 +912,6 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, dl = (struct media_download*)qe->data; md_dl_collect_supernodes(inst, dl); - dl->last_activity_tb = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: download started media=%02x%02x... blocks=%d supers=%d", MDL_ID, dl->media_id[0], dl->media_id[1], dl->num_blocks, dl->super_count); diff --git a/src/media_delivery/media_download.h b/src/media_delivery/media_download.h index 79bf91ee..25b021e7 100644 --- a/src/media_delivery/media_download.h +++ b/src/media_delivery/media_download.h @@ -2,7 +2,10 @@ * Состояние хранит inst->md. При открытии подключения через conn_mgr загрузчик получает CM handle. * Найденный ETCP_CONN используется без нового handle; при отсутствии conn_mgr запрос отправляется best-effort без handle. * Старт копирует метаданные result; callbacks, обработчики и файловые операции выполняются в uasync-потоке. - * Failover/таймер простоя ограничивают повторы; done сообщает результат, а не гарантирует доставку другим участникам. */ + * QUERY: одна пачка до 32 блоков в полёте, отдельный UUID/дедлайн, источники только своей группы и автор. + * До четырёх кандидатов на блок; до десяти используемых узлов, слоты переиспользуются после освобождения. + * Повторы ограничены источниками и бюджетом провалов блока; OVERLOADED соблюдает retry_after. + * done сообщает результат, а не гарантирует доставку другим участникам. */ #ifndef MEDIA_DOWNLOAD_H #define MEDIA_DOWNLOAD_H @@ -33,7 +36,7 @@ enum md_block_state { MD_BLK_CONN, /* conn_mgr_open в процессе, BLOCK_REQ ещё не отправлен */ MD_BLK_REQ, /* BLOCK_REQ отправлен, ждём первый чанк */ MD_BLK_RECV, /* чанки идут */ - MD_BLK_WAIT, /* перегружен/нет свободного держателя — ретрай позже */ + MD_BLK_WAIT, /* Причина в wait_reason: поиск, занятые слоты или retry_at_tb. */ MD_BLK_DONE, /* BLOCK_DONE + сигнатура валидна */ MD_BLK_FAILED /* попытки исчерпаны */ }; @@ -83,7 +86,6 @@ struct media_download { int err; uint64_t super_nodes[10]; int super_count; - int super_current; uint64_t announce_node; /* Адресат HAVE_BLOCK, независимый от текущего QUERY. */ uint8_t query_id[16]; /* UUID ожидающей попытки; поздние ответы игнорируются. */ uint64_t query_node; @@ -92,8 +94,7 @@ struct media_download { int query_blocks[MEDIA_QUERY_MAX_BLOCKS]; /* Индексы пачки в blocks, в порядке QUERY. */ void* pump_timer; /* Отложенная обработка CM; отменяется до освобождения dl. */ uint64_t author_node_id; /* fallback, когда у суперноды нет инфо */ - void* watchdog_timer; /* таймер простоя (2с) */ - uint64_t last_activity_tb; /* любая входящая активность (chunk/done/query_resp) */ + void* watchdog_timer; /* Ближайший QUERY/простой блока/retry_at; отменяется до done_cb. */ int num_peers; struct media_download_peer peers[MD_MAX_PEERS]; void (*done_cb)(void* arg, int err); @@ -120,6 +121,9 @@ int media_download_cancel(struct UTUN_INSTANCE* inst, const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk); /* ── handlers for incoming data, called from md_etcp_recv dispatch ── */ +/* Полный ответ пачки. group_id/from_node — из router; несовпадающие/истёкшие попытки игнорируются. + * Повреждённый пакет не меняет состояние; ошибка поиска продолжает обход источников. + * Может завершить загрузку и вызвать done_cb; после возврата прежний dl может быть освобождён. */ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t from_node, const uint8_t* data, size_t len); void media_download_handle_chunk(struct UTUN_INSTANCE* inst, @@ -139,9 +143,9 @@ void md_dl_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_i enum conn_mgr_event event, void* arg); /* ── тестовые хуки (для детерминированных юнит-тестов с «фейковым» временем) ── - * md_dl_check_stall — проверка простоя: возвращает 0 (активна) / -1 (провал, нужно finish). + * md_dl_check_stall — обработать QUERY/простой/ожидание, запустить готовые блоки; 0 / -1 (нужен finish). * md_dl_failover_block — переход блока к следующему держателю: 0, включая DONE/FAILED и ожидание новых держателей; - * -1 — исчерпаны попытки или не удалось начать запрос (лимит peers, ошибка conn_mgr_open). + * -1 — исчерпаны попытки или локальная ошибка; занятые peer-слоты возвращают 0. * md_dl_dump — дамп состояния загрузки в лог (DEBUG_CATEGORY_MEDIA). */ int md_dl_check_stall(struct media_download* dl, uint64_t now_tb); int md_dl_failover_block(struct media_download* dl, int bi, uint64_t now_tb); diff --git a/tests/test_media_delivery_integration.c b/tests/test_media_delivery_integration.c index 358d8a04..6638f30f 100644 --- a/tests/test_media_delivery_integration.c +++ b/tests/test_media_delivery_integration.c @@ -118,10 +118,34 @@ static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); int rc = etcp_route_send(inst, TOPO_GROUP_UTUN, dst, e, 1, 0); - if (rc != 0) { u_free(e->dgram); queue_entry_free(e); } return rc; } +static int query_received, query_valid; +static void (*saved_media_recv)(struct ETCP_CONN*, struct ll_entry*); + +/* Перехватываем только ожидаемый ответ; прочие пакеты обрабатывает настоящий сервис. */ +static void query_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { + if (entry->len < ROUTER_SVC_PAYLOAD_OFF + 1 || entry->dgram[ROUTER_SVC_PAYLOAD_OFF] != MEDIA_SUBCMD_QUERY_RESP) { + saved_media_recv(conn, entry); return; + } + query_received++; + size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF; + uint64_t source, group; + memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8); memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); + if (len == MEDIA_QUERY_RESP_HDR_SIZE + sizeof(struct media_pkt_query_resp_entry)) { + const struct media_pkt_query_resp* r = (const struct media_pkt_query_resp*)(entry->dgram + ROUTER_SVC_PAYLOAD_OFF); + const struct media_pkt_query_resp_entry* e = (const struct media_pkt_query_resp_entry*)((const uint8_t*)r + MEDIA_QUERY_RESP_HDR_SIZE); + uint8_t block[16], request[16]; + for (int i = 0; i < 16; i++) block[i] = (uint8_t)(0x10 + i); + memset(request, 0x91, 16); + query_valid = source == g_nid_a && group == TOPO_GROUP_UTUN && r->status == MEDIA_QUERY_OK && r->num_blocks == 1 + && !memcmp(r->request_id, request, 16) && !memcmp(r->media_id, block, 16) && !memcmp(e->block_id, block, 16) + && e->num_holders == 1 && e->holders[0] == g_nid_a; + } + queue_dgram_free(entry); queue_entry_free(entry); +} + /* ── tests ── */ static void phase1_serve_reg(void) { @@ -181,12 +205,15 @@ static void phase3_query(void) { memcpy(q->media_id, block_uuid, 16); q->num_blocks = 1; memcpy(pkt + MEDIA_QUERY_HDR_SIZE, block_uuid, 16); - - msend(g_b, g_nid_a, pkt, sizeof(pkt)); - /* QUERY_RESP arrives asynchronously — poll */ - int ok = 0, poll_count = 0; - while (poll_count < 200) { uasync_poll(g_ua, POLL_MS); poll_count++; } - (void)ok; OK(); /* smoke test: doesn't crash */ + media_delivery_send_have_block(g_a, TOPO_GROUP_UTUN, g_nid_a, block_uuid, block_uuid, 0); + saved_media_recv = g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY]; + g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY] = query_recv; + query_received = query_valid = 0; + int rc = msend(g_b, g_nid_a, pkt, sizeof(pkt)), poll_count = 0; + while (!query_received && poll_count++ < 2000) uasync_poll(g_ua, POLL_MS); + g_b->router_bindings.callbacks[ETCP_RT_ID_MEDIA_DELIVERY] = saved_media_recv; + if (!rc && query_received == 1 && query_valid) OK(); + else FAIL("send=%d received=%d valid=%d", rc, query_received, query_valid); } }