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

Define bounded media QUERY batches and complete responses

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 18:31:12 +02:00
1 parent cecbc682a0
commit 6267973d62
2 files changed
+71 -81
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/media_delivery/media_delivery.c
+53 -72
View File
@@ -121,24 +121,34 @@ static int md_ba_complete_block(sqlite3* db, const uint8_t* block_uuid,
return md_ba_insert(db, block_uuid, group_id, node_id, chunk, ts, 1);
}
static int md_ba_find_nodes(sqlite3* db, const uint8_t* block_uuid,
uint64_t* out_node_ids, int max) {
const char* sql =
"SELECT node_id, timestamp FROM block_availability WHERE block_uuid=? ORDER BY timestamp DESC LIMIT ?";
/* Выбрать ограниченный список кандидатов: свой готовый блок, затем свежие HAVE/PROCESSING.
* Исключения применяются до LIMIT; processing сохраняется для промежуточной раздачи. */
static int md_query_holders(struct media_delivery_ctx* md, uint64_t group_id, uint64_t requester,
const uint8_t* media_id, struct media_pkt_query_resp_entry* result) {
sqlite3* db = md->db;
const char* sql = "SELECT node_id FROM ("
" SELECT node_id,0 AS own,timestamp FROM block_availability WHERE block_uuid=? AND group_id=?"
" UNION ALL SELECT node_id,1 AS own,timestamp FROM media_files WHERE block_id=? AND media_id=? AND node_id=?"
") WHERE node_id!=0 AND node_id!=? GROUP BY node_id ORDER BY MAX(own) DESC,MAX(timestamp) DESC,node_id ASC LIMIT ?";
sqlite3_stmt* st = NULL;
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;
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: QUERY prepare failed: %s", MD_ID, sqlite3_errmsg(db)); return -1;
}
sqlite3_bind_blob(st, 1, block_uuid, 16, SQLITE_STATIC);
sqlite3_bind_int(st, 2, max);
int n = 0;
while (n < max && sqlite3_step(st) == SQLITE_ROW) {
out_node_ids[n] = (uint64_t)sqlite3_column_int64(st, 0);
n++;
int rc = sqlite3_bind_blob(st, 1, result->block_id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)group_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, result->block_id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 4, media_id, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 5, (sqlite3_int64)md->self_node_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 6, (sqlite3_int64)requester);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 7, MEDIA_QUERY_MAX_HOLDERS);
if (rc == SQLITE_OK) {
while ((rc = sqlite3_step(st)) == SQLITE_ROW)
result->holders[result->num_holders++] = (uint64_t)sqlite3_column_int64(st, 0);
}
sqlite3_finalize(st);
return n;
int final_rc = sqlite3_finalize(st);
if (rc == SQLITE_DONE && final_rc == SQLITE_OK) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: QUERY SQL failed rc=%d finalize=%d: %s", MD_ID, rc, final_rc, sqlite3_errmsg(db));
return -1;
}
static int md_ba_get_since(sqlite3* db, uint64_t since_id,
@@ -417,7 +427,6 @@ static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_n
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: md_send FAIL rc=%d dst=0x%016llx group=0x%016llx",
MD_ID, rc, (unsigned long long)dst_node_id, (unsigned long long)group_id);
u_free(e->dgram); queue_entry_free(e);
}
return rc;
}
@@ -452,63 +461,34 @@ static void md_handle_serve_reg(struct media_delivery_ctx* md, uint64_t from_nod
MD_ID, (unsigned long long)from_node, (unsigned long long)pkt->group_id);
}
static void md_handle_query(struct media_delivery_ctx* md, uint64_t from_node,
static void md_handle_query(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node,
const uint8_t* data, size_t len) {
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;
uint8_t* bids = (uint8_t*)data + MEDIA_QUERY_HDR_SIZE;
uint8_t resp[2048]; int off = 0;
resp[off++] = MEDIA_SUBCMD_QUERY_RESP;
int num_pos = off; off += 2;
int count = 0;
for (int i = 0; i < nb && off + 30 <= (int)sizeof(resp); i++) {
uint8_t* block_id = bids + i * 16;
uint64_t nodes[10]; int nf = 0;
/* 1) block_availability (supernode or HAVE_BLOCK reports) */
nf = md_ba_find_nodes(md->db, block_id, nodes, 10);
/* 2) media_files: собственные блоки (автор-не-суперузел тоже отдаёт файлы).
Проверяем всегда — иначе при наличии чужих записей в block_availability
собственный блок автора не попал бы в список держателей. */
{
const char* sql = "SELECT 1 FROM media_files WHERE block_id=? AND node_id=? LIMIT 1";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(md->db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_blob(st, 1, block_id, 16, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)md->self_node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
int dup = 0;
for (int j = 0; j < nf; j++) if (nodes[j] == md->self_node_id) { dup = 1; break; }
if (!dup && nf < 10) nodes[nf++] = md->self_node_id;
}
sqlite3_finalize(st);
}
}
for (int j = 0; j < nf && off + 30 <= (int)sizeof(resp); j++) {
if (nodes[j] == from_node) continue; /* сам себе держателем не бывает */
uint16_t rtt = 0;
memcpy(resp + off, &nodes[j], 8); off += 8;
memcpy(resp + off, &rtt, 2); off += 2;
memcpy(resp + off, block_id, 16); off += 16;
uint32_t ch = q->num_blocks;
memcpy(resp + off, &ch, 4); off += 4;
count++;
if (len < MEDIA_QUERY_HDR_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: short QUERY bytes=%zu", MD_ID, len); return;
}
const struct media_pkt_query* q = (const struct media_pkt_query*)data;
if (!q->num_blocks || q->num_blocks > MEDIA_QUERY_MAX_BLOCKS
|| len != MEDIA_QUERY_HDR_SIZE + (size_t)q->num_blocks * 16) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: invalid QUERY blocks=%u bytes=%zu", MD_ID, q->num_blocks, len); return;
}
uint8_t response[MEDIA_QUERY_RESP_HDR_SIZE + MEDIA_QUERY_MAX_BLOCKS * sizeof(struct media_pkt_query_resp_entry)] = {0};
struct media_pkt_query_resp* r = (struct media_pkt_query_resp*)response;
r->subcmd = MEDIA_SUBCMD_QUERY_RESP;
memcpy(r->media_id, q->media_id, 16); memcpy(r->request_id, q->request_id, 16);
r->num_blocks = q->num_blocks;
struct media_pkt_query_resp_entry* results = (struct media_pkt_query_resp_entry*)(response + MEDIA_QUERY_RESP_HDR_SIZE);
for (unsigned i = 0; i < q->num_blocks; i++) {
memcpy(results[i].block_id, data + MEDIA_QUERY_HDR_SIZE + i * 16, 16);
if (!md->db || md_query_holders(md, group_id, from_node, q->media_id, &results[i])) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: QUERY batch failed group=%016llx block=%u db=%p",
MD_ID, (unsigned long long)group_id, i, (void*)md->db);
r->status = MEDIA_QUERY_DB_ERROR; r->num_blocks = 0; break;
}
}
uint16_t nc = (uint16_t)count;
memcpy(resp + num_pos, &nc, 2);
uint64_t resp_group = q->group_id ? q->group_id : TOPO_GROUP_UTUN;
md_send(md->inst, resp_group, from_node, resp, (size_t)off);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: QUERY from 0x%016llx → %d entries (super=%d)",
MD_ID, (unsigned long long)from_node, count, md->is_supernode);
int rc = md_send(md->inst, group_id, from_node, response,
MEDIA_QUERY_RESP_HDR_SIZE + (size_t)r->num_blocks * sizeof(*results));
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: QUERY peer=%016llx group=%016llx blocks=%u status=%u send=%d",
MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, q->num_blocks, r->status, rc);
}
static void md_handle_have_block(struct media_delivery_ctx* md, uint64_t from_node,
@@ -1256,8 +1236,9 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
size_t len = entry->len;
/* Единый формат router_deliver: [svc_id][src_node_id(8)][dst_node_id(8)][subcmd][...] */
if (len < ROUTER_SVC_PAYLOAD_OFF + 1) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t from_node;
uint64_t from_node, group_id;
memcpy(&from_node, data + ROUTER_SVC_SRC_OFF, 8);
memcpy(&group_id, data + ROUTER_SVC_GROUP_OFF, 8);
uint8_t subcmd = data[ROUTER_SVC_PAYLOAD_OFF];
data += ROUTER_SVC_PAYLOAD_OFF; len -= ROUTER_SVC_PAYLOAD_OFF;
@@ -1277,7 +1258,7 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
}
break;
case MEDIA_SUBCMD_QUERY:
md_handle_query(md, from_node, data, len);
md_handle_query(md, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_HAVE_BLOCK:
md_handle_have_block(md, from_node, data, len);
@@ -1299,7 +1280,7 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SUPER_HELLO ignored — not supernode", MD_ID);
break;
case MEDIA_SUBCMD_QUERY_RESP:
media_download_handle_query_resp(inst, data, len);
media_download_handle_query_resp(inst, group_id, from_node, data, len);
break;
case MEDIA_SUBCMD_HAVE_BLOCK_ACK:
/* handled by download module */
src/media_delivery/media_delivery_proto.h
+18 -9
View File
@@ -36,6 +36,13 @@ enum {
#define MEDIA_ERR_NOT_SUPERNODE 0xFE
/* Один QUERY — одна полная пачка результатов, включая блоки без держателей.
* Группу и end-to-end отправителя задаёт router. Пределы общие для обеих сторон. */
#define MEDIA_QUERY_MAX_BLOCKS 32
#define MEDIA_QUERY_MAX_HOLDERS 4
#define MEDIA_QUERY_OK 0
#define MEDIA_QUERY_DB_ERROR 1
/* ── packed-структуры пакетов ── */
#pragma pack(push, 1)
@@ -58,28 +65,30 @@ struct media_pkt_serve_leave {
struct media_pkt_query {
uint8_t subcmd; // MEDIA_SUBCMD_QUERY
uint64_t group_id;
uint8_t media_id[16];
uint16_t num_blocks;
uint8_t request_id[16]; // Новый случайный UUID для каждой попытки; не средство аутентификации.
uint16_t num_blocks; // 1..MEDIA_QUERY_MAX_BLOCKS
// далее: block_ids[num_blocks * 16]
};
#define MEDIA_QUERY_HDR_SIZE (sizeof(struct media_pkt_query)) // 1+8+16+2=27
#define MEDIA_QUERY_HDR_SIZE (sizeof(struct media_pkt_query)) // 35
struct media_pkt_query_resp_entry {
uint64_t node_id;
uint16_t rtt; // cumulative_rtt из topo_node суперузла (0.1ms)
uint8_t block_id[16];
uint32_t chunk;
uint8_t num_holders; // 0..MEDIA_QUERY_MAX_HOLDERS; ноль — держатели не найдены.
uint64_t holders[MEDIA_QUERY_MAX_HOLDERS]; // Уникальные ненулевые узлы; неиспользуемые слоты нулевые.
};
struct media_pkt_query_resp {
uint8_t subcmd; // MEDIA_SUBCMD_QUERY_RESP
uint16_t num_entries;
// далее: entries[num_entries * sizeof(media_pkt_query_resp_entry)]
uint8_t media_id[16];
uint8_t request_id[16]; // Точное эхо QUERY, включая ответ с ошибкой.
uint8_t status; // MEDIA_QUERY_OK / MEDIA_QUERY_DB_ERROR
uint16_t num_blocks; // OK: столько же, сколько в QUERY; ошибка: 0.
// Далее результаты всех блоков в порядке запроса; частичных успешных ответов нет.
};
#define MEDIA_QUERY_RESP_HDR_SIZE (sizeof(struct media_pkt_query_resp)) // 1+2=3
#define MEDIA_QUERY_RESP_HDR_SIZE (sizeof(struct media_pkt_query_resp)) // 36; максимум с записями — 1604 байта.
struct media_pkt_have_block {
uint8_t subcmd; // MEDIA_SUBCMD_HAVE_BLOCK