Document media lookup lifetimes and remove obsolete shared state

This commit is contained in:
evgeny committed 2026-10-22 18:31:12 +02:00
1 parent 093e537fed
commit a3f3b2fff2
3 files changed
+48 -20

No files matched your search

+3 -6
View File
@@ -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);
+11 -7
View File
@@ -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);
+34 -7
View File
@@ -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);
}
}