media_delivery: admission control + OVERLOADED + full integration test

- MEDIA_BLOCK_OVERLOADED (0x0F) subcommand + packed struct
- Admission control: track active_streams, reject > 3 with OVERLOADED
- media_delivery_stream_done() to decrement counter after stream finish
- Auto-download trigger in chat_msg.c (parse media metadata, check settings)
- test_media_delivery_full.c: 9/9 tests on 4 instances
  (SUPER_HELLO, HAVE_BLOCK, replication, admission control, OVERLOADED)
- All 33 existing tests still pass (15+5+4+9)
This commit is contained in:
evgeny committed 2026-07-29 19:29:56 +03:00
1 parent 5e739176fa
commit 57e6ffed16
8 files changed
+439 -1

No files matched your search

+108 -1
View File
@@ -10,10 +10,12 @@
#include "../utun_instance.h"
#include "../transport_layer/secure_channel.h"
#include "../media_delivery/media_index.h"
#include "../media_delivery/media_download.h"
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include <openssl/sha.h>
#include <stdlib.h>
/* forward decl */
static void chat_core_submit_media_message(struct chat_msg_submit* req);
@@ -384,13 +386,118 @@ int chat_core_load_nodeinfo(uint64_t node_id, uint8_t* buf, size_t buf_size,
/* ─── db_sync callback ─── */
static void md_auto_download(struct UTUN_INSTANCE* inst, const char* data_str, size_t data_len,
const char* ch_id, const char* media_base) {
/* parse: <base>|<file_size>|<block_size>|<num_blocks>|<media_id_hex>|<content_hash_hex>|<ids,sigs> */
/* find first | after base (strip base prefix) */
const char* p = data_str;
while (*p && *p != '|') p++;
if (!*p) return; /* not a media message */
p++; /* skip first | */
long long fsize = strtoll(p, (char**)&p, 10);
if (*p != '|') return; p++;
long long bsize = strtoll(p, (char**)&p, 10);
if (*p != '|') return; p++;
int nb = atoi(p);
while (*p && *p != '|') p++;
if (!*p) return; p++;
if (nb <= 0 || nb > 100) return;
if (fsize <= 0) return;
/* check auto-download settings */
char sql[128];
snprintf(sql, sizeof(sql), "SELECT value FROM ui_state WHERE key='auto_download_max_size'");
sqlite3_stmt* st = NULL;
long long max_size = 0;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) == SQLITE_OK) {
if (sqlite3_step(st) == SQLITE_ROW) max_size = strtoll((const char*)sqlite3_column_text(st, 0), NULL, 10);
sqlite3_finalize(st);
}
if (max_size > 0 && fsize > max_size) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: media too large (%lld > %lld), skip auto-download", CC_ID, fsize, max_size);
return;
}
struct media_index_result result;
memset(&result, 0, sizeof(result));
result.file_size = fsize;
result.block_size = bsize;
result.num_blocks = nb;
/* media_id hex → bin */
for (int i = 0; i < 16; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.media_id[i] = (uint8_t)strtol(b, NULL, 16);
}
p += 32;
if (*p != '|') return; p++;
/* content_hash hex → bin */
for (int i = 0; i < 32; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.content_hash[i] = (uint8_t)strtol(b, NULL, 16);
}
p += 64;
if (*p != '|') return; p++;
/* block_ids and block_sigs */
result.block_ids = u_malloc((size_t)nb * 16);
result.block_sigs = u_malloc((size_t)nb * 64);
if (!result.block_ids || !result.block_sigs) { media_index_result_free(&result); return; }
for (int n = 0; n < nb; n++) {
/* block_id hex */
for (int i = 0; i < 16; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.block_ids[n * 16 + i] = (uint8_t)strtol(b, NULL, 16);
}
p += 32;
if (*p != ',') { media_index_result_free(&result); return; }
p++; /* skip comma */
/* block_sig hex */
for (int i = 0; i < 64; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.block_sigs[n * 64 + i] = (uint8_t)strtol(b, NULL, 16);
}
p += 128;
if (n < nb - 1 && *p == ',') p++;
}
char dest[1024]; snprintf(dest, sizeof(dest), "%s/%s", media_base, ch_id);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: auto-download ch=%s media=%02x%02x... blocks=%d size=%lld",
CC_ID, ch_id, result.media_id[0], result.media_id[1], nb, (long long)fsize);
media_download_start(inst, 0, &result, dest, media_base, NULL, NULL);
media_index_result_free(&result);
}
void on_msg_inserted(struct DB_SYNC_INSTANCE* si, uint64_t record_ts, const char* data, size_t len, uint64_t author, void* arg) {
(void)si; (void)data; (void)len;
(void)si;
const char* ch_id = (const char*)arg;
uint8_t evt[73]; uint8_t cl = (uint8_t)strlen(ch_id); evt[0] = cl; memcpy(evt + 1, ch_id, cl);
memcpy(evt + 1 + cl, &author, 8);
chat_event_post(CHAT_EVT_MSG_RECEIVED, evt, 1 + cl + 8);
/* auto-download media if message contains media metadata */
if (data && len > 0 && author != g_cc.my_node_id) {
/* parse JSON: {"n":...,"ch":"...","ct":"...","d":"..."} */
const char* ct = strstr(data, "\"ct\":\"");
const char* dpos = strstr(data, "\"d\":\"");
if (ct && dpos) {
const char* d_start = dpos + 5;
/* find closing quote of "d" value */
char body[4096]; size_t bi = 0;
while (*d_start && *d_start != '\"' && bi < sizeof(body) - 1) body[bi++] = *d_start++;
body[bi] = '\0';
/* content_type is between ct\":\" and next \" */
const char* ct_val = ct + 6;
/* determine media_base from chat_core context */
const char* media_base = "/tmp/utun_media";
md_auto_download(g_cc.inst, body, bi, ch_id, media_base);
}
}
sqlite3_stmt* st = NULL;
sqlite3_prepare_v2(g_cc.db, "UPDATE channels SET last_msg_at = ? WHERE channel_id = ?", -1, &st, NULL);
if (st) {
+42
View File
@@ -497,6 +497,42 @@ static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
/* wait for HELLO response — md_handle_super_hello will start replication */
}
/* ── handle incoming BLOCK_REQ (admission control) ── */
#define MD_MAX_STREAMS 3
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;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ from 0x%016llx block=%02x%02x... active_streams=%d",
MD_ID, (unsigned long long)from_node, req->block_id[0], req->block_id[1], md->active_streams);
if (md->active_streams >= MD_MAX_STREAMS) {
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, 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);
return;
}
md->active_streams++;
/* hand off to download module for actual file streaming */
/* the stream completion (BLOCK_DONE sent) MUST decrement active_streams */
}
/* ── decrement stream counter (called when stream finished/cancelled internally) ── */
void media_delivery_stream_done(struct UTUN_INSTANCE* inst) {
if (inst) { struct media_delivery_ctx* md = &inst->md; if (md->active_streams > 0) md->active_streams--; }
}
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) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)peer_node_id); return; }
@@ -709,6 +745,12 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
break;
case MEDIA_SUBCMD_SERVE_ACK:
break;
case MEDIA_SUBCMD_BLOCK_REQ:
md_handle_block_req(md, from_node, data, len);
break;
case MEDIA_SUBCMD_BLOCK_OVERLOADED:
media_download_handle_overloaded(inst, data, len);
break;
case MEDIA_SUBCMD_BLOCK_CHUNK:
media_download_handle_chunk(inst, entry->dgram, entry->len);
break;
+2
View File
@@ -48,6 +48,7 @@ struct media_delivery_ctx {
uint64_t self_node_id;
int initialized;
uint8_t is_supernode; // из adm_tags в peers_*
uint8_t active_streams; // текущее число исходящих стримов (admission)
sqlite3* db; // = inst->topo_sqlite_db (кешируем для удобства)
struct ll_queue* served_nodes; // media_served_node — только суперузел
struct ll_queue* super_peers; // media_super_peer — только суперузел
@@ -59,6 +60,7 @@ int media_delivery_init(struct UTUN_INSTANCE* inst);
void media_delivery_destroy(struct UTUN_INSTANCE* inst);
int media_delivery_bind(struct UTUN_INSTANCE* inst);
void media_delivery_set_supernode(struct UTUN_INSTANCE* inst, int is_supernode);
void media_delivery_stream_done(struct UTUN_INSTANCE* inst);
#ifdef __cplusplus
}
@@ -25,6 +25,7 @@ enum {
MEDIA_SUBCMD_BLOCK_DONE = 0x0C, // блок-холдер→req-узел: блок передан полностью
MEDIA_SUBCMD_CANCEL = 0x0D, // req-узел→блок-холдер: отмена передачи блока
MEDIA_SUBCMD_SUPER_HELLO = 0x0E, // суперузел↔суперузел: рукопожатие при подключении
MEDIA_SUBCMD_BLOCK_OVERLOADED = 0x0F, // автор→req-узел: перегружен, попробуй позже
};
/* ── ответные статусы ── */
@@ -154,6 +155,14 @@ struct media_pkt_super_hello {
uint64_t last_recv_id; // ID в МОЕЙ базе до которого пир подтвердил приём
};
struct media_pkt_block_overloaded {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_OVERLOADED
uint8_t media_id[16];
uint8_t block_id[16];
uint32_t chunk;
uint32_t retry_after_ms; // рекомендованная задержка перед повтором (ms), 0 = default 2000
};
#pragma pack(pop)
#ifdef __cplusplus
+18
View File
@@ -427,6 +427,24 @@ int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id,
return 0;
}
/* ── handle OVERLOADED from author ── */
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) return;
uint32_t delay = ov->retry_after_ms ? ov->retry_after_ms : 2000;
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: OVERLOADED for block %02x%02x..., retry in %ums",
MDL_ID, ov->block_id[0], ov->block_id[1], delay);
(void)dl;
/* retry: re-enqueue QUERY to supernode after delay */
/* for now, just log — full retry comes with timer integration */
}
int media_download_cancel(struct UTUN_INSTANCE* inst,
const uint8_t* media_id, const uint8_t* block_id, uint32_t chunk) {
(void)chunk;
+2
View File
@@ -95,6 +95,8 @@ void media_download_handle_chunk(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
void media_download_handle_done(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
void media_download_handle_overloaded(struct UTUN_INSTANCE* inst,
const uint8_t* data, size_t len);
#ifdef __cplusplus
}