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

build: extract libutun.a from src/ + add db_sync module + etcp API, BBR improvements

Browse Source 
- Add libutun.a (noinst) containing all src/*.c except utun.c
- utun binary now links libutun.a + libuasync.a
- tests/Makefile.am simplified: all tests link libutun.a instead of
  manual .o file lists (removed ~70 lines of boilerplate)
- Add db_sync module: distributed content-addressed DB with LMDB
  backend, peer-to-peer sync via etcp_router (svc 0x20)
- etcp API: etcp_send_in_order(), etcp_clear_inflight_queues(),
  verify_packet_signature(), BBR getters
- BBR: adaptive threshold (1/4 of pipe-min instead of hardcoded)
- route_bgp/etcp_connect: STCP transport support in NODEINFO
- conn_mgr: NODEINFO_ADDRS doc for nat_type/transport/proto fields
- Cleanup: removed AUDIT_FINDINGS.txt, filelist.txt, nat.txt
This commit is contained in:
Evgeny committed 2026-07-06 11:42:13 +03:00
1 parent 7139969acd
commit 8756f561be
23 files changed
+1508 -786
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Makefile.am
+58 -1
View File
@@ -1,4 +1,5 @@
bin_PROGRAMS = utun
noinst_LIBRARIES = libutun.a
# Sources that are always compiled
utun_CORE_SOURCES = \
@@ -14,6 +15,7 @@ utun_CORE_SOURCES = \
route_node_lmdb.c \
route_connectivity.c \
conn_mgr.c \
db_sync.c \
routing.c \
tun_if.c \
tun_route.c \
@@ -53,10 +55,64 @@ utun_CORE_SOURCES = \
lwip_tcp/lwip_tcp_in.c \
lwip_tcp/lwip_tcp_out.c
# libutun: all core sources except main()
libutun_a_SOURCES = \
utun_instance.c \
config_parser.c \
config_updater.c \
route_lib.c \
route6_lib.c \
route_bgp.c \
route_ping.c \
route_node.c \
route_node_lmdb.c \
route_connectivity.c \
conn_mgr.c \
db_sync.c \
routing.c \
tun_if.c \
tun_route.c \
tun_linux.c \
tun_freebsd.c \
tun_windows.c \
etcp.c \
etcp_connections.c \
etcp_bbr.c \
etcp_loadbalancer.c \
etcp_debug.c \
etcp_dump.c \
secure_channel.c \
crc32.c \
stcp_link.c \
stcp.c \
stcp_server.c \
stcp_client.c \
pkt_normalizer.c \
packet_dump.c \
etcp_api.c \
etcp_connect.c \
control_server.c \
msg_transport.c \
firewall.c \
eim_nat.c \
nat_transport.c \
dummynet.c \
proxy/tcp_proxy_client.c \
etcp_router.c \
proxy/tcp_proxy_server.c \
proxy/udp_proxy.c \
proxy/socks_proxy.c \
proxy/icmp_proxy.c \
lwip_tcp/lwip_pbuf.c \
lwip_tcp/lwip_tcp.c \
lwip_tcp/lwip_tcp_in.c \
lwip_tcp/lwip_tcp_out.c
libutun_a_CFLAGS = $(utun_CFLAGS)
# Platform-specific TUN libs (Windows only)
utun_TUN_LIBS = @TUN_LIBS@
utun_SOURCES = $(utun_CORE_SOURCES) $(utun_TUN_SOURCES)
utun_SOURCES = utun.c
# Include paths
utun_CFLAGS = \
@@ -67,6 +123,7 @@ utun_CFLAGS = \
# Libraries
utun_LDADD = \
libutun.a \
$(top_builddir)/lib/libuasync.a \
-lpthread \
-lm \
src/_db_arch.txt
+22
View File
@@ -0,0 +1,22 @@
Архитектура таблицы c быстрой репликацией между несколькими узлами:
1. В конец таблицы каждый узел может самостоятельно добавлять данные (со своим timestamp, желательно время правильное)
2. узлы между собой синхронизируются, распространяя обновления по соседям
3. TTL: изменение имеет время жизни. узел удаляет собственные записи которые не были отправлены никому в течении суток
4. Формат записи:
ID (64bit monotonic autoincrement)
chain_hash (256bit) - цепочка хешей. считается так: chain_hash[index+1]=hash(chain_hash[index],ID[index],timestamp[index],datahash[index]). если хеш совпадает - это признак того что все строки выше синхронизированы.
timestamp (64bit) - время создания записи. строки должны сортироваться по возрастанию concat(datahash(младшая часть числа,64 bit)+timestamp(старшая часть))
datahash (64bit) - хеш данных этой строки
data (varchar) - данные в формате json
алгоритм репликации (2 узла):
узел A запрашивает синхронизацию и передает свой last id
test_id= min(mast id, peer last id)
узел B передает: test_id, chain_hash(test id), hash(test id-1), hash(test id-2), hash(test id-4), hash(test id-8) итд 16,32,..., до первого элемента включитально (лимитируем вылетевший индекс первым элементом)
узел A - сравнивает, находит проверяемый диапазон id, отправляет другому узлу 16 (можно больше) хешей (линейно разбив проверяемый диапазон на более короткие поддиапазоны). таким образом узлы уточняют первый ид который не совпал.
Когда первая различающияся запись найдена, узел отправляет хеш этой и n (например 32) последующих записей (если записей много). если записей мало (<4) то узел отправляет сразу содержимое записей.
(надо додумать алгоритм, чтобы оптимизировать количество итераций - лучше передать больше данных за раз чем много итераций с ожиданием ответной стороны)
src/config_parser.c
+10
View File
@@ -399,6 +399,14 @@ static int parse_global(const char *key, const char *value, struct global_config
if (strcmp(key, "db_path") == 0) {
return assign_string(global->db_path, sizeof(global->db_path), value);
}
if (strcmp(key, "db_sync_enabled") == 0) {
global->db_sync_enabled = atoi(value);
return 0;
}
if (strcmp(key, "db_sync_ttl") == 0) {
global->db_sync_ttl = (uint32_t)atoi(value);
return 0;
}
/* debug_categories key is deprecated - use [debug] section instead */
if (strcmp(key, "enable_timestamp") == 0) {
global->enable_timestamp = atoi(value);
@@ -792,6 +800,8 @@ static struct utun_config* parse_config_internal(FILE *fp, const char *filename)
cfg->global.firewall_bypass_all = 0;
cfg->global.control_allows = NULL;
cfg->global.control_allow_count = 0;
cfg->global.db_sync_enabled = 0;
cfg->global.db_sync_ttl = 86400;
section_type_t cur_section = SECTION_UNKNOWN;
struct CFG_SERVER *cur_server = NULL;
src/config_parser.h
+2
View File
@@ -115,6 +115,8 @@ struct global_config {
// Debug and logging configuration
char log_file[256]; // Path to log file (empty = stdout)
char db_path[256]; // Path to LMDB nodeinfo database (empty = disabled)
int db_sync_enabled; // 1 = enable distributed DB sync (default: 0)
uint32_t db_sync_ttl; // TTL for unsent records in seconds (default: 86400)
char debug_level[16]; // debug level: error, warn, info, debug, trace
int enable_timestamp; // enable timestamps in logs
int enable_function_names; // enable function names in logs
src/db_sync.c
+940
View File
@@ -0,0 +1,940 @@
// db_sync.c — Distributed content-addressed table with LMDB + peer sync via etcp_router
#include "db_sync.h"
#include "etcp_api.h"
#include "etcp.h"
#include "etcp_router.h"
#include "utun_instance.h"
#include "route_bgp.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/u_async.h"
#include "../lib/sha256.h"
#include "../lib/liblmdb/lmdb.h"
#include "../lib/platform_compat.h"
#include <string.h>
#define DEBUG_CATEGORY_DB_SYNC DEBUG_CATEGORY_DEBUG
// ---- LMDB key/value layout ----
// Key: 16 bytes = timestamp:8 BE || datahash:8 BE (sorted by timestamp first)
// Value: id:8 || chain_hash:32 || creator:8 || flags:1 || data_len:4 || data:variable
#define DB_VAL_OFF_ID 0
#define DB_VAL_OFF_CHAIN_HASH 8
#define DB_VAL_OFF_CREATOR 40
#define DB_VAL_OFF_FLAGS 48
#define DB_VAL_OFF_DATA_LEN 49
#define DB_VAL_OFF_DATA 53
#define DB_VAL_HDR_SIZE 53
// ---- Module state ----
struct DB_SYNC {
struct UTUN_INSTANCE* inst;
MDB_env* env;
MDB_dbi dbi_records;
MDB_dbi dbi_meta;
uint64_t next_id;
uint64_t last_connected_tb; // timebase when any peer was last connected (for TTL)
uint64_t last_timestamp_us; // гарантия монотонности timestamp
struct DB_SYNC_PEER {
uint64_t node_id;
uint8_t synced; // 0=not, 1=in_progress, 2=synced
}* peers;
int peer_count;
int peer_capacity;
void* ttl_timer;
void* peer_check_timer;
uint8_t enabled;
};
// ---- Forward declarations ----
static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);
static void db_sync_on_new_conn(struct ETCP_CONN* conn, void* arg);
static void db_sync_on_conn_up(struct ETCP_CONN* conn, void* arg);
static void db_sync_on_conn_down(struct ETCP_CONN* conn, void* arg);
static void db_sync_peer_check_cb(void* arg);
static void db_sync_ttl_cleanup_cb(void* arg);
static int db_sync_send(struct DB_SYNC* db, uint64_t dst_node_id, const uint8_t* payload, size_t len);
static void db_sync_initiate_sync(struct DB_SYNC* db, uint64_t peer_node_id);
// ---- LMDB helpers ----
static int db_lmdb_open(struct DB_SYNC* db, const char* path) {
int rc = mdb_env_create(&db->env);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: mdb_env_create failed: %s", mdb_strerror(rc)); return -1; }
rc = mdb_env_set_mapsize(db->env, DB_SYNC_DEFAULT_MAPSIZE);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: mdb_env_set_mapsize failed: %s", mdb_strerror(rc)); mdb_env_close(db->env); db->env=NULL; return -1; }
rc = mdb_env_set_maxdbs(db->env, 3);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: mdb_env_set_maxdbs failed: %s", mdb_strerror(rc)); mdb_env_close(db->env); db->env=NULL; return -1; }
rc = mdb_env_open(db->env, path, 0, 0644);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: mdb_env_open(%s) failed: %s", path, mdb_strerror(rc)); mdb_env_close(db->env); db->env=NULL; return -1; }
MDB_txn* txn;
rc = mdb_txn_begin(db->env, NULL, 0, &txn);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: txn_begin failed: %s", mdb_strerror(rc)); mdb_env_close(db->env); db->env=NULL; return -1; }
rc = mdb_dbi_open(txn, "records", MDB_CREATE, &db->dbi_records);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: dbi_open(records) failed: %s", mdb_strerror(rc)); mdb_txn_abort(txn); mdb_env_close(db->env); db->env=NULL; return -1; }
rc = mdb_dbi_open(txn, "meta", MDB_CREATE, &db->dbi_meta);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: dbi_open(meta) failed: %s", mdb_strerror(rc)); mdb_txn_abort(txn); mdb_env_close(db->env); db->env=NULL; return -1; }
rc = mdb_txn_commit(txn);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: txn_commit failed: %s", mdb_strerror(rc)); mdb_env_close(db->env); db->env=NULL; return -1; }
// Read next_id from meta
MDB_txn* rt;
rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &rt);
if (rc == MDB_SUCCESS) {
MDB_val key, data; key.mv_size = 7; key.mv_data = (char*)"next_id";
rc = mdb_get(rt, db->dbi_meta, &key, &data);
if (rc == MDB_SUCCESS && data.mv_size >= 8) db->next_id = *(uint64_t*)data.mv_data;
else db->next_id = 1;
mdb_txn_abort(rt);
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: LMDB opened at %s, next_id=%llu", path, (unsigned long long)db->next_id);
return 0;
}
static void db_lmdb_close(struct DB_SYNC* db) {
if (!db->env) return;
mdb_env_close(db->env); db->env = NULL;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: LMDB closed");
}
// Build 16-byte key: timestamp:8 BE || datahash:8 BE
static void db_key_build(uint8_t key[16], uint64_t timestamp, uint64_t datahash) {
uint64_t ts_be = htobe64(timestamp), dh_be = htobe64(datahash);
memcpy(key, &ts_be, 8); memcpy(key + 8, &dh_be, 8);
}
// Get chain_hash at a given sorted position (0-based) via cursor. Returns 0 on success.
static int db_chain_hash_at(struct DB_SYNC* db, uint32_t pos, uint8_t chain_hash_out[32]) {
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: chain_hash_at txn_begin failed: %s", mdb_strerror(rc)); return -1; }
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc) { mdb_txn_abort(txn); return -1; }
MDB_val key, val;
rc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
for (uint32_t i = 0; rc == 0 && i < pos; i++) rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
if (rc == 0 && val.mv_size >= DB_VAL_HDR_SIZE) memcpy(chain_hash_out, (uint8_t*)val.mv_data + DB_VAL_OFF_CHAIN_HASH, 32);
else { mdb_cursor_close(cursor); mdb_txn_abort(txn); return -1; }
mdb_cursor_close(cursor); mdb_txn_abort(txn);
return 0;
}
// Get count of records
static uint32_t db_count(struct DB_SYNC* db) {
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rc) return 0;
MDB_stat stat; rc = mdb_stat(txn, db->dbi_records, &stat);
mdb_txn_abort(txn);
return rc == 0 ? (uint32_t)stat.ms_entries : 0;
}
// Get datahash at given sorted position
static int db_datahash_at(struct DB_SYNC* db, uint32_t pos, uint64_t* datahash) {
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rc) return -1;
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc) { mdb_txn_abort(txn); return -1; }
MDB_val key, val; rc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
for (uint32_t i=0; rc==0 && i<pos; i++) rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
if (rc==0) {
*datahash = be64toh(*(uint64_t*)((uint8_t*)key.mv_data+8));
mdb_cursor_close(cursor); mdb_txn_abort(txn); return 0;
}
mdb_cursor_close(cursor); mdb_txn_abort(txn); return -1;
}
// ---- SHA256 helpers ----
static void db_sha256(const uint8_t* data, size_t len, uint8_t hash[32]) {
SC_SHA256_CTX ctx; sc_sha256_init(&ctx); sc_sha256_update(&ctx, data, len); sc_sha256_final(&ctx, hash);
}
// Compute datahash: first 64 bits of SHA256(json_data)
static uint64_t db_datahash(const uint8_t* data, size_t len) {
uint8_t hash[32]; db_sha256(data, len, hash);
uint64_t dh; memcpy(&dh, hash, 8); return dh;
}
// Compute chain_hash for a record
static void db_chain_hash_compute(const uint8_t prev_chain_hash[32], uint64_t id, uint64_t timestamp, uint64_t datahash, uint8_t out[32]) {
uint8_t buf[32+8+8+8]; // prev_chain_hash:32 + id:8 + timestamp:8 + datahash:8
memcpy(buf, prev_chain_hash, 32);
memcpy(buf+32, &id, 8); memcpy(buf+40, &timestamp, 8); memcpy(buf+48, &datahash, 8);
db_sha256(buf, 56, out);
}
// Get previous record's chain_hash (record before the given key in sorted order)
static int db_prev_chain_hash(struct DB_SYNC* db, const uint8_t key[16], uint8_t chain_hash[32]) {
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rc) return -1;
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc) { mdb_txn_abort(txn); return -1; }
MDB_val search_key, val; search_key.mv_size = 16; search_key.mv_data = (void*)key;
rc = mdb_cursor_get(cursor, &search_key, &val, MDB_SET_RANGE);
if (rc == 0) {
rc = mdb_cursor_get(cursor, &search_key, &val, MDB_PREV);
if (rc == 0 && val.mv_size >= DB_VAL_HDR_SIZE) {
memcpy(chain_hash, (uint8_t*)val.mv_data + DB_VAL_OFF_CHAIN_HASH, 32);
mdb_cursor_close(cursor); mdb_txn_abort(txn); return 0;
}
}
// No previous record → use zero hash
memset(chain_hash, 0, 32);
mdb_cursor_close(cursor); mdb_txn_abort(txn);
return 0;
}
// ---- Insert record ----
static int db_record_insert(struct DB_SYNC* db, uint64_t id, uint64_t timestamp, uint64_t datahash, const char* json, size_t json_len) {
uint8_t key[16]; db_key_build(key, timestamp, datahash);
// Check duplicate
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, 0, &txn);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: insert txn_begin failed: %s", mdb_strerror(rc)); return -1; }
MDB_val dup_key, dup_val; dup_key.mv_size = 16; dup_key.mv_data = key;
rc = mdb_get(txn, db->dbi_records, &dup_key, &dup_val);
if (rc == MDB_SUCCESS) { mdb_txn_abort(txn); return 1; } // already exists
// Compute chain_hash
uint8_t prev_ch[32]; db_prev_chain_hash(db, key, prev_ch);
uint8_t chain_hash[32]; db_chain_hash_compute(prev_ch, id, timestamp, datahash, chain_hash);
// Build value
size_t val_size = DB_VAL_HDR_SIZE + json_len;
uint8_t* val_buf = u_malloc(val_size);
if (!val_buf) { mdb_txn_abort(txn); return -1; }
uint64_t creator = db->inst->node_id;
memcpy(val_buf + DB_VAL_OFF_ID, &id, 8);
memcpy(val_buf + DB_VAL_OFF_CHAIN_HASH, chain_hash, 32);
memcpy(val_buf + DB_VAL_OFF_CREATOR, &creator, 8);
val_buf[DB_VAL_OFF_FLAGS] = 0;
*(uint32_t*)(val_buf + DB_VAL_OFF_DATA_LEN) = (uint32_t)json_len;
if (json_len > 0) memcpy(val_buf + DB_VAL_OFF_DATA, json, json_len);
MDB_val val; val.mv_size = val_size; val.mv_data = val_buf;
MDB_val k; k.mv_size = 16; k.mv_data = key;
rc = mdb_put(txn, db->dbi_records, &k, &val, 0);
u_free(val_buf);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: mdb_put failed: %s", mdb_strerror(rc)); mdb_txn_abort(txn); return -1; }
// Update chain_hash for subsequent records (out-of-order insert case)
// Check if there are records AFTER this key
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc == 0) {
MDB_val ckey, cval; ckey.mv_size = 16; ckey.mv_data = key;
rc = mdb_cursor_get(cursor, &ckey, &cval, MDB_SET);
if (rc == 0) {
uint8_t running_ch[32]; memcpy(running_ch, chain_hash, 32);
uint64_t running_id = id;
uint64_t running_ts = timestamp;
uint64_t running_dh = datahash;
// Move to next
while (mdb_cursor_get(cursor, &ckey, &cval, MDB_NEXT) == 0) {
if (cval.mv_size < DB_VAL_HDR_SIZE) continue;
uint64_t n_id = *(uint64_t*)((uint8_t*)cval.mv_data + DB_VAL_OFF_ID);
uint64_t n_ts = be64toh(*(uint64_t*)ckey.mv_data);
uint64_t n_dh = be64toh(*(uint64_t*)((uint8_t*)ckey.mv_data + 8));
uint8_t new_ch[32]; db_chain_hash_compute(running_ch, n_id, n_ts, n_dh, new_ch);
memcpy((uint8_t*)cval.mv_data + DB_VAL_OFF_CHAIN_HASH, new_ch, 32);
mdb_cursor_put(cursor, &ckey, &cval, MDB_CURRENT);
memcpy(running_ch, new_ch, 32);
}
}
mdb_cursor_close(cursor);
}
// Update next_id in meta
MDB_val mk, md; mk.mv_size = 7; mk.mv_data = (char*)"next_id"; md.mv_size = 8; md.mv_data = &id; id++;
mdb_put(txn, db->dbi_meta, &mk, &md, 0);
db->next_id = id;
rc = mdb_txn_commit(txn);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: insert commit failed: %s", mdb_strerror(rc)); return -1; }
return 0;
}
// ---- Send ----
static int db_sync_send(struct DB_SYNC* db, uint64_t dst_node_id, const uint8_t* payload, size_t len) {
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync_send: queue_entry_new failed"); return -1; }
uint8_t* buf = u_malloc(len + 1);
if (!buf) { queue_entry_free(entry); return -1; }
buf[0] = ETCP_RT_ID_DB_SYNC; memcpy(buf + 1, payload, len);
entry->dgram = buf; entry->len = len + 1;
int ret = etcp_route_send(db->inst, dst_node_id, entry, 0);
// etcp_route_send always takes ownership and frees entry in all code paths
return ret;
}
// ---- Peer management ----
static struct DB_SYNC_PEER* db_peer_find(struct DB_SYNC* db, uint64_t node_id) {
for (int i=0; i<db->peer_count; i++) if (db->peers[i].node_id == node_id) return &db->peers[i];
return NULL;
}
static struct DB_SYNC_PEER* db_peer_add(struct DB_SYNC* db, uint64_t node_id) {
struct DB_SYNC_PEER* p = db_peer_find(db, node_id);
if (p) return p;
if (db->peer_count >= db->peer_capacity) {
int nc = db->peer_capacity ? db->peer_capacity * 2 : 8;
struct DB_SYNC_PEER* np = u_realloc(db->peers, nc * sizeof(struct DB_SYNC_PEER));
if (!np) return NULL;
db->peers = np; db->peer_capacity = nc;
}
p = &db->peers[db->peer_count++];
p->node_id = node_id; p->synced = 0;
return p;
}
// ---- Sync protocol: INIT_SYNC (A→B) ----
static void db_handle_init_sync(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 36) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: INIT_SYNC too short %zu from %016llx", len, (unsigned long long)src_node_id); return; }
uint32_t peer_count = *(uint32_t*)payload;
const uint8_t* peer_last_ch = payload + 4;
uint32_t my_count = db_count(db);
uint32_t test_pos = (peer_count < my_count ? peer_count : my_count);
if (test_pos > 0) test_pos--;
// Send INIT_RESP
uint8_t resp[4096];
uint32_t resp_off = 0;
resp[resp_off++] = DB_MSG_INIT_RESP;
memcpy(resp+resp_off, &test_pos, 4); resp_off += 4;
uint8_t my_ch[32];
if (my_count == 0) {
memset(my_ch, 0, 32);
} else if (db_chain_hash_at(db, test_pos, my_ch) != 0) {
memset(my_ch, 0, 32);
}
memcpy(resp+resp_off, my_ch, 32); resp_off += 32;
// Always include sparse_count at position 37 (after test_pos + chain_hash)
// For complete case: sparse_count=0, no sparse entries follow
// For sparse case: sparse entries start at position 38
// Check if chain_hash matches
if (memcmp(my_ch, peer_last_ch, 32) == 0) {
resp[resp_off++] = 0; // sparse_count = 0
db_sync_send(db, src_node_id, resp, resp_off);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: INIT_RESP complete to %016llx test_pos=%u", (unsigned long long)src_node_id, test_pos);
return;
}
// Skip sparse_count slot for now, write it at the end
uint32_t sparse_count_pos = resp_off;
resp_off++; // placeholder for sparse_count
// Build sparse hashes: chain_hash at positions test_pos-1, -2, -4, -8, -16, ...
int sparse_count = 0;
for (uint32_t k = 0; k < 16; k++) {
uint32_t step = (uint32_t)(1u << k); // 1, 2, 4, 8, ...
if (test_pos < step) break;
uint32_t pos = test_pos - step;
if (db_chain_hash_at(db, pos, my_ch) != 0) break;
if (resp_off + 36 > sizeof(resp)) break;
memcpy(resp+resp_off, &pos, 4); resp_off += 4;
memcpy(resp+resp_off, my_ch, 32); resp_off += 32;
sparse_count++;
}
resp[sparse_count_pos] = (uint8_t)sparse_count;
db_sync_send(db, src_node_id, resp, resp_off);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: INIT_RESP to %016llx test_pos=%u sparse=%d", (unsigned long long)src_node_id, test_pos, sparse_count);
}
// ---- Sync protocol: INIT_RESP (B→A) ----
static void db_handle_init_resp(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 37) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: INIT_RESP too short %zu from %016llx", len, (unsigned long long)src_node_id); return; }
uint32_t test_pos = *(uint32_t*)payload;
const uint8_t* peer_ch = payload + 4;
uint8_t sparse_count = payload[36];
uint8_t my_ch[32];
if (db_chain_hash_at(db, test_pos, my_ch) != 0) { memset(my_ch, 0, 32); }
// Check if complete
if (sparse_count == 0 && memcmp(my_ch, peer_ch, 32) == 0) {
struct DB_SYNC_PEER* p = db_peer_find(db, src_node_id);
if (p) p->synced = 2;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: sync complete with %016llx", (unsigned long long)src_node_id);
return;
}
// If peer has empty table (chain_hash is all zeros), send everything
{
uint8_t zero[32]; memset(zero, 0, 32);
if (memcmp(peer_ch, zero, 32) == 0 && sparse_count == 0) {
uint32_t my_count = db_count(db);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: peer %016llx has empty table, sending all %u records", (unsigned long long)src_node_id, my_count);
// Send all records in batches
uint32_t sent = 0;
while (sent < my_count) {
uint32_t batch = my_count - sent;
if (batch > DB_SEND_DATA_MAX) batch = DB_SEND_DATA_MAX;
uint8_t sdbuf[8192]; uint32_t off=0;
sdbuf[off++] = DB_MSG_SEND_DATA; memcpy(sdbuf+off, &sent, 4); off+=4;
uint16_t* rcp = (uint16_t*)(sdbuf+off); off+=2; uint16_t rc=0;
MDB_txn* txn; int rrc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rrc==0) { MDB_cursor* cursor; rrc=mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rrc==0) { MDB_val key,val; rrc=mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
uint32_t cur=0; while (rrc==0 && cur < sent) { rrc=mdb_cursor_get(cursor,&key,&val,MDB_NEXT); cur++; }
while (rrc==0 && rc < batch && off+val.mv_size+16 < 8000) {
uint64_t ts = be64toh(*(uint64_t*)key.mv_data); uint64_t dh = be64toh(*(uint64_t*)((uint8_t*)key.mv_data+8));
uint64_t rid = *(uint64_t*)((uint8_t*)val.mv_data+DB_VAL_OFF_ID); uint32_t dlen = *(uint32_t*)((uint8_t*)val.mv_data+DB_VAL_OFF_DATA_LEN);
memcpy(sdbuf+off, &rid,8); off+=8; memcpy(sdbuf+off, &ts,8); off+=8; memcpy(sdbuf+off, &dh,8); off+=8;
memcpy(sdbuf+off, &dlen,4); off+=4; if (dlen>0) { memcpy(sdbuf+off, (uint8_t*)val.mv_data+DB_VAL_OFF_DATA, dlen); off+=dlen; }
rc++; cur++; rrc=mdb_cursor_get(cursor, &key, &val, MDB_NEXT); }
mdb_cursor_close(cursor); } mdb_txn_abort(txn); }
*rcp = rc; sent += rc;
db_sync_send(db, src_node_id, sdbuf, off);
if (rc == 0) break; // no more records
}
return;
}
}
if (memcmp(my_ch, peer_ch, 32) == 0) {
struct DB_SYNC_PEER* p = db_peer_find(db, src_node_id);
if (p) p->synced = 2;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: chain_hash match with %016llx at pos %u — synced", (unsigned long long)src_node_id, test_pos);
// Send our extra records if any (beyond test_pos)
uint32_t my_count = db_count(db);
if (my_count > test_pos + 1) {
// We have more records — send them
uint32_t from = test_pos + 1;
uint32_t remaining = my_count - from;
uint32_t batch = remaining < DB_SEND_DATA_MAX ? remaining : DB_SEND_DATA_MAX;
// Continue in SEND_DATA handler...
}
return;
}
// Find divergence range using sparse hashes
uint32_t div_start = 0, div_end = test_pos;
const uint8_t* sparse = payload + 37;
for (int i = 0; i < sparse_count && sparse + 36 <= payload + len; i++) {
uint32_t pos = *(uint32_t*)sparse;
const uint8_t* peer_sparse_ch = sparse + 4;
uint8_t my_sparse_ch[32];
if (db_chain_hash_at(db, pos, my_sparse_ch) == 0) {
if (memcmp(my_sparse_ch, peer_sparse_ch, 32) == 0) { if (pos + 1 > div_start) div_start = pos + 1; }
else { if (pos < div_end) div_end = pos; }
}
sparse += 36;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: divergence with %016llx range [%u, %u]", (unsigned long long)src_node_id, div_start, div_end);
if (div_end - div_start <= 1) {
// Divergence is small — request data directly via REFINE round
uint8_t ref[512]; uint32_t ref_off = 0;
ref[ref_off++] = DB_MSG_REFINE;
memcpy(ref+ref_off, &div_start, 4); ref_off += 4;
memcpy(ref+ref_off, &div_end, 4); ref_off += 4;
uint8_t hcount = 0; ref[ref_off++] = hcount; // no hashes, just requesting data
db_sync_send(db, src_node_id, ref, ref_off);
} else {
// Send REFINE with our hashes
uint8_t ref[512]; uint32_t ref_off = 0;
ref[ref_off++] = DB_MSG_REFINE;
memcpy(ref+ref_off, &div_start, 4); ref_off += 4;
memcpy(ref+ref_off, &div_end, 4); ref_off += 4;
uint32_t range = div_end - div_start;
uint8_t count = range < DB_REFINE_HASHES ? (uint8_t)range : DB_REFINE_HASHES;
ref[ref_off++] = count;
for (uint8_t i = 0; i < count; i++) {
uint32_t pos = div_start + (range * i / count);
uint64_t dh;
if (db_datahash_at(db, pos, &dh) == 0) {
memcpy(ref+ref_off, &pos, 4); ref_off += 4;
memcpy(ref+ref_off, &dh, 8); ref_off += 8;
}
}
db_sync_send(db, src_node_id, ref, ref_off);
}
}
// ---- Sync protocol: REFINE ----
static void db_handle_refine(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 9) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: REFINE too short %zu", len); return; }
uint32_t from = *(uint32_t*)payload;
uint32_t to = *(uint32_t*)(payload+4);
uint8_t hcount = payload[8];
if (hcount == 0) {
// Peer requests data in range [from, to]
uint32_t my_count = db_count(db);
uint32_t send_from = from;
uint32_t send_count = (to - from + 1) < DB_SEND_DATA_MAX ? (to - from + 1) : DB_SEND_DATA_MAX;
if (send_from >= my_count) return; // nothing to send
// Build SEND_DATA
uint8_t* buf = u_malloc(8192);
if (!buf) return;
uint32_t off = 0; buf[off++] = DB_MSG_SEND_DATA;
memcpy(buf+off, &send_from, 4); off += 4;
uint16_t rec_count = 0;
uint16_t* cnt_ptr = (uint16_t*)(buf+off); off += 2;
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rc == 0) {
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc == 0) {
MDB_val key, val; rc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
uint32_t cur = 0;
while (rc == 0 && cur < send_from) { rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT); cur++; }
while (rc == 0 && rec_count < send_count && off + val.mv_size + 16 < 8000) {
uint64_t ts = be64toh(*(uint64_t*)key.mv_data);
uint64_t dh = be64toh(*(uint64_t*)((uint8_t*)key.mv_data + 8));
uint64_t rid = *(uint64_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_ID);
uint32_t dlen = *(uint32_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_DATA_LEN);
const uint8_t* dptr = (uint8_t*)val.mv_data + DB_VAL_OFF_DATA;
memcpy(buf+off, &rid, 8); off += 8;
memcpy(buf+off, &ts, 8); off += 8;
memcpy(buf+off, &dh, 8); off += 8;
memcpy(buf+off, &dlen, 4); off += 4;
if (dlen > 0) { memcpy(buf+off, dptr, dlen); off += dlen; }
rec_count++;
rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
}
mdb_cursor_close(cursor);
}
mdb_txn_abort(txn);
}
*cnt_ptr = rec_count;
db_sync_send(db, src_node_id, buf, off);
u_free(buf);
return;
}
// Peer sent hashes at positions — find mismatch
uint32_t first_mismatch = to + 1;
const uint8_t* hp = payload + 9;
for (uint8_t i = 0; i < hcount && hp + 12 <= payload + len; i++) {
uint32_t pos = *(uint32_t*)hp;
uint64_t peer_dh = *(uint64_t*)(hp + 4);
uint64_t my_dh;
if (db_datahash_at(db, pos, &my_dh) == 0 && my_dh == peer_dh) {
if (pos >= from) { if (pos + 1 < first_mismatch) first_mismatch = pos + 1; }
} else {
if (pos < first_mismatch) first_mismatch = pos;
}
hp += 12;
}
// Send SEND_DATA from first_mismatch
uint32_t my_count = db_count(db);
uint32_t send_count = 4;
if (first_mismatch > to || first_mismatch >= my_count) { send_count = 0; }
uint8_t sdbuf[8192]; uint32_t off = 0;
sdbuf[off++] = DB_MSG_SEND_DATA;
memcpy(sdbuf+off, &first_mismatch, 4); off += 4;
uint16_t rc_count = 0;
uint16_t* rcp = (uint16_t*)(sdbuf+off); off += 2;
if (send_count > 0) {
MDB_txn* txn; int rrc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rrc == 0) {
MDB_cursor* cursor; rrc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rrc == 0) {
MDB_val key, val; rrc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
uint32_t cur = 0;
while (rrc == 0 && cur < first_mismatch) { rrc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT); cur++; }
while (rrc == 0 && rc_count < send_count && off + val.mv_size + 16 < 8000) {
uint64_t ts = be64toh(*(uint64_t*)key.mv_data);
uint64_t dh = be64toh(*(uint64_t*)((uint8_t*)key.mv_data + 8));
uint64_t rid = *(uint64_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_ID);
uint32_t dlen = *(uint32_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_DATA_LEN);
const uint8_t* dptr = (uint8_t*)val.mv_data + DB_VAL_OFF_DATA;
memcpy(sdbuf+off, &rid, 8); off += 8;
memcpy(sdbuf+off, &ts, 8); off += 8;
memcpy(sdbuf+off, &dh, 8); off += 8;
memcpy(sdbuf+off, &dlen, 4); off += 4;
if (dlen > 0) { memcpy(sdbuf+off, dptr, dlen); off += dlen; }
rc_count++; rrc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
}
mdb_cursor_close(cursor);
}
mdb_txn_abort(txn);
}
}
*rcp = rc_count;
db_sync_send(db, src_node_id, sdbuf, off);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: SEND_DATA to %016llx from=%u count=%u", (unsigned long long)src_node_id, first_mismatch, rc_count);
}
// ---- Sync protocol: SEND_DATA ----
static void db_handle_send_data(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 6) return;
uint32_t from = *(uint32_t*)payload;
uint16_t count = *(uint16_t*)(payload+4);
const uint8_t* ptr = payload + 6;
for (uint16_t i = 0; i < count; i++) {
if (ptr + 28 > payload + len) break;
uint64_t rid = *(uint64_t*)ptr; ptr += 8;
uint64_t ts = *(uint64_t*)ptr; ptr += 8;
uint64_t dh = *(uint64_t*)ptr; ptr += 8;
uint32_t dlen = *(uint32_t*)ptr; ptr += 4;
if (ptr + dlen > payload + len) break;
db_record_insert(db, rid, ts, dh, (const char*)ptr, dlen);
ptr += dlen;
}
// Send our extra records too (bidirectional)
uint32_t my_count = db_count(db);
struct DB_SYNC_PEER* p = db_peer_find(db, src_node_id);
uint32_t peer_known = from + count;
if (my_count > peer_known && p && p->synced == 1) {
// Send our records beyond what peer knows
uint32_t send_count = my_count - peer_known;
if (send_count > DB_SEND_DATA_MAX) send_count = DB_SEND_DATA_MAX;
uint8_t sdbuf[8192]; uint32_t off=0;
sdbuf[off++] = DB_MSG_SEND_DATA; memcpy(sdbuf+off, &peer_known, 4); off+=4;
uint16_t* rcp = (uint16_t*)(sdbuf+off); off+=2;
uint16_t rc=0;
MDB_txn* txn; int rrc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
if (rrc==0) {
MDB_cursor* cursor; rrc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rrc==0) {
MDB_val key,val; rrc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
uint32_t cur=0;
while (rrc==0 && cur < peer_known) { rrc=mdb_cursor_get(cursor,&key,&val,MDB_NEXT); cur++; }
while (rrc==0 && rc<send_count && off+val.mv_size+16 < 8000) {
uint64_t rts = be64toh(*(uint64_t*)key.mv_data);
uint64_t rdh = be64toh(*(uint64_t*)((uint8_t*)key.mv_data+8));
uint64_t rrid = *(uint64_t*)((uint8_t*)val.mv_data+DB_VAL_OFF_ID);
uint32_t rdlen = *(uint32_t*)((uint8_t*)val.mv_data+DB_VAL_OFF_DATA_LEN);
const uint8_t* rdptr = (uint8_t*)val.mv_data+DB_VAL_OFF_DATA;
memcpy(sdbuf+off, &rrid,8); off+=8; memcpy(sdbuf+off, &rts,8); off+=8;
memcpy(sdbuf+off, &rdh,8); off+=8; memcpy(sdbuf+off, &rdlen,4); off+=4;
if (rdlen>0) { memcpy(sdbuf+off, rdptr, rdlen); off+=rdlen; }
rc++; rrc=mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
}
mdb_cursor_close(cursor);
}
mdb_txn_abort(txn);
}
*rcp = rc;
db_sync_send(db, src_node_id, sdbuf, off);
}
// Verify sync complete
uint32_t new_count = db_count(db);
uint8_t my_last_ch[32];
if (new_count > 0 && db_chain_hash_at(db, new_count - 1, my_last_ch) == 0) {
uint8_t sd[37]; sd[0] = DB_MSG_SYNC_DONE;
memcpy(sd+1, &new_count, 4); memcpy(sd+5, my_last_ch, 32);
db_sync_send(db, src_node_id, sd, 37);
}
if (p) p->synced = 2;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: received %u records from %016llx", count, (unsigned long long)src_node_id);
}
// ---- Sync protocol: SYNC_DONE ----
static void db_handle_sync_done(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 36) return;
uint32_t peer_count = *(uint32_t*)payload;
const uint8_t* peer_ch = payload + 4;
uint32_t my_count = db_count(db);
uint8_t my_ch[32];
if (my_count > 0) db_chain_hash_at(db, my_count - 1, my_ch);
else memset(my_ch, 0, 32);
if (my_count != peer_count || memcmp(my_ch, peer_ch, 32) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "db_sync: SYNC_DONE mismatch with %016llx my=%u peer=%u — re-syncing",
(unsigned long long)src_node_id, my_count, peer_count);
db_sync_initiate_sync(db, src_node_id);
return;
}
struct DB_SYNC_PEER* p = db_peer_find(db, src_node_id);
if (p) p->synced = 2;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: sync confirmed with %016llx count=%u", (unsigned long long)src_node_id, my_count);
}
// ---- Sync protocol: ACK_PUSH ----
static void db_handle_ack_push(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
(void)src_node_id;
if (len < 16) return;
uint64_t dh = *(uint64_t*)payload;
uint64_t ts = *(uint64_t*)(payload + 8);
uint8_t key[16]; db_key_build(key, ts, dh);
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, 0, &txn);
if (rc) return;
MDB_val k, val; k.mv_size = 16; k.mv_data = key;
rc = mdb_get(txn, db->dbi_records, &k, &val);
if (rc == 0 && val.mv_size >= DB_VAL_HDR_SIZE) {
uint64_t creator = *(uint64_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_CREATOR);
uint8_t flags = *(uint8_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_FLAGS);
if (creator == db->inst->node_id && !(flags & DB_REC_FLAG_WAS_SENT)) {
flags |= DB_REC_FLAG_WAS_SENT;
*(uint8_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_FLAGS) = flags;
mdb_put(txn, db->dbi_records, &k, &val, 0);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: ACK_PUSH mark sent dh=%016llx ts=%llu from %016llx", (unsigned long long)dh, (unsigned long long)ts, (unsigned long long)src_node_id);
}
}
mdb_txn_commit(txn);
}
// ---- Sync protocol: PUSH ----
static void db_handle_push(struct DB_SYNC* db, uint64_t src_node_id, const uint8_t* payload, size_t len) {
if (len < 28) return;
const uint8_t* ptr = payload;
uint64_t rid = *(uint64_t*)ptr; ptr += 8;
uint64_t ts = *(uint64_t*)ptr; ptr += 8;
uint64_t dh = *(uint64_t*)ptr; ptr += 8;
uint32_t dlen = *(uint32_t*)ptr; ptr += 4;
if (ptr + dlen > payload + len) return;
int ret = db_record_insert(db, rid, ts, dh, (const char*)ptr, dlen);
if (ret == 0) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: PUSH inserted from %016llx id=%llu dh=%016llx", (unsigned long long)src_node_id, (unsigned long long)rid, (unsigned long long)dh);
uint8_t ack[17]; ack[0] = DB_MSG_ACK_PUSH; memcpy(ack+1, &dh, 8); memcpy(ack+9, &ts, 8);
db_sync_send(db, src_node_id, ack, 17);
}
}
// ---- Main receive callback ----
static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || entry->len < 2) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; }
struct DB_SYNC* db = conn ? conn->instance->db_sync : NULL;
if (!db || !db->enabled) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t src_node_id = conn ? conn->peer_node_id : 0;
uint8_t type = entry->dgram[1];
const uint8_t* payload = entry->dgram + 2;
size_t plen = entry->len - 2;
switch (type) {
case DB_MSG_INIT_SYNC: db_handle_init_sync(db, src_node_id, payload, plen); break;
case DB_MSG_INIT_RESP: db_handle_init_resp(db, src_node_id, payload, plen); break;
case DB_MSG_REFINE: db_handle_refine(db, src_node_id, payload, plen); break;
case DB_MSG_SEND_DATA: db_handle_send_data(db, src_node_id, payload, plen); break;
case DB_MSG_PUSH: db_handle_push(db, src_node_id, payload, plen); break;
case DB_MSG_ACK_PUSH: db_handle_ack_push(db, src_node_id, payload, plen); break;
case DB_MSG_SYNC_DONE: db_handle_sync_done(db, src_node_id, payload, plen); break;
default: DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "db_sync: unknown msg type 0x%02x from %016llx", type, (unsigned long long)src_node_id); break;
}
queue_dgram_free(entry); queue_entry_free(entry);
}
// ---- Connection callbacks ----
static void db_sync_on_new_conn(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn || !conn->instance || !conn->instance->db_sync) return;
etcp_conn_add_up_cbk(conn, db_sync_on_conn_up, NULL);
etcp_conn_add_down_cbk(conn, db_sync_on_conn_down, NULL);
}
static void db_sync_on_conn_up(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn || !conn->instance || !conn->instance->db_sync) return;
struct DB_SYNC* db = conn->instance->db_sync;
uint64_t peer_id = conn->peer_node_id;
if (peer_id == 0 || peer_id == db->inst->node_id) return;
db->last_connected_tb = get_time_tb();
struct DB_SYNC_PEER* p = db_peer_add(db, peer_id);
if (p && p->synced == 0) {
p->synced = 1;
db_sync_initiate_sync(db, peer_id);
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: peer up %016llx", (unsigned long long)peer_id);
}
static void db_sync_on_conn_down(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn || !conn->instance || !conn->instance->db_sync) return;
struct DB_SYNC* db = conn->instance->db_sync;
uint64_t peer_id = conn->peer_node_id;
if (peer_id == 0) return;
struct DB_SYNC_PEER* p = db_peer_find(db, peer_id);
if (p) p->synced = 0;
int any_up = 0;
for (int i = 0; i < db->peer_count; i++) { if (db->peers[i].synced >= 1) { any_up = 1; break; } }
if (!any_up) db->last_connected_tb = 0;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: peer down %016llx", (unsigned long long)peer_id);
}
// ---- Initiate sync ----
static void db_sync_initiate_sync(struct DB_SYNC* db, uint64_t peer_node_id) {
uint32_t my_count = db_count(db);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: initiate_sync to %016llx my_count=%u", (unsigned long long)peer_node_id, my_count);
uint8_t msg[37]; msg[0] = DB_MSG_INIT_SYNC;
memcpy(msg+1, &my_count, 4);
if (my_count > 0) {
if (db_chain_hash_at(db, my_count-1, msg+5) != 0) memset(msg+5, 0, 32);
} else {
memset(msg+5, 0, 32);
}
db_sync_send(db, peer_node_id, msg, 37);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: INIT_SYNC → %016llx my_count=%u", (unsigned long long)peer_node_id, my_count);
}
// ---- Periodic timers ----
static void db_sync_peer_check_cb(void* arg) {
struct DB_SYNC* db = (struct DB_SYNC*)arg;
if (!db || !db->enabled) return;
struct ROUTE_BGP* bgp = db->inst->bgp;
if (!bgp) { db->peer_check_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_PEER_CHECK_INTERVAL * 10000u, db, db_sync_peer_check_cb, "db_sync_peer"); return; }
// Check all known peers with active connections
struct ll_entry* e = bgp->senders_list->head;
while (e) {
struct ROUTE_BGP_CONN_ITEM* item = (struct ROUTE_BGP_CONN_ITEM*)e->data;
if (item->conn && item->conn->peer_node_id != 0 && item->conn->links_up) {
uint64_t pid = item->conn->peer_node_id;
if (pid != db->inst->node_id) {
struct DB_SYNC_PEER* p = db_peer_add(db, pid);
if (p && p->synced == 0) { p->synced = 1; db_sync_initiate_sync(db, pid); }
}
}
e = e->next;
}
db->peer_check_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_PEER_CHECK_INTERVAL * 10000u, db, db_sync_peer_check_cb, "db_sync_peer");
}
static void db_sync_ttl_cleanup_cb(void* arg) {
struct DB_SYNC* db = (struct DB_SYNC*)arg;
if (!db || !db->enabled || !db->env) { if (db) db->ttl_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_TTL_INTERVAL * 10000u, db, db_sync_ttl_cleanup_cb, "db_sync_ttl"); return; }
uint64_t my_id = db->inst->node_id;
uint64_t now_us = get_time_us();
uint64_t ttl_us = (uint64_t)(db->inst->config->global.db_sync_ttl) * 1000000uLL;
uint64_t cutoff = now_us - ttl_us;
MDB_txn* txn; int rc = mdb_txn_begin(db->env, NULL, 0, &txn);
if (rc != 0) { db->ttl_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_TTL_INTERVAL * 10000u, db, db_sync_ttl_cleanup_cb, "db_sync_ttl"); return; }
MDB_cursor* cursor; rc = mdb_cursor_open(txn, db->dbi_records, &cursor);
if (rc == 0) {
MDB_val key, val; rc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);
while (rc == 0) {
if (val.mv_size >= DB_VAL_HDR_SIZE) {
uint64_t creator = *(uint64_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_CREATOR);
uint8_t flags = *(uint8_t*)((uint8_t*)val.mv_data + DB_VAL_OFF_FLAGS);
uint64_t ts = be64toh(*(uint64_t*)key.mv_data);
if (creator == my_id && !(flags & DB_REC_FLAG_WAS_SENT) && ts < cutoff) {
mdb_cursor_del(cursor, 0);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: TTL deleted unsent record ts=%llu dh=%016llx", (unsigned long long)ts, (unsigned long long)be64toh(*(uint64_t*)((uint8_t*)key.mv_data+8)));
}
}
rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
}
mdb_cursor_close(cursor);
}
mdb_txn_commit(txn);
db->ttl_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_TTL_INTERVAL * 10000u, db, db_sync_ttl_cleanup_cb, "db_sync_ttl");
}
// ---- Public API ----
int db_sync_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync_init: NULL instance"); return -1; }
struct DB_SYNC* db = u_calloc(1, sizeof(struct DB_SYNC));
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync_init: u_calloc failed"); return -1; }
db->inst = inst; db->last_connected_tb = 0; db->last_timestamp_us = 0;
if (!inst->config->global.db_sync_enabled) {
db->enabled = 0; inst->db_sync = db;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: disabled by config");
return 0;
}
db->enabled = 1;
inst->db_sync = db;
// Open LMDB
const char* db_path = inst->config->global.db_path;
char sync_path[512];
if (db_path[0]) {
snprintf(sync_path, sizeof(sync_path), "%s/sync", db_path);
} else {
snprintf(sync_path, sizeof(sync_path), "/tmp/utun_db_sync");
}
if (db_lmdb_open(db, sync_path) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "db_sync_init: LMDB open failed, sync disabled");
db->enabled = 0;
}
// Register in etcp_router
etcp_router_bind(inst, ETCP_RT_ID_DB_SYNC, db_sync_recv_cb);
// Subscribe to ETCP connection events
etcp_add_new_conn_cbk(inst, db_sync_on_new_conn, NULL);
// For existing connections: add up/down callbacks
{
struct ETCP_CONN* conn = inst->connections;
while (conn) {
etcp_conn_add_up_cbk(conn, db_sync_on_conn_up, NULL);
etcp_conn_add_down_cbk(conn, db_sync_on_conn_down, NULL);
conn = conn->next;
}
}
// Start periodic timers
db->peer_check_timer = uasync_set_timeout(inst->ua, DB_SYNC_PEER_CHECK_INTERVAL * 10000u, db, db_sync_peer_check_cb, "db_sync_peer");
db->ttl_timer = uasync_set_timeout(inst->ua, DB_SYNC_TTL_INTERVAL * 10000u, db, db_sync_ttl_cleanup_cb, "db_sync_ttl");
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: initialized (enabled=%d)", db->enabled);
return 0;
}
void db_sync_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->db_sync) return;
struct DB_SYNC* db = inst->db_sync;
inst->db_sync = NULL;
etcp_router_unbind(inst, ETCP_RT_ID_DB_SYNC);
if (db->peer_check_timer) { uasync_cancel_timeout(inst->ua, db->peer_check_timer); db->peer_check_timer = NULL; }
if (db->ttl_timer) { uasync_cancel_timeout(inst->ua, db->ttl_timer); db->ttl_timer = NULL; }
// Remove up/down callbacks from existing connections
{ struct ETCP_CONN* conn = inst->connections; while (conn) { etcp_conn_remove_up_cbk(conn, db_sync_on_conn_up, NULL); etcp_conn_remove_down_cbk(conn, db_sync_on_conn_down, NULL); conn = conn->next; } }
db_lmdb_close(db);
if (db->peers) u_free(db->peers);
u_free(db);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: destroyed");
}
int db_sync_insert_len(struct UTUN_INSTANCE* inst, const char* json_data, size_t len) {
if (!inst || !inst->db_sync || !json_data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync_insert: invalid args"); return -1; }
struct DB_SYNC* db = inst->db_sync;
if (!db->enabled) return -1;
// Compute datahash and timestamp
uint64_t datahash = db_datahash((const uint8_t*)json_data, len);
uint64_t now_us = get_time_us();
// Guarantee monotonicity
if (now_us <= db->last_timestamp_us) now_us = db->last_timestamp_us + 1;
db->last_timestamp_us = now_us;
uint64_t timestamp = now_us;
uint64_t id = db->next_id;
int ret = db_record_insert(db, id, timestamp, datahash, json_data, len);
if (ret != 0) return ret; // 1 = duplicate, -1 = error
// Broadcast PUSH to all synced peers
uint8_t pbuf[2048];
uint32_t off = 0; pbuf[off++] = DB_MSG_PUSH;
memcpy(pbuf+off, &id, 8); off += 8;
memcpy(pbuf+off, &timestamp, 8); off += 8;
memcpy(pbuf+off, &datahash, 8); off += 8;
memcpy(pbuf+off, &len, 4); off += 4;
if (len > 0 && off + len <= sizeof(pbuf)) { memcpy(pbuf+off, json_data, len); off += len; }
for (int i = 0; i < db->peer_count; i++) if (db->peers[i].synced >= 1 && db->peers[i].node_id != inst->node_id) db_sync_send(db, db->peers[i].node_id, pbuf, off);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: insert id=%llu dh=%016llx ts=%llu len=%zu", (unsigned long long)id, (unsigned long long)datahash, (unsigned long long)timestamp, len);
return 0;
}
int db_sync_insert(struct UTUN_INSTANCE* inst, const char* json_data) {
if (!json_data) return -1;
return db_sync_insert_len(inst, json_data, strlen(json_data));
}
uint32_t db_sync_count(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->db_sync) return 0;
return db_count(inst->db_sync);
}
src/db_sync.h
+41
View File
@@ -0,0 +1,41 @@
// db_sync.h — Distributed content-addressed table with LMDB storage and peer sync via etcp_router
#ifndef DB_SYNC_H
#define DB_SYNC_H
#include <stdint.h>
#include <stddef.h>
struct UTUN_INSTANCE;
// etcp_router service ID
#define ETCP_RT_ID_DB_SYNC 0x20
// Message types
#define DB_MSG_INIT_SYNC 0x01
#define DB_MSG_INIT_RESP 0x02
#define DB_MSG_REFINE 0x03
#define DB_MSG_SEND_DATA 0x04
#define DB_MSG_PUSH 0x05
#define DB_MSG_ACK_PUSH 0x06
#define DB_MSG_SYNC_DONE 0x07
// Defaults
#define DB_SYNC_DEFAULT_TTL 86400
#define DB_SYNC_DEFAULT_MAPSIZE (100UL * 1024 * 1024)
#define DB_SYNC_PEER_CHECK_INTERVAL 5
#define DB_SYNC_TTL_INTERVAL 3600
// Record flags
#define DB_REC_FLAG_WAS_SENT 0x01
// Sync protocol constants
#define DB_REFINE_HASHES 16
#define DB_SEND_DATA_MAX 32
int db_sync_init(struct UTUN_INSTANCE* inst);
void db_sync_destroy(struct UTUN_INSTANCE* inst);
int db_sync_insert(struct UTUN_INSTANCE* inst, const char* json_data);
int db_sync_insert_len(struct UTUN_INSTANCE* inst, const char* json_data, size_t len);
uint32_t db_sync_count(struct UTUN_INSTANCE* inst);
#endif // DB_SYNC_H
src/etcp.c
+13 -6
View File
@@ -257,8 +257,9 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n
etcp->log_name, etcp, etcp->mtu, etcp->next_tx_id);
// Вызываем callback для нового соединения если установлен
if (instance && instance->etcp_new_conn_cbk) {
instance->etcp_new_conn_cbk(etcp, instance->etcp_new_conn_arg);
if (instance) {
struct etcp_cbk_entry* cbe = instance->new_conn_cbks;
while (cbe) { cbe->fn(etcp, cbe->arg); cbe = cbe->next; }
}
return etcp;
@@ -266,14 +267,15 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n
static void etcp_on_up(struct ETCP_CONN* etcp) {
// if (!etcp->initialized) return;
DEBUG_WARN(DEBUG_CATEGORY_BGP, "[%s] UP links_up=%d initialized=%d", etcp->log_name, etcp->links_up, etcp->initialized);
if (etcp->up_cbk) etcp->up_cbk(etcp, etcp->up_arg);
struct etcp_cbk_entry* cbe = etcp->up_cbks;
while (cbe) { cbe->fn(etcp, cbe->arg); cbe = cbe->next; }
}
static void etcp_on_down(struct ETCP_CONN* etcp) {
DEBUG_WARN(DEBUG_CATEGORY_BGP, "[%s] DOWN links_up=%d", etcp->log_name, etcp->links_up);
if (etcp->down_cbk) etcp->down_cbk(etcp, etcp->down_arg);
struct etcp_cbk_entry* cbe = etcp->down_cbks;
while (cbe) { cbe->fn(etcp, cbe->arg); cbe = cbe->next; }
}
@@ -332,6 +334,11 @@ void etcp_connection_close(struct ETCP_CONN* etcp) {
drain_and_free_queue(&etcp->recv_q);
drain_and_free_queue(&etcp->ack_q);
// Free callback chains
{ struct etcp_cbk_entry* cbe = etcp->ready_cbks; while (cbe) { struct etcp_cbk_entry* n = cbe->next; u_free(cbe); cbe = n; } etcp->ready_cbks = NULL; }
{ struct etcp_cbk_entry* cbe = etcp->up_cbks; while (cbe) { struct etcp_cbk_entry* n = cbe->next; u_free(cbe); cbe = n; } etcp->up_cbks = NULL; }
{ struct etcp_cbk_entry* cbe = etcp->down_cbks; while (cbe) { struct etcp_cbk_entry* n = cbe->next; u_free(cbe); cbe = n; } etcp->down_cbks = NULL; }
// Free memory pools after all elements are returned
if (etcp->inflight_pool) {
memory_pool_destroy(etcp->inflight_pool);
@@ -527,7 +534,7 @@ void etcp_conn_ready(struct ETCP_CONN* conn) {
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] Connection ready", conn->log_name);
// Вызываем callback если установлен
if (conn->ready_cbk) conn->ready_cbk(conn, conn->ready_arg);
{ struct etcp_cbk_entry* cbe = conn->ready_cbks; while (cbe) { cbe->fn(conn, cbe->arg); cbe = cbe->next; } }
if (conn->links_up) etcp_on_up(conn);
}
src/etcp.h
+9 -7
View File
@@ -21,6 +21,11 @@ struct stcp_link; // forward declaration
struct UTUN_INSTANCE;
struct ETCP_CONN;
typedef void (*etcp_on_conn_ready)(struct ETCP_CONN* conn, void* arg);
struct etcp_cbk_entry {
etcp_on_conn_ready fn;
void* arg;
struct etcp_cbk_entry* next;
};
struct UASYNC;
uint16_t get_current_timestamp(void);
@@ -184,13 +189,10 @@ struct ETCP_CONN {
uint8_t tx_state; // 0 - n/a, 1 - data_wait (queues empty), 2 - link_wait (link busy)
uint8_t links_up; // 0 - канал не готов для передачи, 1 - канал готов для передачи (хотя бы один линк не down)
// Callback for ready notification
etcp_on_conn_ready ready_cbk; // callback при готовности соединения
void* ready_arg; // аргумент для ready_cbk
etcp_on_conn_ready up_cbk; // callback при готовности соединения
void* up_arg; // аргумент для ready_cbk
etcp_on_conn_ready down_cbk; // callback при готовности соединения
void* down_arg; // аргумент для ready_cbk
// Callback chains for ready/up/down notifications
struct etcp_cbk_entry* ready_cbks; // цепочка callback'ов при готовности соединения
struct etcp_cbk_entry* up_cbks; // цепочка callback'ов при поднятии канала
struct etcp_cbk_entry* down_cbks; // цепочка callback'ов при падении канала
void (*bgp_ready_cbk)(struct ETCP_CONN* conn); // вызывается когда BGP готов (завершён или пропущен)
uint32_t cnt_ack_hit_inf; // счетчик удлений из inflight
src/etcp_api.c
+49 -4
View File
@@ -7,23 +7,68 @@
#include "pkt_normalizer.h"
#include "utun_instance.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include <string.h>
#define DEBUG_CATEGORY_ETCP_API DEBUG_CATEGORY_ETCP
void etcp_conn_set_ready_cbk(struct ETCP_CONN* e, etcp_cbk_fn fn, void* arg) {
if (e) { e->ready_cbk = fn; e->ready_arg = arg; }
if (!e) return;
struct etcp_cbk_entry* entry = e->ready_cbks;
while (entry) { struct etcp_cbk_entry* next = entry->next; u_free(entry); entry = next; }
e->ready_cbks = NULL;
if (fn) etcp_conn_add_ready_cbk(e, fn, arg);
}
void etcp_conn_set_up_cbk(struct ETCP_CONN* e, etcp_cbk_fn fn, void* arg) {
if (e) { e->up_cbk = fn; e->up_arg = arg; }
if (!e) return;
struct etcp_cbk_entry* entry = e->up_cbks;
while (entry) { struct etcp_cbk_entry* next = entry->next; u_free(entry); entry = next; }
e->up_cbks = NULL;
if (fn) etcp_conn_add_up_cbk(e, fn, arg);
}
void etcp_conn_set_down_cbk(struct ETCP_CONN* e, etcp_cbk_fn fn, void* arg) {
if (e) { e->down_cbk = fn; e->down_arg = arg; }
if (!e) return;
struct etcp_cbk_entry* entry = e->down_cbks;
while (entry) { struct etcp_cbk_entry* next = entry->next; u_free(entry); entry = next; }
e->down_cbks = NULL;
if (fn) etcp_conn_add_down_cbk(e, fn, arg);
}
void etcp_set_new_conn_cbk(struct UTUN_INSTANCE* inst, etcp_cbk_fn fn, void* arg) {
if (inst) { inst->etcp_new_conn_cbk = fn; inst->etcp_new_conn_arg = arg; }
if (!inst) return;
struct etcp_cbk_entry* entry = inst->new_conn_cbks;
while (entry) { struct etcp_cbk_entry* next = entry->next; u_free(entry); entry = next; }
inst->new_conn_cbks = NULL;
if (fn) etcp_add_new_conn_cbk(inst, fn, arg);
}
static void etcp_cbk_add_to_chain(struct etcp_cbk_entry** head, etcp_cbk_fn fn, void* arg) {
if (!head || !fn) return;
struct etcp_cbk_entry* e = u_malloc(sizeof(struct etcp_cbk_entry));
if (!e) return;
e->fn = fn; e->arg = arg; e->next = *head;
*head = e;
}
static void etcp_cbk_remove_from_chain(struct etcp_cbk_entry** head, etcp_cbk_fn fn, void* arg) {
if (!head || !fn) return;
struct etcp_cbk_entry** p = head;
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct etcp_cbk_entry* rm = *p;
*p = rm->next; u_free(rm); return;
}
p = &(*p)->next;
}
}
void etcp_conn_add_ready_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_add_to_chain(&conn->ready_cbks, fn, arg); }
void etcp_conn_remove_ready_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_remove_from_chain(&conn->ready_cbks, fn, arg); }
void etcp_conn_add_up_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_add_to_chain(&conn->up_cbks, fn, arg); }
void etcp_conn_remove_up_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_remove_from_chain(&conn->up_cbks, fn, arg); }
void etcp_conn_add_down_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_add_to_chain(&conn->down_cbks, fn, arg); }
void etcp_conn_remove_down_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg) { if (conn) etcp_cbk_remove_from_chain(&conn->down_cbks, fn, arg); }
void etcp_add_new_conn_cbk(struct UTUN_INSTANCE* inst, etcp_cbk_fn fn, void* arg) { if (inst) etcp_cbk_add_to_chain(&inst->new_conn_cbks, fn, arg); }
void etcp_remove_new_conn_cbk(struct UTUN_INSTANCE* inst, etcp_cbk_fn fn, void* arg) { if (inst) etcp_cbk_remove_from_chain(&inst->new_conn_cbks, fn, arg); }
void etcp_set_routing_exchange_state(struct ETCP_CONN* conn, uint8_t new_state) {
if (!conn) return;
conn->routing_exchange_active = new_state;
src/etcp_api.h
+9
View File
@@ -132,6 +132,15 @@ void etcp_conn_set_ready_cbk(struct ETCP_CONN* conn, etcp_cbk_fn callback_fn, vo
void etcp_conn_set_up_cbk (struct ETCP_CONN* conn, etcp_cbk_fn callback_fn, void* arg);
void etcp_conn_set_down_cbk (struct ETCP_CONN* conn, etcp_cbk_fn callback_fn, void* arg);
void etcp_conn_add_ready_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_conn_remove_ready_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_conn_add_up_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_conn_remove_up_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_conn_add_down_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_conn_remove_down_cbk(struct ETCP_CONN* conn, etcp_cbk_fn fn, void* arg);
void etcp_add_new_conn_cbk(struct UTUN_INSTANCE* inst, etcp_cbk_fn fn, void* arg);
void etcp_remove_new_conn_cbk(struct UTUN_INSTANCE* inst, etcp_cbk_fn fn, void* arg);
void etcp_set_routing_exchange_state(struct ETCP_CONN* conn, uint8_t new_state);
/**
src/etcp_connect.c
+2 -4
View File
@@ -192,8 +192,7 @@ static void connect_settle_timeout_cb(void* arg) {
(unsigned long long)ctx->node_id);
stcp_link_close(ctx->tcp_link); ctx->tcp_link = NULL;
}
ctx->conn->ready_cbk = NULL;
ctx->conn->ready_arg = NULL;
etcp_conn_remove_ready_cbk(ctx->conn, connect_ready_cb, ctx);
connect_deliver(ctx, ETCP_CONNECT_LATE);
connect_cancel(ctx);
}
@@ -281,8 +280,7 @@ int etcp_connect(struct UTUN_INSTANCE* inst, struct NODEINFO_Q* node,
cn->cb = cb; cn->arg = arg; cn->flags = flags; ctx->cb_list = cn;
}
conn->ready_cbk = connect_ready_cb;
conn->ready_arg = ctx;
etcp_conn_add_ready_cbk(conn, connect_ready_cb, ctx);
conn->bgp_ready_cbk = connect_bgp_ready_cb;
connect_create_links_v4(ctx, node);
src/etcp_connections.c
+3 -2
View File
@@ -40,8 +40,9 @@ static void tcp_server_on_link(struct stcp_link *link, void *arg) {
conn->next = inst->connections;
inst->connections = conn;
inst->connections_count++;
if (inst->etcp_new_conn_cbk) inst->etcp_new_conn_cbk(conn, inst->etcp_new_conn_arg);
if (conn->ready_cbk) conn->ready_cbk(conn, conn->ready_arg);
struct etcp_cbk_entry* cbe = inst->new_conn_cbks;
while (cbe) { cbe->fn(conn, cbe->arg); cbe = cbe->next; }
{ struct etcp_cbk_entry* rcb = conn->ready_cbks; while (rcb) { rcb->fn(conn, rcb->arg); rcb = rcb->next; } }
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "TCP server new conn=%p total=%d", (void*)conn, inst->connections_count);
}
src/route_bgp.c
+3 -3
View File
@@ -241,8 +241,8 @@ static void route_bgp_etcp_conn_cbk(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (conn && conn->instance && conn->instance->bgp) {
DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP set callbacks: %s", conn->log_name);
etcp_conn_set_up_cbk(conn, route_bgp_on_conn_up, conn->instance->bgp);
etcp_conn_set_down_cbk(conn, route_bgp_on_conn_down, conn->instance->bgp);
etcp_conn_add_up_cbk(conn, route_bgp_on_conn_up, conn->instance->bgp);
etcp_conn_add_down_cbk(conn, route_bgp_on_conn_down, conn->instance->bgp);
}
}
@@ -316,7 +316,7 @@ struct ROUTE_BGP* route_bgp_init(struct UTUN_INSTANCE* instance) {
etcp_bind(instance, ETCP_ID_ROUTE_ENTRY, route_bgp_receive_cbk);
// Устанавливаем callback для новых ETCP соединений
etcp_set_new_conn_cbk(instance, route_bgp_etcp_conn_cbk, NULL);
etcp_add_new_conn_cbk(instance, route_bgp_etcp_conn_cbk, NULL);
DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP module initialized (NODEINFO based routing)");
return bgp;
src/utun_instance.c
+7
View File
@@ -12,6 +12,7 @@
#include "etcp_connections.h"
#include "etcp.h"
#include "conn_mgr.h"
#include "db_sync.h"
#include "stcp_server.h"
#include "control_server.h"
#include "msg_transport.h"
@@ -179,6 +180,9 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u
return -1;
}
// db_sync — распределённая таблица с репликацией (после etcp_router)
db_sync_init(instance);
// Bind DATA handler via etcp_router (after etcp_router_init)
if (routing_bind(instance) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_ROUTING, "Failed to bind DATA via etcp_router");
@@ -417,6 +421,9 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
nat_transport_destroy(instance);
}
// Cleanup db_sync (before etcp_router_destroy)
db_sync_destroy(instance);
// Cleanup etcp_router
etcp_router_destroy(instance);
src/utun_instance.h
+5 -3
View File
@@ -35,6 +35,7 @@ struct msg_transport;
struct PING_CONTEXT;
struct CONN_MGR;
struct ETCP_CONNECT;
struct DB_SYNC;
struct NETWORK_ENTRY {
uint64_t id; // 56-bit (offset 0 = index key)
@@ -77,9 +78,9 @@ struct UTUN_INSTANCE {
struct ETCP_CONN* connections;// linked-list
int connections_count; // Number of connections
// Callback for new ETCP connections
etcp_new_conn_fn etcp_new_conn_cbk;
void* etcp_new_conn_arg;
// Callback chain for new ETCP connections
struct etcp_cbk_entry* new_conn_cbks;
void* test_user_ptr; // Generic user pointer (used by tests)
struct memory_pool* data_pool;// для входных-выходных данных пакета
struct memory_pool* pkt_pool;
@@ -122,6 +123,7 @@ struct UTUN_INSTANCE {
struct ETCP_ROUTER_BINDINGS router_bindings;
struct ll_queue* router_conns;
struct CONN_MGR* conn_mgr; // Connection Manager (может быть NULL)
struct DB_SYNC* db_sync; // Distributed DB sync (может быть NULL)
// TCP proxy server (exit node)
struct tcp_proxy_server tcp_proxy_server;