From b4b11d608633e1d9b1a12f7525d0c2816aae3f44 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 20:54:37 +0200 Subject: [PATCH] Implement atomic group-scoped media catalog replication --- src/media_delivery/media_delivery.c | 1239 ++++++++++++++------------- 1 file changed, 651 insertions(+), 588 deletions(-) diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index af170d4f..83dbdc9e 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -30,14 +30,17 @@ static void md_block_wake(struct media_block_source* source, struct stream_ctx* executing); 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); static void md_on_props_changed(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg); -static int md_super_start(struct media_delivery_ctx* md); -static void md_super_stop(struct media_delivery_ctx* md); -static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super_peer* peer); static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id); static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len); -static struct media_super_peer* md_super_peer_add(struct media_delivery_ctx* md, uint64_t node_id, uint64_t group_id); +static int md_send_packet(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, + const uint8_t* data, size_t len, int force); +static void md_repl_pump(void* arg); +static void md_repl_watchdog(void* arg); +static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id); +static int md_group_role(struct media_delivery_ctx* md, uint64_t group_id, uint64_t node_id); +static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_id, + enum conn_mgr_event event, void* arg); /* ── helpers ── */ @@ -50,77 +53,77 @@ static const char* md_subcmd_name(uint8_t sc) { case MEDIA_SUBCMD_QUERY_RESP: return "QUERY_RESP"; case MEDIA_SUBCMD_HAVE_BLOCK: return "HAVE_BLOCK"; case MEDIA_SUBCMD_HAVE_BLOCK_ACK: return "HAVE_BLOCK_ACK"; - case MEDIA_SUBCMD_SUPER_REPL: return "SUPER_REPL"; - case MEDIA_SUBCMD_SUPER_ACK: return "SUPER_ACK"; + case MEDIA_SUBCMD_SYNC_REQ: return "SYNC_REQ"; + case MEDIA_SUBCMD_SYNC_RESP: return "SYNC_RESP"; case MEDIA_SUBCMD_BLOCK_REQ: return "BLOCK_REQ"; case MEDIA_SUBCMD_BLOCK_CHUNK: return "BLOCK_CHUNK"; case MEDIA_SUBCMD_BLOCK_DONE: return "BLOCK_DONE"; case MEDIA_SUBCMD_CANCEL: return "CANCEL"; - case MEDIA_SUBCMD_SUPER_HELLO: return "SUPER_HELLO"; + case MEDIA_SUBCMD_CHANGED: return "CHANGED"; case MEDIA_SUBCMD_BLOCK_PROCESSING: return "BLOCK_PROCESSING"; case MEDIA_SUBCMD_BLOCK_RELAY_FULL: return "BLOCK_RELAY_FULL"; default: return "???"; } } -/* ── SQL helper: insert/update block_availability ── */ - -static int md_ba_insert(sqlite3* db, const uint8_t* block_uuid, uint64_t group_id, - uint64_t node_id, uint32_t chunk, int64_t ts, int status) { - const char* sql = - "INSERT INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status)" - " VALUES(?,?,?,?,?,?)"; - sqlite3_stmt* st = NULL; - 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); - sqlite3_bind_int(st, 4, (int)chunk); - sqlite3_bind_int64(st, 5, (sqlite3_int64)ts); - sqlite3_bind_int(st, 6, status); - int rc = sqlite3_step(st); - sqlite3_finalize(st); - 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; +/* Все новые SQL-пути проверяют и завершение statement: ошибка finalize не является успехом. */ +static int md_sql_finish(sqlite3* db, sqlite3_stmt* st, int rc, const char* phase) { + int final_rc = sqlite3_finalize(st); + if (rc == SQLITE_DONE && final_rc == SQLITE_OK) return 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: SQL phase=%s rc=%d finalize=%d error=%s", + MD_ID, phase, rc, final_rc, sqlite3_errmsg(db)); + return -1; } -/* ── complete block: DELETE processing records + INSERT completed with new id ── */ +static int md_sql_exec(sqlite3* db, const char* sql) { + int rc = sqlite3_exec(db, sql, NULL, NULL, NULL); + if (rc == SQLITE_OK) return 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: SQL rc=%d command=%s error=%s", MD_ID, rc, sql, sqlite3_errmsg(db)); + return -1; +} -static int md_ba_complete_block(sqlite3* db, const uint8_t* block_uuid, - uint64_t group_id, uint64_t node_id, - uint32_t chunk, int64_t ts) { - /* 1. DELETE processing records (status=0) for this (block_uuid, node_id) */ - const char* del_sql = "DELETE FROM block_availability WHERE block_uuid=? AND node_id=? AND status=0"; +/* Чужая транзакция не даёт гарантии commit; её владелец должен завершить работу до репликации. */ +static int md_sql_begin(sqlite3* db, const char* sql) { + if (!db || !sqlite3_get_autocommit(db)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: cannot own transaction db=%p", MD_ID, (void*)db); return -1; + } + return md_sql_exec(db, sql); +} + +/* Вызывается внутри собственной транзакции. Максимум (status,timestamp) идемпотентен; + * REPLACE выполняется только при изменении, поэтому повтор не расходует AUTOINCREMENT ID. */ +static int md_ba_merge(sqlite3* db, const uint8_t block_id[16], uint64_t group_id, + uint64_t node_id, uint32_t chunk, int64_t ts, int status) { sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(db, del_sql, -1, &st, NULL) == SQLITE_OK) { - sqlite3_bind_blob(st, 1, block_uuid, 16, SQLITE_STATIC); - sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id); - sqlite3_step(st); - sqlite3_finalize(st); + int rc = sqlite3_prepare_v2(db, + "SELECT node_id,chunk,timestamp,status FROM block_availability WHERE group_id=? AND block_uuid=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, block_id, 16, SQLITE_STATIC); + int unchanged = 0, conflict = 0; + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + if ((uint32_t)sqlite3_column_int64(st, 1) != chunk) conflict = 1; + if ((uint64_t)sqlite3_column_int64(st, 0) != node_id) continue; + int old_status = sqlite3_column_int(st, 3); + if (old_status > status || (old_status == status && sqlite3_column_int64(st, 2) >= ts)) unchanged = 1; } - - /* 2. check if completed already exists — avoid duplicate insert */ - const char* chk_sql = "SELECT 1 FROM block_availability WHERE block_uuid=? AND node_id=? AND status=1"; - if (sqlite3_prepare_v2(db, chk_sql, -1, &st, NULL) == SQLITE_OK) { - sqlite3_bind_blob(st, 1, block_uuid, 16, SQLITE_STATIC); - sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id); - int exists = (sqlite3_step(st) == SQLITE_ROW); - sqlite3_finalize(st); - if (exists) { - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: ba_complete_block: completed already exists for node=0x%016llx, skipping", - MD_ID, (unsigned long long)node_id); - return 0; - } + if (md_sql_finish(db, st, rc, "merge lookup")) return -1; + if (conflict) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: contradictory chunk group=%llx holder=%llx chunk=%u block=%02x%02x", + MD_ID, (unsigned long long)group_id, (unsigned long long)node_id, chunk, block_id[0], block_id[1]); + return -1; } - - /* 3. INSERT completed record (status=1, new id via AUTOINCREMENT) */ - return md_ba_insert(db, block_uuid, group_id, node_id, chunk, ts, 1); + if (unchanged) return 0; + st = NULL; + rc = sqlite3_prepare_v2(db, "INSERT OR REPLACE INTO block_availability" + "(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,?,?,?,?,?)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, 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_int64(st, 3, (sqlite3_int64)node_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, chunk); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 5, ts); + if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 6, status); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + return md_sql_finish(db, st, rc, "merge insert") ? -1 : 1; } /* Выбрать ограниченный список кандидатов: свой готовый блок, затем свежие HAVE/PROCESSING. @@ -130,7 +133,7 @@ static int md_query_holders(struct media_delivery_ctx* md, uint64_t group_id, ui 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=?" + " UNION ALL SELECT node_id,1 AS own,timestamp FROM media_files WHERE block_id=? AND media_id=? AND node_id=? AND chat_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) { @@ -141,8 +144,10 @@ static int md_query_holders(struct media_delivery_ctx* md, uint64_t group_id, ui 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); + char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 6, channel, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 7, (sqlite3_int64)requester); + if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 8, 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); @@ -153,73 +158,20 @@ static int md_query_holders(struct media_delivery_ctx* md, uint64_t group_id, ui return -1; } -static int md_ba_get_since(sqlite3* db, uint64_t since_id, - uint8_t* out_buf, int max_bytes) { - const char* sql = - "SELECT id,block_uuid,group_id,node_id,chunk,timestamp,status" - " 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) { - 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) { - if (off + 56 > max_bytes) break; - int64_t id = sqlite3_column_int64(st, 0); - memcpy(out_buf + off, &id, 8); off += 8; - memcpy(out_buf + off, sqlite3_column_blob(st, 1), 16); off += 16; - int64_t gid = sqlite3_column_int64(st, 2); memcpy(out_buf + off, &gid, 8); off += 8; - int64_t nid = sqlite3_column_int64(st, 3); memcpy(out_buf + off, &nid, 8); off += 8; - int32_t ch = sqlite3_column_int(st, 4); memcpy(out_buf + off, &ch, 4); off += 4; - int64_t ts = sqlite3_column_int64(st, 5); memcpy(out_buf + off, &ts, 8); off += 8; - int32_t sts = sqlite3_column_int(st, 6); memcpy(out_buf + off, &sts, 4); off += 4; - } - sqlite3_finalize(st); - return off; -} - -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) { - 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); - sqlite3_finalize(st); - return v; -} - -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) { - 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); - sqlite3_finalize(st); -} - /* ── super_peer find/create ── */ -static struct media_super_peer* md_super_peer_find(struct media_delivery_ctx* md, uint64_t node_id) { +static struct media_super_peer* md_super_peer_find(struct media_delivery_ctx* md, uint64_t node_id, uint64_t group_id) { if (!md->super_peers) return NULL; - struct ll_entry* e = queue_find_data_by_index(md->super_peers, &node_id); + uint64_t key[2] = {node_id, group_id}; + struct ll_entry* e = queue_find_data_by_index(md->super_peers, key); return e ? (struct media_super_peer*)e->data : NULL; } static struct media_super_peer* md_super_peer_add(struct media_delivery_ctx* md, uint64_t node_id, uint64_t group_id) { - struct media_super_peer* sp = md_super_peer_find(md, node_id); + struct media_super_peer* sp = md_super_peer_find(md, node_id, group_id); if (sp) return sp; if (!md->super_peers) { - md->super_peers = queue_new(md->inst->ua, 64, offsetof(struct media_super_peer, peer_node_id), 8, "md_super"); + md->super_peers = queue_new(md->inst->ua, 64, offsetof(struct media_super_peer, peer_node_id), 16, "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)); @@ -230,16 +182,23 @@ static struct media_super_peer* md_super_peer_add(struct media_delivery_ctx* md, sp->group_id = group_id; sp->timeout_tb = MEDIA_REPL_TIMEOUT_TB; sp->md = md; - memcpy(qe->data, &node_id, 8); queue_data_put_with_index(md->super_peers, qe); + sp->timeout_timer = uasync_set_timeout(md->inst->ua, MEDIA_REPL_WATCHDOG_TB, sp, md_repl_watchdog, "md_repl"); + if (!sp->timeout_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: replication watchdog allocation failed", MD_ID); + queue_remove_data(md->super_peers, qe); queue_entry_free(qe); return NULL; + } return sp; } -static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id) { +static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id, uint64_t group_id) { if (!md->super_peers) return; - struct ll_entry* e = queue_find_data_by_index(md->super_peers, &node_id); + uint64_t key[2] = {node_id, group_id}; + struct ll_entry* e = queue_find_data_by_index(md->super_peers, key); if (!e) return; struct media_super_peer* sp = (struct media_super_peer*)e->data; + etcp_router_cancel_send_ready(md->inst, group_id, node_id, ETCP_RT_ID_MEDIA_DELIVERY, &sp->waiter); + if (sp->pump) uasync_call_soon_cancel(md->inst->ua, sp->pump); if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; } if (sp->connection) { conn_mgr_close(sp->connection); sp->connection = NULL; } DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: super peer released node=%llx group=%llx", MD_ID, @@ -396,78 +355,201 @@ void md_file_load_dec(struct media_delivery_ctx* md, const uint8_t* media_id, ui } } -/* ── replication timers ── */ +/* Wire-числа читаются побайтно: packed-поля могут быть невыровнены, endian одинаков на всех платформах. */ +static uint64_t md_read_le(const void* ptr, unsigned bytes) { + const uint8_t* p = ptr; + uint64_t v = 0; + for (unsigned i = 0; i < bytes; i++) v |= (uint64_t)p[i] << (i * 8); + return v; +} -static void md_repl_retrans_cb(void* arg) { - struct media_super_peer* peer = (struct media_super_peer*)arg; +static void md_write_le(void* ptr, uint64_t value, unsigned bytes) { + uint8_t* p = ptr; + for (unsigned i = 0; i < bytes; i++) p[i] = (uint8_t)(value >> (i * 8)); +} + +static void md_repl_schedule(struct media_super_peer* peer) { + if (!peer->pump) peer->pump = uasync_call_soon(peer->md->inst->ua, peer, md_repl_pump); + if (!peer->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: replication pump allocation failed", MD_ID); +} + +/* Курсор читается заново для каждой новой страницы; ошибки БД не превращаются в «курсор 0». */ +static int md_repl_request(struct media_super_peer* peer) { struct media_delivery_ctx* md = peer->md; - if (!md || !peer->connected) return; - peer->timeout_timer = NULL; - peer->timeout_tb = peer->timeout_tb < MEDIA_REPL_TIMEOUT_MAX_TB / 2 - ? peer->timeout_tb * 2 : peer->timeout_tb; - if (peer->timeout_tb > MEDIA_REPL_TIMEOUT_MAX_TB) - peer->timeout_tb = MEDIA_REPL_TIMEOUT_MAX_TB; - md_super_repl_send(md, peer); -} - -static void md_repl_schedule_retrans(struct media_delivery_ctx* md, struct media_super_peer* peer) { - if (peer->timeout_timer) uasync_cancel_timeout(md->inst->ua, peer->timeout_timer); - peer->timeout_timer = uasync_set_timeout(md->inst->ua, peer->timeout_tb, peer, - md_repl_retrans_cb, "md_repl_retrans"); -} - -/* ── replication send ── */ - -static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super_peer* peer) { - if (!md->db || !peer->connected || !peer->hello_done) return; - - while (peer->inflight_count < MEDIA_MAX_REPL_INFLIGHT) { - uint8_t buf[560]; - int nb = md_ba_get_since(md->db, peer->peer_last_recv_id, buf, (int)sizeof(buf)); - if (nb == 0) { - if (peer->timeout_timer) uasync_cancel_timeout(md->inst->ua, peer->timeout_timer); - peer->timeout_timer = NULL; - return; + struct media_pkt_sync_req req = {.subcmd = MEDIA_SUBCMD_SYNC_REQ}; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, + "SELECT source_epoch,last_recv_id FROM super_sync WHERE group_id=? AND peer_node_id=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)peer->group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)peer->peer_node_id); + if (rc == SQLITE_OK) { + rc = sqlite3_step(st); + if (rc == SQLITE_ROW) { + if (sqlite3_column_bytes(st, 0) != 16 || sqlite3_column_int64(st, 1) < 0) rc = SQLITE_CORRUPT; + else { + memcpy(req.source_epoch, sqlite3_column_blob(st, 0), 16); + md_write_le(&req.after_id, (uint64_t)sqlite3_column_int64(st, 1), 8); + rc = sqlite3_step(st); + } } - - /* build packet: subcmd + seq + num + raw rows */ - uint8_t pkt[600]; - int off = 0; - pkt[off++] = MEDIA_SUBCMD_SUPER_REPL; - - int num_entries = nb / 56; - int64_t max_id = 0; - const uint8_t* rp = buf; - for (int i = 0; i < num_entries; i++) { - int64_t id; memcpy(&id, rp, 8); - if (id > max_id) max_id = id; - rp += 56; - } - - uint32_t seq = (uint32_t)max_id; memcpy(pkt + off, &seq, 4); off += 4; - uint16_t num = (uint16_t)num_entries; memcpy(pkt + off, &num, 2); off += 2; - memcpy(pkt + off, buf, (size_t)nb); off += nb; - - if (md_send(md->inst, peer->group_id, peer->peer_node_id, pkt, (size_t)off) == 0) { - peer->inflight_count++; - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SUPER_REPL sent to 0x%016llx seq=%u entries=%d", - MD_ID, (unsigned long long)peer->peer_node_id, seq, num_entries); - } else { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: SUPER_REPL send failed to 0x%016llx", - MD_ID, (unsigned long long)peer->peer_node_id); - return; - } - - if (peer->inflight_count >= MEDIA_MAX_REPL_INFLIGHT) break; } + if (md_sql_finish(md->db, st, rc, "read RX cursor") || media_index_generate_uuid(req.request_id)) return -1; + peer->request = req; + peer->request_pending = peer->request_send = 1; peer->dirty = 0; + peer->deadline_tb = get_time_tb() + peer->timeout_tb; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SYNC request group=%llx peer=%llx uuid=%02x%02x epoch=%02x%02x after=%llu", + MD_ID, (unsigned long long)peer->group_id, (unsigned long long)peer->peer_node_id, + req.request_id[0], req.request_id[1], req.source_epoch[0], req.source_epoch[1], + (unsigned long long)md_read_le(&req.after_id, 8)); + return 0; +} - if (peer->inflight_count > 0) md_repl_schedule_retrans(md, peer); +/* Один read snapshot: high-water не уменьшается после DELETE, страницы изолированы по router-группе. */ +static size_t md_repl_page(struct media_super_peer* peer, uint8_t packet[MEDIA_SYNC_PACKET_MAX]) { + struct media_delivery_ctx* md = peer->md; + const struct media_pkt_sync_req* req = &peer->response; + struct media_pkt_sync_resp* r = (struct media_pkt_sync_resp*)packet; + memset(packet, 0, MEDIA_SYNC_PACKET_MAX); + r->subcmd = MEDIA_SUBCMD_SYNC_RESP; memcpy(r->request_id, req->request_id, 16); + memcpy(r->source_epoch, md->source_epoch, 16); + r->status = MEDIA_SYNC_DB_ERROR; + if (md_sql_begin(md->db, "BEGIN")) return sizeof(*r); + sqlite3_stmt* st = NULL; + uint64_t high = 0; + int rc = sqlite3_prepare_v2(md->db, "SELECT seq FROM sqlite_sequence WHERE name='block_availability'", -1, &st, NULL); + if (rc == SQLITE_OK) { + rc = sqlite3_step(st); + if (rc == SQLITE_ROW) { + if (sqlite3_column_int64(st, 0) < 0) rc = SQLITE_CORRUPT; + else { high = (uint64_t)sqlite3_column_int64(st, 0); rc = sqlite3_step(st); } + } + } + if (md_sql_finish(md->db, st, rc, "snapshot high-water")) goto failed; + uint64_t from = memcmp(req->source_epoch, md->source_epoch, 16) ? 0 : md_read_le(&req->after_id, 8); + if (from > high) { + if (md_sql_exec(md->db, "COMMIT")) goto failed; + r->status = MEDIA_SYNC_CURSOR_INVALID; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: source cursor ahead group=%llx peer=%llx from=%llu high=%llu", MD_ID, + (unsigned long long)peer->group_id, (unsigned long long)peer->peer_node_id, + (unsigned long long)from, (unsigned long long)high); + return sizeof(*r); + } + st = NULL; + rc = sqlite3_prepare_v2(md->db, "SELECT id,block_uuid,node_id,chunk,timestamp,status FROM block_availability" + " WHERE group_id=? AND id>? AND id<=? ORDER BY id LIMIT 33", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)peer->group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)from); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)high); + struct media_pkt_sync_entry* rows = (struct media_pkt_sync_entry*)(packet + sizeof(*r)); + unsigned count = 0; + uint64_t last = from; + int more = 0; + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + if (count == MEDIA_SYNC_MAX_ENTRIES) { more = 1; rc = SQLITE_DONE; break; } + sqlite3_int64 id = sqlite3_column_int64(st, 0), chunk = sqlite3_column_int64(st, 3); + sqlite3_int64 ts = sqlite3_column_int64(st, 4); + int status = sqlite3_column_int(st, 5); + if (id <= (sqlite3_int64)last || (uint64_t)id > high || sqlite3_column_bytes(st, 1) != 16 + || !sqlite3_column_int64(st, 2) || chunk < 0 || chunk > UINT32_MAX || ts < 0 || status < 0 || status > 1) { + rc = SQLITE_CORRUPT; break; + } + struct media_pkt_sync_entry* row = &rows[count++]; + md_write_le(&row->id, id, 8); memcpy(row->block_id, sqlite3_column_blob(st, 1), 16); + md_write_le(&row->node_id, (uint64_t)sqlite3_column_int64(st, 2), 8); + md_write_le(&row->chunk, chunk, 4); md_write_le(&row->timestamp, ts, 8); row->status = (uint8_t)status; + last = (uint64_t)id; + } + if (md_sql_finish(md->db, st, rc, "snapshot rows") || md_sql_exec(md->db, "COMMIT")) goto failed; + r->status = MEDIA_SYNC_OK; r->more = (uint8_t)more; + md_write_le(&r->from_id, from, 8); md_write_le(&r->next_id, more ? last : high, 8); + md_write_le(&r->num_entries, count, 2); + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SYNC page group=%llx peer=%llx uuid=%02x%02x epoch=%02x%02x from=%llu next=%llu n=%u more=%d", + MD_ID, (unsigned long long)peer->group_id, (unsigned long long)peer->peer_node_id, + r->request_id[0], r->request_id[1], r->source_epoch[0], r->source_epoch[1], + (unsigned long long)from, (unsigned long long)(more ? last : high), count, more); + return sizeof(*r) + count * sizeof(*rows); +failed: + md_sql_exec(md->db, "ROLLBACK"); + return sizeof(*r); +} + +/* Watchdog и waiter только планируют pump; это безопасно при синхронном callback очереди. */ +static void md_repl_ready(void* arg) { md_repl_schedule(arg); } + +static void md_repl_pump(void* arg) { + struct media_super_peer* peer = arg; + struct media_delivery_ctx* md = peer->md; + peer->pump = NULL; + if (!md->initialized || md_group_role(md, peer->group_id, md->self_node_id) != 1 + || md_group_role(md, peer->group_id, peer->peer_node_id) != 1) return; + etcp_router_cancel_send_ready(md->inst, peer->group_id, peer->peer_node_id, ETCP_RT_ID_MEDIA_DELIVERY, &peer->waiter); + /* Максимум три пакета за вызов: ответ, свой запрос и подсказка. */ + for (int budget = 0; budget < 3; budget++) { + if (!peer->response_pending && !(peer->connected && (peer->request_send || peer->changed))) return; + if (!etcp_router_send_q_has_room(md->inst, peer->group_id, peer->peer_node_id, ETCP_RT_ID_MEDIA_DELIVERY)) { + etcp_router_on_send_ready(md->inst, peer->group_id, peer->peer_node_id, ETCP_RT_ID_MEDIA_DELIVERY, + &peer->waiter, md_repl_ready, peer); + return; + } + uint8_t packet[MEDIA_SYNC_PACKET_MAX]; + const uint8_t* data = packet; + size_t len = 1; + int kind = peer->response_pending ? 0 : peer->request_send ? 1 : 2; + if (kind == 0) len = md_repl_page(peer, packet); + else if (kind == 1) { data = (const uint8_t*)&peer->request; len = sizeof(peer->request); } + else packet[0] = MEDIA_SUBCMD_CHANGED; + if (md_send_packet(md->inst, peer->group_id, peer->peer_node_id, data, len, 0)) return; + if (kind == 0) peer->response_pending = 0; + else if (kind == 1) peer->request_send = 0; + else peer->changed = 0; + } +} + +static void md_repl_watchdog(void* arg) { + struct media_super_peer* peer = arg; + peer->timeout_timer = NULL; + uint64_t now = get_time_tb(); + if (peer->connected && now >= peer->deadline_tb) { + if (peer->request_pending) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SYNC retry group=%llx peer=%llx uuid=%02x%02x delay=%u", + MD_ID, (unsigned long long)peer->group_id, (unsigned long long)peer->peer_node_id, + peer->request.request_id[0], peer->request.request_id[1], peer->timeout_tb); + peer->timeout_tb = peer->timeout_tb < MEDIA_REPL_TIMEOUT_MAX_TB / 2 + ? peer->timeout_tb * 2 : MEDIA_REPL_TIMEOUT_MAX_TB; + peer->request_send = 1; + } else if (md_repl_request(peer)) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: SYNC request initialization failed", MD_ID); + peer->deadline_tb = now + peer->timeout_tb; + } else if (!peer->connection && now >= peer->deadline_tb) { + md_super_connect(peer->md, peer->peer_node_id, peer->group_id); + peer->deadline_tb = now + peer->timeout_tb; + } + if (peer->response_pending || (peer->connected && (peer->request_send || peer->changed))) md_repl_schedule(peer); + peer->timeout_timer = uasync_set_timeout(peer->md->inst->ua, MEDIA_REPL_WATCHDOG_TB, peer, md_repl_watchdog, "md_repl"); + if (!peer->timeout_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: replication watchdog lost; releasing peer", MD_ID); + md_super_peer_remove(peer->md, peer->peer_node_id, peer->group_id); + } +} + +/* Только после commit: потеря CHANGED безопасна благодаря периодической сверке. */ +static void md_repl_notify(struct media_delivery_ctx* md, uint64_t group_id) { + for (struct ll_entry* e = md->super_peers ? md->super_peers->head : NULL; e; e = e->next) { + struct media_super_peer* peer = (struct media_super_peer*)e->data; + if (peer->group_id != group_id) continue; + peer->changed = 1; md_repl_schedule(peer); + } } /* ── send via etcp_router helper ── */ static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len) { + return md_send_packet(inst, group_id, dst_node_id, data, len, 1); +} + +/* Репликация использует force=0; остальные команды сохраняют существующую политику очереди. */ +static int md_send_packet(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, + const uint8_t* data, size_t len, int force) { if (!inst || !data || len == 0) return -1; struct TOPO_GROUP* grp = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; struct ETCP_CONN* conn = grp ? topo_group_find_conn_for_node(grp, dst_node_id) : NULL; @@ -475,12 +557,12 @@ static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_n DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: no conn to 0x%016llx group=0x%016llx subcmd=0x%02x — packet queued, no route", MD_ID, (unsigned long long)dst_node_id, (unsigned long long)group_id, (unsigned)data[0]); struct ll_entry* e = queue_entry_new(0); - if (!e) return -1; + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: send entry allocation failed", MD_ID); return -1; } e->dgram = u_malloc(len + 1); - if (!e->dgram) { queue_entry_free(e); return -1; } + if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: send payload allocation failed", MD_ID); queue_entry_free(e); return -1; } e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1); - int rc = etcp_route_send(inst, group_id, dst_node_id, e, 1, 0); + int rc = etcp_route_send(inst, group_id, dst_node_id, e, force, 0); 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); @@ -548,198 +630,171 @@ static void md_handle_query(struct media_delivery_ctx* md, uint64_t group_id, ui 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, - const uint8_t* data, size_t len) { - 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_complete_block(md->db, hb->block_id, hb->group_id, from_node, hb->chunk, hb->timestamp); - - uint8_t ack[sizeof(struct media_pkt_have_block_ack)]; - struct media_pkt_have_block_ack* ha = (struct media_pkt_have_block_ack*)ack; - memset(ack, 0, sizeof(ack)); - ha->subcmd = MEDIA_SUBCMD_HAVE_BLOCK_ACK; - memcpy(ha->media_id, hb->media_id, 16); - memcpy(ha->block_id, hb->block_id, 16); - ha->status = (rc == 0) ? 0 : 1; - - md_send(md->inst, hb->group_id, from_node, ack, sizeof(ack)); - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: HAVE_BLOCK from 0x%016llx rc=%d", - MD_ID, (unsigned long long)from_node, rc); - - if (rc == 0) { - /* replicate to connected super-peers */ - struct ll_entry* se = md->super_peers ? md->super_peers->head : NULL; - while (se) { - struct media_super_peer* sp = (struct media_super_peer*)se->data; - if (sp->connected && sp->hello_done && sp->inflight_count < MEDIA_MAX_REPL_INFLIGHT) - md_super_repl_send(md, sp); - se = se->next; +/* Локальное HAVE/PROCESSING проходит тот же merge, что и страница, с самостоятельным commit. */ +static void md_handle_announcement(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len, int status) { + if (len != MEDIA_HAVE_BLOCK_SIZE || !from_node || md_group_role(md, group_id, md->self_node_id) != 1) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: rejected announcement group=%llx peer=%llx bytes=%zu", MD_ID, + (unsigned long long)group_id, (unsigned long long)from_node, len); return; + } + const struct media_pkt_have_block* hb = (const struct media_pkt_have_block*)data; + if (hb->group_id != group_id || hb->timestamp < 0) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: announcement group/timestamp mismatch", MD_ID); return; + } + int changed = -1; + if (!md_sql_begin(md->db, "BEGIN IMMEDIATE")) { + changed = md_ba_merge(md->db, hb->block_id, group_id, from_node, hb->chunk, hb->timestamp, status); + if (changed < 0 || md_sql_exec(md->db, "COMMIT")) { + md_sql_exec(md->db, "ROLLBACK"); changed = -1; } } + if (status) { + struct media_pkt_have_block_ack ack = {.subcmd = MEDIA_SUBCMD_HAVE_BLOCK_ACK, + .status = changed < 0 ? 2 : changed ? 0 : 1}; + memcpy(ack.media_id, hb->media_id, 16); memcpy(ack.block_id, hb->block_id, 16); + md_send(md->inst, group_id, from_node, (const uint8_t*)&ack, sizeof(ack)); + } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: announcement group=%llx holder=%llx status=%d changed=%d", + MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, status, changed); + if (changed > 0) md_repl_notify(md, group_id); } -/* ── handle BLOCK_PROCESSING: node started downloading a block, insert as processing (status=0) ── */ - -static void md_handle_block_processing(struct media_delivery_ctx* md, uint64_t from_node, - const uint8_t* data, size_t len) { - if (len < MEDIA_HAVE_BLOCK_SIZE) return; - if (!md->db) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_PROCESSING but no DB", MD_ID); return; } - struct media_pkt_have_block* hb = (struct media_pkt_have_block*)data; - - /* INSERT processing (status=0) with new id */ - int rc = md_ba_insert(md->db, hb->block_id, hb->group_id, from_node, hb->chunk, hb->timestamp, 0); - if (rc != 0) { - DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_PROCESSING insert failed for node=0x%016llx (duplicate/constraint?)", - MD_ID, (unsigned long long)from_node); +/* До выделения контекста проверяем точный формат и подписанные роли именно router-группы. */ +static void md_handle_sync_req(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (len != MEDIA_SYNC_REQ_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed SYNC_REQ bytes=%zu", MD_ID, len); return; } - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_PROCESSING from 0x%016llx block=%02x%02x...", - MD_ID, (unsigned long long)from_node, hb->block_id[0], hb->block_id[1]); - - /* replicate to connected super-peers */ - struct ll_entry* se = md->super_peers ? md->super_peers->head : NULL; - while (se) { - struct media_super_peer* sp = (struct media_super_peer*)se->data; - if (sp->connected && sp->hello_done && sp->inflight_count < MEDIA_MAX_REPL_INFLIGHT) - md_super_repl_send(md, sp); - se = se->next; + const struct media_pkt_sync_req* q = (const struct media_pkt_sync_req*)data; + if (md_read_le(&q->after_id, 8) > INT64_MAX) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: invalid SYNC_REQ cursor", MD_ID); return; } + struct media_super_peer* peer = md_super_peer_add(md, from_node, group_id); + if (!peer) return; + peer->response = *q; peer->response_pending = 1; + md_super_connect(md, from_node, group_id); + md_repl_schedule(peer); } -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) 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; - const uint8_t* ep = data + MEDIA_SUPER_REPL_HDR_SIZE; - - for (int i = 0; i < (int)num; i++) { - int64_t id; memcpy(&id, ep, 8); ep += 8; - if (id > max_id) max_id = id; - uint8_t block_uuid[16]; memcpy(block_uuid, ep, 16); ep += 16; - int64_t gid; memcpy(&gid, ep, 8); ep += 8; - int64_t nid; memcpy(&nid, ep, 8); ep += 8; - int32_t ch; memcpy(&ch, ep, 4); ep += 4; - int64_t ts; memcpy(&ts, ep, 8); ep += 8; - int32_t st; memcpy(&st, ep, 4); ep += 4; - if (st == 1) { - md_ba_complete_block(md->db, block_uuid, (uint64_t)gid, (uint64_t)nid, (uint32_t)ch, ts); - } else { - /* processing: INSERT OR IGNORE (дубликаты при репликации — норма) */ - const char* ign = "INSERT OR IGNORE INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp,status) VALUES(?,?,?,?,?,0)"; - sqlite3_stmt* igs = NULL; - if (sqlite3_prepare_v2(md->db, ign, -1, &igs, NULL) == SQLITE_OK) { - sqlite3_bind_blob(igs, 1, block_uuid, 16, SQLITE_STATIC); - sqlite3_bind_int64(igs, 2, (sqlite3_int64)gid); - sqlite3_bind_int64(igs, 3, (sqlite3_int64)nid); - sqlite3_bind_int(igs, 4, (int)ch); - sqlite3_bind_int64(igs, 5, (sqlite3_int64)ts); - sqlite3_step(igs); - sqlite3_finalize(igs); +/* Ошибка/дубликат не меняет ожидающий запрос. Только полный commit открывает следующую страницу. */ +static void md_handle_sync_resp(struct media_delivery_ctx* md, uint64_t group_id, uint64_t from_node, + const uint8_t* data, size_t len) { + if (len < MEDIA_SYNC_RESP_HDR_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: short SYNC_RESP bytes=%zu", MD_ID, len); return; + } + const struct media_pkt_sync_resp* r = (const struct media_pkt_sync_resp*)data; + struct media_super_peer* peer = md_super_peer_find(md, from_node, group_id); + if (!peer || !peer->request_pending || memcmp(r->request_id, peer->request.request_id, 16)) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: stale SYNC_RESP group=%llx peer=%llx uuid=%02x%02x", MD_ID, + (unsigned long long)group_id, (unsigned long long)from_node, r->request_id[0], r->request_id[1]); return; + } + unsigned count = (unsigned)md_read_le(&r->num_entries, 2); + uint64_t from = md_read_le(&r->from_id, 8), next = md_read_le(&r->next_id, 8); + static const uint8_t zero[16] = {0}; + if (count > MEDIA_SYNC_MAX_ENTRIES || len != sizeof(*r) + count * sizeof(struct media_pkt_sync_entry) + || r->more > 1 || r->status > MEDIA_SYNC_CURSOR_INVALID || !memcmp(r->source_epoch, zero, 16)) goto malformed; + if (r->status != MEDIA_SYNC_OK) { + if (count || r->more || from || next) goto malformed; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SYNC error group=%llx peer=%llx status=%u uuid=%02x%02x", + MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, + r->status, r->request_id[0], r->request_id[1]); + if (r->status == MEDIA_SYNC_CURSOR_INVALID) { + if (memcmp(r->source_epoch, peer->request.source_epoch, 16)) goto malformed; + if (md_sql_begin(md->db, "BEGIN IMMEDIATE")) return; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, "DELETE FROM super_sync WHERE group_id=? AND peer_node_id=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)from_node); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (md_sql_finish(md->db, st, rc, "reset ahead cursor") || md_sql_exec(md->db, "COMMIT")) { + md_sql_exec(md->db, "ROLLBACK"); return; } + peer->request_pending = peer->request_send = 0; + peer->timeout_tb = MEDIA_REPL_TIMEOUT_MAX_TB; + peer->deadline_tb = get_time_tb() + peer->timeout_tb; } - } - - md_ss_set(md->db, from_node, (uint64_t)max_id); - - uint8_t ack[sizeof(struct media_pkt_super_ack)]; - struct media_pkt_super_ack* sa = (struct media_pkt_super_ack*)ack; - sa->subcmd = MEDIA_SUBCMD_SUPER_ACK; - sa->ack_seq = (uint32_t)max_id; - struct media_super_peer* peer = md_super_peer_find(md, from_node); - md_send(md->inst, peer ? peer->group_id : TOPO_GROUP_UTUN, from_node, ack, sizeof(ack)); - - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SUPER_REPL from 0x%016llx entries=%d max_id=%lld", - MD_ID, (unsigned long long)from_node, num, (long long)max_id); -} - -static void md_handle_super_ack(struct media_delivery_ctx* md, uint64_t from_node, - const uint8_t* data, size_t len) { - if (len < sizeof(struct media_pkt_super_ack)) return; - 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) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%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--; - peer->timeout_tb = MEDIA_REPL_TIMEOUT_TB; - - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SUPER_ACK from 0x%016llx seq=%u inflight=%d", - MD_ID, (unsigned long long)from_node, sa->ack_seq, peer->inflight_count); - - md_super_repl_send(md, peer); -} - -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)) 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, sh->group_id); - if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)from_node); return; } - - /* группа связи определяется HELLO-пакетом (может отличаться от connect-time, напр. при ручном HELLO) */ - peer->group_id = sh->group_id; - peer->peer_last_recv_id = sh->last_recv_id; - peer->connected = 1; - peer->hello_done = 1; - - /* send our HELLO back */ - uint64_t my_last_recv = md_ss_get(md->db, from_node); - struct media_pkt_super_hello resp; - resp.subcmd = MEDIA_SUBCMD_SUPER_HELLO; - resp.last_recv_id = my_last_recv; - resp.group_id = peer->group_id; - md_send(md->inst, peer->group_id, from_node, (const uint8_t*)&resp, sizeof(resp)); - - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: SUPER_HELLO from 0x%016llx (peer_last_recv=%lld, my_last_recv=%lld)", - MD_ID, (unsigned long long)from_node, (long long)sh->last_recv_id, (long long)my_last_recv); - - /* start replication */ - peer->inflight_count = 0; - md_super_repl_send(md, peer); -} - -/* ── supernode connection ── */ - -static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_id, - enum conn_mgr_event event, void* arg) { - struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg; (void)h; (void)group_id; - struct media_super_peer* peer = md_super_peer_find(md, node_id); - if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: conn_cb for unknown peer 0x%016llx", MD_ID, (unsigned long long)node_id); return; } - - if (event != CONN_EVENT_UP) { - peer->connected = peer->hello_done = 0; - if (peer->timeout_timer) { uasync_cancel_timeout(md->inst->ua, peer->timeout_timer); peer->timeout_timer = NULL; } - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: connect to supernode 0x%016llx failed rc=%d", - MD_ID, (unsigned long long)node_id, event); return; } + uint64_t expected = memcmp(r->source_epoch, peer->request.source_epoch, 16) + ? 0 : md_read_le(&peer->request.after_id, 8); + if (from != expected || next < from || next > INT64_MAX || (r->more && count != MEDIA_SYNC_MAX_ENTRIES)) goto malformed; + const struct media_pkt_sync_entry* rows = (const struct media_pkt_sync_entry*)(data + sizeof(*r)); + uint64_t last = from; + for (unsigned i = 0; i < count; i++) { + uint64_t id = md_read_le(&rows[i].id, 8); + if (id <= last || id > next || !md_read_le(&rows[i].node_id, 8) + || md_read_le(&rows[i].timestamp, 8) > INT64_MAX || rows[i].status > 1) goto malformed; + for (unsigned j = 0; j < i; j++) { + if (memcmp(rows[i].block_id, rows[j].block_id, 16)) continue; + if (!memcmp(&rows[i].node_id, &rows[j].node_id, 8) + || memcmp(&rows[i].chunk, &rows[j].chunk, 4)) goto malformed; + } + last = id; + } + if (r->more && last != next) goto malformed; + if (md_sql_begin(md->db, "BEGIN IMMEDIATE")) return; + int changed = 0; + for (unsigned i = 0; i < count; i++) { + int applied = md_ba_merge(md->db, rows[i].block_id, group_id, md_read_le(&rows[i].node_id, 8), + (uint32_t)md_read_le(&rows[i].chunk, 4), (int64_t)md_read_le(&rows[i].timestamp, 8), + rows[i].status); + if (applied < 0) { md_sql_exec(md->db, "ROLLBACK"); return; } + changed += applied; + } + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, "INSERT OR REPLACE INTO super_sync" + "(group_id,peer_node_id,source_epoch,last_recv_id) VALUES(?,?,?,?)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)from_node); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, r->source_epoch, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)next); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (md_sql_finish(md->db, st, rc, "save RX cursor") || md_sql_exec(md->db, "COMMIT")) { + md_sql_exec(md->db, "ROLLBACK"); return; + } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SYNC committed group=%llx peer=%llx uuid=%02x%02x epoch=%02x%02x from=%llu next=%llu n=%u changed=%d more=%u dirty=%u", + MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, r->request_id[0], r->request_id[1], + r->source_epoch[0], r->source_epoch[1], (unsigned long long)from, (unsigned long long)next, + count, changed, r->more, peer->dirty); + int again = r->more || peer->dirty; + peer->request_pending = peer->request_send = 0; peer->timeout_tb = MEDIA_REPL_TIMEOUT_TB; + peer->deadline_tb = get_time_tb() + MEDIA_REPL_POLL_TB; + if (changed) md_repl_notify(md, group_id); + if (again) { + if (md_repl_request(peer)) peer->deadline_tb = get_time_tb() + peer->timeout_tb; + else md_repl_schedule(peer); + } + return; +malformed: + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed SYNC_RESP group=%llx peer=%llx bytes=%zu count=%u status=%u", + MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, len, count, r->status); +} - peer->connected = 1; - - /* send SUPER_HELLO */ - uint64_t my_last = md_ss_get(md->db, node_id); - struct media_pkt_super_hello hello; - hello.subcmd = MEDIA_SUBCMD_SUPER_HELLO; - hello.last_recv_id = my_last; - hello.group_id = peer->group_id; - - if (md_send(md->inst, peer->group_id, node_id, (const uint8_t*)&hello, sizeof(hello)) == 0) { - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%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); +/* CM владеет путём целиком: raw ETCP DOWN не отменяет исправный непрямой путь. */ +static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_id, + enum conn_mgr_event event, void* arg) { + struct media_delivery_ctx* md = arg; + struct media_super_peer* peer = md_super_peer_find(md, node_id, group_id); + if (!peer || (peer->connection && h && peer->connection != h)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: stale replication CM callback", MD_ID); return; + } + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: replication CM group=%llx peer=%llx event=%d", MD_ID, + (unsigned long long)group_id, (unsigned long long)node_id, event); + peer->connected = event == CONN_EVENT_UP; + peer->request_pending = peer->request_send = 0; + if (event == CONN_EVENT_UP) { + peer->timeout_tb = MEDIA_REPL_TIMEOUT_TB; + if (md_repl_request(peer)) peer->deadline_tb = get_time_tb() + peer->timeout_tb; + md_repl_schedule(peer); + return; + } + peer->response_pending = 0; + etcp_router_cancel_send_ready(md->inst, group_id, node_id, ETCP_RT_ID_MEDIA_DELIVERY, &peer->waiter); + if (peer->pump) { uasync_call_soon_cancel(md->inst->ua, peer->pump); peer->pump = NULL; } + if (event == CONN_EVENT_TIMEOUT) { + if (peer->connection) conn_mgr_close(peer->connection); + peer->connection = NULL; peer->deadline_tb = get_time_tb() + peer->timeout_tb; } - /* wait for HELLO response — md_handle_super_hello will start replication */ } /* ── handle incoming BLOCK_REQ (admission control + streaming) ── */ @@ -1025,133 +1080,150 @@ void media_delivery_stream_done(struct UTUN_INSTANCE* inst) { } static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id) { + if (!peer_node_id || peer_node_id == md->self_node_id) return; struct media_super_peer* peer = md_super_peer_add(md, peer_node_id, group_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; } - if (peer->connection) return; - - struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id); - if (!grp || !grp->conn_mgr) { - DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: no group/conn_mgr for supernode connect", MD_ID); - return; - } + if (peer->connection || get_time_tb() < peer->deadline_tb) return; DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: connecting to supernode 0x%016llx", MD_ID, (unsigned long long)peer_node_id); - if (conn_mgr_open(grp->instance, grp->group_id, peer_node_id, md_super_conn_cb, md, &peer->connection) < 0) + if (conn_mgr_open(md->inst, group_id, peer_node_id, md_super_conn_cb, md, &peer->connection) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: supernode CM request failed node=%llx group=%llx", MD_ID, (unsigned long long)peer_node_id, (unsigned long long)group_id); + peer->deadline_tb = get_time_tb() + peer->timeout_tb; + } } -/* ── supernode start/stop ── */ +/* Подписанные adm_tags — источник роли; node_type/deleted являются производными кешами. */ +static int md_tags_super(const char* tags) { + char value[16], deleted[16]; + return tags && json_flat_get(tags, "supernode", value, sizeof(value)) == 0 && !strcmp(value, "yes") + && !(json_flat_get(tags, "deleted", deleted, sizeof(deleted)) == 0 && !strcmp(deleted, "yes")); +} -static int md_super_start(struct media_delivery_ctx* md) { - 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); - - struct ll_entry* gle = md->inst->topo_groups->group_list->head; - while (gle) { - struct TOPO_GROUP* g = (struct TOPO_GROUP*)gle; - if (g->group_type != TOPO_GROUP_TYPE_CHAT) { gle = gle->next; continue; } - - char sql[256]; - snprintf(sql, sizeof(sql), - "SELECT node_id FROM peers_%s WHERE node_type=4 AND node_id!=%llu", - g->channel_id, (unsigned long long)md->self_node_id); - - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(md->db, sql, -1, &st, NULL) == SQLITE_OK) { - while (sqlite3_step(st) == SQLITE_ROW) { - uint64_t pid = (uint64_t)sqlite3_column_int64(st, 0); - md_super_connect(md, pid, g->group_id); - } - sqlite3_finalize(st); +/* 1/0 — актуальная роль в существующей CHAT-группе; -1 — ошибка SQL, не повод удалять кеш. */ +static int md_group_role(struct media_delivery_ctx* md, uint64_t group_id, uint64_t node_id) { + struct TOPO_GROUP* g = md->inst->topo_groups ? topo_groups_find(md->inst->topo_groups, group_id) : NULL; + if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || g->stopping || !g->channel_id[0]) return 0; + char sql[256]; + sqlite3_snprintf(sizeof(sql), sql, "SELECT adm_tags FROM \"peers_%w\" WHERE node_id=?", g->channel_id); + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); + int role = 0; + if (rc == SQLITE_OK) { + rc = sqlite3_step(st); + if (rc == SQLITE_ROW) { + role = md_tags_super((const char*)sqlite3_column_text(st, 0)); rc = sqlite3_step(st); } - gle = gle->next; + } + return md_sql_finish(md->db, st, rc, "read group role") ? -1 : role; +} + +/* Удалённые данные и RX-позиции всегда очищаются вместе; собственные объявления сохраняются при потере роли. */ +static int md_purge_group(struct media_delivery_ctx* md, uint64_t group_id, int all) { + if (md_sql_begin(md->db, "BEGIN IMMEDIATE")) return -1; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, + "DELETE FROM block_availability WHERE group_id=? AND (? OR node_id!=?)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)group_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 2, all); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)md->self_node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (md_sql_finish(md->db, st, rc, "purge group rows")) goto failed; + int removed = sqlite3_changes(md->db); + st = NULL; + rc = sqlite3_prepare_v2(md->db, "DELETE FROM super_sync WHERE group_id=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)group_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (md_sql_finish(md->db, st, rc, "purge group cursors") || md_sql_exec(md->db, "COMMIT")) goto failed; + if (removed) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: group cache reset group=%llx rows=%d all=%d", + MD_ID, (unsigned long long)group_id, removed, all); + return 0; +failed: + md_sql_exec(md->db, "ROLLBACK"); return -1; +} + +/* Единственный суперузел тоже должен реплицировать собственные зарегистрированные блоки. + * Стабильный timestamp из media_files исключает новые ID при каждом перечитывании ролей. */ +static int md_seed_local(struct media_delivery_ctx* md, uint64_t group_id) { + char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id); + if (md_sql_begin(md->db, "BEGIN IMMEDIATE")) return -1; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(md->db, "SELECT block_id,chunk,timestamp FROM media_files WHERE chat_id=? AND node_id=?", + -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)md->self_node_id); + int changed = 0; + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + sqlite3_int64 chunk = sqlite3_column_int64(st, 1), ts = sqlite3_column_int64(st, 2); + if (sqlite3_column_bytes(st, 0) != 16 || chunk < 0 || chunk > UINT32_MAX || ts < 0) { rc = SQLITE_CORRUPT; break; } + int applied = md_ba_merge(md->db, sqlite3_column_blob(st, 0), group_id, md->self_node_id, (uint32_t)chunk, ts, 1); + if (applied < 0) { rc = SQLITE_ERROR; break; } + changed += applied; + } + if (md_sql_finish(md->db, st, rc, "seed own blocks") || md_sql_exec(md->db, "COMMIT")) { + md_sql_exec(md->db, "ROLLBACK"); return -1; + } + if (changed) { + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: own blocks seeded group=%llx changed=%d", MD_ID, + (unsigned long long)group_id, changed); + md_repl_notify(md, group_id); } return 0; } -static void md_super_stop(struct media_delivery_ctx* md) { - 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); - - /* delete foreign blocks */ - char sql[128]; - snprintf(sql, sizeof(sql), - "DELETE FROM block_availability WHERE node_id!=%llu", - (unsigned long long)md->self_node_id); - sqlite3_exec(md->db, sql, NULL, NULL, NULL); - - /* clear super_peers */ - if (md->super_peers) { - while (md->super_peers->head) - md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id); - queue_free(md->super_peers); - md->super_peers = NULL; +/* Согласование идемпотентно, в том числе при новой группе после init. Ошибка роли сохраняет прежний контекст. */ +void media_delivery_refresh_roles(struct UTUN_INSTANCE* inst) { + if (!inst || !inst->md.initialized || !inst->topo_groups) return; + struct media_delivery_ctx* md = &inst->md; + if (!sqlite3_get_autocommit(md->db)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: role reconciliation deferred during SQL transaction", MD_ID); return; } -} - -/* ── callbacks ── */ - -static void md_on_bgp_node(struct TOPO_GROUP* group, uint64_t node_id, int event, void* arg) { - struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg; - if (!md->is_supernode || !md->db) return; - - if (event == TOPO_NODE_EVENT_REMOVE) { - char sql[128]; - snprintf(sql, sizeof(sql), - "DELETE FROM block_availability WHERE node_id=%llu", - (unsigned long long)node_id); - sqlite3_exec(md->db, sql, NULL, NULL, NULL); - md_super_peer_remove(md, node_id); - return; + for (struct ll_entry* e = md->super_peers ? md->super_peers->head : NULL; e;) { + struct media_super_peer* peer = (struct media_super_peer*)e->data; + e = e->next; + int own = md_group_role(md, peer->group_id, md->self_node_id); + int other = md_group_role(md, peer->group_id, peer->peer_node_id); + if (own == 0 || other == 0) md_super_peer_remove(md, peer->peer_node_id, peer->group_id); } - - /* check if this node is a supernode */ - struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, node_id); - struct TOPO_NODE* ni = nq ? topo_node_registry_find(group->instance->topo_groups, nq->node_id) : NULL; - if (!nq || !ni) return; - - /* for CHAT groups, check node_type via DB */ - if (group->group_type == TOPO_GROUP_TYPE_CHAT && group->channel_id[0]) { + int any = 0; + for (struct ll_entry* e = inst->topo_groups->group_list->head; e; e = e->next) { + struct TOPO_GROUP* group = (struct TOPO_GROUP*)e; + if (group->group_type != TOPO_GROUP_TYPE_CHAT || group->stopping) continue; + int own = md_group_role(md, group->group_id, md->self_node_id); + if (own < 0) { any |= md->is_supernode; continue; } + if (!own) { md_purge_group(md, group->group_id, 0); continue; } + any = 1; + md_seed_local(md, group->group_id); char sql[256]; - snprintf(sql, sizeof(sql), - "SELECT node_type FROM peers_%s WHERE node_id=%llu", - group->channel_id, (unsigned long long)node_id); + sqlite3_snprintf(sizeof(sql), sql, "SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=?", group->channel_id); sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(md->db, sql, -1, &st, NULL) == SQLITE_OK) { - int nt = 0; - if (sqlite3_step(st) == SQLITE_ROW) nt = sqlite3_column_int(st, 0); - sqlite3_finalize(st); - if (nt == 4) md_super_connect(md, node_id, group->group_id); + int rc = sqlite3_prepare_v2(md->db, sql, -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)md->self_node_id); + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + if (md_tags_super((const char*)sqlite3_column_text(st, 1))) + md_super_connect(md, (uint64_t)sqlite3_column_int64(st, 0), group->group_id); } + md_sql_finish(md->db, st, rc, "enumerate supernodes"); } + if (md->is_supernode != any) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: aggregate supernode role=%d", MD_ID, any); + md->is_supernode = (uint8_t)any; } -/* adm_tags хранится как JSON: {"supernode":"yes",...}. Проверяем через json_flat_get, - * т.к. strstr(tags,"supernode=yes") не совпадает с JSON-форматом. */ -static int md_adm_tags_is_supernode(const char* adm_tags) { - char buf[16]; - return adm_tags && json_flat_get(adm_tags, "supernode", buf, sizeof(buf)) == 0 - && strcmp(buf, "yes") == 0; +static void md_reconcile_timer(void* arg) { + struct media_delivery_ctx* md = arg; + md->reconcile_timer = NULL; + media_delivery_refresh_roles(md->inst); + md->reconcile_timer = uasync_set_timeout(md->inst->ua, MEDIA_REPL_POLL_TB, md, md_reconcile_timer, "md_roles"); + if (!md->reconcile_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: role reconciliation timer allocation failed", MD_ID); } +/* member_sync сообщает и изменение подписанных свойств, и BGP-события; REMOVE маршрута не удаляет участника. */ static void md_on_props_changed(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg) { - struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg; - int is_super = md_adm_tags_is_supernode(adm_tags); - - if (node_id == md->self_node_id) { - media_delivery_set_supernode(md->inst, is_super); - } else if (md->is_supernode) { - if (is_super) md_super_connect(md, node_id, strtoull(channel_id, NULL, 10)); - else md_super_peer_remove(md, node_id); - } + (void)node_id; (void)adm_tags; (void)channel_id; + struct media_delivery_ctx* md = arg; + media_delivery_refresh_roles(md->inst); } static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) { @@ -1166,11 +1238,6 @@ static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) { struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &node_id); if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); } } - { struct media_super_peer* sp = md_super_peer_find(md, node_id); - if (sp) { - sp->connected = 0; sp->hello_done = 0; sp->inflight_count = 0; - if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; } - } } break; default: break; } @@ -1202,6 +1269,14 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: recv %s(%02x) from 0x%016llx len=%zu", MD_ID, md_subcmd_name(subcmd), subcmd, (unsigned long long)from_node, len); + if (subcmd == MEDIA_SUBCMD_SYNC_REQ || subcmd == MEDIA_SUBCMD_SYNC_RESP || subcmd == MEDIA_SUBCMD_CHANGED) { + if (!from_node || from_node == md->self_node_id || md_group_role(md, group_id, md->self_node_id) != 1 + || md_group_role(md, group_id, from_node) != 1) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: replication role rejected group=%llx peer=%llx cmd=%u", MD_ID, + (unsigned long long)group_id, (unsigned long long)from_node, subcmd); + queue_dgram_free(entry); queue_entry_free(entry); return; + } + } switch (subcmd) { case MEDIA_SUBCMD_SERVE_REG: md_handle_serve_reg(md, from_node, data, len); @@ -1218,24 +1293,27 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { 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); + md_handle_announcement(md, group_id, from_node, data, len, 1); break; case MEDIA_SUBCMD_BLOCK_PROCESSING: - if (md->is_supernode) md_handle_block_processing(md, from_node, data, len); - else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: BLOCK_PROCESSING ignored — not supernode", MD_ID); + md_handle_announcement(md, group_id, from_node, data, len, 0); break; - case MEDIA_SUBCMD_SUPER_REPL: - if (md->is_supernode) md_handle_super_repl(md, from_node, data, len); - else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SUPER_REPL ignored — not supernode", MD_ID); + case MEDIA_SUBCMD_SYNC_REQ: + md_handle_sync_req(md, group_id, from_node, data, len); break; - case MEDIA_SUBCMD_SUPER_ACK: - if (md->is_supernode) md_handle_super_ack(md, from_node, data, len); - else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SUPER_ACK ignored — not supernode", MD_ID); + case MEDIA_SUBCMD_SYNC_RESP: + md_handle_sync_resp(md, group_id, from_node, data, len); break; - case MEDIA_SUBCMD_SUPER_HELLO: - if (md->is_supernode) md_handle_super_hello(md, from_node, data, len); - else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: SUPER_HELLO ignored — not supernode", MD_ID); + case MEDIA_SUBCMD_CHANGED: { + if (len != 1) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: malformed CHANGED bytes=%zu", MD_ID, len); break; } + struct media_super_peer* peer = md_super_peer_find(md, from_node, group_id); + if (!peer) { md_super_connect(md, from_node, group_id); peer = md_super_peer_find(md, from_node, group_id); } + if (peer) { + if (peer->request_pending) peer->dirty = 1; + else peer->deadline_tb = 0; + } break; + } case MEDIA_SUBCMD_QUERY_RESP: media_download_handle_query_resp(inst, group_id, from_node, data, len); break; @@ -1271,38 +1349,53 @@ static void md_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) { /* ── table creation ── */ +/* Версия относится только к вспомогательному каталогу, файлы/чаты при переходе не затрагиваются. */ static int media_delivery_create_tables(struct UTUN_INSTANCE* inst) { sqlite3* db = inst->topo_sqlite_db; - if (!db) return 0; - - const char* sql_ba = - "CREATE TABLE IF NOT EXISTS block_availability (" - " id INTEGER PRIMARY KEY AUTOINCREMENT," - " block_uuid BLOB NOT NULL," - " group_id INTEGER NOT NULL," - " node_id INTEGER NOT NULL," - " chunk INTEGER NOT NULL," - " timestamp INTEGER NOT NULL," - " status INTEGER NOT NULL DEFAULT 1);"; // 0=processing, 1=completed - if (sqlite3_exec(db, sql_ba, NULL, NULL, NULL) != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: CREATE block_availability failed: %s", MD_ID, sqlite3_errmsg(db)); - return -1; + if (md_sql_begin(db, "BEGIN IMMEDIATE")) return -1; + if (md_sql_exec(db, "CREATE TABLE IF NOT EXISTS media_replication_meta" + "(singleton INTEGER PRIMARY KEY CHECK(singleton=1),version INTEGER NOT NULL,source_epoch BLOB NOT NULL)")) goto failed; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "SELECT version,source_epoch," + "(SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('block_availability','super_sync'))" + " FROM media_replication_meta WHERE singleton=1", -1, &st, NULL); + uint8_t epoch[16] = {0}; + int reset = 1; + if (rc == SQLITE_OK) { + rc = sqlite3_step(st); + if (rc == SQLITE_ROW) { + if (sqlite3_column_int(st, 0) == 1 && sqlite3_column_bytes(st, 1) == 16 && sqlite3_column_int(st, 2) == 2) { + memcpy(epoch, sqlite3_column_blob(st, 1), 16); reset = 0; + } + rc = sqlite3_step(st); + } } - sqlite3_exec(db, "CREATE UNIQUE INDEX IF NOT EXISTS idx_ba_uuid_node ON block_availability(block_uuid, node_id);", NULL, NULL, NULL); - sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS idx_ba_id ON block_availability(id);", NULL, NULL, NULL); - sqlite3_exec(db, "CREATE INDEX IF NOT EXISTS idx_ba_node_id ON block_availability(node_id);", NULL, NULL, NULL); - - const char* sql_ss = - "CREATE TABLE IF NOT EXISTS super_sync (" - " peer_node_id INTEGER PRIMARY KEY," - " last_recv_id INTEGER NOT NULL DEFAULT 0);"; - if (sqlite3_exec(db, sql_ss, NULL, NULL, NULL) != SQLITE_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: CREATE super_sync failed: %s", MD_ID, sqlite3_errmsg(db)); - return -1; + if (md_sql_finish(db, st, rc, "catalog metadata")) goto failed; + if (reset) { + if (media_index_generate_uuid(epoch) || md_sql_exec(db, + "DROP TABLE IF EXISTS block_availability; DROP TABLE IF EXISTS super_sync;" + "CREATE TABLE block_availability(" + "id INTEGER PRIMARY KEY AUTOINCREMENT,block_uuid BLOB NOT NULL CHECK(length(block_uuid)=16)," + "group_id INTEGER NOT NULL,node_id INTEGER NOT NULL CHECK(node_id!=0)," + "chunk INTEGER NOT NULL CHECK(chunk BETWEEN 0 AND 4294967295)," + "timestamp INTEGER NOT NULL CHECK(timestamp>=0),status INTEGER NOT NULL CHECK(status IN (0,1))," + "UNIQUE(group_id,block_uuid,node_id));" + "CREATE INDEX idx_ba_group_id ON block_availability(group_id,id);" + "CREATE TABLE super_sync(group_id INTEGER NOT NULL,peer_node_id INTEGER NOT NULL," + "source_epoch BLOB NOT NULL CHECK(length(source_epoch)=16),last_recv_id INTEGER NOT NULL CHECK(last_recv_id>=0)," + "PRIMARY KEY(group_id,peer_node_id));")) goto failed; + st = NULL; + rc = sqlite3_prepare_v2(db, "INSERT OR REPLACE INTO media_replication_meta VALUES(1,1,?)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, epoch, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + if (md_sql_finish(db, st, rc, "write catalog epoch")) goto failed; } - - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: tables created", MD_ID); + if (md_sql_exec(db, "COMMIT")) goto failed; + memcpy(inst->md.source_epoch, epoch, 16); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: catalog initialized reset=%d epoch=%02x%02x", MD_ID, reset, epoch[0], epoch[1]); return 0; +failed: + md_sql_exec(db, "ROLLBACK"); return -1; } /* ── public API ── */ @@ -1329,31 +1422,13 @@ int media_delivery_init(struct UTUN_INSTANCE* inst) { etcp_add_conn_status_cbk(inst, md_on_conn_status, md); member_sync_add_props_cbk(inst, md_on_props_changed, md); - /* determine initial supernode state and subscribe to BGP callbacks */ - struct ll_entry* gle = inst->topo_groups ? inst->topo_groups->group_list->head : NULL; - while (gle) { - struct TOPO_GROUP* g = (struct TOPO_GROUP*)gle; - topo_group_add_node_cbk(g, md_on_bgp_node, md); - if (g->group_type == TOPO_GROUP_TYPE_CHAT && g->channel_id[0] && inst->topo_sqlite_db) { - char sql[256]; - snprintf(sql, sizeof(sql), - "SELECT adm_tags FROM peers_%s WHERE node_id=%llu", - g->channel_id, (unsigned long long)inst->node_id); - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) == SQLITE_OK) { - if (sqlite3_step(st) == SQLITE_ROW) { - const char* tags = (const char*)sqlite3_column_text(st, 0); - if (md_adm_tags_is_supernode(tags)) { md->is_supernode = 1; } - } - sqlite3_finalize(st); - } - } - gle = gle->next; - } - - if (md->is_supernode) md_super_start(md); - md->initialized = 1; + media_delivery_refresh_roles(inst); + md->reconcile_timer = uasync_set_timeout(inst->ua, MEDIA_REPL_POLL_TB, md, md_reconcile_timer, "md_roles"); + if (!md->reconcile_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: role reconciliation timer allocation failed", MD_ID); + media_delivery_destroy(inst); return -1; + } DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: initialized node_id=0x%016llx super=%d", MD_ID, (unsigned long long)md->self_node_id, md->is_supernode); return 0; @@ -1364,23 +1439,18 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) { struct media_delivery_ctx* md = &inst->md; md->initialized = 0; + if (md->reconcile_timer) { uasync_cancel_timeout(inst->ua, md->reconcile_timer); md->reconcile_timer = NULL; } while (md->streams) stream_free(md->streams); etcp_router_unbind(inst, ETCP_RT_ID_MEDIA_DELIVERY); etcp_remove_conn_status_cbk(inst, md_on_conn_status, md); member_sync_remove_props_cbk(inst, md_on_props_changed, md); - /* unsubscribe from BGP callbacks */ - struct ll_entry* gle = inst->topo_groups ? inst->topo_groups->group_list->head : NULL; - while (gle) { - struct TOPO_GROUP* g = (struct TOPO_GROUP*)gle; - topo_group_remove_node_cbk(g, md_on_bgp_node, md); - gle = gle->next; - } - if (md->super_peers) { - while (md->super_peers->head) - md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id); + while (md->super_peers->head) { + struct media_super_peer* peer = (struct media_super_peer*)md->super_peers->head->data; + md_super_peer_remove(md, peer->peer_node_id, peer->group_id); + } queue_free(md->super_peers); md->super_peers = NULL; } @@ -1429,11 +1499,12 @@ void media_delivery_cancel_channel(struct UTUN_INSTANCE* inst, uint64_t group_id struct media_super_peer* sp = (struct media_super_peer*)se->data; struct ll_entry* next = se->next; if (sp->group_id == group_id) - md_super_peer_remove(md, sp->peer_node_id); + md_super_peer_remove(md, sp->peer_node_id, group_id); se = next; } } + md_purge_group(md, group_id, 1); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cancel_channel group=%016llx downloads=%d", MD_ID, (unsigned long long)group_id, nm); } @@ -1442,30 +1513,21 @@ int media_delivery_bind(struct UTUN_INSTANCE* inst) { return media_delivery_init(inst); } -/* Собирает уникальных суперузлов (node_type=4, не сам) всех CHAT-групп. */ -static int md_collect_supernodes(struct UTUN_INSTANCE* inst, uint64_t supers[10]) { - int nsup = 0; - struct ll_entry* gle = inst->topo_groups ? inst->topo_groups->group_list->head : NULL; - while (gle) { - struct TOPO_GROUP* g = (struct TOPO_GROUP*)gle; - if (g->group_type == TOPO_GROUP_TYPE_CHAT && g->channel_id[0] && inst->topo_sqlite_db) { - char sql[256]; - snprintf(sql, sizeof(sql), "SELECT node_id FROM peers_%s WHERE node_type=4 AND node_id!=%llu", - g->channel_id, (unsigned long long)inst->node_id); - sqlite3_stmt* st = NULL; - if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) == SQLITE_OK) { - while (sqlite3_step(st) == SQLITE_ROW && nsup < 10) { - uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0); - int dup = 0; - for (int i = 0; i < nsup; i++) if (supers[i] == nid) { dup = 1; break; } - if (!dup) supers[nsup++] = nid; - } - sqlite3_finalize(st); - } - } - gle = gle->next; +/* Адресаты только этой группы, роли читаем из подписанных adm_tags, не из node_type. */ +static int md_collect_supernodes(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t supers[10]) { + struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL; + if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || !g->channel_id[0]) return 0; + char sql[256]; + sqlite3_snprintf(sizeof(sql), sql, "SELECT node_id,adm_tags FROM \"peers_%w\" WHERE node_id!=?", g->channel_id); + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(inst->md.db, sql, -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); + int count = 0; + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + uint64_t node = (uint64_t)sqlite3_column_int64(st, 0); + if (node && md_tags_super((const char*)sqlite3_column_text(st, 1)) && count < 10) supers[count++] = node; } - return nsup; + return md_sql_finish(inst->md.db, st, rc, "announcement recipients") ? 0 : count; } void media_delivery_send_have_block(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, @@ -1483,6 +1545,21 @@ void media_delivery_send_have_block(struct UTUN_INSTANCE* inst, uint64_t group_i memcpy(hb.block_id, block_id, 16); hb.chunk = chunk; hb.timestamp = (int64_t)time(NULL); + /* Зарегистрированный блок переанонсируется с прежним временем, не создавая новое событие. */ + sqlite3_stmt* st = NULL; + char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group_id); + int rc = sqlite3_prepare_v2(inst->md.db, "SELECT timestamp FROM media_files" + " WHERE chat_id=? AND media_id=? AND block_id=? AND node_id=? AND chunk=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, media_id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, block_id, 16, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 5, chunk); + if (rc == SQLITE_OK) { + rc = sqlite3_step(st); + if (rc == SQLITE_ROW) { hb.timestamp = sqlite3_column_int64(st, 0); rc = sqlite3_step(st); } + } + if (md_sql_finish(inst->md.db, st, rc, "read own announcement timestamp")) return; uint8_t smsg[64]; size_t soff = 0; memcpy(smsg + soff, hb.block_id, 16); soff += 16; @@ -1507,7 +1584,8 @@ void media_delivery_announce_media(struct UTUN_INSTANCE* inst, uint64_t group_id struct media_delivery_ctx* md = &inst->md; if (!md->initialized) return; - uint64_t supers[10]; int nsup = md_collect_supernodes(inst, supers); + md_seed_local(md, group_id); + uint64_t supers[10]; int nsup = md_collect_supernodes(inst, group_id, supers); if (nsup == 0) { DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: announce — no supernodes for group 0x%016llx", MD_ID, (unsigned long long)group_id); @@ -1529,8 +1607,7 @@ void media_delivery_announce_local_blocks(struct UTUN_INSTANCE* inst) { sqlite3* db = inst->topo_sqlite_db; if (!db) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: announce_local_blocks — no DB", MD_ID); return; } - uint64_t supers[10]; int nsup = md_collect_supernodes(inst, supers); - if (nsup == 0) { DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: announce_local_blocks — no supernodes", MD_ID); return; } + media_delivery_refresh_roles(inst); sqlite3_stmt* st = NULL; if (sqlite3_prepare_v2(db, "SELECT media_id, block_id, chunk, chat_id FROM media_files WHERE node_id=?", @@ -1549,25 +1626,11 @@ void media_delivery_announce_local_blocks(struct UTUN_INSTANCE* inst) { if (!media_id || !block_id || !chat_id || !chat_id[0]) continue; uint64_t gid = strtoull(chat_id, NULL, 10); if (gid == 0) continue; + uint64_t supers[10]; int nsup = md_collect_supernodes(inst, gid, supers); for (int s = 0; s < nsup; s++) media_delivery_send_have_block(inst, gid, supers[s], media_id, block_id, chunk); announced++; } sqlite3_finalize(st); - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: announce_local_blocks done, blocks=%d supers=%d", - MD_ID, announced, nsup); -} - -void media_delivery_set_supernode(struct UTUN_INSTANCE* inst, int is_supernode) { - if (!inst) return; - struct media_delivery_ctx* md = &inst->md; - if (is_supernode && !md->is_supernode) { - md->is_supernode = 1; - md_super_start(md); - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: supernode ENABLED", MD_ID); - } else if (!is_supernode && md->is_supernode) { - md->is_supernode = 0; - md_super_stop(md); - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: supernode DISABLED", MD_ID); - } + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: announce_local_blocks done, blocks=%d", MD_ID, announced); }