diff --git a/src/chat/chat_msg.c b/src/chat/chat_msg.c index 80ade16e..f7c47073 100644 --- a/src/chat/chat_msg.c +++ b/src/chat/chat_msg.c @@ -158,6 +158,7 @@ void chat_core_submit_trampoline(void* arg) { struct media_submit_ctx { struct media_submit_ctx* next; + struct media_destination* destination; // Общая с загрузчиком резервация пути до завершения workers. int delete_pending; struct UTUN_INSTANCE* inst; char channel_id[64]; @@ -185,6 +186,7 @@ static void media_submit_finish(struct media_submit_ctx* ctx, int success) { while (*slot && *slot != ctx) slot = &(*slot)->next; if (*slot) *slot = ctx->next; else DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media submission missing from owner ch=%s", channel); + media_delivery_release(inst, ctx->destination); u_free(ctx); if (delete_pending) { for (struct media_submit_ctx* pending = cc->media_submissions; pending; pending = pending->next) @@ -435,6 +437,8 @@ void chat_core_submit_media_message(struct UTUN_INSTANCE* inst, struct chat_msg_ DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: media context allocation failed", CC_ID); chat_event_submitted(inst, req->request_id, 0); return; } + mctx->destination = media_delivery_reserve(inst, destination); + if (!mctx->destination) { u_free(mctx); chat_event_submitted(inst, req->request_id, 0); return; } mctx->inst = inst; mctx->request_id = req->request_id; mctx->next = cc->media_submissions; cc->media_submissions = mctx; diff --git a/src/media_delivery/media_download.c b/src/media_delivery/media_download.c index 67f683c6..2b7d0efa 100644 --- a/src/media_delivery/media_download.c +++ b/src/media_delivery/media_download.c @@ -194,6 +194,20 @@ static void md_dl_send_block_processing(struct UTUN_INSTANCE* inst, struct media /* ── запрос блока у текущего держателя ── */ +static void md_dl_end_attempt(struct media_download* dl, int bi) { + struct media_download_block* b = &dl->blocks[bi]; + if ((b->state == MD_BLK_REQ || b->state == MD_BLK_RECV) && b->holder_idx >= 0 && b->holder_idx < b->num_holders) { + struct media_pkt_cancel can = {.subcmd = MEDIA_SUBCMD_CANCEL, .chunk = (uint32_t)bi}; + memcpy(can.media_id, dl->media_id, 16); memcpy(can.block_id, b->block_id, 16); + memcpy(can.transfer_id, b->transfer_id, 16); + md_dl_send(dl->inst, dl->group_id, b->holders[b->holder_idx], (const uint8_t*)&can, sizeof(can)); + } + if (b->source) { + if (!b->source->verified) md_block_abort(b->source); + md_block_release(b->source); b->source = NULL; + } +} + static int md_dl_request_block(struct media_download* dl, int bi) { struct media_download_block* b = &dl->blocks[bi]; uint64_t now = get_time_tb(); @@ -246,6 +260,7 @@ static int md_dl_pump(struct media_download* dl) { int md_dl_failover_block(struct media_download* dl, int bi, uint64_t now_tb) { struct media_download_block* b = &dl->blocks[bi]; if (b->state == MD_BLK_DONE || b->state == MD_BLK_FAILED) return 0; + md_dl_end_attempt(dl, bi); if (md_dl_block_inflight(b) && dl->inflight > 0) dl->inflight--; b->state = MD_BLK_IDLE; b->bytes_received = 0; b->last_progress_tb = now_tb; b->retry_at_tb = 0; do { @@ -310,12 +325,14 @@ static void md_dl_finish(struct media_download* dl, int err) { dl->inflight, dl->num_peers, dl->query_count); dl->active = 0; dl->err = err; + queue_remove_data(md->downloads, dl->entry); if (dl->pump_timer) { uasync_call_soon_cancel(inst->ua, dl->pump_timer); dl->pump_timer = NULL; } if (dl->watchdog_timer) { uasync_cancel_timeout(inst->ua, dl->watchdog_timer); dl->watchdog_timer = NULL; } for (int pi = 0; pi < dl->num_peers; pi++) { if (dl->peers[pi].cm_handle) { conn_mgr_close(dl->peers[pi].cm_handle); dl->peers[pi].cm_handle = NULL; } } - queue_remove_data(md->downloads, dl->entry); + for (int bi = 0; bi < dl->num_blocks; bi++) md_dl_end_attempt(dl, bi); + media_delivery_release(inst, dl->destination); dl->destination = NULL; md_dl_release(dl); } @@ -535,93 +552,65 @@ invalid: DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: QUERY_RESP invalid block/holder list", MDL_ID); } -/* ── handle incoming BLOCK_CHUNK ── */ - -void media_download_handle_chunk(struct UTUN_INSTANCE* inst, - const uint8_t* data, size_t len) { - if (!data || len < MEDIA_BLOCK_CHUNK_HDR_SIZE) return; - - const uint8_t* d = data; - struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)data; - struct media_download* dl = md_dl_find(inst, ch->media_id); - if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: CHUNK for unknown media %02x%02x...", MDL_ID, ch->media_id[0], ch->media_id[1]); return; } - if (!dl->active) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: CHUNK for inactive download", MDL_ID); return; } - - int bi = md_dl_block_index(dl, ch->block_id); - if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: CHUNK for unknown block %02x%02x...", MDL_ID, ch->block_id[0], ch->block_id[1]); return; } - - struct media_download_block* b = &dl->blocks[bi]; - size_t wlen = ch->data_len; - - char tmp[2048]; - snprintf(tmp, sizeof(tmp), "%s.chunk_%d", dl->dest_path, bi); - FILE* f = fopen(tmp, "rb+"); - if (!f) f = fopen(tmp, "wb"); - if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed for chunk write: %s", MDL_ID, tmp, strerror(errno)); return; } - if (len >= MEDIA_BLOCK_CHUNK_HDR_SIZE + wlen) { - fseeko(f, (off_t)ch->offset, SEEK_SET); - fwrite(d + MEDIA_BLOCK_CHUNK_HDR_SIZE, 1, wlen, f); - } else { - DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: CHUNK truncated: len=%zu need=%zu+%zu", - MDL_ID, len, (size_t)MEDIA_BLOCK_CHUNK_HDR_SIZE, (size_t)wlen); - } - fclose(f); - - uint32_t off = ch->offset; - uint32_t end = off + (uint32_t)wlen; - if (end > b->bytes_received) b->bytes_received = end; - 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 */ - { - uint32_t mb_off = off >> 20; - uint32_t mb_end = end >> 20; - int64_t blk_size = b->expected_size > 0 ? b->expected_size : dl->block_size; - if (off == 0 || mb_end != mb_off || (blk_size > 0 && end >= (uint32_t)blk_size)) { - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: CHUNK blk=%d off=%u +%zu =%u/%lld (%.0f%%)%s", - MDL_ID, bi, off, wlen, end, (long long)blk_size, - blk_size > 0 ? 100.0 * (double)end / (double)blk_size : 0.0, - (blk_size > 0 && end >= (uint32_t)blk_size) ? " [data complete]" : ""); - } - } - - /* forward to relay downstreams */ - struct relay_block_ctx* rc = md_relay_find(&inst->md, ch->block_id); - if (rc) { - rc->file_offset += wlen; - for (int i = 0; i < rc->downstream_count; i++) { - struct relay_downstream* ds = &rc->downstream[i]; - if (ds->sent_offset >= rc->file_offset) continue; - FILE* rf = fopen(tmp, "rb"); - if (!rf) continue; - while (ds->sent_offset < rc->file_offset) { - fseeko(rf, (off_t)ds->sent_offset, SEEK_SET); - size_t to_read = rc->file_offset - ds->sent_offset; - if (to_read > 1024) to_read = 1024; - uint8_t rbuf[1024]; - size_t rd = fread(rbuf, 1, to_read, rf); - if (rd == 0) break; - uint8_t rpkt[MEDIA_BLOCK_CHUNK_HDR_SIZE + 1024]; - struct media_pkt_block_chunk* rch = (struct media_pkt_block_chunk*)rpkt; - memset(rch, 0, MEDIA_BLOCK_CHUNK_HDR_SIZE); - rch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; - memcpy(rch->media_id, ch->media_id, 16); - memcpy(rch->block_id, ch->block_id, 16); - rch->chunk = ch->chunk; - rch->offset = (uint32_t)ds->sent_offset; - rch->data_len = (uint16_t)rd; - memcpy(rpkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, rbuf, rd); - if (md_dl_send(inst, ds->group_id, ds->node_id, rpkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd) == 0) - ds->sent_offset += rd; - else break; - } - fclose(rf); - } +/* Проверяем всю принадлежность попытки до доступа к файлу/таймерам. chunk задаёт индекс O(1). */ +static struct media_download* md_dl_match(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, + const uint8_t media_id[16], const uint8_t block_id[16], const uint8_t transfer_id[16], uint32_t chunk) { + struct media_download* dl = md_dl_find(inst, media_id); + if (dl && dl->active && dl->group_id == group && chunk < (uint32_t)dl->num_blocks) { + struct media_download_block* b = &dl->blocks[chunk]; + if ((b->state == MD_BLK_REQ || b->state == MD_BLK_RECV) && b->holder_idx >= 0 + && b->holder_idx < b->num_holders && b->holders[b->holder_idx] == peer + && !memcmp(b->block_id, block_id, 16) && !memcmp(b->transfer_id, transfer_id, 16)) return dl; } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: stale block packet group=%016llx peer=%016llx chunk=%u attempt=%02x%02x", + MDL_ID, (unsigned long long)group, (unsigned long long)peer, chunk, transfer_id[0], transfer_id[1]); + return NULL; } -/* ── handle incoming BLOCK_DONE ── */ +static void md_dl_reject_block(struct media_download* dl, int bi) { + if (md_dl_failover_block(dl, bi, get_time_tb()) < 0 || md_dl_query_pump(dl, get_time_tb()) < 0 + || md_dl_pump(dl) < 0 || md_dl_arm_watchdog(dl) < 0) md_dl_finish(dl, -1); +} + +void media_download_handle_chunk(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (!data || len < MEDIA_BLOCK_CHUNK_HDR_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: truncated CHUNK header bytes=%zu", MDL_ID, len); return; + } + const struct media_pkt_block_chunk* ch = (const struct media_pkt_block_chunk*)data; + struct media_download* dl = md_dl_match(inst, group_id, from_node, ch->media_id, ch->block_id, ch->transfer_id, ch->chunk); + if (!dl) return; + struct media_download_block* b = &dl->blocks[ch->chunk]; + if (!ch->data_len || ch->data_len > MEDIA_BLOCK_DATA_MAX || len != sizeof(*ch) + ch->data_len + || ch->offset > b->expected_size || ch->data_len > b->expected_size - ch->offset) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: invalid CHUNK block=%u bytes=%zu offset=%u length=%u expected=%u", + MDL_ID, ch->chunk, len, ch->offset, ch->data_len, b->expected_size); + md_dl_reject_block(dl, (int)ch->chunk); return; + } + uint32_t end = ch->offset + ch->data_len; + if (end <= b->bytes_received) return; // Полный дубль не продлевает таймаут и не публикует байты заново. + if (ch->offset != b->bytes_received) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: CHUNK gap/overlap block=%u offset=%u prefix=%u length=%u", + MDL_ID, ch->chunk, ch->offset, b->bytes_received, ch->data_len); + md_dl_reject_block(dl, (int)ch->chunk); return; + } + FILE* file = b->source ? file_io_open_update(b->source->storage.temporary) : NULL; + int failed = !file; + if (file) { + if (fseeko(file, (off_t)ch->offset, SEEK_SET) || fwrite(data + sizeof(*ch), 1, ch->data_len, file) != ch->data_len) + failed = 1; + if (fclose(file)) failed = 1; + } + if (failed) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: CHUNK local write failed block=%u offset=%u length=%u errno=%d", + MDL_ID, ch->chunk, ch->offset, ch->data_len, errno); + md_dl_finish(dl, -1); return; + } + b->bytes_received = end; b->state = MD_BLK_RECV; b->last_progress_tb = get_time_tb(); + md_block_publish(b->source, end, 0); + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: CHUNK committed block=%u prefix=%u/%u attempt=%02x%02x", + MDL_ID, ch->chunk, end, b->expected_size, b->transfer_id[0], b->transfer_id[1]); +} /* Регистрирует скачанные блоки в media_files (location = собранный файл), чтобы узел * мог отдавать их как держатель. Вызывается после успешной сборки и проверки хеша. */ @@ -640,209 +629,158 @@ static int md_dl_register_servable(struct media_download* dl) { u_free(ids); return rc; } -void media_download_handle_done(struct UTUN_INSTANCE* inst, - const uint8_t* data, size_t len) { - if (!data || len < MEDIA_BLOCK_DONE_SIZE) return; - - struct media_pkt_block_done* bd = (struct media_pkt_block_done*)data; - struct media_download* dl = md_dl_find(inst, bd->media_id); - if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE for unknown media", MDL_ID); return; } - if (!dl->active) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE for inactive download", MDL_ID); return; } - - int bi = md_dl_block_index(dl, bd->block_id); - if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE for unknown block", MDL_ID); return; } +/* Запись назначения начинается только после проверки блоков. До publish старое имя сохраняет прежние байты. */ +static int md_dl_assemble(struct media_download* dl) { + struct file_io_temp output = {0}; + int failed = file_io_temp_open(&output, dl->dest_path); + uint64_t total = 0; + uint8_t buffer[65536]; + for (int bi = 0; bi < dl->num_blocks && !failed; bi++) { + struct media_download_block* b = &dl->blocks[bi]; + struct file_io_info info; + if (!b->source || !b->source->verified || file_io_inspect(b->source->storage.temporary, 0, &info) + || info.type != FILE_IO_REGULAR || info.size != b->expected_size) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: assembly source invalid block=%d", MDL_ID, bi); failed = 1; break; + } + FILE* file = file_io_open_read(b->source->storage.temporary); + if (!file) { failed = 1; break; } + uint32_t left = b->expected_size; + while (left && !failed) { + size_t count = left < sizeof(buffer) ? left : sizeof(buffer); + if (fread(buffer, 1, count, file) != count || fwrite(buffer, 1, count, output.file) != count) failed = 1; + else { left -= (uint32_t)count; total += count; } + } + if (fclose(file)) failed = 1; + } + if (!failed && total != (uint64_t)dl->file_size) failed = 1; + if (output.file && file_io_temp_finish(&output)) failed = 1; + uint8_t hash[32]; + if (!failed && (ma_sha256_file(output.temporary, hash) || memcmp(hash, dl->content_hash, 32))) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: assembled content hash mismatch/read error path=%s", MDL_ID, dl->dest_path); + failed = 1; + } + if (!failed && file_io_temp_publish(&output, output.temporary, dl->dest_path)) failed = 1; + file_io_temp_discard(&output); + if (failed) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: assembly failed bytes=%llu/%lld path=%s errno=%d", + MDL_ID, (unsigned long long)total, (long long)dl->file_size, dl->dest_path, errno); return -1; + } + dl->assembled = 1; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: verified file published path=%s bytes=%llu", + MDL_ID, dl->dest_path, (unsigned long long)total); + return md_dl_register_servable(dl); +} +void media_download_handle_done(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (!data || len != MEDIA_BLOCK_DONE_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed DONE bytes=%zu", MDL_ID, len); return; + } + const struct media_pkt_block_done* done = (const struct media_pkt_block_done*)data; + struct media_download* dl = md_dl_match(inst, group_id, from_node, done->media_id, done->block_id, + done->transfer_id, done->chunk); + if (!dl) return; + int bi = (int)done->chunk; struct media_download_block* b = &dl->blocks[bi]; - if (b->state == MD_BLK_DONE) return; - - /* verify block data: Ed25519 signature from author, or size-only fallback */ - int sig_ok = 0; - uint8_t author_ed25519_pubkey[32] = {0}; - if (dl->author_node_id && inst->topo_sqlite_db) { - sqlite3_stmt* st = NULL; - sqlite3_prepare_v2(inst->topo_sqlite_db, - "SELECT ed25519_pubkey FROM nodes WHERE node_id=?", -1, &st, NULL); - if (st) { sqlite3_bind_int64(st, 1, (sqlite3_int64)dl->author_node_id); - if (sqlite3_step(st) == SQLITE_ROW) { const void* pk = sqlite3_column_blob(st, 0); int plen = sqlite3_column_bytes(st, 0); if (pk && plen == 32) memcpy(author_ed25519_pubkey, pk, 32); } - sqlite3_finalize(st); } + if (done->status != MEDIA_BLOCK_OK || done->total_size != b->expected_size || b->bytes_received != b->expected_size) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: incomplete DONE block=%d status=%u wire=%u prefix=%u expected=%u", + MDL_ID, bi, done->status, done->total_size, b->bytes_received, b->expected_size); + md_dl_reject_block(dl, bi); return; } - char tmp[2048]; snprintf(tmp, sizeof(tmp), "%s.chunk_%d", dl->dest_path, bi); - FILE* f = fopen(tmp, "rb"); - if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed", MDL_ID, tmp); } - else { - fseeko(f, 0, SEEK_END); off_t fsz = ftello(f); fseeko(f, 0, SEEK_SET); - if (fsz == (off_t)bd->total_size && fsz > 0) { - if (author_ed25519_pubkey[0]) { - uint8_t* buf = u_malloc((size_t)fsz); - if (buf) { - size_t rd = fread(buf, 1, (size_t)fsz, f); - if (rd == (size_t)fsz) { - EVP_PKEY* apk = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, author_ed25519_pubkey, 32); - EVP_MD_CTX* vctx = apk ? EVP_MD_CTX_new() : NULL; - if (vctx) { - sig_ok = (EVP_DigestVerifyInit(vctx, NULL, NULL, NULL, apk) == 1 - && EVP_DigestVerify(vctx, dl->block_sigs + bi * 64, 64, buf, rd) == 1) ? 1 : 0; - EVP_MD_CTX_free(vctx); - } - if (apk) EVP_PKEY_free(apk); - } - u_free(buf); - } - if (!sig_ok) DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_DONE Ed25519 sig fail block=%d", MDL_ID, bi); - } else { - sig_ok = (memcmp(bd->block_sig, dl->block_sigs + bi * 64, 64) == 0) ? 1 : 0; - } - } else { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_DONE size mismatch: fsz=%lld expected=%u", MDL_ID, (long long)fsz, bd->total_size); } - fclose(f); + struct file_io_info info; + if (!b->source || b->source->failed || file_io_inspect(b->source->storage.temporary, 0, &info) + || info.type != FILE_IO_REGULAR || info.size != b->expected_size) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: DONE local file invalid block=%d", MDL_ID, bi); md_dl_finish(dl, -1); return; } - - if (!sig_ok) { - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_DONE invalid blk=%d — failover", MDL_ID, bi); - if (md_dl_failover_block(dl, bi, get_time_tb()) < 0) { md_dl_finish(dl, -1); return; } - if (md_dl_pump(dl) < 0) md_dl_finish(dl, -1); - return; + uint8_t* buffer = u_malloc(b->expected_size); + FILE* file = buffer ? file_io_open_read(b->source->storage.temporary) : NULL; + int failed = !file; + if (file) { + if (fread(buffer, 1, b->expected_size, file) != b->expected_size) failed = 1; + if (fclose(file)) failed = 1; } - - b->state = MD_BLK_DONE; - if (dl->inflight > 0) dl->inflight--; - dl->blocks_validated++; - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE block=%d total=%d/%d", - MDL_ID, bi, dl->blocks_validated, dl->num_blocks); - + if (failed) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: DONE local read/allocation failed block=%d size=%u errno=%d", + MDL_ID, bi, b->expected_size, errno); + u_free(buffer); md_dl_finish(dl, -1); return; + } + int signature = sc_ed25519_verify(dl->author_pubkey, buffer, b->expected_size, dl->block_sigs + (size_t)bi * 64); + u_free(buffer); + if (signature != SC_OK) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: DONE author signature invalid block=%d rc=%d", MDL_ID, bi, signature); + md_dl_reject_block(dl, bi); return; + } + b->state = MD_BLK_DONE; dl->inflight--; dl->blocks_validated++; + md_block_publish(b->source, b->expected_size, 1); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: block verified block=%d progress=%d/%d", MDL_ID, bi, dl->blocks_validated, dl->num_blocks); dl->refs++; if (dl->progress_cb) dl->progress_cb(dl->progress_arg, dl->blocks_validated, dl->num_blocks); - if (!dl->active) { md_dl_release(dl); return; } - - md_dl_send_have_block(inst, dl, bi); - - /* forward BLOCK_DONE to relay downstreams */ - { - struct relay_block_ctx* rc = md_relay_find(&inst->md, dl->blocks[bi].block_id); - if (rc) { - for (int i = 0; i < rc->downstream_count; i++) { - struct media_pkt_block_done rd; - memset(&rd, 0, sizeof(rd)); - rd.subcmd = MEDIA_SUBCMD_BLOCK_DONE; - memcpy(rd.media_id, dl->media_id, 16); - memcpy(rd.block_id, dl->blocks[bi].block_id, 16); - rd.chunk = (uint32_t)bi; - rd.total_size = (uint32_t)rc->file_offset; - md_dl_send(inst, rc->downstream[i].group_id, rc->downstream[i].node_id, (const uint8_t*)&rd, sizeof(rd)); - } - md_relay_remove(&inst->md, dl->blocks[bi].block_id); - } + if (dl->active) { + md_dl_send_have_block(inst, dl, bi); + if (dl->blocks_validated == dl->num_blocks) md_dl_finish(dl, md_dl_assemble(dl)); + else if (md_dl_pump(dl) < 0) md_dl_finish(dl, -1); } - - /* собрать файл, если все блоки готовы */ - if (dl->blocks_validated == dl->num_blocks) { - FILE* out = fopen(dl->dest_path, "wb"); - if (!out) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed for assembly: %s", MDL_ID, dl->dest_path, strerror(errno)); } - else { - for (int n = 0; n < dl->num_blocks; n++) { - char tmp2[2048]; - snprintf(tmp2, sizeof(tmp2), "%s.chunk_%d", dl->dest_path, n); - FILE* cf = fopen(tmp2, "rb"); - if (cf) { - uint8_t buf[65536]; - size_t rd; - while ((rd = fread(buf, 1, sizeof(buf), cf)) > 0) fwrite(buf, 1, rd, out); - fclose(cf); - remove(tmp2); - } - } - fclose(out); - } - { int hash_zero = 1; for (int hi = 0; hi < 32; hi++) if (dl->content_hash[hi]) { hash_zero = 0; break; } - if (!hash_zero) { - uint8_t fhash[32]; ma_sha256_file(dl->dest_path, fhash); - if (memcmp(fhash, dl->content_hash, 32) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: content_hash mismatch for %s", MDL_ID, dl->dest_path); remove(dl->dest_path); md_dl_finish(dl, -1); md_dl_release(dl); return; } - } } - dl->assembled = 1; - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: file assembled: %s", MDL_ID, dl->dest_path); - int registered = md_dl_register_servable(dl); - md_dl_finish(dl, registered); - md_dl_release(dl); - return; - } - - /* освободился слот — добрать ожидающие блоки */ - if (md_dl_pump(dl) < 0) md_dl_finish(dl, -1); md_dl_release(dl); } -/* ── handle OVERLOADED ── */ - -void media_download_handle_overloaded(struct UTUN_INSTANCE* inst, - const uint8_t* data, size_t len) { - if (!data || len < sizeof(struct media_pkt_block_overloaded)) return; - struct media_pkt_block_overloaded* ov = (struct media_pkt_block_overloaded*)data; - - struct media_download* dl = md_dl_find(inst, ov->media_id); - if (!dl || !dl->active) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: OVERLOADED — no active download", MDL_ID); return; } - - int bi = md_dl_block_index(dl, ov->block_id); - if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: OVERLOADED for unknown block", MDL_ID); return; } - - struct media_download_block* b = &dl->blocks[bi]; - if (b->state == MD_BLK_DONE || b->state == MD_BLK_FAILED) return; - - uint32_t delay = ov->retry_after_ms ? ov->retry_after_ms : 2000; - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: OVERLOADED blk=%d retry_after=%ums holders=%d hidx=%d", - MDL_ID, bi, delay, b->num_holders, b->holder_idx); - - if (b->state == MD_BLK_CONN || b->state == MD_BLK_REQ || b->state == MD_BLK_RECV) { - if (dl->inflight > 0) dl->inflight--; +void media_download_handle_overloaded(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (!data || len != sizeof(struct media_pkt_block_overloaded)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed OVERLOADED bytes=%zu", MDL_ID, len); return; } + const struct media_pkt_block_overloaded* ov = (const struct media_pkt_block_overloaded*)data; + struct media_download* dl = md_dl_match(inst, group_id, from_node, ov->media_id, ov->block_id, ov->transfer_id, ov->chunk); + if (!dl) return; + struct media_download_block* b = &dl->blocks[ov->chunk]; + uint32_t delay = ov->retry_after_ms ? ov->retry_after_ms : 2000; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: OVERLOADED block=%u peer=%016llx delay=%u", + MDL_ID, ov->chunk, (unsigned long long)from_node, delay); + md_dl_end_attempt(dl, (int)ov->chunk); + dl->inflight--; b->bytes_received = 0; b->state = MD_BLK_WAIT; b->wait_reason = MD_WAIT_RETRY; b->retry_at_tb = get_time_tb() + (uint64_t)delay * 10; - b->last_progress_tb = get_time_tb(); - b->holder_idx++; - if (b->holder_idx < b->num_holders) { - /* пробуем следующий держатель сразу (перегруз — не провал блока) */ + if (++b->holder_idx < b->num_holders) { b->retry_at_tb = 0; - if (md_dl_request_block(dl, bi) < 0 && md_dl_failover_block(dl, bi, get_time_tb()) < 0) { + if (md_dl_request_block(dl, (int)ov->chunk) < 0 + && (dl->err || md_dl_failover_block(dl, (int)ov->chunk, get_time_tb()) < 0)) { md_dl_finish(dl, -1); return; } - } else { - b->holder_idx = 0; - } + } else b->holder_idx = 0; if (md_dl_arm_watchdog(dl) < 0) md_dl_finish(dl, -1); } -/* ── handle RELAY_FULL: добавляем присланные узлы и повторяем запрос ── */ - -void media_download_handle_relay_full(struct UTUN_INSTANCE* inst, - const uint8_t* media_id, const uint8_t* block_id, - uint32_t chunk, const uint64_t* node_ids, int num_nodes) { - struct media_download* dl = md_dl_find(inst, media_id); - if (!dl || !dl->active) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL — no active download", MDL_ID); return; } - - int bi = md_dl_block_index(dl, block_id); - if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL — unknown block", MDL_ID); return; } - (void)chunk; - - struct media_download_block* b = &dl->blocks[bi]; - if (b->state == MD_BLK_DONE || b->state == MD_BLK_FAILED) return; - - if (b->state == MD_BLK_CONN || b->state == MD_BLK_REQ || b->state == MD_BLK_RECV) { - if (dl->inflight > 0) dl->inflight--; +void media_download_handle_relay_full(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (!data || len < MEDIA_RELAY_FULL_HDR_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: truncated RELAY_FULL bytes=%zu", MDL_ID, len); return; } - b->state = MD_BLK_IDLE; b->retry_at_tb = 0; - - /* добавить новых держателей (уникальные) */ - for (int ni = 0; ni < num_nodes && ni < MD_MAX_HOLDERS_PER_BLOCK; ni++) { - int dup = 0; - for (int h = 0; h < b->num_holders; h++) if (b->holders[h] == node_ids[ni]) { dup = 1; break; } - if (!dup && b->num_holders < MD_MAX_HOLDERS_PER_BLOCK) b->holders[b->num_holders++] = node_ids[ni]; + const struct media_pkt_block_relay_full* full = (const struct media_pkt_block_relay_full*)data; + unsigned count = full->num_relay_nodes; + if (!count || count > MD_MAX_RELAY_DOWNSTREAM || len != sizeof(*full) + count * sizeof(uint64_t)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed RELAY_FULL bytes=%zu nodes=%u", MDL_ID, len, count); return; } - /* стартуем с первого нового держателя */ - b->holder_idx = 0; - if (b->num_holders > 0) { - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL blk=%d → %d holders, retry", MDL_ID, bi, b->num_holders); - if (md_dl_request_block(dl, bi) < 0 && md_dl_failover_block(dl, bi, get_time_tb()) < 0) md_dl_finish(dl, -1); - } else { - b->state = MD_BLK_WAIT; + struct media_download* dl = md_dl_match(inst, group_id, from_node, full->media_id, full->block_id, + full->transfer_id, full->chunk); + if (!dl) return; + uint64_t nodes[MD_MAX_RELAY_DOWNSTREAM]; + memcpy(nodes, data + sizeof(*full), count * sizeof(uint64_t)); + for (unsigned i = 0; i < count; i++) { + if (!nodes[i] || nodes[i] == inst->node_id || nodes[i] == from_node) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL invalid node index=%u", MDL_ID, i); return; + } + for (unsigned j = 0; j < i; j++) if (nodes[i] == nodes[j]) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL duplicate node index=%u", MDL_ID, i); return; + } } + struct media_download_block* b = &dl->blocks[full->chunk]; + md_dl_end_attempt(dl, (int)full->chunk); + dl->inflight--; b->bytes_received = 0; b->num_holders = 0; b->holder_idx = 0; b->retry_at_tb = 0; + for (unsigned i = 0; i < count && b->num_holders < MD_MAX_HOLDERS_PER_BLOCK; i++) + b->holders[b->num_holders++] = nodes[i]; + b->state = MD_BLK_IDLE; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL block=%u holders=%d", MDL_ID, full->chunk, b->num_holders); + if (md_dl_request_block(dl, (int)full->chunk) < 0 + && (dl->err || md_dl_failover_block(dl, (int)full->chunk, get_time_tb()) < 0)) md_dl_finish(dl, -1); } /* ── public API ── */ @@ -861,13 +799,15 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, struct media_delivery_ctx* md = &inst->md; if (!md->initialized) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: start before media_delivery init", MDL_ID); return -1; } - char location[MEDIA_PATH_MAX]; - if (result->file_size <= 0 || result->block_size <= 0 || result->block_size > UINT32_MAX + char location[MEDIA_PATH_MAX], normalized[MEDIA_PATH_MAX]; + uint8_t author_pubkey[32]; + if (result->file_size <= 0 || result->block_size <= 0 || result->block_size > MEDIA_BLOCK_SIZE_MAX || result->num_blocks <= 0 || (uint64_t)result->num_blocks > UINT32_MAX / sizeof(struct media_download_block) || (uint64_t)result->num_blocks > UINT32_MAX / 64 || (result->file_size - 1) / result->block_size + 1 != result->num_blocks || !result->block_ids || !result->block_sigs || media_index_location(dest_path, media_base, location) - || media_index_set_base(inst->topo_sqlite_db, media_base)) { + || media_index_set_base(inst->topo_sqlite_db, media_base) || media_index_path(media_base, location, normalized) + || topo_node_sqlite_get_ed25519_pubkey(inst->topo_sqlite_db, author_node_id, author_pubkey)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: invalid file geometry/path/base before download", MDL_ID); return -1; } @@ -877,16 +817,15 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, return -1; } char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id); - if (media_index_track_preparation(inst->topo_sqlite_db, channel, dest_path, media_base)) return -1; struct media_download* dl = u_calloc(1, sizeof(*dl)); if (!dl) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_calloc for download failed", MDL_ID); return -1; } memcpy(dl->media_id, result->media_id, 16); dl->group_id = group_id; - snprintf(dl->dest_path, sizeof(dl->dest_path), "%s", dest_path); - md_mkdir_parent(dl->dest_path); - snprintf(dl->media_base, sizeof(dl->media_base), "%s", media_base); + snprintf(dl->dest_path, sizeof(dl->dest_path), "%s", normalized); + if (media_index_get_base(inst->topo_sqlite_db, dl->media_base)) { u_free(dl); return -1; } + memcpy(dl->author_pubkey, author_pubkey, 32); dl->num_blocks = result->num_blocks; dl->file_size = result->file_size; dl->block_size = result->block_size; @@ -925,11 +864,16 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(download) failed", MDL_ID); u_free(dl->blocks); u_free(dl->block_sigs); u_free(dl); return -1; } memcpy(qe->data, dl, sizeof(*dl)); u_free(dl); - queue_data_put_with_index(md->downloads, qe); - dl = (struct media_download*)qe->data; dl->entry = qe; dl->refs = 1; + dl->destination = media_delivery_reserve(inst, dl->dest_path); + if (!dl->destination || media_index_track_preparation(inst->topo_sqlite_db, channel, dl->dest_path, dl->media_base)) { + media_delivery_release(inst, dl->destination); + u_free(dl->blocks); u_free(dl->block_sigs); queue_entry_free(qe); return -1; + } + md_mkdir_parent(dl->dest_path); + queue_data_put_with_index(md->downloads, qe); md_dl_collect_supernodes(inst, dl); @@ -942,21 +886,10 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, int media_download_cancel(struct UTUN_INSTANCE* inst, const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk) { - (void)chunk; + (void)block_id; (void)chunk; struct media_download* dl = md_dl_find(inst, media_id); if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cancel — download not found", MDL_ID); return -1; } - struct media_pkt_cancel can; - memset(&can, 0, sizeof(can)); - can.subcmd = MEDIA_SUBCMD_CANCEL; - memcpy(can.media_id, media_id, 16); - if (block_id) memcpy(can.block_id, block_id, 16); - can.chunk = chunk; - - for (int pi = 0; pi < dl->num_peers; pi++) - if (dl->peers[pi].connected) - md_dl_send(inst, dl->group_id, dl->peers[pi].node_id, (const uint8_t*)&can, sizeof(can)); - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: download cancelled", MDL_ID); md_dl_finish(dl, -2); return 0;