jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Validate received blocks and publish complete verified media atomically

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 19:14:12 +02:00
1 parent 4edf72ea14
commit d0a401e166
2 files changed
+223 -286
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_msg.c
+4
View File
@@ -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;
src/media_delivery/media_download.c
+219 -286
View File
@@ -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;