From 9b4974f02d85a4ac9ef1ec2eba3d51fcfbc54e5d Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 19:11:26 +0200 Subject: [PATCH] Unify media source and relay transmission with owned buffers --- src/media_delivery/media_delivery.c | 691 ++++++++++++--------------- src/routing_layer/topo_node_sqlite.c | 20 +- 2 files changed, 324 insertions(+), 387 deletions(-) diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index bd30093e..389f73a3 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -20,12 +20,14 @@ #include "../lib/sqlite3.h" #include #include +#include +#include "../lib/platform_compat.h" #define MD_ID "media_delivery" /* ── forward decls ── */ -static void md_relay_cancel_block(struct media_delivery_ctx* md, struct relay_block_ctx* block); +static void md_block_wake(struct media_block_source* source); static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry); static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg); static void md_on_bgp_node(struct TOPO_GROUP* group, uint64_t node_id, int event, void* arg); @@ -246,44 +248,95 @@ static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id queue_entry_free(e); } -/* ── relay block context helpers ── */ +/* Назначения общие для загрузок и исходящих фоновых подготовок. */ +struct media_destination { + struct media_destination* next; + char path[MEDIA_PATH_MAX]; +}; -struct relay_block_ctx* md_relay_find(struct media_delivery_ctx* md, const uint8_t* block_id) { - if (!md->relay_blocks) return NULL; - struct ll_entry* e = queue_find_data_by_index(md->relay_blocks, block_id); - return e ? (struct relay_block_ctx*)e->data : NULL; -} - -struct relay_block_ctx* md_relay_add(struct media_delivery_ctx* md, - const uint8_t* block_id, - const uint8_t* media_id, - uint32_t chunk) { - struct relay_block_ctx* rc = md_relay_find(md, block_id); - if (rc) return rc; - if (!md->relay_blocks) { - md->relay_blocks = queue_new(md->inst->ua, 64, offsetof(struct relay_block_ctx, block_id), 16, "md_relay"); - if (!md->relay_blocks) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(relay_blocks) failed", MD_ID); return NULL; } +struct media_destination* media_delivery_reserve(struct UTUN_INSTANCE* inst, const char* path) { + char base[MEDIA_BASE_MAX], location[MEDIA_PATH_MAX], normalized[MEDIA_PATH_MAX]; + if (!inst || !inst->md.initialized || media_index_get_base(inst->topo_sqlite_db, base) + || media_index_location(path, base, location) || media_index_path(base, location, normalized)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destination reservation failed path=%s", MD_ID, path ? path : "(null)"); + return NULL; } - struct ll_entry* qe = queue_entry_new(sizeof(struct relay_block_ctx)); - if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(relay_block) failed", MD_ID); return NULL; } - rc = (struct relay_block_ctx*)qe->data; - memset(rc, 0, sizeof(*rc)); - memcpy(rc->block_id, block_id, 16); - memcpy(rc->media_id, media_id, 16); - rc->chunk = chunk; - rc->inst = md->inst; - memcpy(qe->data, block_id, 16); - queue_data_put_with_index(md->relay_blocks, qe); - return rc; + for (struct media_destination* d = inst->md.destinations; d; d = d->next) { + if (!strcmp(d->path, normalized)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: destination busy path=%s", MD_ID, normalized); return NULL; + } + } + struct media_destination* d = u_calloc(1, sizeof(*d)); + if (!d) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destination allocation failed", MD_ID); return NULL; } + memcpy(d->path, normalized, strlen(normalized) + 1); + d->next = inst->md.destinations; inst->md.destinations = d; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: reserved destination=%s", MD_ID, normalized); + return d; } -void md_relay_remove(struct media_delivery_ctx* md, const uint8_t* block_id) { - if (!md->relay_blocks) return; - struct ll_entry* e = queue_find_data_by_index(md->relay_blocks, block_id); - if (!e) return; - md_relay_cancel_block(md, (struct relay_block_ctx*)e->data); - queue_remove_data(md->relay_blocks, e); - queue_entry_free(e); +void media_delivery_release(struct UTUN_INSTANCE* inst, struct media_destination* destination) { + if (!destination) return; + struct media_destination** slot = &inst->md.destinations; + while (*slot && *slot != destination) slot = &(*slot)->next; + if (!*slot) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: unknown destination reservation", MD_ID); return; } + *slot = destination->next; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: released destination=%s", MD_ID, destination->path); + u_free(destination); +} + +static void md_block_detach(struct media_block_source* source) { + if (!source->indexed) return; + queue_remove_data(source->md->block_sources, source->entry); + source->indexed = 0; +} + +static struct media_block_source* md_block_find(struct media_delivery_ctx* md, const uint8_t block_id[16]) { + struct ll_entry* e = md->block_sources ? queue_find_data_by_index(md->block_sources, block_id) : NULL; + return e ? (struct media_block_source*)e->data : NULL; +} + +void md_block_release(struct media_block_source* source) { + if (!source || --source->refs) return; + md_block_detach(source); + if (source->owned) file_io_temp_discard(&source->storage); + queue_entry_free(source->entry); +} + +struct media_block_source* md_block_create(struct media_delivery_ctx* md, uint64_t group_id, + const uint8_t media_id[16], const uint8_t block_id[16], uint32_t chunk, uint32_t size, const char* destination) { + if (!md->block_sources) + md->block_sources = queue_new(md->inst->ua, 64, offsetof(struct media_block_source, block_id), 16, "md_sources"); + if (!md->block_sources) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: source index allocation failed", MD_ID); return NULL; } + struct ll_entry* entry = queue_entry_new(sizeof(struct media_block_source)); + if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: block source allocation failed", MD_ID); return NULL; } + struct media_block_source* source = (struct media_block_source*)entry->data; + memset(source, 0, sizeof(*source)); + source->entry = entry; source->md = md; source->refs = 1; source->owned = 1; + source->group_id = group_id; source->chunk = chunk; source->size = size; + memcpy(source->media_id, media_id, 16); memcpy(source->block_id, block_id, 16); + if (file_io_temp_open(&source->storage, destination) || file_io_temp_finish(&source->storage)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: block temp creation failed chunk=%u", MD_ID, chunk); + md_block_release(source); return NULL; + } + struct media_block_source* previous = md_block_find(md, block_id); + if (previous) md_block_detach(previous); + source->indexed = 1; queue_data_put_with_index(md->block_sources, entry); + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: block source created chunk=%u size=%u path=%s", + MD_ID, chunk, size, source->storage.temporary); + return source; +} + +void md_block_publish(struct media_block_source* source, uint32_t available, int verified) { + source->available = available; source->verified = verified != 0; + md_block_wake(source); +} + +void md_block_abort(struct media_block_source* source) { + if (!source || source->failed) return; + source->failed = 1; md_block_detach(source); + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: block source aborted chunk=%u available=%u/%u", + MD_ID, source->chunk, source->available, source->size); + md_block_wake(source); } /* ── file_load helpers (per-file source download tracking) ── */ @@ -688,377 +741,255 @@ static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64 /* ── handle incoming BLOCK_REQ (admission control + streaming) ── */ #define MD_MAX_STREAMS 3 -#define MD_CHUNK_SIZE 1024 struct stream_ctx { struct stream_ctx* next; struct media_delivery_ctx* md; - uint64_t group_id; - uint64_t dst_node_id; - uint8_t media_id[16]; - uint8_t block_id[16]; - uint32_t chunk; - FILE* file; - uint64_t offset; - uint64_t remaining; - uint64_t block_start; - size_t block_data_len; + struct media_block_source* source; + uint64_t group_id, dst_node_id; + uint8_t transfer_id[16]; + uint32_t offset; + uint64_t last_progress_tb; + void* pump; + void* timer; struct queue_waiter_handle waiter; }; +static void stream_pump(void* arg); + static void stream_free(struct stream_ctx* sc) { struct media_delivery_ctx* md = sc->md; etcp_router_cancel_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sc->waiter); - struct stream_ctx** p = &md->streams; - while (*p && *p != sc) p = &(*p)->next; - if (*p) *p = sc->next; - md_file_load_dec(md, sc->media_id, sc->dst_node_id); + if (sc->pump) uasync_call_soon_cancel(md->inst->ua, sc->pump); + if (sc->timer) uasync_cancel_timeout(md->inst->ua, sc->timer); + struct stream_ctx** slot = &md->streams; + while (*slot && *slot != sc) slot = &(*slot)->next; + if (*slot) *slot = sc->next; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: TX released peer=%016llx chunk=%u sent=%u/%u attempt=%02x%02x", + MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk, sc->offset, sc->source->size, + sc->transfer_id[0], sc->transfer_id[1]); + md_file_load_dec(md, sc->source->media_id, sc->dst_node_id); media_delivery_stream_done(md->inst); - if (sc->file) fclose(sc->file); + md_block_release(sc->source); u_free(sc); } -static void stream_send_chunk_cb(struct ll_queue* q, void* arg) { - (void)q; - struct stream_ctx* sc = (struct stream_ctx*)arg; +static int stream_schedule(struct stream_ctx* sc) { + if (!sc->pump) sc->pump = uasync_call_soon(sc->md->inst->ua, sc, stream_pump); + if (sc->pump) return 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot schedule TX peer=%016llx chunk=%u", + MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk); + return -1; +} + +static void stream_ready(struct ll_queue* queue, void* arg) { + (void)queue; + struct stream_ctx* sc = arg; + if (stream_schedule(sc)) stream_free(sc); +} + +static void md_block_wake(struct media_block_source* source) { + struct stream_ctx* sc = source->md->streams; + while (sc) { + struct stream_ctx* next = sc->next; + if (sc->source == source && stream_schedule(sc)) stream_free(sc); + sc = next; + } +} + +static void stream_watchdog(void* arg) { + struct stream_ctx* sc = arg; + sc->timer = NULL; + uint64_t timeout = sc->md->dl_stall_timeout_tb > 0 ? (uint64_t)sc->md->dl_stall_timeout_tb : MEDIA_DL_STALL_TIMEOUT_TB; + if (get_time_tb() - sc->last_progress_tb >= timeout) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: TX stalled peer=%016llx chunk=%u offset=%u available=%u verified=%u", + MD_ID, (unsigned long long)sc->dst_node_id, sc->source->chunk, sc->offset, + sc->source->available, sc->source->verified); + stream_free(sc); return; + } + sc->timer = uasync_set_timeout(sc->md->inst->ua, 2000, sc, stream_watchdog, "md_tx"); + if (!sc->timer || stream_schedule(sc)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX watchdog scheduling failed", MD_ID); stream_free(sc); + } +} + +/* Единственный путь выдачи готового диапазона и растущего temp. Waiter может сработать синхронно: + * callback только ставит pump, не входит в него рекурсивно. DONE проходит тот же backpressure, что DATA. */ +static void stream_pump(void* arg) { + struct stream_ctx* sc = arg; struct media_delivery_ctx* md = sc->md; - if (!md->initialized || sc->remaining == 0) { stream_free(sc); return; } - - while (sc->remaining > 0) { - size_t to_read = sc->remaining < MD_CHUNK_SIZE ? (size_t)sc->remaining : MD_CHUNK_SIZE; - uint8_t buf[MD_CHUNK_SIZE]; - fseeko(sc->file, (off_t)sc->offset, SEEK_SET); - size_t rd = fread(buf, 1, to_read, sc->file); - if (rd == 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "stream read failed"); stream_free(sc); return; } - - uint8_t pkt[MEDIA_BLOCK_CHUNK_HDR_SIZE + MD_CHUNK_SIZE]; - struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)pkt; - memset(ch, 0, MEDIA_BLOCK_CHUNK_HDR_SIZE); - ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; - memcpy(ch->media_id, sc->media_id, 16); - memcpy(ch->block_id, sc->block_id, 16); - ch->chunk = sc->chunk; - ch->offset = (uint32_t)(sc->offset - sc->block_start); - ch->data_len = (uint16_t)rd; - memcpy(pkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, buf, rd); - - int snd = md_send(md->inst, sc->group_id, sc->dst_node_id, pkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd); - if (snd != 0) { - DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: stream chunk send FAILED rc=%d — abort stream chunk=%d to 0x%016llx", - MD_ID, snd, sc->chunk, (unsigned long long)sc->dst_node_id); - stream_free(sc); - return; - } - - sc->offset += rd; - sc->remaining -= rd; - { - uint64_t sent = sc->offset - sc->block_start; - uint64_t mb_off = (sent - rd) >> 20; - uint64_t mb_end = sent >> 20; - if (sent == rd || mb_end != mb_off || sent >= sc->block_data_len) - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: stream chunk=%d sent=%llu/%zu (%.0f%%) to 0x%016llx", - MD_ID, sc->chunk, (unsigned long long)sent, sc->block_data_len, - sc->block_data_len ? 100.0 * (double)sent / (double)sc->block_data_len : 0.0, - (unsigned long long)sc->dst_node_id); - } - - if (sc->remaining == 0) break; - + struct media_block_source* source = sc->source; + sc->pump = NULL; + if (!md->initialized) { stream_free(sc); return; } + etcp_router_cancel_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sc->waiter); + for (int budget = 0; budget < 4; budget++) { + if (!source->failed && sc->offset == source->available && !source->verified) return; if (!etcp_router_send_q_has_room(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY)) { - etcp_router_on_send_ready(md->inst, sc->group_id, - sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, - &sc->waiter, stream_send_chunk_cb, sc); + etcp_router_on_send_ready(md->inst, sc->group_id, sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, + &sc->waiter, stream_ready, sc); return; } + if (source->failed || (source->verified && sc->offset == source->size)) { + struct media_pkt_block_done done = {.subcmd = MEDIA_SUBCMD_BLOCK_DONE, .chunk = source->chunk, + .total_size = source->size, .status = source->failed ? MEDIA_BLOCK_UNAVAILABLE : MEDIA_BLOCK_OK}; + memcpy(done.media_id, source->media_id, 16); memcpy(done.block_id, source->block_id, 16); + memcpy(done.transfer_id, sc->transfer_id, 16); + if (!md_send(md->inst, sc->group_id, sc->dst_node_id, (const uint8_t*)&done, sizeof(done))) + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX DONE peer=%016llx chunk=%u bytes=%u status=%u", + MD_ID, (unsigned long long)sc->dst_node_id, source->chunk, sc->offset, done.status); + stream_free(sc); return; + } + uint32_t count = source->available - sc->offset; + if (count > MEDIA_BLOCK_DATA_MAX) count = MEDIA_BLOCK_DATA_MAX; + uint8_t packet[MEDIA_BLOCK_CHUNK_HDR_SIZE + MEDIA_BLOCK_DATA_MAX]; + struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)packet; + memset(ch, 0, sizeof(*ch)); ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; + memcpy(ch->media_id, source->media_id, 16); memcpy(ch->block_id, source->block_id, 16); + memcpy(ch->transfer_id, sc->transfer_id, 16); + ch->chunk = source->chunk; ch->offset = sc->offset; ch->data_len = (uint16_t)count; + FILE* file = file_io_open_read(source->storage.temporary); + int io_error = !file; + if (file) { + if (fseeko(file, (off_t)(source->base + sc->offset), SEEK_SET) + || fread(packet + MEDIA_BLOCK_CHUNK_HDR_SIZE, 1, count, file) != count) io_error = 1; + if (fclose(file)) io_error = 1; + } + if (io_error) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX read failed path=%s chunk=%u offset=%u count=%u errno=%d", + MD_ID, source->storage.temporary, source->chunk, sc->offset, count, errno); + source->failed = 1; md_block_detach(source); md_block_wake(source); + continue; + } + if (md_send(md->inst, sc->group_id, sc->dst_node_id, packet, sizeof(*ch) + count)) { + stream_free(sc); return; + } + sc->offset += count; sc->last_progress_tb = get_time_tb(); + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: TX peer=%016llx chunk=%u offset=%u available=%u/%u verified=%u", + MD_ID, (unsigned long long)sc->dst_node_id, source->chunk, sc->offset, + source->available, source->size, source->verified); } - - /* all data sent — send BLOCK_DONE */ - { - uint32_t chunk = sc->chunk; - uint8_t done_pkt[MEDIA_BLOCK_DONE_SIZE]; - struct media_pkt_block_done* bd = (struct media_pkt_block_done*)done_pkt; - memset(bd, 0, sizeof(*bd)); - bd->subcmd = MEDIA_SUBCMD_BLOCK_DONE; - memcpy(bd->media_id, sc->media_id, 16); - memcpy(bd->block_id, sc->block_id, 16); - bd->chunk = chunk; - bd->total_size = (uint32_t)sc->block_data_len; - - md_send(md->inst, sc->group_id, sc->dst_node_id, done_pkt, MEDIA_BLOCK_DONE_SIZE); - stream_free(sc); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_DONE sent chunk=%d total=%u", - MD_ID, chunk, bd->total_size); - } + if (stream_schedule(sc)) stream_free(sc); } -/* ── relay chunk forwarding ── */ - -struct relay_fwd_ctx { - struct relay_fwd_ctx* next; - struct media_delivery_ctx* md; - uint64_t dst_node_id; - uint64_t group_id; - struct relay_block_ctx* rc; - int ds_idx; // индекс в rc->downstream[] - struct queue_waiter_handle waiter; -}; - -static void relay_fwd_free(struct relay_fwd_ctx* fc) { - etcp_router_cancel_send_ready(fc->md->inst, fc->group_id, fc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &fc->waiter); - struct relay_fwd_ctx** p = &fc->md->forwards; - while (*p && *p != fc) p = &(*p)->next; - if (*p) *p = fc->next; - u_free(fc); -} - -static void md_relay_cancel_block(struct media_delivery_ctx* md, struct relay_block_ctx* block) { - struct relay_fwd_ctx* fc = md->forwards; - while (fc) { - struct relay_fwd_ctx* next = fc->next; - if (fc->rc == block) relay_fwd_free(fc); - fc = next; - } -} - -static void md_relay_fwd_cb(struct ll_queue* q, void* arg) { - (void)q; - struct relay_fwd_ctx* fc = (struct relay_fwd_ctx*)arg; - struct relay_block_ctx* rc = fc->rc; - struct relay_downstream* ds = &rc->downstream[fc->ds_idx]; - - if (ds->sent_offset >= rc->file_offset) { relay_fwd_free(fc); return; } - - FILE* f = fopen(rc->chunk_file, "rb"); - if (!f) { relay_fwd_free(fc); return; } - fseeko(f, (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 buf[1024]; - size_t rd = fread(buf, 1, to_read, f); - fclose(f); - if (rd == 0) { relay_fwd_free(fc); return; } - - uint8_t pkt[MEDIA_BLOCK_CHUNK_HDR_SIZE + 1024]; - struct media_pkt_block_chunk* ch = (struct media_pkt_block_chunk*)pkt; - memset(ch, 0, MEDIA_BLOCK_CHUNK_HDR_SIZE); - ch->subcmd = MEDIA_SUBCMD_BLOCK_CHUNK; - memcpy(ch->media_id, rc->media_id, 16); - memcpy(ch->block_id, rc->block_id, 16); - ch->chunk = rc->chunk; - ch->offset = (uint32_t)ds->sent_offset; - ch->data_len = (uint16_t)rd; - memcpy(pkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, buf, rd); - - size_t pkt_len = MEDIA_BLOCK_CHUNK_HDR_SIZE + rd; - if (md_send(fc->md->inst, fc->group_id, fc->dst_node_id, pkt, pkt_len) == 0) { - ds->sent_offset += rd; - } - - if (ds->sent_offset < rc->file_offset) { - etcp_router_on_send_ready(fc->md->inst, fc->group_id, - fc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY, - &fc->waiter, md_relay_fwd_cb, fc); +static void md_block_reply(struct media_delivery_ctx* md, uint64_t group, uint64_t peer, + const struct media_pkt_block_req* request, int overloaded) { + if (overloaded) { + struct media_pkt_block_overloaded reply = {.subcmd = MEDIA_SUBCMD_BLOCK_OVERLOADED, + .chunk = request->chunk, .retry_after_ms = 2000}; + memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16); + memcpy(reply.transfer_id, request->transfer_id, 16); + md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply)); } else { - relay_fwd_free(fc); + struct media_pkt_block_done reply = {.subcmd = MEDIA_SUBCMD_BLOCK_DONE, .chunk = request->chunk, + .status = MEDIA_BLOCK_UNAVAILABLE}; + memcpy(reply.media_id, request->media_id, 16); memcpy(reply.block_id, request->block_id, 16); + memcpy(reply.transfer_id, request->transfer_id, 16); + md_send(md->inst, group, peer, (const uint8_t*)&reply, sizeof(reply)); } } -static void md_relay_catchup(struct media_delivery_ctx* md, struct relay_block_ctx* rc, - uint64_t dst_node_id, uint64_t group_id, int ds_idx) { - struct relay_downstream* ds = &rc->downstream[ds_idx]; - if (rc->file_offset == 0 || ds->sent_offset >= rc->file_offset) return; - - struct relay_fwd_ctx* fc = u_calloc(1, sizeof(*fc)); - if (!fc) return; - fc->next = md->forwards; md->forwards = fc; - fc->md = md; fc->dst_node_id = dst_node_id; fc->group_id = group_id; - fc->rc = rc; fc->ds_idx = ds_idx; - memset(&fc->waiter, 0, sizeof(fc->waiter)); - md_relay_fwd_cb(NULL, fc); -} - -static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_node, - const uint8_t* data, size_t len) { - if (len < MEDIA_BLOCK_REQ_SIZE) return; - struct media_pkt_block_req* req = (struct media_pkt_block_req*)data; - uint64_t group_id = req->group_id ? req->group_id : TOPO_GROUP_UTUN; - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ from 0x%016llx group=%016llx block=%02x%02x... active_streams=%d", - MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, req->block_id[0], req->block_id[1], md->active_streams); - - int limit_reached = md->active_streams >= MD_MAX_STREAMS; - md->active_streams++; - - if (limit_reached) { - md->active_streams--; /* undo */ - struct media_pkt_block_overloaded ov; - ov.subcmd = MEDIA_SUBCMD_BLOCK_OVERLOADED; - memcpy(ov.media_id, req->media_id, 16); - memcpy(ov.block_id, req->block_id, 16); - ov.chunk = req->chunk; - ov.retry_after_ms = 2000; - md_send(md->inst, group_id, from_node, (const uint8_t*)&ov, sizeof(ov)); - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ OVERLOADED (streams=%d) to 0x%016llx", - MD_ID, md->active_streams, (unsigned long long)from_node); +static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (len != MEDIA_BLOCK_REQ_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed BLOCK_REQ bytes=%zu", MD_ID, len); return; + } + const struct media_pkt_block_req* req = (const struct media_pkt_block_req*)data; + int relay_count = 0; + struct media_block_source* source = md_block_find(md, req->block_id); + if (source && (source->group_id != group_id || source->chunk != req->chunk + || memcmp(source->media_id, req->media_id, 16))) source = NULL; + for (struct stream_ctx* sc = md->streams; sc; sc = sc->next) { + if (sc->source == source) relay_count++; + if (sc->group_id != group_id || sc->dst_node_id != from_node || memcmp(sc->transfer_id, req->transfer_id, 16)) continue; + if (sc->source->chunk != req->chunk || memcmp(sc->source->media_id, req->media_id, 16) + || memcmp(sc->source->block_id, req->block_id, 16)) + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: conflicting BLOCK_REQ attempt peer=%016llx", MD_ID, (unsigned long long)from_node); return; } - - /* establish direct connection if not already connected */ - if (!instance_find_conn(md->inst, from_node) && group_id != TOPO_GROUP_UTUN) { - struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id); - if (!grp) grp = topo_groups_get_default(md->inst->topo_groups); - if (grp && grp->conn_mgr) { - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: no direct conn to 0x%016llx — connecting via conn_mgr", - MD_ID, (unsigned long long)from_node); - conn_mgr_open(grp->instance, grp->group_id, from_node, NULL, NULL, NULL); + struct md_file_load* load = md_file_load_find(md, req->media_id); + if (md->active_streams >= MD_MAX_STREAMS || relay_count >= MD_MAX_RELAY_DOWNSTREAM + || (load && md->max_downloads_per_file && load->active_downloads >= md->max_downloads_per_file)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ admission busy peer=%016llx active=%u file=%d relay=%d", + MD_ID, (unsigned long long)from_node, md->active_streams, load ? load->active_downloads : 0, relay_count); + md_block_reply(md, group_id, from_node, req, 1); return; + } + if (source) source->refs++; + else { + sqlite3_stmt* st = NULL; + const char* sql = "SELECT location,chunk_size,chunk,offset,file_size,chat_id FROM media_files" + " WHERE block_id=? AND media_id=? AND node_id=? LIMIT 1"; + int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, req->block_id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, req->media_id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)md->self_node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + char base[MEDIA_BASE_MAX], path[MEDIA_PATH_MAX], channel[32]; + snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id); + const char* location = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 0) : NULL; + int64_t size = rc == SQLITE_ROW ? sqlite3_column_int64(st, 1) : 0; + int64_t offset = rc == SQLITE_ROW ? sqlite3_column_int64(st, 3) : -1; + int64_t file_size = rc == SQLITE_ROW ? sqlite3_column_int64(st, 4) : 0; + const char* chat = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 5) : NULL; + int valid = location && sqlite3_column_bytes(st, 0) == (int)strlen(location) + && chat && sqlite3_column_bytes(st, 5) == (int)strlen(chat) && !strcmp(chat, channel) + && sqlite3_column_int64(st, 2) == req->chunk && size > 0 && size <= MEDIA_BLOCK_SIZE_MAX + && offset >= 0 && offset < file_size && size <= file_size - offset + && !media_index_get_base(md->db, base) && !media_index_path(base, location, path); + int final_rc = sqlite3_finalize(st); + struct file_io_info info; + if (!valid || final_rc != SQLITE_OK || file_io_inspect(path, 1, &info) + || info.type != FILE_IO_REGULAR || info.size != file_size) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_REQ unavailable peer=%016llx chunk=%u sql=%d finalize=%d", + MD_ID, (unsigned long long)from_node, req->chunk, rc, final_rc); + md_block_reply(md, group_id, from_node, req, 0); return; } - } - - /* find file in media_files DB */ - sqlite3* db = md->db; - if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ no DB", MD_ID); md->active_streams--; return; } - - const char* sql = "SELECT location,chunk_size,chunk,offset,file_size FROM media_files WHERE block_id=? AND node_id=? LIMIT 1"; - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ prepare failed: %s", MD_ID, sqlite3_errmsg(db)); md->active_streams--; return; } - sqlite3_bind_blob(st, 1, req->block_id, 16, SQLITE_STATIC); - sqlite3_bind_int64(st, 2, (sqlite3_int64)md->inst->node_id); - if (sqlite3_step(st) != SQLITE_ROW) { - sqlite3_finalize(st); - /* fallback: check relay_blocks — maybe we are downloading this block and can relay */ - struct relay_block_ctx* rc = md_relay_find(md, req->block_id); - if (rc && rc->downstream_count < MD_MAX_RELAY_DOWNSTREAM) { - int dsi = rc->downstream_count++; - rc->downstream[dsi].node_id = from_node; - rc->downstream[dsi].group_id = group_id; - rc->downstream[dsi].sent_offset = 0; - memset(&rc->downstream[dsi].waiter, 0, sizeof(rc->downstream[dsi].waiter)); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: relay downstream[%d] added node=0x%016llx file_offset=%llu", - MD_ID, dsi, (unsigned long long)from_node, (unsigned long long)rc->file_offset); - md_relay_catchup(md, rc, from_node, group_id, dsi); - md->active_streams--; return; + struct ll_entry* entry = queue_entry_new(sizeof(*source)); + if (!entry) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: indexed source allocation failed", MD_ID); + md_block_reply(md, group_id, from_node, req, 0); return; } - if (rc) { - /* relay full — send RELAY_FULL with downstream list */ - uint8_t rfp[256]; int roff = 0; - rfp[roff++] = MEDIA_SUBCMD_BLOCK_RELAY_FULL; - memcpy(rfp + roff, req->media_id, 16); roff += 16; - memcpy(rfp + roff, req->block_id, 16); roff += 16; - uint32_t ck = req->chunk; memcpy(rfp + roff, &ck, 4); roff += 4; - uint16_t nrn = (uint16_t)rc->downstream_count; - memcpy(rfp + roff, &nrn, 2); roff += 2; - for (int i = 0; i < rc->downstream_count; i++) { - memcpy(rfp + roff, &rc->downstream[i].node_id, 8); roff += 8; - } - md_send(md->inst, group_id, from_node, rfp, (size_t)roff); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: relay full for block=%02x%02x..., sent %d downstream nodes to 0x%016llx", - MD_ID, req->block_id[0], req->block_id[1], rc->downstream_count, (unsigned long long)from_node); - } else { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ block_id=%02x%02x... not found in media_files", - MD_ID, req->block_id[0], req->block_id[1]); - } - md->active_streams--; return; + source = (struct media_block_source*)entry->data; memset(source, 0, sizeof(*source)); + source->entry = entry; source->md = md; source->refs = 1; source->verified = 1; + source->group_id = group_id; source->chunk = req->chunk; source->size = source->available = (uint32_t)size; + source->base = (uint64_t)offset; + memcpy(source->block_id, req->block_id, 16); memcpy(source->media_id, req->media_id, 16); + memcpy(source->storage.temporary, path, strlen(path) + 1); } - - const char* loc = (const char*)sqlite3_column_text(st, 0); - int64_t chunk_size = sqlite3_column_int64(st, 1); - int chunk = sqlite3_column_int(st, 2); - int64_t offset = sqlite3_column_int64(st, 3); - int64_t file_size = sqlite3_column_int64(st, 4); - - /* compute chunk start and size: offset column хранит n * block_size - (для последнего усечённого блока chunk * chunk_size даёт неверный offset) */ - int64_t block_start = offset; - int64_t block_end = block_start + chunk_size; - if (block_end > file_size) block_end = file_size; - int64_t block_length = block_end - block_start; - - /* find media_base */ - char media_base[MEDIA_BASE_MAX], path[MEDIA_PATH_MAX]; - if (!loc || sqlite3_column_bytes(st, 0) != (int)strlen(loc) || media_index_get_base(db, media_base) - || media_index_path(media_base, loc, path)) { - DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: invalid stored media path", MD_ID); - sqlite3_finalize(st); md->active_streams--; return; - } - FILE* f = file_io_open_read(path); - sqlite3_finalize(st); - - if (!f) { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: cannot open file %s for streaming", MD_ID, path); - md->active_streams--; /* undo pre-increment */ - return; - } - - /* per-file download limit: redirect to current downloaders if at capacity */ - if (md->max_downloads_per_file > 0) { - struct md_file_load* fl = md_file_load_find(md, req->media_id); - if (fl && fl->active_downloads >= (int)md->max_downloads_per_file) { - /* source is at capacity — send RELAY_FULL with current downloader list */ - uint8_t rfp[256]; int roff = 0; - rfp[roff++] = MEDIA_SUBCMD_BLOCK_RELAY_FULL; - memcpy(rfp + roff, req->media_id, 16); roff += 16; - memcpy(rfp + roff, req->block_id, 16); roff += 16; - uint32_t ck = req->chunk; memcpy(rfp + roff, &ck, 4); roff += 4; - uint16_t nrn = (uint16_t)fl->active_downloads; - memcpy(rfp + roff, &nrn, 2); roff += 2; - for (int i = 0; i < fl->active_downloads && i < 10; i++) { - memcpy(rfp + roff, &fl->downloader_ids[i], 8); roff += 8; - } - md_send(md->inst, group_id, from_node, rfp, (size_t)roff); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: source at capacity (%d/%d) for media=%02x%02x..., redirecting to %d nodes", - MD_ID, fl->active_downloads, md->max_downloads_per_file, - req->media_id[0], req->media_id[1], fl->active_downloads); - fclose(f); - md->active_streams--; return; - } - } - md_file_load_inc(md, req->media_id, from_node); - struct stream_ctx* sc = u_calloc(1, sizeof(*sc)); - if (!sc) { fclose(f); md->active_streams--; md_file_load_dec(md, req->media_id, from_node); return; } - sc->md = md; sc->next = md->streams; md->streams = sc; - sc->group_id = group_id; - sc->dst_node_id = from_node; - memcpy(sc->media_id, req->media_id, 16); - memcpy(sc->block_id, req->block_id, 16); - sc->chunk = req->chunk; - sc->file = f; - sc->offset = (uint64_t)block_start; - sc->remaining = (uint64_t)block_length; - sc->block_start = (uint64_t)block_start; - sc->block_data_len = (size_t)block_length; - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: streaming file=%s chunk=%d start=%lld len=%lld group=%016llx to 0x%016llx", - MD_ID, path, chunk, (long long)block_start, (long long)block_length, (unsigned long long)group_id, (unsigned long long)from_node); - md->streams_started++; /* monotonic — for test admission checks */ - - /* start streaming — first chunk immediately, then backpressure */ - memset(&sc->waiter, 0, sizeof(sc->waiter)); - stream_send_chunk_cb(NULL, sc); + if (!sc || !etcp_router_conn_get(md->inst, group_id, from_node, ETCP_RT_ID_MEDIA_DELIVERY) + || md_file_load_inc(md, req->media_id, from_node) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot start TX peer=%016llx chunk=%u", MD_ID, + (unsigned long long)from_node, req->chunk); + u_free(sc); md_block_release(source); md_block_reply(md, group_id, from_node, req, 0); return; + } + sc->md = md; sc->source = source; sc->group_id = group_id; sc->dst_node_id = from_node; + memcpy(sc->transfer_id, req->transfer_id, 16); sc->last_progress_tb = get_time_tb(); + sc->next = md->streams; md->streams = sc; md->active_streams++; md->streams_started++; + sc->timer = uasync_set_timeout(md->inst->ua, 2000, sc, stream_watchdog, "md_tx"); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX started peer=%016llx group=%016llx chunk=%u size=%u available=%u relay=%u", + MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, source->chunk, + source->size, source->available, source->owned); + if (!sc->timer || stream_schedule(sc)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: TX startup scheduling failed", MD_ID); stream_free(sc); + } } -/* ── handle RELAY_FULL: relay node is overloaded, retry one of its downstream nodes ── */ - -static void md_handle_block_relay_full(struct media_delivery_ctx* md, uint64_t from_node, - const uint8_t* data, size_t len) { - if (len < MEDIA_RELAY_FULL_HDR_SIZE) return; - struct media_pkt_block_relay_full* rf = (struct media_pkt_block_relay_full*)data; - uint16_t nrn = rf->num_relay_nodes; - if (nrn > MD_MAX_RELAY_DOWNSTREAM) nrn = MD_MAX_RELAY_DOWNSTREAM; - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: RELAY_FULL from 0x%016llx block=%02x%02x... nodes=%u", - MD_ID, (unsigned long long)from_node, rf->block_id[0], rf->block_id[1], nrn); - - /* forward to media_download layer to retry from one of the listed nodes */ - const uint8_t* rp = data + MEDIA_RELAY_FULL_HDR_SIZE; - uint64_t node_ids[MD_MAX_RELAY_DOWNSTREAM]; - for (int i = 0; i < (int)nrn && i < MD_MAX_RELAY_DOWNSTREAM; i++) { - memcpy(&node_ids[i], rp, 8); rp += 8; +static void md_handle_cancel(struct media_delivery_ctx* md, uint64_t group, uint64_t peer, + const uint8_t* data, size_t len) { + if (len != sizeof(struct media_pkt_cancel)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed CANCEL bytes=%zu", MD_ID, len); return; } - media_download_handle_relay_full(md->inst, rf->media_id, rf->block_id, - rf->chunk, node_ids, (int)nrn); + const struct media_pkt_cancel* can = (const struct media_pkt_cancel*)data; + for (struct stream_ctx* sc = md->streams; sc; sc = sc->next) { + if (sc->group_id == group && sc->dst_node_id == peer && sc->source->chunk == can->chunk + && !memcmp(sc->transfer_id, can->transfer_id, 16) && !memcmp(sc->source->media_id, can->media_id, 16) + && !memcmp(sc->source->block_id, can->block_id, 16)) { + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: TX cancelled peer=%016llx chunk=%u", MD_ID, + (unsigned long long)peer, can->chunk); + stream_free(sc); return; + } + } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: stale CANCEL peer=%016llx chunk=%u", MD_ID, (unsigned long long)peer, can->chunk); } /* ── decrement stream counter (called when stream finished/cancelled internally) ── */ @@ -1288,22 +1219,22 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { case MEDIA_SUBCMD_SERVE_ACK: break; case MEDIA_SUBCMD_BLOCK_REQ: - md_handle_block_req(md, from_node, data, len); + md_handle_block_req(md, group_id, from_node, data, len); break; case MEDIA_SUBCMD_BLOCK_OVERLOADED: - media_download_handle_overloaded(inst, data, len); + media_download_handle_overloaded(inst, group_id, from_node, data, len); break; case MEDIA_SUBCMD_BLOCK_RELAY_FULL: - md_handle_block_relay_full(md, from_node, data, len); + media_download_handle_relay_full(inst, group_id, from_node, data, len); break; case MEDIA_SUBCMD_BLOCK_CHUNK: - media_download_handle_chunk(inst, data, len); + media_download_handle_chunk(inst, group_id, from_node, data, len); break; case MEDIA_SUBCMD_BLOCK_DONE: - media_download_handle_done(inst, data, len); + media_download_handle_done(inst, group_id, from_node, data, len); break; case MEDIA_SUBCMD_CANCEL: - /* TODO: handle cancel from remote */ + md_handle_cancel(md, group_id, from_node, data, len); break; default: DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: unknown subcmd 0x%02x from 0x%016llx", @@ -1408,7 +1339,6 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) { md->initialized = 0; while (md->streams) stream_free(md->streams); - while (md->forwards) relay_fwd_free(md->forwards); etcp_router_unbind(inst, ETCP_RT_ID_MEDIA_DELIVERY); etcp_remove_conn_status_cbk(inst, md_on_conn_status, md); @@ -1436,9 +1366,10 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) { } queue_free(md->downloads); md->downloads = NULL; } - if (md->relay_blocks) { queue_free(md->relay_blocks); md->relay_blocks = NULL; } + if (md->block_sources) { queue_free(md->block_sources); md->block_sources = NULL; } if (md->file_loads) { queue_free(md->file_loads); md->file_loads = NULL; } + if (md->destinations) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: destroy with active destination writers", MD_ID); memset(md, 0, sizeof(*md)); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: destroyed", MD_ID); } diff --git a/src/routing_layer/topo_node_sqlite.c b/src/routing_layer/topo_node_sqlite.c index a6328810..66261fb1 100644 --- a/src/routing_layer/topo_node_sqlite.c +++ b/src/routing_layer/topo_node_sqlite.c @@ -907,18 +907,24 @@ int topo_node_sqlite_node_get_online(sqlite3* db, uint64_t node_id) { int topo_node_sqlite_get_ed25519_pubkey(sqlite3* db, uint64_t node_id, uint8_t pubkey_out[32]) { - if (!db || !pubkey_out) return -1; + if (!db || !pubkey_out) { + DEBUG_ERROR(DEBUG_CATEGORY_ROUTING, "get_ed25519_pubkey: invalid arguments node=%016llx", (unsigned long long)node_id); + return -1; + } sqlite3_stmt* stmt = NULL; - if (sqlite3_prepare_v2(db, "SELECT ed25519_pubkey FROM nodes WHERE node_id=?", - -1, &stmt, NULL) != SQLITE_OK) return -1; - sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id); + int sql_rc = sqlite3_prepare_v2(db, "SELECT ed25519_pubkey FROM nodes WHERE node_id=?", -1, &stmt, NULL); + if (sql_rc == SQLITE_OK) sql_rc = sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id); + if (sql_rc == SQLITE_OK) sql_rc = sqlite3_step(stmt); int rc = -1; - if (sqlite3_step(stmt) == SQLITE_ROW) { + if (sql_rc == SQLITE_ROW) { const void* b = sqlite3_column_blob(stmt, 0); int bytes = sqlite3_column_bytes(stmt, 0); - if (b && bytes >= 32) { uint64_t chk; memcpy(&chk, b, 8); if (chk != 0) { memcpy(pubkey_out, b, 32); rc = 0; } } + if (b && bytes == 32) { memcpy(pubkey_out, b, 32); rc = 0; } } - sqlite3_finalize(stmt); + int final_rc = sqlite3_finalize(stmt); + if (final_rc != SQLITE_OK) rc = -1; + if (rc) DEBUG_WARN(DEBUG_CATEGORY_ROUTING, "get_ed25519_pubkey: unavailable node=%016llx sql=%d finalize=%d: %s", + (unsigned long long)node_id, sql_rc, final_rc, sqlite3_errmsg(db)); return rc; }