jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
e75e3750aa825486cbf12677b9c94eeea008deb1
utun2/src/chat/merkle_sync.c
T
evgeny 7b73d99b5b chat: передавать group_id в бинарном виде в протоколе синхронизации 
db_sync/merkle_sync/chat_sync шлют group_id (uint64 BE) вместо текстового
channel_id/namespace. Приёмная сторона находит группу/таблицу через
topo_groups_find(group_id) и требует существующую CHAT-группу. db_sync:
instance key заменён с SHA256(ch_id) на strtoull(ch_id)=group_id.
2026-09-04 09:02:46 +03:00

913 lines
38 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#include "merkle_sync.h"
#include "../utun_instance.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../routing_layer/topo_group.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../../lib/ll_queue.h"
#include "../../lib/platform_compat.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#define MS_ID "merkle_sync"
#define MS_WAL_INTERVAL_MS 60000
#define MS_PENDING_MAX 128
struct bucket_entry { uint8_t level; uint8_t prefix_bytes; uint64_t prefix; };
enum { SESS_SYNCING, SESS_SYNCED };
enum { MS_PEND_WAITING = 0, MS_PEND_RESOLVED = 1 };
struct ms_pending {
uint8_t level;
uint64_t prefix;
int8_t parent;
int sub_count;
uint8_t state;
};
struct ms_session {
struct ms_session* next;
struct merkle_sync* ms;
char ns[64];
uint64_t peer;
uint8_t sess_state;
merkle_sync_done_cb done_cb;
void* cb_arg;
struct ms_pending pending[MS_PENDING_MAX];
int pending_count;
struct ll_queue* out_q;
struct queue_waiter_handle waiter;
};
struct merkle_sync {
struct UTUN_INSTANCE* inst;
uint8_t svc_id;
const struct merkle_sync_data_ops* ops;
void* data_ctx;
struct ms_session* sessions;
uint8_t initialized;
void* wal_timer;
void* wal_wait; /* standby_wait handle (Android) */
};
/* ── DB access ── */
static sqlite3* _db(struct UTUN_INSTANCE* inst) {
return inst ? inst->topo_sqlite_db : NULL;
}
/* ── Prefix arithmetic ── */
uint64_t merkle_sync_level_prefix(uint64_t key, uint8_t level) {
if (level == 0) return 0;
int shift = 64 - (int)level * 5; if (shift < 0) shift = 0;
return (key >> shift) << shift;
}
uint8_t merkle_sync_prefix_bytes(uint8_t level) { int bits = level * 5; return (uint8_t)((bits + 7) / 8); }
static void _prefix_write(uint8_t* out, uint64_t prefix, uint8_t pb) {
for (int i = 0; i < (int)pb; i++)
out[i] = (uint8_t)(prefix >> (8 * (7 - i)));
}
static uint64_t _prefix_read(const uint8_t* in, uint8_t pb) {
uint64_t v = 0;
for (uint8_t i = 0; i < pb && i < 8; i++) v = (v << 8) | in[i];
return v << (8 * (8 - (int)pb));
}
static int _popcount_u32(uint32_t v) { return __builtin_popcount(v); }
/* ── Ensure merkle_tree_hash table ── */
static void _ensure_table(struct merkle_sync* ms) {
sqlite3* db = _db(ms->inst); if (!db) return;
sqlite3_exec(db,
"CREATE TABLE IF NOT EXISTS merkle_tree_hash ("
" namespace TEXT NOT NULL,"
" level INTEGER NOT NULL CHECK(level BETWEEN 1 AND 5),"
" prefix64 INTEGER NOT NULL,"
" hash BLOB NOT NULL,"
" member_count INTEGER NOT NULL,"
" PRIMARY KEY (namespace, level, prefix64))",
NULL, NULL, NULL);
}
/* ── Tree operations ── */
static int _recompute_bucket(struct merkle_sync* ms, const char* ns,
uint8_t level, uint64_t prefix64) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
sqlite3* db = _db(ms->inst); if (!db) return -1;
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
int count = ms->ops->update_bucket_hash(ms->data_ctx, ns, level, prefix64, ctx);
if (count < 0) { DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → error", MS_ID); EVP_MD_CTX_free(ctx); return -1; }
if (count == 0) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → empty", MS_ID);
EVP_MD_CTX_free(ctx);
sqlite3_stmt* cs = NULL;
sqlite3_prepare_v2(db,
"SELECT COUNT(*) FROM merkle_tree_hash WHERE namespace=? AND level=? AND prefix64=?", -1, &cs, NULL);
if (cs) { sqlite3_bind_text(cs, 1, ns, -1, SQLITE_STATIC); sqlite3_bind_int(cs, 2, level);
sqlite3_bind_int64(cs, 3, (sqlite3_int64)prefix64);
if (sqlite3_step(cs) == SQLITE_ROW && sqlite3_column_int(cs, 0) > 0) {
sqlite3_finalize(cs);
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → cleared (was non-empty)", MS_ID);
sqlite3_stmt* ds = NULL;
sqlite3_prepare_v2(db,
"DELETE FROM merkle_tree_hash WHERE namespace=? AND level=? AND prefix64=?", -1, &ds, NULL);
if (ds) { sqlite3_bind_text(ds, 1, ns, -1, SQLITE_STATIC); sqlite3_bind_int(ds, 2, level);
sqlite3_bind_int64(ds, 3, (sqlite3_int64)prefix64); sqlite3_step(ds); sqlite3_finalize(ds); }
return 1;
} else { sqlite3_finalize(cs); }
}
return 0;
}
uint8_t new_hash[MT_HASH_SIZE];
EVP_DigestFinal_ex(ctx, new_hash, NULL);
EVP_MD_CTX_free(ctx);
const uint8_t* old_hash = merkle_sync_get_hash(ms->inst, ns, level, prefix64);
if (memcmp(old_hash, new_hash, MT_HASH_SIZE) == 0) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → unchanged (count=%d)", MS_ID, count);
return 0;
}
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: → saved, count=%d", MS_ID, count);
sqlite3_stmt* is = NULL;
sqlite3_prepare_v2(db,
"INSERT OR REPLACE INTO merkle_tree_hash(namespace, level, prefix64, hash, member_count)"
" VALUES(?,?,?,?,?)", -1, &is, NULL);
if (is) {
sqlite3_bind_text(is, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(is, 2, level);
sqlite3_bind_int64(is, 3, (sqlite3_int64)prefix64);
sqlite3_bind_blob(is, 4, new_hash, MT_HASH_SIZE, SQLITE_STATIC);
sqlite3_bind_int(is, 5, count);
sqlite3_step(is); sqlite3_finalize(is);
}
return 1;
}
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recompute_path called but msync not initialized", MS_ID); return -1; }
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: ns=%s key=%016llx", MS_ID, ns, (unsigned long long)key);
int changed = 0;
for (uint8_t level = 1; level <= MT_MAX_LEVEL; level++) {
int r = _recompute_bucket(ms, ns, level, merkle_sync_level_prefix(key, level));
if (r < 0) return -1;
if (r > 0) changed = 1;
}
return changed;
}
const uint8_t* merkle_sync_get_hash(struct UTUN_INSTANCE* inst, const char* ns,
uint8_t level, uint64_t prefix64) {
static uint8_t zero[MT_HASH_SIZE];
sqlite3* db = _db(inst); if (!db) { memset(zero, 0, MT_HASH_SIZE); return zero; }
sqlite3_stmt* stmt = NULL;
sqlite3_prepare_v2(db,
"SELECT hash FROM merkle_tree_hash WHERE namespace=? AND level=? AND prefix64=?",
-1, &stmt, NULL);
if (!stmt) { memset(zero, 0, MT_HASH_SIZE); return zero; }
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, level);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)prefix64);
memset(zero, 0, MT_HASH_SIZE);
if (sqlite3_step(stmt) == SQLITE_ROW) {
const void* h = sqlite3_column_blob(stmt, 0);
if (h) memcpy(zero, h, MT_HASH_SIZE);
}
sqlite3_finalize(stmt);
return zero;
}
static int _get_level_hashes(struct merkle_sync* ms, const char* ns,
uint8_t level, uint64_t prefix, uint8_t prefix_bytes,
uint32_t* bitmap, uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE]) {
(void)prefix_bytes;
memset(hashes, 0, sizeof(uint8_t) * MT_BUCKETS * MT_HASH_SIZE);
*bitmap = 0;
if (level >= MT_MAX_LEVEL) return 0;
sqlite3* db = _db(ms->inst); if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: _get_level_hashes — db is NULL", MS_ID); return -1; }
int next_shift = 64 - ((int)level + 1) * 5; if (next_shift < 0) next_shift = 0;
sqlite3_stmt* stmt = NULL;
int query_level = (int)(level + 1);
const char* sql = (level == 0)
? "SELECT prefix64, hash FROM merkle_tree_hash"
" WHERE namespace=? AND level=? ORDER BY prefix64"
: "SELECT prefix64, hash FROM merkle_tree_hash"
" WHERE namespace=? AND level=? AND prefix64>=? AND prefix64<=?"
" ORDER BY prefix64";
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK || !stmt) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: _get_level_hashes SQL error ns=%s L%d — %s", MS_ID, ns, level, sqlite3_errmsg(db));
return -1;
}
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, query_level);
if (level > 0) {
uint64_t range_end = prefix | (0x1FULL << (next_shift > 0 ? next_shift : 0));
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)prefix);
sqlite3_bind_int64(stmt, 4, (sqlite3_int64)range_end);
}
while (sqlite3_step(stmt) == SQLITE_ROW) {
sqlite3_int64 row_pfx = sqlite3_column_int64(stmt, 0);
const void* h = sqlite3_column_blob(stmt, 1);
if (!h || sqlite3_column_bytes(stmt, 1) < MT_HASH_SIZE) continue;
int bucket = (int)(((uint64_t)row_pfx >> next_shift) & 0x1F);
*bitmap |= (1u << bucket);
memcpy(hashes[bucket], h, MT_HASH_SIZE);
}
sqlite3_finalize(stmt);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: _get_level_hashes ns=%s L%d/P%016llx bitmap=%08x",
MS_ID, ns, level, (unsigned long long)prefix, *bitmap);
return 0;
}
/* ── group_id в wire: [group_id:8] вместо [ns_len:1][ns] ── */
static uint64_t ms_ns_to_group_id(const char* ns) {
return strtoull(ns, NULL, 10);
}
static void ms_write_group_id(uint8_t* p, const char* ns) {
uint64_t gb = htobe64(ms_ns_to_group_id(ns));
memcpy(p, &gb, 8);
}
/* group_id → ns через группу канала (требует существующую CHAT-группу). 0=ок, -1=нет группы */
static int ms_group_id_to_ns(struct UTUN_INSTANCE* inst, uint64_t group_id, char* ns, size_t sz) {
if (!inst || !inst->topo_groups) return -1;
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, group_id);
if (!g || !g->channel_id[0]) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv — no CHAT group for group=%016llx, drop", MS_ID, (unsigned long long)group_id);
return -1;
}
snprintf(ns, sz, "%s", g->channel_id);
return 0;
}
/* ── Send helpers ── */
static struct ETCP_CONN* ms_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: find_conn — inst->connections is NULL", MS_ID); return NULL; }
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: find_conn — node=%016llx NOT FOUND", MS_ID, (unsigned long long)node_id); return NULL; }
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d",
MS_ID, (unsigned long long)node_id, 1, conn->initialized, conn->links_up);
if (conn->initialized && conn->links_up) return conn;
return NULL;
}
/* ── Per-session out queue (broadcast/push с backpressure) ── */
static void ms_session_outq_free(struct ms_session* s) {
if (!s) return;
if (s->out_q && s->ms && s->ms->inst) {
struct ETCP_CONN* conn = ms_find_conn_for_node(s->ms->inst, s->peer);
if (conn && conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &s->waiter);
}
if (s->out_q) {
struct ll_entry* e;
while ((e = queue_data_get(s->out_q)) != NULL) {
if (e->dgram) u_free(e->dgram);
queue_entry_free(e);
}
queue_free(s->out_q);
s->out_q = NULL;
}
}
/* Инициализировать out-очередь сессии (вызывается при создании сессии). */
static void ms_session_setup(struct merkle_sync* ms, struct ms_session* s) {
s->ms = ms;
s->out_q = queue_new(ms->inst->ua, 0, 0, 0, "ms_out");
if (!s->out_q) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: out_q queue_new failed", MS_ID);
}
static void ms_outq_drain_cb(struct ll_queue* q, void* arg) {
struct ms_session* s = (struct ms_session*)arg;
struct merkle_sync* ms = s ? s->ms : NULL;
if (!ms || !s->out_q) return;
struct ll_entry* e = queue_data_get(s->out_q);
if (!e) return;
struct ETCP_CONN* conn = ms_find_conn_for_node(ms->inst, s->peer);
if (!conn) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: outq drain — no conn for peer=%016llx, drop %u bytes",
MS_ID, (unsigned long long)s->peer, e->len);
if (e->dgram) u_free(e->dgram);
queue_entry_free(e);
} else {
int r = etcp_send(conn, e);
if (r != 0) { if (e->dgram) u_free(e->dgram); queue_entry_free(e); }
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: outq drain peer=%016llx len=%u rc=%d left=%d",
MS_ID, (unsigned long long)s->peer, e->len, r, queue_entry_count(s->out_q));
}
if (queue_entry_count(s->out_q) > 0 && conn && conn->send_input_q)
queue_waiter_wait(conn->send_input_q, &s->waiter, ms_outq_drain_cb, s);
}
/* Собрать [svc_id][payload] и положить в out_q сессии (с backpressure на send_input_q). */
static int ms_outq_enqueue(struct ms_session* s, const uint8_t* payload, size_t plen) {
struct merkle_sync* ms = s ? s->ms : NULL;
if (!ms || !s->out_q) return -1;
struct ETCP_CONN* conn = ms_find_conn_for_node(ms->inst, s->peer);
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: outq enqueue — no conn for peer=%016llx", MS_ID, (unsigned long long)s->peer); return -1; }
uint8_t* buf = u_malloc(1 + plen);
if (!buf) return -1;
buf[0] = ms->svc_id;
memcpy(buf + 1, payload, plen);
struct ll_entry* e = queue_entry_new(0);
if (!e) { u_free(buf); return -1; }
e->dgram = buf; e->len = (uint16_t)(1 + plen);
int was_empty = (queue_entry_count(s->out_q) == 0);
if (queue_data_put(s->out_q, e) != 0) { u_free(buf); queue_entry_free(e); return -1; }
if (was_empty && conn->send_input_q)
queue_waiter_wait(conn->send_input_q, &s->waiter, ms_outq_drain_cb, s);
return 0;
}
static int _send_msg(struct merkle_sync* ms, uint64_t peer,
const uint8_t* payload, size_t len) {
if (!ms->inst || len < 1) return -1;
uint8_t* buf = u_malloc(1 + len);
if (!buf) return -1;
buf[0] = ms->svc_id;
memcpy(buf + 1, payload, len);
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { u_free(buf); return -1; }
entry->dgram = buf; entry->len = 1 + len;
struct ETCP_CONN* conn = ms_find_conn_for_node(ms->inst, peer);
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: no conn for node %016llx", MS_ID, (unsigned long long)peer); u_free(buf); queue_entry_free(entry); return -1; }
int r = etcp_send(conn, entry);
if (r != 0) { u_free(buf); queue_entry_free(entry); }
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: send OK peer=%016llx len=%zu rc=%d", MS_ID, (unsigned long long)peer, len, r);
return r;
}
static int _broadcast_to_session(struct merkle_sync* ms, struct ms_session* s,
const char* ns, const uint8_t* item_data, size_t item_len) {
size_t sz = 1 + 8 + 1 + 1 + 1 + 1 + 2 + item_len;
uint8_t* buf = u_malloc(sz); if (!buf) return -1;
uint8_t* p = buf;
ms_write_group_id(p, ns); p += 8;
*p++ = 0x01; /* MSG_HASHES */
*p++ = 0; /* level=0 */
*p++ = 0; /* prefix_bytes=0 */
*p++ = 1; /* is_data=1 */
uint16_t dlen = (uint16_t)item_len; memcpy(p, &dlen, 2); p += 2;
memcpy(p, item_data, item_len); p += item_len;
int r = ms_outq_enqueue(s, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
static int _send_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
uint8_t level, uint64_t prefix, uint8_t prefix_bytes, int is_data) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: send_hashes peer=%016llx ns=%s L%d P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
size_t max_sz = 1 + 8 + 1 + 1 + prefix_bytes + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 2 + 65536;
uint8_t* buf = u_malloc(max_sz);
if (!buf) return -1;
uint8_t* p = buf;
ms_write_group_id(p, ns); p += 8;
*p++ = 0x01; /* MS_MSG_HASHES */
*p++ = level;
*p++ = prefix_bytes; _prefix_write(p, prefix, prefix_bytes); p += prefix_bytes;
*p++ = (uint8_t)(is_data ? 1 : 0);
if (is_data) {
size_t mlen = 65536; uint8_t* mbuf = u_malloc(mlen);
if (mbuf) {
if (ms->ops->get_items(ms->data_ctx, ns, level, prefix, prefix_bytes, mbuf, &mlen) == 0) {
uint16_t dlen = (uint16_t)mlen; memcpy(p, &dlen, 2); p += 2;
memcpy(p, mbuf, mlen); p += mlen;
} else { uint16_t zero = 0; memcpy(p, &zero, 2); p += 2; }
u_free(mbuf);
}
} else {
uint32_t bitmap; uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, level, prefix, prefix_bytes, &bitmap, hashes);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: send_hashes peer=%016llx ns=%s level=%d prefix=%016llx bitmap=%08x", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, bitmap);
memcpy(p, &bitmap, 4); p += 4;
for (int i = 0; i < MT_BUCKETS; i++)
if (bitmap & (1u << i)) { memcpy(p, hashes[i], MT_HASH_SIZE); p += MT_HASH_SIZE; }
}
int r = _send_msg(ms, peer, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
static int _send_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
struct bucket_entry* buckets, int count) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: send_batch peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
size_t max_sz = 1 + 8 + 1 + (size_t)count * (1 + 1 + 8 + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 65536);
uint8_t* buf = u_malloc(max_sz);
if (!buf) return -1;
uint8_t* p = buf;
ms_write_group_id(p, ns); p += 8;
*p++ = 0x03; /* MS_MSG_BATCH */
*p++ = (uint8_t)count;
for (int i = 0; i < count; i++) {
*p++ = buckets[i].level;
*p++ = buckets[i].prefix_bytes;
_prefix_write(p, buckets[i].prefix, buckets[i].prefix_bytes); p += buckets[i].prefix_bytes;
EVP_MD_CTX* tctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(tctx, EVP_sha256(), NULL);
int bcount = ms->ops->update_bucket_hash(ms->data_ctx, ns,
buckets[i].level, buckets[i].prefix, tctx);
if (bcount < 0) bcount = 0;
EVP_MD_CTX_free(tctx);
int is_terminal = (buckets[i].level >= MT_MAX_LEVEL || bcount < 8);
*p++ = (uint8_t)(is_terminal ? 1 : 0);
if (is_terminal) {
size_t mlen = 65536; uint8_t* mbuf = u_malloc(mlen);
if (mbuf) {
if (ms->ops->get_items(ms->data_ctx, ns, buckets[i].level,
buckets[i].prefix, buckets[i].prefix_bytes, mbuf, &mlen) == 0) {
uint16_t dlen = (uint16_t)mlen; memcpy(p, &dlen, 2); p += 2;
memcpy(p, mbuf, mlen); p += mlen;
} else { uint16_t z = 0; memcpy(p, &z, 2); p += 2; }
u_free(mbuf);
}
} else {
uint32_t bm; uint8_t hs[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, buckets[i].level,
buckets[i].prefix, buckets[i].prefix_bytes, &bm, hs);
memcpy(p, &bm, 4); p += 4;
for (int j = 0; j < MT_BUCKETS; j++)
if (bm & (1u << j)) { memcpy(p, hs[j], MT_HASH_SIZE); p += MT_HASH_SIZE; }
}
}
int r = _send_msg(ms, peer, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
/* ── Sessions ── */
static struct ms_session* _session_find(struct merkle_sync* ms, uint64_t peer, const char* ns) {
for (struct ms_session* s = ms->sessions; s; s = s->next)
if (s->peer == peer && strcmp(s->ns, ns) == 0) return s;
return NULL;
}
static int _find_pending(struct ms_session* s, uint8_t level, uint64_t prefix) {
for (int i = 0; i < s->pending_count; i++)
if (s->pending[i].level == level && s->pending[i].prefix == prefix)
return i;
return -1;
}
static void _session_done(struct ms_session* s, int result) {
if (s->sess_state == SESS_SYNCED) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: session ALREADY DONE peer=%016llx ns=%s", MS_ID, (unsigned long long)s->peer, s->ns); return; }
merkle_sync_done_cb cb = s->done_cb; void* arg = s->cb_arg;
s->done_cb = NULL; s->cb_arg = NULL;
if (result == MT_OK) { s->sess_state = SESS_SYNCED; DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: session SYNCED peer=%016llx ns=%s", MS_ID, (unsigned long long)s->peer, s->ns); }
else { s->sess_state = SESS_SYNCING; DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: session FAILED peer=%016llx ns=%s result=%d", MS_ID, (unsigned long long)s->peer, s->ns, result); }
if (cb) cb(s->peer, s->ns, result, arg);
}
/* ── Recv handlers ── */
static void _handle_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen, int parent_idx) {
if (plen < 3) return;
uint8_t level = pl[0]; uint8_t pb = pl[1];
if (level > MT_MAX_LEVEL || pb > 8 || (size_t)(3 + pb) > plen) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: handle_hashes bad level=%u prefix_bytes=%u plen=%zu", MS_ID, level, pb, plen); return; }
uint64_t prefix = _prefix_read(pl + 2, pb);
uint8_t is_data = pl[2 + pb];
const uint8_t* payload = pl + 3 + pb;
size_t paylen = plen - 3 - pb;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: handle_hashes peer=%016llx ns=%s L%d/P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
struct ms_session* s = _session_find(ms, peer, ns);
if (is_data) {
if (paylen < 2) return;
uint16_t dlen; memcpy(&dlen, payload, 2);
const uint8_t* pd = payload + 2;
if (paylen - 2 < dlen) return;
ms->ops->apply_items(ms->data_ctx, ns, peer, pd, dlen);
if (s && s->sess_state == SESS_SYNCING) _session_done(s, MT_OK);
return;
}
if (!s) {
s = u_calloc(1, sizeof(*s)); if (!s) return;
snprintf(s->ns, sizeof(s->ns), "%s", ns); s->peer = peer;
s->next = ms->sessions; ms->sessions = s;
ms_session_setup(ms, s);
} else {
s->pending_count = 0;
}
s->sess_state = SESS_SYNCING;
if (paylen < 4) return;
uint32_t remote_bm; memcpy(&remote_bm, payload, 4);
const uint8_t* rh = payload + 4;
uint32_t local_bm; uint8_t lh[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, level, prefix, pb, &local_bm, lh);
uint32_t differs = remote_bm ^ local_bm;
for (int i = 0; i < MT_BUCKETS; i++) {
if (!(local_bm & (1u << i)) && !(remote_bm & (1u << i))) continue;
if ((local_bm & (1u << i)) && (remote_bm & (1u << i))) {
int rh_idx = 0;
for (int j = 0; j < i; j++) if (remote_bm & (1u << j)) rh_idx++;
if (memcmp(rh + (size_t)rh_idx * MT_HASH_SIZE, lh[i], MT_HASH_SIZE) != 0)
differs |= (1u << i);
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: handle_hashes peer=%016llx ns=%s level=%d remote_bm=%08x local_bm=%08x differs=%08x is_data=%d",
MS_ID, (unsigned long long)peer, ns, level, remote_bm, local_bm, differs, is_data);
if (differs == 0 && level == 0) {
_send_hashes(ms, peer, ns, 0, 0, 0, 1);
_session_done(s, MT_OK);
return;
}
if (differs == 0) return;
int next_shift = 64 - ((int)level + 1) * 5;
struct bucket_entry requests[MT_BUCKETS]; int rcount = 0;
for (int i = 0; i < MT_BUCKETS && rcount < 32; i++) {
if (!(differs & (1u << i))) continue;
uint8_t nl = (uint8_t)(level + 1); if (nl > MT_MAX_LEVEL) nl = MT_MAX_LEVEL;
requests[rcount].level = nl;
requests[rcount].prefix_bytes = merkle_sync_prefix_bytes(nl);
requests[rcount].prefix = prefix | ((uint64_t)i << (next_shift > 0 ? next_shift : 0));
rcount++;
}
if (rcount > 0) {
if (s) {
for (int i = 0; i < rcount && s->pending_count < MS_PENDING_MAX; i++) {
int idx = s->pending_count++;
s->pending[idx].level = requests[i].level;
s->pending[idx].prefix = requests[i].prefix;
s->pending[idx].parent = (int8_t)parent_idx;
s->pending[idx].sub_count = 0;
s->pending[idx].state = MS_PEND_WAITING;
}
if (parent_idx >= 0 && parent_idx < MS_PENDING_MAX)
s->pending[parent_idx].sub_count++;
}
size_t rs = 1 + 8 + 1;
for (int i = 0; i < rcount; i++) rs += 1 + 1 + requests[i].prefix_bytes + 1;
uint8_t* rbuf = u_malloc(rs);
if (rbuf) {
uint8_t* wr = rbuf;
ms_write_group_id(wr, ns); wr += 8;
*wr++ = 0x02; /* MS_MSG_REQUEST */
*wr++ = (uint8_t)rcount;
for (int i = 0; i < rcount; i++) {
*wr++ = requests[i].level;
*wr++ = requests[i].prefix_bytes;
_prefix_write(wr, requests[i].prefix, requests[i].prefix_bytes);
wr += requests[i].prefix_bytes;
EVP_MD_CTX* tctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(tctx, EVP_sha256(), NULL);
int cnt = ms->ops->update_bucket_hash(ms->data_ctx, ns,
requests[i].level, requests[i].prefix, tctx);
if (cnt < 0) cnt = 0;
EVP_MD_CTX_free(tctx);
*wr++ = (uint8_t)(cnt < 8 || requests[i].level >= MT_MAX_LEVEL ? 1 : 0);
}
_send_msg(ms, peer, rbuf, (size_t)(wr - rbuf));
u_free(rbuf);
}
} else {
if (s) _session_done(s, MT_OK);
}
}
static void _handle_request(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t off = 0;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: handle_request peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
struct bucket_entry buckets[32]; int bc = 0;
for (uint8_t i = 0; i < count && bc < 32; i++) {
if (off + 2 > plen - 1) break;
uint8_t lvl = bp[off++]; uint8_t pb_i = bp[off++];
if (off + pb_i > plen - 1) break;
uint64_t pr = _prefix_read(bp + off, pb_i); off += pb_i;
buckets[bc].level = lvl; buckets[bc].prefix_bytes = pb_i; buckets[bc].prefix = pr; bc++;
off++; /* skip is_terminal byte */
}
if (bc > 0) {
_send_batch(ms, peer, ns, buckets, bc);
}
}
static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t rem = plen - 1;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
for (uint8_t i = 0; i < count && rem >= 3; i++) {
uint8_t lvl = bp[0]; uint8_t pb_i = bp[1]; rem -= 2; bp += 2;
if (rem < pb_i + 1) break;
uint64_t pr = _prefix_read(bp, pb_i); bp += pb_i; rem -= pb_i;
uint8_t is_data = *bp++; rem--;
if (is_data && rem >= 4) {
uint16_t dlen; memcpy(&dlen, bp, 2); bp += 2; rem -= 2;
if (rem < dlen) break;
ms->ops->apply_items(ms->data_ctx, ns, peer, bp, dlen);
bp += dlen; rem -= dlen;
struct ms_session* sb = _session_find(ms, peer, ns);
if (sb) { int pidx = _find_pending(sb, lvl, pr); if (pidx >= 0) sb->pending[pidx].state = MS_PEND_RESOLVED; }
} else if (!is_data && rem >= 4) {
struct ms_session* sc = _session_find(ms, peer, ns);
int pidx = sc ? _find_pending(sc, lvl, pr) : -1;
if (pidx >= 0) sc->pending[pidx].state = MS_PEND_RESOLVED;
uint8_t sub_pl[4096]; size_t sub_len = 0;
sub_pl[sub_len++] = lvl;
sub_pl[sub_len++] = merkle_sync_prefix_bytes(lvl);
_prefix_write(sub_pl + sub_len, pr, merkle_sync_prefix_bytes(lvl));
sub_len += merkle_sync_prefix_bytes(lvl);
sub_pl[sub_len++] = 0; /* is_data=0 */
uint32_t bm; memcpy(&bm, bp, 4); bp += 4; rem -= 4;
memcpy(sub_pl + sub_len, &bm, 4); sub_len += 4;
int nh = _popcount_u32(bm);
if (rem >= (size_t)nh * MT_HASH_SIZE) {
memcpy(sub_pl + sub_len, bp, (size_t)nh * MT_HASH_SIZE);
sub_len += (size_t)nh * MT_HASH_SIZE;
bp += (size_t)nh * MT_HASH_SIZE; rem -= (size_t)nh * MT_HASH_SIZE;
}
_handle_hashes(ms, peer, ns, sub_pl, sub_len, pidx);
}
}
{
struct ms_session* s = _session_find(ms, peer, ns);
if (!s) return;
int all_done = (s->pending_count > 0);
for (int i = 0; i < s->pending_count; i++) {
if (s->pending[i].state != MS_PEND_RESOLVED) { all_done = 0; break; }
}
if (all_done && s->sess_state == SESS_SYNCING) {
_send_hashes(ms, peer, ns, 0, 0, 0, 1);
_session_done(s, MT_OK);
}
}
}
static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || entry->len < 10) {
if (entry) { if (entry->dgram) u_free(entry->dgram); queue_entry_free(entry); }
return;
}
struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized) { u_free(entry->dgram); queue_entry_free(entry); return; }
uint64_t peer = conn ? conn->peer_node_id : 0;
const uint8_t* d = entry->dgram;
size_t dlen = entry->len;
uint64_t group_id = be64toh(*(const uint64_t*)(d + 1));
char ns[64];
if (ms_group_id_to_ns(inst, group_id, ns, sizeof(ns)) != 0) { u_free(entry->dgram); queue_entry_free(entry); return; }
uint8_t type = d[9];
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv type=%02x from=%016llx group=%016llx ns=%s len=%zu", MS_ID, type, (unsigned long long)peer, (unsigned long long)group_id, ns, dlen);
const uint8_t* pl = d + 10;
size_t plen = dlen - 10;
switch (type) {
case 0x01: _handle_hashes(ms, peer, ns, pl, plen, -1); break;
case 0x02: _handle_request(ms, peer, ns, pl, plen); break;
case 0x03: _handle_batch(ms, peer, ns, pl, plen); break;
case 0x04:
if (plen >= 9 && ms->ops->apply_update) {
uint64_t key; memcpy(&key, pl, 8);
uint8_t utype = pl[8];
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv ITEM_UPDATE key=%016llx type=%02x from=%016llx ns=%s",
MS_ID, (unsigned long long)key, utype, (unsigned long long)peer, ns);
ms->ops->apply_update(ms->data_ctx, ns, key, utype, pl + 9, plen - 9);
}
break;
}
u_free(entry->dgram); queue_entry_free(entry);
}
/* ── Push lightweight update to synced peers ── */
void merkle_sync_push_update(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t key, uint8_t type, const uint8_t* data, size_t len) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized || !ns || !data) return;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: push_update ns=%s key=%016llx type=%02x len=%zu",
MS_ID, ns, (unsigned long long)key, type, len);
size_t pkt = 1 + 8 + 1 + 8 + 1 + len;
uint8_t* buf = u_malloc(pkt);
if (!buf) return;
uint8_t* p = buf;
ms_write_group_id(p, ns); p += 8;
*p++ = 0x04; /* MSG_ITEM_UPDATE */
memcpy(p, &key, 8); p += 8;
*p++ = type;
memcpy(p, data, len);
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->sess_state != SESS_SYNCED || strcmp(s->ns, ns) != 0) continue;
ms_outq_enqueue(s, buf, pkt);
}
u_free(buf);
}
/* ── Broadcast to synced peers ── */
void merkle_sync_broadcast(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t from_peer, const uint8_t* data, size_t len) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized || !ns || !data || len < 2) return;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: broadcast ns=%s from=%016llx len=%zu",
MS_ID, ns, (unsigned long long)from_peer, len);
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->peer == from_peer || strcmp(s->ns, ns) != 0) continue;
_broadcast_to_session(ms, s, ns, data, len);
}
}
int merkle_sync_send_to(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t peer, const uint8_t* data, size_t len) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized || !ns || !data || len < 2) return -1;
struct ms_session* s = _session_find(ms, peer, ns);
if (!s) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: send_to — no session peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns); return -1; }
return _broadcast_to_session(ms, s, ns, data, len);
}
/* ── Background consistency check ── */
int merkle_sync_bg_check(struct UTUN_INSTANCE* inst, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized) return -1;
sqlite3* db = _db(inst); if (!db || !ns) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bg_check — db or ns is NULL db=%p ns=%s", MS_ID, (void*)db, ns ? ns : "(null)"); return -1; }
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db,
"SELECT prefix64 FROM merkle_tree_hash WHERE namespace=? AND level=? LIMIT 1",
-1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, MT_MAX_LEVEL);
if (sqlite3_step(stmt) != SQLITE_ROW) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bg_check — no hash row for ns=%s", MS_ID, ns); sqlite3_finalize(stmt); return -1; }
uint64_t pref = (uint64_t)sqlite3_column_int64(stmt, 0);
sqlite3_finalize(stmt);
uint8_t old_hash[MT_HASH_SIZE];
memcpy(old_hash, merkle_sync_get_hash(inst, ns, MT_MAX_LEVEL, pref), MT_HASH_SIZE);
_recompute_bucket(ms, ns, MT_MAX_LEVEL, pref);
const uint8_t* new_hash = merkle_sync_get_hash(inst, ns, MT_MAX_LEVEL, pref);
if (memcmp(old_hash, new_hash, MT_HASH_SIZE) != 0) {
for (uint8_t lv = (uint8_t)(MT_MAX_LEVEL - 1); lv >= 1; lv--)
_recompute_bucket(ms, ns, lv, merkle_sync_level_prefix(pref, lv));
return 1;
}
return 0;
}
static void _wal_timer_cb(void* arg) {
struct merkle_sync* ms = (struct merkle_sync*)arg;
if (!ms || !ms->initialized || !ms->inst) return;
#ifdef UTUN_HAVE_STANDBY
if (standby_get_sleep_tb() > 0) {
ms->wal_wait = standby_wait(ms, _wal_timer_cb);
return;
}
ms->wal_wait = NULL;
#endif
sqlite3* db = _db(ms->inst); if (!db) { ms->wal_timer = NULL; return; }
int pn = 0, ckpt = 0;
sqlite3_wal_checkpoint_v2(db, NULL, SQLITE_CHECKPOINT_PASSIVE, &pn, &ckpt);
if (pn > 0 || ckpt > 0)
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WAL checkpoint done: %d frames in log, %d checkpointed", MS_ID, pn, ckpt);
ms->wal_timer = uasync_set_timeout(ms->inst->ua, (uint32_t)(MS_WAL_INTERVAL_MS * 10),
ms, _wal_timer_cb, "ms_wal");
}
/* ── Public lifecycle ── */
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
const struct merkle_sync_data_ops* ops, void* data_ctx) {
if (!inst || !ops) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init svc=%02x", MS_ID, svc_id);
struct merkle_sync* ms = u_calloc(1, sizeof(*ms));
if (!ms) return -1;
ms->inst = inst; ms->svc_id = svc_id; ms->ops = ops; ms->data_ctx = data_ctx;
ms->initialized = 1;
inst->msync = ms;
_ensure_table(ms);
etcp_bind(inst, svc_id, _recv_cb);
ms->wal_timer = uasync_set_timeout(inst->ua, (uint32_t)(MS_WAL_INTERVAL_MS * 10),
ms, _wal_timer_cb, "ms_wal");
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialized svc=%02x", MS_ID, svc_id);
return 0;
}
void merkle_sync_destroy(struct UTUN_INSTANCE* inst) {
if (!inst) return;
struct merkle_sync* ms = inst->msync;
if (!ms) return;
ms->initialized = 0; inst->msync = NULL;
etcp_unbind(inst, ms->svc_id);
if (ms->wal_timer) { uasync_cancel_timeout(inst->ua, ms->wal_timer); ms->wal_timer = NULL; }
#ifdef UTUN_HAVE_STANDBY
if (ms->wal_wait) { standby_wait_cancel(ms->wal_wait); ms->wal_wait = NULL; }
#endif
struct ms_session* s = ms->sessions;
while (s) { struct ms_session* next = s->next;
ms_session_outq_free(s);
u_free(s); s = next; }
u_free(ms);
}
/* ── Public async sync ── */
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ns, merkle_sync_done_cb done_cb, void* arg) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!inst || !ns || !ms || !ms->initialized) return -1;
_ensure_table(ms);
struct ms_session* s = _session_find(ms, peer, ns);
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: start peer=%016llx ns=%s new=%d", MS_ID, (unsigned long long)peer, ns, s ? 0 : 1);
if (!s) {
s = u_calloc(1, sizeof(*s));
if (!s) return -1;
snprintf(s->ns, sizeof(s->ns), "%s", ns);
s->peer = peer;
s->next = ms->sessions;
ms->sessions = s;
ms_session_setup(ms, s);
} else {
s->pending_count = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: start OVERWRITE peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns);
}
s->sess_state = SESS_SYNCING;
s->done_cb = done_cb; s->cb_arg = arg;
_send_hashes(ms, peer, ns, 0, 0, 0, 0);
return 0;
}
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ns) return;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cancel peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns);
struct ms_session** p = &ms->sessions;
while (*p) {
struct ms_session* s = *p;
if (s->peer == peer && strcmp(s->ns, ns) == 0) {
*p = s->next;
ms_session_outq_free(s);
u_free(s);
return;
}
p = &(*p)->next;
}
}
Reference in new issue View Git Blame Copy Permalink