From 6267973d62a880ce5c0d082bd3a4b2dc7cef93e4 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 18:06:41 +0200 Subject: [PATCH] Define bounded media QUERY batches and complete responses --- src/media_delivery/media_delivery.c | 125 +++++++++------------- src/media_delivery/media_delivery_proto.h | 27 +++-- 2 files changed, 71 insertions(+), 81 deletions(-) diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index ac199fb6..bd30093e 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -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 */ diff --git a/src/media_delivery/media_delivery_proto.h b/src/media_delivery/media_delivery_proto.h index f5bd8385..b930a4b1 100644 --- a/src/media_delivery/media_delivery_proto.h +++ b/src/media_delivery/media_delivery_proto.h @@ -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