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

db_sync v2: positional sync per-peer (synced_pos), datahash instead of chain_hash, no cascade on PUSH, single cascade after SEND_DATA batch, startup chain verification. PRAGMA synchronous=NORMAL + wal_autocheckpoint=10000 for performance.

Browse Source
This commit is contained in:
Evgeny committed 2026-07-16 12:31:52 +03:00
1 parent 0479c309bb
commit bb425c6722
5 files changed
+849 -169
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/db_sync.c
+302 -165
View File
@@ -39,7 +39,8 @@ static void db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si,
struct SI_PEER {
uint64_t node_id;
uint8_t synced;
uint32_t synced_pos;
uint8_t sync_state;
};
struct DB_SYNC_INSTANCE {
@@ -108,6 +109,8 @@ static int db_sqlite_open(struct DB_SYNC* db, const char* path)
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: WAL pragma: %s", err);
sqlite3_free(err);
}
sqlite3_exec(db->db, "PRAGMA synchronous=NORMAL", NULL, NULL, NULL);
sqlite3_exec(db->db, "PRAGMA wal_autocheckpoint=10000", NULL, NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "db_sync: SQLite opened at %s", path);
return 0;
}
@@ -243,7 +246,8 @@ static struct SI_PEER* si_peer_add(struct DB_SYNC_INSTANCE* si,
}
p = &si->peers[si->peer_count++];
p->node_id = node_id;
p->synced = 0;
p->synced_pos = 0;
p->sync_state = 0;
return p;
}
@@ -311,6 +315,23 @@ static int db_datahash_at(struct DB_SYNC_INSTANCE* si,
return 0;
}
static uint32_t si_find_pos(struct DB_SYNC_INSTANCE* si,
uint64_t ts, uint64_t dh)
{
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
"SELECT COUNT(*) FROM \"%s\""
" WHERE timestamp<?1"
" OR (timestamp=?1 AND datahash<?2)") != SQLITE_OK)
return 0;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)dh);
uint32_t pos = (sqlite3_step(stmt) == SQLITE_ROW)
? (uint32_t)sqlite3_column_int64(stmt, 0) : 0;
sqlite3_finalize(stmt);
return pos;
}
static int db_prev_chain_hash(struct DB_SYNC_INSTANCE* si,
uint64_t ts, uint64_t dh, uint8_t out[32])
{
@@ -338,10 +359,83 @@ static int db_prev_chain_hash(struct DB_SYNC_INSTANCE* si,
return 0;
}
static void db_cascade_from(struct DB_SYNC_INSTANCE* si, uint32_t from_pos)
{
sqlite3* db = SI_DB(si);
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC,
"cascade_from BEGIN: %s", sqlite3_errmsg(db));
return;
}
uint8_t prev_ch[32];
memset(prev_ch, 0, 32);
if (from_pos > 0) {
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
"SELECT chain_hash FROM \"%s\""
" ORDER BY timestamp, datahash"
" LIMIT 1 OFFSET ?") == SQLITE_OK)
{
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)(from_pos - 1));
if (sqlite3_step(stmt) == SQLITE_ROW) {
const void* b = sqlite3_column_blob(stmt, 0);
if (b && sqlite3_column_bytes(stmt, 0) >= 32)
memcpy(prev_ch, b, 32);
}
sqlite3_finalize(stmt);
}
}
sqlite3_stmt* sel, *upd;
if (si_prep(si, &sel,
"SELECT id,timestamp,datahash FROM \"%s\""
" ORDER BY timestamp,datahash"
" LIMIT -1 OFFSET ?") != SQLITE_OK)
{
sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL);
return;
}
sqlite3_bind_int64(sel, 1, (sqlite3_int64)from_pos);
if (si_prep(si, &upd,
"UPDATE \"%s\" SET chain_hash=?"
" WHERE timestamp=? AND datahash=?") != SQLITE_OK)
{
sqlite3_finalize(sel);
sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL);
return;
}
uint8_t ch[32];
while (sqlite3_step(sel) == SQLITE_ROW) {
sqlite3_int64 n_id = sqlite3_column_int64(sel, 0);
sqlite3_int64 n_ts = sqlite3_column_int64(sel, 1);
sqlite3_int64 n_dh = sqlite3_column_int64(sel, 2);
db_chain_hash_compute(prev_ch, (uint64_t)n_id,
(uint64_t)n_ts, (uint64_t)n_dh, ch);
sqlite3_reset(upd);
sqlite3_bind_blob(upd, 1, ch, 32, SQLITE_STATIC);
sqlite3_bind_int64(upd, 2, n_ts);
sqlite3_bind_int64(upd, 3, n_dh);
sqlite3_step(upd);
memcpy(prev_ch, ch, 32);
}
sqlite3_finalize(upd);
sqlite3_finalize(sel);
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC,
"cascade_from COMMIT: %s", sqlite3_errmsg(db));
}
static int db_record_insert(struct DB_SYNC_INSTANCE* si,
uint64_t id, uint64_t ts, uint64_t dh,
const char* json, size_t jlen,
const uint8_t* sig, size_t sig_len)
const uint8_t* sig, size_t sig_len,
int do_cascade)
{
sqlite3* db = SI_DB(si);
sqlite3_stmt* stmt;
@@ -410,6 +504,7 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
}
// Recompute chain hashes for all subsequent records
if (do_cascade) {
sqlite3_stmt* sel;
if (si_prep(si, &sel,
"SELECT timestamp,datahash,id FROM \"%s\""
@@ -451,6 +546,7 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
}
sqlite3_finalize(upd);
sqlite3_finalize(sel);
}
si->next_id = id + 1;
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
@@ -575,14 +671,13 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si,
uint64_t src,
const uint8_t* p, size_t len)
{
if (len < 36) {
if (len < 4) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC,
"INIT_SYNC too short %zu from %016llx",
len, (unsigned long long)src);
return;
}
uint32_t pc = *(uint32_t*)p;
const uint8_t* plc = p + 4;
uint32_t mc = db_count(si);
uint32_t tp = (pc < mc ? pc : mc);
if (tp > 0)
@@ -593,24 +688,11 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si,
resp[off++] = DB_MSG_INIT_RESP;
memcpy(resp + off, &tp, 4); off += 4;
uint8_t my_ch[32];
if (mc == 0)
memset(my_ch, 0, 32);
else if (db_chain_hash_at(si, tp, my_ch) != 0)
memset(my_ch, 0, 32);
memcpy(resp + off, my_ch, 32); off += 32;
uint64_t my_dh = 0;
if (mc > 0)
db_datahash_at(si, tp, &my_dh);
memcpy(resp + off, &my_dh, 8); off += 8;
// Exact match — sync complete
if (memcmp(my_ch, plc, 32) == 0) {
resp[off++] = 0;
db_sync_send(si, src, resp, off);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"INIT_RESP complete to %016llx tp=%u",
(unsigned long long)src, tp);
return;
}
// Sparse hashes — binary search
uint32_t sp = off;
off++;
int scnt = 0;
@@ -619,12 +701,13 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si,
if (tp < step)
break;
uint32_t pos = tp - step;
if (db_chain_hash_at(si, pos, my_ch) != 0)
uint64_t sdh;
if (db_datahash_at(si, pos, &sdh) != 0)
break;
if (off + 36 > sizeof(resp))
if (off + 12 > sizeof(resp))
break;
memcpy(resp + off, &pos, 4); off += 4;
memcpy(resp + off, my_ch, 32); off += 32;
memcpy(resp + off, &pos, 4); off += 4;
memcpy(resp + off, &sdh, 8); off += 8;
scnt++;
}
resp[sp] = (uint8_t)scnt;
@@ -638,122 +721,119 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si,
uint64_t src,
const uint8_t* p, size_t len)
{
if (len < 37) {
if (len < 13) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC,
"INIT_RESP too short %zu from %016llx",
len, (unsigned long long)src);
return;
}
uint32_t tp = *(uint32_t*)p;
const uint8_t* pc = p + 4;
uint8_t sc = p[36];
uint64_t peer_dh;
memcpy(&peer_dh, p + 4, 8);
uint8_t sc = p[12];
uint8_t my_ch[32];
if (db_chain_hash_at(si, tp, my_ch) != 0)
memset(my_ch, 0, 32);
uint64_t my_dh = 0;
db_datahash_at(si, tp, &my_dh);
// Exact match at tp — sync complete
if (sc == 0 && memcmp(my_ch, pc, 32) == 0) {
if (sc == 0 && my_dh == peer_dh) {
struct SI_PEER* sp = si_peer_find(si, src);
if (sp)
sp->synced = 2;
if (sp) {
sp->synced_pos = tp;
sp->sync_state = 2;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"sync complete with %016llx",
(unsigned long long)src);
return;
}
// Peer has empty DB — send all records
{
uint8_t z[32];
memset(z, 0, 32);
if (memcmp(pc, z, 32) == 0 && sc == 0) {
uint32_t mc = db_count(si);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"peer %016llx empty, sending all %u",
(unsigned long long)src, mc);
uint32_t sent = 0;
while (sent < mc) {
uint32_t b = mc - sent;
if (b > DB_SEND_DATA_MAX)
b = DB_SEND_DATA_MAX;
if (peer_dh == 0 && sc == 0) {
uint32_t mc = db_count(si);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"peer %016llx empty, sending all %u",
(unsigned long long)src, mc);
uint32_t sent = 0;
while (sent < mc) {
uint32_t b = mc - sent;
if (b > DB_SEND_DATA_MAX)
b = 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 rc = 0;
uint16_t* rcp = (uint16_t*)(sdbuf + off); off += 2;
uint8_t sdbuf[8192];
uint32_t off = 0;
sdbuf[off++] = DB_MSG_SEND_DATA;
memcpy(sdbuf + off, &sent, 4); off += 4;
uint16_t rc = 0;
uint16_t* rcp = (uint16_t*)(sdbuf + off); off += 2;
sqlite3_stmt* stmt;
si_prep(si, &stmt,
"SELECT id,timestamp,datahash,data,author_signature"
" FROM \"%s\" ORDER BY timestamp,datahash"
" LIMIT ? OFFSET ?");
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)b);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)sent);
while (sqlite3_step(stmt) == SQLITE_ROW) {
uint64_t rid = (uint64_t)sqlite3_column_int64(stmt, 0);
uint64_t rts = (uint64_t)sqlite3_column_int64(stmt, 1);
uint64_t rdh = (uint64_t)sqlite3_column_int64(stmt, 2);
const uint8_t* rd =
(const uint8_t*)sqlite3_column_blob(stmt, 3);
uint32_t rdl = (uint32_t)sqlite3_column_bytes(stmt, 3);
if (!rd) rdl = 0;
const uint8_t* rsig =
(const uint8_t*)sqlite3_column_blob(stmt, 4);
int rsl = sqlite3_column_bytes(stmt, 4);
if (!rsig) rsl = 0;
sqlite3_stmt* stmt;
si_prep(si, &stmt,
"SELECT id,timestamp,datahash,data,author_signature"
" FROM \"%s\" ORDER BY timestamp,datahash"
" LIMIT ? OFFSET ?");
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)b);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)sent);
while (sqlite3_step(stmt) == SQLITE_ROW) {
uint64_t rid = (uint64_t)sqlite3_column_int64(stmt, 0);
uint64_t rts = (uint64_t)sqlite3_column_int64(stmt, 1);
uint64_t rdh = (uint64_t)sqlite3_column_int64(stmt, 2);
const uint8_t* rd =
(const uint8_t*)sqlite3_column_blob(stmt, 3);
uint32_t rdl = (uint32_t)sqlite3_column_bytes(stmt, 3);
if (!rd) rdl = 0;
const uint8_t* rsig =
(const uint8_t*)sqlite3_column_blob(stmt, 4);
int rsl = sqlite3_column_bytes(stmt, 4);
if (!rsig) rsl = 0;
int rec_sz = 28 + rdl + 1 + rsl;
if (off + rec_sz > (int)sizeof(sdbuf))
break;
memcpy(sdbuf + off, &rid, 8); off += 8;
memcpy(sdbuf + off, &rts, 8); off += 8;
memcpy(sdbuf + off, &rdh, 8); off += 8;
memcpy(sdbuf + off, &rdl, 4); off += 4;
if (rdl > 0) {
memcpy(sdbuf + off, rd, rdl); off += rdl;
}
sdbuf[off++] = (uint8_t)rsl;
if (rsl > 0) {
memcpy(sdbuf + off, rsig, rsl); off += rsl;
}
rc++;
}
sqlite3_finalize(stmt);
*rcp = rc;
sent += rc;
db_sync_send(si, src, sdbuf, off);
if (rc == 0)
int rec_sz = 28 + rdl + 1 + rsl;
if (off + rec_sz > (int)sizeof(sdbuf))
break;
memcpy(sdbuf + off, &rid, 8); off += 8;
memcpy(sdbuf + off, &rts, 8); off += 8;
memcpy(sdbuf + off, &rdh, 8); off += 8;
memcpy(sdbuf + off, &rdl, 4); off += 4;
if (rdl > 0) {
memcpy(sdbuf + off, rd, rdl); off += rdl;
}
sdbuf[off++] = (uint8_t)rsl;
if (rsl > 0) {
memcpy(sdbuf + off, rsig, rsl); off += rsl;
}
rc++;
}
return;
sqlite3_finalize(stmt);
*rcp = rc;
sent += rc;
db_sync_send(si, src, sdbuf, off);
if (rc == 0)
break;
}
return;
}
// Prefix match — sync confirmed
if (memcmp(my_ch, pc, 32) == 0) {
if (my_dh == peer_dh) {
struct SI_PEER* sp = si_peer_find(si, src);
if (sp)
sp->synced = 2;
if (sp) {
sp->synced_pos = tp;
sp->sync_state = 2;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"chain_hash match with %016llx at %u",
"dh match with %016llx at %u",
(unsigned long long)src, tp);
return;
}
// Binary search for divergence point
uint32_t ds = 0, de = tp;
const uint8_t* spr = p + 37;
for (int i = 0; i < sc && spr + 36 <= p + len; i++) {
const uint8_t* spr = p + 13;
for (int i = 0; i < sc && spr + 12 <= p + len; i++) {
uint32_t pos = *(uint32_t*)spr;
const uint8_t* pch = spr + 4;
uint8_t mch[32];
if (db_chain_hash_at(si, pos, mch) == 0) {
if (memcmp(mch, pch, 32) == 0) {
uint64_t pdh;
memcpy(&pdh, spr + 4, 8);
uint64_t mdh;
if (db_datahash_at(si, pos, &mdh) == 0) {
if (mdh == pdh) {
if (pos + 1 > ds)
ds = pos + 1;
}
@@ -762,14 +842,13 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si,
de = pos;
}
}
spr += 36;
spr += 12;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"divergence with %016llx range [%u,%u]",
(unsigned long long)src, ds, de);
if (de - ds <= 1) {
// Small divergence — REFINE with 0 hashes
uint8_t ref[512];
uint32_t roff = 0;
ref[roff++] = DB_MSG_REFINE;
@@ -779,7 +858,6 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si,
db_sync_send(si, src, ref, roff);
}
else {
// Sparse REFINE
uint8_t ref[512];
uint32_t roff = 0;
ref[roff++] = DB_MSG_REFINE;
@@ -952,6 +1030,10 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si,
uint16_t count = *(uint16_t*)(p + 4);
const uint8_t* ptr = p + 6;
struct SI_PEER* sp = si_peer_find(si, src);
uint32_t fix_from = sp ? sp->synced_pos : 0;
uint16_t received = 0;
for (uint16_t i = 0; i < count; i++) {
uint64_t rid, rts, rdh;
uint32_t rdlen;
@@ -963,56 +1045,59 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si,
break;
int ret = db_record_insert(si, rid, rts, rdh,
(const char*)rdata, rdlen,
rsig, rsiglen);
rsig, rsiglen, 0);
if (ret >= 0)
received++;
if (ret == 0 && si->on_insert)
si->on_insert(si, (const char*)rdata, rdlen,
src, si->on_insert_arg);
}
uint32_t mc = db_count(si);
struct SI_PEER* sp = si_peer_find(si, src);
uint32_t pk = from + count;
db_cascade_from(si, fix_from);
// Request next batch if more records expected
if (mc > pk && sp && sp->synced == 1) {
if (sp && received > 0)
sp->synced_pos = from + received - 1;
uint32_t mc = db_count(si);
uint32_t pk = from + received;
if (mc > pk && sp && sp->sync_state == 1) {
uint32_t scnt = mc - pk;
if (scnt > DB_SEND_DATA_MAX)
scnt = DB_SEND_DATA_MAX;
si_send_data_batch(si, src, pk, scnt, 1);
}
// Send SYNC_DONE when we've received all
uint32_t nc = db_count(si);
uint8_t lch[32];
if (nc > 0 && db_chain_hash_at(si, nc - 1, lch) == 0) {
uint8_t sd[37];
sd[0] = DB_MSG_SYNC_DONE;
memcpy(sd + 1, &nc, 4);
memcpy(sd + 5, lch, 32);
db_sync_send(si, src, sd, 37);
}
uint64_t ldh = 0;
if (nc > 0)
db_datahash_at(si, nc - 1, &ldh);
uint8_t sd[13];
sd[0] = DB_MSG_SYNC_DONE;
memcpy(sd + 1, &nc, 4);
memcpy(sd + 5, &ldh, 8);
db_sync_send(si, src, sd, 13);
if (sp)
sp->synced = 2;
sp->sync_state = 2;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"received %u records from %016llx",
count, (unsigned long long)src);
received, (unsigned long long)src);
}
static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si,
uint64_t src, const uint8_t* p, size_t len)
{
if (len < 36)
if (len < 12)
return;
uint32_t pc = *(uint32_t*)p;
const uint8_t* pch = p + 4;
uint64_t pdh;
memcpy(&pdh, p + 4, 8);
uint32_t mc = db_count(si);
uint8_t mch[32];
uint64_t mdh = 0;
if (mc > 0)
db_chain_hash_at(si, mc - 1, mch);
else
memset(mch, 0, 32);
db_datahash_at(si, mc - 1, &mdh);
if (mc != pc || memcmp(mch, pch, 32) != 0) {
if (mc != pc || mdh != pdh) {
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC,
"SYNC_DONE mismatch %016llx my=%u peer=%u",
(unsigned long long)src, mc, pc);
@@ -1020,8 +1105,10 @@ static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si,
return;
}
struct SI_PEER* sp = si_peer_find(si, src);
if (sp)
sp->synced = 2;
if (sp) {
sp->synced_pos = (mc < pc ? mc : pc) > 0 ? (mc < pc ? mc : pc) - 1 : 0;
sp->sync_state = 2;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"sync confirmed with %016llx count=%u",
(unsigned long long)src, mc);
@@ -1075,8 +1162,13 @@ static void db_handle_push(struct DB_SYNC_INSTANCE* si, uint64_t src,
int ret = db_record_insert(si, rid, rts, rdh,
(const char*)rdata, rdlen,
rsig, rsiglen);
rsig, rsiglen, 0);
if (ret == 0) {
uint32_t ins_pos = si_find_pos(si, rts, rdh);
for (int j = 0; j < si->peer_count; j++) {
if (si->peers[j].synced_pos >= ins_pos)
si->peers[j].synced_pos = ins_pos > 0 ? ins_pos - 1 : 0;
}
if (si->on_insert)
si->on_insert(si, (const char*)rdata, rdlen,
src, si->on_insert_arg);
@@ -1206,8 +1298,8 @@ static void db_sync_on_conn_up(struct ETCP_CONN* conn, void* arg)
if (!si->enabled)
continue;
struct SI_PEER* p = si_peer_add(si, pid);
if (p && p->synced == 0 && conn->initialized && conn->links_up) {
p->synced = 1;
if (p && p->sync_state == 0 && conn->initialized && conn->links_up) {
p->sync_state = 1;
db_sync_initiate_sync(si, pid);
}
}
@@ -1231,13 +1323,13 @@ static void db_sync_on_conn_down(struct ETCP_CONN* conn, void* arg)
for (int i = 0; i < db->instance_count; i++) {
struct SI_PEER* p = si_peer_find(&db->instances[i], pid);
if (p)
p->synced = 0;
p->sync_state = 0;
}
int any = 0;
for (int i = 0; i < db->instance_count; i++) {
for (int j = 0; j < db->instances[i].peer_count; j++) {
if (db->instances[i].peers[j].synced >= 1) {
if (db->instances[i].peers[j].sync_state >= 1) {
any = 1;
goto cd_done;
}
@@ -1262,22 +1354,57 @@ static void db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si,
"initiate_sync to %016llx my=%u",
(unsigned long long)pid, mc);
uint8_t msg[37];
uint8_t msg[5];
msg[0] = DB_MSG_INIT_SYNC;
memcpy(msg + 1, &mc, 4);
if (mc > 0) {
if (db_chain_hash_at(si, mc - 1, msg + 5) != 0)
memset(msg + 5, 0, 32);
}
else {
memset(msg + 5, 0, 32);
}
db_sync_send(si, pid, msg, 37);
db_sync_send(si, pid, msg, 5);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"INIT_SYNC -> %016llx my=%u",
(unsigned long long)pid, mc);
}
static void db_verify_chain(struct DB_SYNC_INSTANCE* si)
{
uint32_t mc = db_count(si);
if (mc == 0)
return;
uint8_t prev_ch[32]; memset(prev_ch, 0, 32);
uint8_t exp_ch[32], stored_ch[32];
int bad_pos = -1;
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
"SELECT id,timestamp,datahash,chain_hash FROM \"%s\""
" ORDER BY timestamp,datahash") != SQLITE_OK)
return;
for (uint32_t pos = 0; sqlite3_step(stmt) == SQLITE_ROW; pos++) {
uint64_t rid = (uint64_t)sqlite3_column_int64(stmt, 0);
uint64_t rts = (uint64_t)sqlite3_column_int64(stmt, 1);
uint64_t rdh = (uint64_t)sqlite3_column_int64(stmt, 2);
const void* b = sqlite3_column_blob(stmt, 3);
if (b && sqlite3_column_bytes(stmt, 3) >= 32)
memcpy(stored_ch, b, 32);
else
memset(stored_ch, 0, 32);
db_chain_hash_compute(prev_ch, rid, rts, rdh, exp_ch);
if (memcmp(exp_ch, stored_ch, 32) != 0) {
bad_pos = (int)pos;
break;
}
memcpy(prev_ch, exp_ch, 32);
}
sqlite3_finalize(stmt);
if (bad_pos >= 0) {
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC,
"chain_hash mismatch at pos %d/%u in %s, recalculating",
bad_pos, mc, SI_TBL(si));
db_cascade_from(si, (uint32_t)bad_pos);
}
}
// ============================================================
// Timers
// ============================================================
@@ -1312,16 +1439,24 @@ static void db_sync_peer_check_cb(void* arg)
&& item->conn->links_up && item->conn->initialized)
{
uint64_t pid = item->conn->peer_node_id;
if (pid != db->inst->node_id) {
struct SI_PEER* p = si_peer_add(si, pid);
if (p && p->synced == 0) {
p->synced = 1;
db_sync_initiate_sync(si, pid);
}
}
if (pid != db->inst->node_id)
si_peer_add(si, pid);
}
e = e->next;
}
struct SI_PEER* best = NULL;
uint32_t min_pos = UINT32_MAX;
for (int j = 0; j < si->peer_count; j++) {
if (si->peers[j].sync_state == 0 && si->peers[j].synced_pos < min_pos) {
min_pos = si->peers[j].synced_pos;
best = &si->peers[j];
}
}
if (best) {
best->sync_state = 1;
db_sync_initiate_sync(si, best->node_id);
}
}
db->peer_check_timer =
@@ -1581,6 +1716,8 @@ struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst,
sqlite3_finalize(stmt);
}
db_verify_chain(si);
si->ttl_timer =
uasync_set_timeout(inst->ua,
DB_SYNC_TTL_INTERVAL * 10000u,
@@ -1597,8 +1734,8 @@ struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst,
&& ce->conn->links_up > 0 && ce->conn->initialized)
{
struct SI_PEER* p = si_peer_add(si, pid);
if (p && p->synced == 0) {
p->synced = 1;
if (p && p->sync_state == 0) {
p->sync_state = 1;
db_sync_initiate_sync(si, pid);
}
}
@@ -1680,7 +1817,7 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si,
uint64_t id = si->next_id;
int ret = db_record_insert(si, id, ts, dh,
json_data, len, sig, sig_len);
json_data, len, sig, sig_len, 1);
if (ret != 0)
return ret;
if (si->on_insert)
@@ -1707,7 +1844,7 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si,
// Push to all synced peers
for (int i = 0; i < si->peer_count; i++) {
if (si->peers[i].synced >= 1
if (si->peers[i].sync_state >= 1
&& si->peers[i].node_id != si->db_sync->inst->node_id)
{
if (db_sync_send(si, si->peers[i].node_id, pbuf, off) >= 0)
src/utun_instance.c
+4 -4
View File
@@ -412,7 +412,10 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
}
instance->etcp_sockets = NULL;
DEBUG_INFO(DEBUG_CATEGORY_MEMORY, "[INSTANCE_DESTROY] ETCP sockets cleanup complete");
// Cleanup db_sync before connections cleanup (db_sync_destroy iterates connections)
db_sync_destroy(instance);
// Cleanup ETCP connections (phase 1 detach + deferred phase 2 via call_soon)
{
struct ll_entry* entry = instance->connections->head;
@@ -469,9 +472,6 @@ 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);