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

media_delivery: add tests + fix queue_new NULL ua and index_offset bugs

Browse Source 
- test_media_delivery_sql: 15 tests (SQL tables, CRUD, packed structs)
- test_media_delivery_download: 5 tests (chunk write/append, sig verify, assembly, cancel)
- Fix: queue_new(NULL) returns NULL — pass inst->ua everywhere
- Fix: queue index offset must be offsetof(struct, key_field) not 0
- Move struct media_download to header for test visibility
This commit is contained in:
evgeny committed 2026-07-29 17:53:46 +03:00
1 parent fb3e34080d
commit a863da7737
6 files changed
+842 -116
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+60 -24
View File
@@ -62,7 +62,10 @@ static int md_ba_insert(sqlite3* db, const uint8_t* block_uuid, uint64_t group_i
"INSERT OR REPLACE INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp)"
" VALUES(?,?,?,?,?)";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return -1;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ba_insert prepare failed: %s", MD_ID, sqlite3_errmsg(db));
return -1;
}
sqlite3_bind_blob(st, 1, block_uuid, 16, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)group_id);
sqlite3_bind_int64(st, 3, (sqlite3_int64)node_id);
@@ -70,7 +73,11 @@ static int md_ba_insert(sqlite3* db, const uint8_t* block_uuid, uint64_t group_i
sqlite3_bind_int64(st, 5, (sqlite3_int64)ts);
int rc = sqlite3_step(st);
sqlite3_finalize(st);
return (rc == SQLITE_DONE) ? 0 : -1;
if (rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ba_insert step failed rc=%d: %s", MD_ID, rc, sqlite3_errmsg(db));
return -1;
}
return 0;
}
static int md_ba_find_nodes(sqlite3* db, const uint8_t* block_uuid,
@@ -78,7 +85,10 @@ static int md_ba_find_nodes(sqlite3* db, const uint8_t* block_uuid,
const char* sql =
"SELECT node_id, timestamp FROM block_availability WHERE block_uuid=? ORDER BY timestamp DESC LIMIT ?";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return 0;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ba_find_nodes prepare failed: %s", MD_ID, sqlite3_errmsg(db));
return 0;
}
sqlite3_bind_blob(st, 1, block_uuid, 16, SQLITE_STATIC);
sqlite3_bind_int(st, 2, max);
int n = 0;
@@ -96,7 +106,10 @@ static int md_ba_get_since(sqlite3* db, uint64_t since_id,
"SELECT id,block_uuid,group_id,node_id,chunk,timestamp"
" FROM block_availability WHERE id > ? ORDER BY id ASC LIMIT 4";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return 0;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ba_get_since prepare failed: %s", MD_ID, sqlite3_errmsg(db));
return 0;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)since_id);
int off = 0;
while (sqlite3_step(st) == SQLITE_ROW) {
@@ -116,7 +129,10 @@ static int md_ba_get_since(sqlite3* db, uint64_t since_id,
static uint64_t md_ss_get(sqlite3* db, uint64_t peer_node_id) {
const char* sql = "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return 0;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ss_get prepare failed: %s", MD_ID, sqlite3_errmsg(db));
return 0;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)peer_node_id);
uint64_t v = 0;
if (sqlite3_step(st) == SQLITE_ROW) v = (uint64_t)sqlite3_column_int64(st, 0);
@@ -127,7 +143,10 @@ static uint64_t md_ss_get(sqlite3* db, uint64_t peer_node_id) {
static void md_ss_set(sqlite3* db, uint64_t peer_node_id, uint64_t last_recv_id) {
const char* sql = "INSERT OR REPLACE INTO super_sync(peer_node_id,last_recv_id) VALUES(?,?)";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return;
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: ss_set prepare failed: %s", MD_ID, sqlite3_errmsg(db));
return;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)peer_node_id);
sqlite3_bind_int64(st, 2, (sqlite3_int64)last_recv_id);
sqlite3_step(st);
@@ -146,11 +165,11 @@ static struct media_super_peer* md_super_peer_add(struct media_delivery_ctx* md,
struct media_super_peer* sp = md_super_peer_find(md, node_id);
if (sp) return sp;
if (!md->super_peers) {
md->super_peers = queue_new(NULL, 64, 0, 8, "md_super");
if (!md->super_peers) return NULL;
md->super_peers = queue_new(md->inst->ua, 64, offsetof(struct media_super_peer, peer_node_id), 8, "md_super");
if (!md->super_peers) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(super_peers) failed", MD_ID); return NULL; }
}
struct ll_entry* qe = queue_entry_new(sizeof(struct media_super_peer));
if (!qe) return NULL;
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(super_peer) failed", MD_ID); return NULL; }
sp = (struct media_super_peer*)qe->data;
memset(sp, 0, sizeof(*sp));
sp->peer_node_id = node_id;
@@ -274,13 +293,13 @@ static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t from_nod
struct media_pkt_serve_reg* pkt = (struct media_pkt_serve_reg*)data;
if (!md->served_nodes) {
md->served_nodes = queue_new(NULL, 64, 0, 8, "md_served");
if (!md->served_nodes) return;
md->served_nodes = queue_new(md->inst->ua, 64, offsetof(struct media_served_node, node_id), 8, "md_served");
if (!md->served_nodes) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(served_nodes) failed", MD_ID); return; }
}
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node);
if (!e) {
struct ll_entry* qe = queue_entry_new(sizeof(struct media_served_node));
if (!qe) return;
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(served_node) failed", MD_ID); return; }
struct media_served_node* sn = (struct media_served_node*)qe->data;
memset(sn, 0, sizeof(*sn));
sn->node_id = from_node; sn->group_id = pkt->group_id;
@@ -298,7 +317,8 @@ static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t from_nod
static void md_handle_query(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < MEDIA_QUERY_HDR_SIZE || !md->db) return;
if (len < MEDIA_QUERY_HDR_SIZE) return;
if (!md->db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: QUERY but no DB", MD_ID); return; }
struct media_pkt_query* q = (struct media_pkt_query*)data;
uint16_t nb = q->num_blocks;
if (nb > 64) nb = 64;
@@ -332,7 +352,8 @@ static void md_handle_query(struct media_delivery_ctx* md, uint64_t from_node,
static void md_handle_have_block(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < MEDIA_HAVE_BLOCK_SIZE || !md->db) return;
if (len < MEDIA_HAVE_BLOCK_SIZE) return;
if (!md->db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: HAVE_BLOCK but no DB", MD_ID); return; }
struct media_pkt_have_block* hb = (struct media_pkt_have_block*)data;
int rc = md_ba_insert(md->db, hb->block_id, hb->group_id, from_node, hb->chunk, hb->timestamp);
@@ -364,7 +385,8 @@ static void md_handle_have_block(struct media_delivery_ctx* md, uint64_t from_no
static void md_handle_super_repl(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < MEDIA_SUPER_REPL_HDR_SIZE || !md->db) return;
if (len < MEDIA_SUPER_REPL_HDR_SIZE) return;
if (!md->db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: SUPER_REPL but no DB", MD_ID); return; }
struct media_pkt_super_repl* rp = (struct media_pkt_super_repl*)data;
uint16_t num = rp->num_entries;
int64_t max_id = 0;
@@ -399,7 +421,7 @@ static void md_handle_super_ack(struct media_delivery_ctx* md, uint64_t from_nod
struct media_pkt_super_ack* sa = (struct media_pkt_super_ack*)data;
struct media_super_peer* peer = md_super_peer_find(md, from_node);
if (!peer) return;
if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: SUPER_ACK from unknown peer 0x%016llx", MD_ID, (unsigned long long)from_node); return; }
peer->peer_last_recv_id = sa->ack_seq;
peer->inflight_count--;
@@ -413,11 +435,12 @@ static void md_handle_super_ack(struct media_delivery_ctx* md, uint64_t from_nod
static void md_handle_super_hello(struct media_delivery_ctx* md, uint64_t from_node,
const uint8_t* data, size_t len) {
if (len < sizeof(struct media_pkt_super_hello) || !md->db) return;
if (len < sizeof(struct media_pkt_super_hello)) return;
if (!md->db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: SUPER_HELLO but no DB", MD_ID); return; }
struct media_pkt_super_hello* sh = (struct media_pkt_super_hello*)data;
struct media_super_peer* peer = md_super_peer_add(md, from_node);
if (!peer) return;
if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)from_node); return; }
peer->peer_last_recv_id = sh->last_recv_id;
peer->connected = 1;
@@ -443,7 +466,7 @@ static void md_handle_super_hello(struct media_delivery_ctx* md, uint64_t from_n
static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg;
struct media_super_peer* peer = md_super_peer_find(md, node_id);
if (!peer) return;
if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: conn_cb for unknown peer 0x%016llx", MD_ID, (unsigned long long)node_id); return; }
if (result != CONN_MGR_OK) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: connect to supernode 0x%016llx failed rc=%d",
@@ -466,13 +489,16 @@ static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
if (md_send(md->inst, node_id, (const uint8_t*)&hello, sizeof(hello)) == 0) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: SUPER_HELLO sent to 0x%016llx last_recv=%lld",
MD_ID, (unsigned long long)node_id, (long long)my_last);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: SUPER_HELLO send failed to 0x%016llx",
MD_ID, (unsigned long long)node_id);
}
/* wait for HELLO response — md_handle_super_hello will start replication */
}
static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id) {
struct media_super_peer* peer = md_super_peer_add(md, peer_node_id);
if (!peer) return;
if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)peer_node_id); return; }
struct TOPO_GROUP* grp = NULL;
/* find group containing this peer */
@@ -495,7 +521,10 @@ static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_i
/* ── supernode start/stop ── */
static int md_super_start(struct media_delivery_ctx* md) {
if (!md->inst || !md->inst->topo_sqlite_db || !md->inst->topo_groups) return -1;
if (!md->inst || !md->inst->topo_sqlite_db || !md->inst->topo_groups) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_start missing inst/db/groups", MD_ID);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: supernode START node_id=0x%016llx",
MD_ID, (unsigned long long)md->self_node_id);
@@ -524,7 +553,7 @@ static int md_super_start(struct media_delivery_ctx* md) {
}
static void md_super_stop(struct media_delivery_ctx* md) {
if (!md->db) return;
if (!md->db) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: super_stop but no DB", MD_ID); return; }
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: supernode STOP, cleaning foreign blocks", MD_ID);
@@ -631,7 +660,7 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
struct UTUN_INSTANCE* inst = conn->instance;
if (!inst) return;
struct media_delivery_ctx* md = &inst->md;
if (!md->initialized) return;
if (!md->initialized) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: recv before init", MD_ID); return; }
const uint8_t* data = entry->dgram;
size_t len = entry->len;
@@ -646,6 +675,8 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
md_handle_serve_reg(md, from_node, data, len);
break;
case MEDIA_SUBCMD_SERVE_LEAVE:
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: SERVE_LEAVE from 0x%016llx",
MD_ID, (unsigned long long)from_node);
if (md->served_nodes) {
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &from_node);
if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); }
@@ -653,18 +684,23 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
break;
case MEDIA_SUBCMD_QUERY:
if (md->is_supernode) md_handle_query(md, from_node, data, len);
else DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: QUERY ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_HAVE_BLOCK:
if (md->is_supernode) md_handle_have_block(md, from_node, data, len);
else DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: HAVE_BLOCK ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_SUPER_REPL:
if (md->is_supernode) md_handle_super_repl(md, from_node, data, len);
else DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: SUPER_REPL ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_SUPER_ACK:
if (md->is_supernode) md_handle_super_ack(md, from_node, data, len);
else DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: SUPER_ACK ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_SUPER_HELLO:
if (md->is_supernode) md_handle_super_hello(md, from_node, data, len);
else DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: SUPER_HELLO ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_QUERY_RESP:
media_download_handle_query_resp(inst, entry->dgram, entry->len);
@@ -727,7 +763,7 @@ static int media_delivery_create_tables(struct UTUN_INSTANCE* inst) {
/* ── public API ── */
int media_delivery_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: init with NULL inst", MD_ID); return -1; }
struct media_delivery_ctx* md = &inst->md;
memset(md, 0, sizeof(*md));
src/media_delivery/media_download.c
+58 -87
View File
@@ -19,39 +19,6 @@
#define MDL_ID "media_download"
/* ── internal download state ── */
struct media_download {
struct ll_entry ll; // индекс по media_id (16 байт)
uint8_t media_id[16];
uint64_t group_id;
char dest_path[1024];
char media_base[512];
int num_blocks;
uint8_t* block_ids; // num_blocks * 16
uint8_t* block_sigs; // num_blocks * 64
uint8_t content_hash[32];
int64_t file_size;
int64_t block_size;
int blocks_received;
int blocks_validated;
int num_peers;
struct media_download_peer peers[10];
uint8_t active;
uint8_t assembled;
int err;
/* supernode list */
uint64_t super_nodes[10];
int super_count;
int super_current;
void* query_timer;
void* timeout_timer;
void (*done_cb)(void* arg, int err);
void* done_arg;
};
/* ── helpers ── */
static struct media_download* md_dl_find(struct UTUN_INSTANCE* inst, const uint8_t* media_id) {
@@ -73,12 +40,12 @@ static void md_dl_free(struct media_download* dl) {
static int md_dl_send(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, size_t len) {
struct ll_entry* e = queue_entry_new(0);
if (!e) return -1;
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new failed for send to 0x%016llx", MDL_ID, (unsigned long long)dst); return -1; }
e->dgram = u_malloc(len);
if (!e->dgram) { queue_entry_free(e); return -1; }
if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_malloc(%zu) failed for send to 0x%016llx", MDL_ID, len, (unsigned long long)dst); queue_entry_free(e); return -1; }
memcpy(e->dgram, data, len); e->len = (uint16_t)len;
int rc = etcp_route_send(inst, 0, dst, e, 1);
if (rc != 0) { u_free(e->dgram); queue_entry_free(e); }
if (rc != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: etcp_route_send failed rc=%d to 0x%016llx", MDL_ID, rc, (unsigned long long)dst); u_free(e->dgram); queue_entry_free(e); }
return rc;
}
@@ -103,7 +70,7 @@ static int md_dl_send_block_req(struct UTUN_INSTANCE* inst, uint64_t dst,
static void md_dl_send_have_block(struct UTUN_INSTANCE* inst, struct media_download* dl, int bi) {
uint64_t super = dl->super_count > 0 ? dl->super_nodes[dl->super_current] : 0;
if (!super) return;
if (!super) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: HAVE_BLOCK but no supernode available", MDL_ID); return; }
struct media_pkt_have_block hb;
memset(&hb, 0, sizeof(hb));
@@ -114,13 +81,15 @@ static void md_dl_send_have_block(struct UTUN_INSTANCE* inst, struct media_downl
hb.chunk = (uint32_t)bi;
hb.timestamp = (int64_t)time(NULL);
/* sign: block_id + chunk + timestamp + my_node_id */
uint8_t smsg[64]; size_t soff = 0;
memcpy(smsg + soff, dl->block_ids + bi * 16, 16); soff += 16;
memcpy(smsg + soff, &hb.chunk, 4); soff += 4;
memcpy(smsg + soff, &hb.timestamp, 8); soff += 8;
uint64_t self = inst->node_id; memcpy(smsg + soff, &self, 8); soff += 8;
sc_ed25519_sign(inst->my_ed25519_privkey, smsg, soff, hb.node_sign);
if (sc_ed25519_sign(inst->my_ed25519_privkey, smsg, soff, hb.node_sign) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: Ed25519 sign failed for HAVE_BLOCK", MDL_ID);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: HAVE_BLOCK to super 0x%016llx block=%d",
MDL_ID, (unsigned long long)super, bi);
@@ -191,7 +160,7 @@ static void md_dl_send_query(struct UTUN_INSTANCE* inst, struct media_download*
size_t pkt_len = MEDIA_QUERY_HDR_SIZE + (size_t)dl->num_blocks * 16;
uint8_t* pkt = u_malloc(pkt_len);
if (!pkt) return;
if (!pkt) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_malloc(%zu) failed for QUERY pkt", MDL_ID, pkt_len); return; }
struct media_pkt_query* q = (struct media_pkt_query*)pkt;
memset(q, 0, sizeof(*q));
q->subcmd = MEDIA_SUBCMD_QUERY;
@@ -221,9 +190,8 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
const uint8_t* bid = entries[0].block_id;
struct media_download* dl = md_dl_find(inst, bid);
if (!dl) {
/* search all blocks */
for (int i = 0; i < ne; i++) { dl = md_dl_find(inst, entries[i].block_id); if (dl) break; }
if (!dl) return;
if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: QUERY_RESP — no matching download", MDL_ID); return; }
}
dl->num_peers = 0;
@@ -260,7 +228,7 @@ void media_download_handle_query_resp(struct UTUN_INSTANCE* inst,
if (g->conn_mgr) { grp = g; break; }
gle = gle->next;
}
if (!grp || !grp->conn_mgr) return;
if (!grp || !grp->conn_mgr) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: QUERY_RESP — no conn_mgr to connect peers", MDL_ID); return; }
for (int pi = 0; pi < dl->num_peers; pi++) {
conn_mgr_connect_node(grp->conn_mgr, dl->peers[pi].node_id, 0, md_dl_conn_cb, dl);
@@ -276,26 +244,27 @@ void media_download_handle_chunk(struct UTUN_INSTANCE* inst,
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 || !dl->active) return;
if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%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_DEBUG, "%s: CHUNK for inactive download", MDL_ID); return; }
/* find block index */
int bi = -1;
for (int i = 0; i < dl->num_blocks; i++) {
if (memcmp(dl->block_ids + i * 16, ch->block_id, 16) == 0) { bi = i; break; }
}
if (bi < 0) return;
if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: CHUNK for unknown block %02x%02x...", MDL_ID, ch->block_id[0], ch->block_id[1]); return; }
/* write chunk to temp file */
char tmp[2048];
snprintf(tmp, sizeof(tmp), "%s.chunk_%d", dl->dest_path, bi);
FILE* f = fopen(tmp, "ab");
if (f) {
size_t wlen = ch->data_len;
if (len >= MEDIA_BLOCK_CHUNK_HDR_SIZE + wlen) {
fwrite(d + MEDIA_BLOCK_CHUNK_HDR_SIZE, 1, wlen, f);
}
fclose(f);
if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed for chunk write", MDL_ID, tmp); return; }
size_t wlen = ch->data_len;
if (len >= MEDIA_BLOCK_CHUNK_HDR_SIZE + wlen) {
fwrite(d + MEDIA_BLOCK_CHUNK_HDR_SIZE, 1, wlen, f);
} else {
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: CHUNK truncated: len=%zu need=%zu+%zu",
MDL_ID, len, (size_t)MEDIA_BLOCK_CHUNK_HDR_SIZE, (size_t)wlen);
}
fclose(f);
}
/* ── handle incoming BLOCK_DONE ── */
@@ -307,13 +276,14 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst,
const uint8_t* d = data;
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 || !dl->active) return;
if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_DONE for unknown media", MDL_ID); return; }
if (!dl->active) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_DONE for inactive download", MDL_ID); return; }
int bi = -1;
for (int i = 0; i < dl->num_blocks; i++) {
if (memcmp(dl->block_ids + i * 16, bd->block_id, 16) == 0) { bi = i; break; }
}
if (bi < 0) return;
if (bi < 0) { DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_DONE for unknown block", MDL_ID); return; }
/* verify signature */
EVP_PKEY* pkey = NULL;
@@ -324,31 +294,27 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst,
char tmp[2048];
snprintf(tmp, sizeof(tmp), "%s.chunk_%d", dl->dest_path, bi);
FILE* f = fopen(tmp, "rb");
if (f) {
fseeko(f, 0, SEEK_END);
off_t fsz = ftello(f);
fseeko(f, 0, SEEK_SET);
uint8_t* buf = u_malloc((size_t)fsz);
if (buf) {
size_t rd = fread(buf, 1, (size_t)fsz, f);
uint64_t node_id = inst->node_id;
uint8_t smsg[128]; size_t soff = 0;
memcpy(smsg + soff, buf, rd); soff += rd;
memcpy(smsg + soff, &node_id, 8); soff += 8;
EVP_DigestVerifyInit(vctx, NULL, EVP_sha256(), NULL, pkey);
/* fallback: compare raw block_sig */
sig_ok = (rd == (size_t)bd->total_size && rd > 0) ? 1 : 0;
if (sig_ok && rd == (size_t)bd->total_size) {
/* basic check: block_sig should match pre-computed */
if (memcmp(bd->block_sig, dl->block_sigs + bi * 64, 64) == 0) sig_ok = 1;
else sig_ok = 0;
}
u_free(buf);
}
fclose(f);
if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed for sig check", MDL_ID, tmp); EVP_MD_CTX_free(vctx); return; }
fseeko(f, 0, SEEK_END);
off_t fsz = ftello(f);
fseeko(f, 0, SEEK_SET);
uint8_t* buf = u_malloc((size_t)fsz);
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_malloc(%lld) failed for sig check", MDL_ID, (long long)fsz); fclose(f); EVP_MD_CTX_free(vctx); return; }
size_t rd = fread(buf, 1, (size_t)fsz, f);
uint64_t node_id = inst->node_id;
uint8_t smsg[128]; size_t soff = 0;
memcpy(smsg + soff, buf, rd); soff += rd;
memcpy(smsg + soff, &node_id, 8); soff += 8;
EVP_DigestVerifyInit(vctx, NULL, EVP_sha256(), NULL, pkey);
sig_ok = (rd == (size_t)bd->total_size && rd > 0) ? 1 : 0;
if (sig_ok && rd == (size_t)bd->total_size) {
if (memcmp(bd->block_sig, dl->block_sigs + bi * 64, 64) == 0) sig_ok = 1;
else sig_ok = 0;
}
EVP_MD_CTX_free(vctx);
u_free(buf);
fclose(f);
}
EVP_MD_CTX_free(vctx);
if (sig_ok) {
dl->blocks_received++;
@@ -363,7 +329,8 @@ void media_download_handle_done(struct UTUN_INSTANCE* inst,
if (dl->blocks_validated == dl->num_blocks) {
/* assemble file */
FILE* out = fopen(dl->dest_path, "wb");
if (out) {
if (!out) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: fopen(%s) failed for assembly", MDL_ID, dl->dest_path); }
else {
for (int n = 0; n < dl->num_blocks; n++) {
char tmp2[2048];
snprintf(tmp2, sizeof(tmp2), "%s.chunk_%d", dl->dest_path, n);
@@ -408,12 +375,16 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id,
const struct media_index_result* result,
const char* dest_path, const char* media_base,
void (*done_cb)(void* arg, int err), void* done_arg) {
if (!inst || !result || !dest_path || !media_base || !done_cb) return -1;
if (!inst || !result || !dest_path || !media_base || !done_cb) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: start with NULL args inst=%p result=%p dest=%p base=%p cb=%p",
MDL_ID, (void*)inst, (void*)result, (void*)dest_path, (void*)media_base, (void*)done_cb);
return -1;
}
struct media_delivery_ctx* md = &inst->md;
if (!md->initialized) return -1;
if (!md->initialized) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: start before media_delivery init", MDL_ID); return -1; }
struct media_download* dl = u_calloc(1, sizeof(*dl));
if (!dl) return -1;
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;
@@ -429,18 +400,18 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id,
dl->block_ids = u_malloc((size_t)dl->num_blocks * 16);
dl->block_sigs = u_malloc((size_t)dl->num_blocks * 64);
if (!dl->block_ids || !dl->block_sigs) { u_free(dl->block_ids); u_free(dl); return -1; }
if (!dl->block_ids || !dl->block_sigs) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: u_malloc for block_ids/sigs failed", MDL_ID); u_free(dl->block_ids); u_free(dl); return -1; }
memcpy(dl->block_ids, result->block_ids, (size_t)dl->num_blocks * 16);
memcpy(dl->block_sigs, result->block_sigs, (size_t)dl->num_blocks * 64);
/* register in downloads queue */
if (!md->downloads) {
md->downloads = queue_new(NULL, 256, 0, 16, "md_dl");
if (!md->downloads) { md_dl_free(dl); u_free(dl); return -1; }
md->downloads = queue_new(inst->ua, 256, offsetof(struct media_download, media_id), 16, "md_dl");
if (!md->downloads) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_new(downloads) failed", MDL_ID); md_dl_free(dl); u_free(dl); return -1; }
}
memcpy(dl->ll.data, dl->media_id, 16);
struct ll_entry* qe = queue_entry_new(sizeof(struct media_download));
if (!qe) { md_dl_free(dl); u_free(dl); return -1; }
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: queue_entry_new(download) failed", MDL_ID); md_dl_free(dl); u_free(dl); return -1; }
memcpy(qe->data, dl, sizeof(*dl));
u_free(dl);
queue_data_put_with_index(md->downloads, qe);
@@ -459,7 +430,7 @@ int media_download_cancel(struct UTUN_INSTANCE* inst,
const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk) {
(void)chunk;
struct media_download* dl = md_dl_find(inst, media_id);
if (!dl) return -1;
if (!dl) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cancel — download not found", MDL_ID); return -1; }
/* send CANCEL to all connected peers */
struct media_pkt_cancel can;
src/media_delivery/media_download.h
+38 -5
View File
@@ -9,6 +9,10 @@ extern "C" {
#include <stdint.h>
#include <stddef.h>
#include <stdio.h>
#include "../lib/ll_queue.h"
struct UTUN_INSTANCE;
struct media_index_result;
struct UTUN_INSTANCE;
struct media_index_result;
@@ -18,16 +22,45 @@ struct media_index_result;
struct media_download_peer {
uint64_t node_id;
uint8_t connected; // 1 = conn_mgr подключён
int num_blocks; // сколько блоков назначено этому узлу
uint8_t connected;
int num_blocks;
struct {
uint8_t block_id[16];
uint8_t started; // 1 = BLOCK_REQ отправлен
uint8_t received; // 1 = BLOCK_DONE получен
uint8_t validated; // 1 = подпись проверена
uint8_t started;
uint8_t received;
uint8_t validated;
} blocks[MD_MAX_BLOCKS_PER_PEER];
};
/* внутреннее состояние загрузки (доступно для тестов) */
struct media_download {
struct ll_entry ll;
uint8_t media_id[16];
uint64_t group_id;
char dest_path[1024];
char media_base[512];
int num_blocks;
uint8_t* block_ids;
uint8_t* block_sigs;
uint8_t content_hash[32];
int64_t file_size;
int64_t block_size;
int blocks_received;
int blocks_validated;
int num_peers;
struct media_download_peer peers[10];
uint8_t active;
uint8_t assembled;
int err;
uint64_t super_nodes[10];
int super_count;
int super_current;
void* query_timer;
void* timeout_timer;
void (*done_cb)(void* arg, int err);
void* done_arg;
};
/* состояние одного стрима на стороне блок-холдера */
struct media_block_stream {
uint8_t media_id[16];