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

Preserve history on verification errors and identify delivery ACKs

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 18:26:58 +01:00
1 parent 26432be0e4
commit fd53b443a0
2 files changed
+362 -313
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/db_sync.c
+357 -308
View File
@@ -37,8 +37,10 @@ static int db_sync_send(struct DB_SYNC_INSTANCE* si, uint64_t dst_node_id, cons
static int db_sync_send_gid(struct DB_SYNC* db, uint64_t node_id, uint64_t group_id, const uint8_t* payload, size_t len);
static int db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si, uint64_t peer_node_id);
static int si_send_data_batch(struct DB_SYNC_INSTANCE* si, uint64_t dst, uint32_t from, uint32_t count, uint32_t vp, uint32_t want_from);
static void si_delivery_update(struct DB_SYNC_INSTANCE* si, uint64_t ts, uint64_t author, uint64_t peer_id);
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t peer_id);
static int db_sync_peer_throttled(struct DB_SYNC* db, struct SI_PEER* p);
static int db_delete_bad_row(struct DB_SYNC_INSTANCE* si, sqlite3_int64 rowid);
static void db_history_deleted(struct DB_SYNC_INSTANCE* si);
// ============================================================
// Structures
@@ -66,6 +68,7 @@ struct DB_SYNC_INSTANCE {
struct {
uint32_t pos;
void* timer;
uint8_t pending; /* Работа сохраняется и при отказе SQL/таймера. */
} recalc;
struct SI_PEER* peers;
int peer_count, peer_capacity;
@@ -298,7 +301,7 @@ static uint64_t db_hash64(const uint8_t* data, size_t len)
}
// Вычисляет цепной хеш: SHA256(prev_chain_hash[32] || id[8] || timestamp[8] || author[8] || sig[64])
static void db_chain_hash_compute(const uint8_t prev_chain_hash[32],
static int db_chain_hash_compute(const uint8_t prev_chain_hash[32],
uint64_t id, uint64_t timestamp,
uint64_t author, const uint8_t author_signature[DB_SIG_SIZE],
uint8_t out[32])
@@ -309,7 +312,8 @@ static void db_chain_hash_compute(const uint8_t prev_chain_hash[32],
memcpy(buf + 40, &timestamp, 8);
memcpy(buf + 48, &author, 8);
memcpy(buf + 56, author_signature, DB_SIG_SIZE);
db_sha256(buf, sizeof(buf), out);
if (SHA256(buf, sizeof(buf), out)) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history chain SHA256 failed"); return -1;
}
// ============================================================
@@ -410,6 +414,15 @@ static struct SI_PEER* si_peer_add(struct DB_SYNC_INSTANCE* si, uint64_t node_id
return p;
}
/* Ошибка раунда не является подтверждением позиции или доставки. */
static void si_abort_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const char* reason) {
struct SI_PEER* peer = si_peer_find(si, src);
if (peer) { peer->sync_state = 0; peer->sync_start_tb = 0; }
db_sync_resume_peer_check(si->db_sync);
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history round aborted tbl=%s peer=%016llx reason=%s; retry scheduled",
SI_TBL(si), (unsigned long long)src, reason);
}
// ============================================================
// Data access
// ============================================================
@@ -424,6 +437,18 @@ static uint32_t db_count(struct DB_SYNC_INSTANCE* si)
return c;
}
/* Внутренний count не подменяет отказ SQL пустой историей. */
static int db_count_checked(struct DB_SYNC_INSTANCE* si, uint32_t* count) {
sqlite3_stmt* stmt = NULL;
int rc = si_prep(si, &stmt, "SELECT COUNT(*) FROM \"%s\"");
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
if (rc == SQLITE_ROW) *count = (uint32_t)sqlite3_column_int64(stmt, 0);
int final_rc = sqlite3_finalize(stmt);
if (rc == SQLITE_ROW && final_rc == SQLITE_OK) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history count tbl=%s rc=%d finalize=%d: %s",
SI_TBL(si), rc, final_rc, sqlite3_errmsg(SI_DB(si))); return -1;
}
// Читает цепной хеш (32 байта) записи на позиции pos в отсортированной таблице
static int db_chain_hash_at(struct DB_SYNC_INSTANCE* si, uint32_t pos, uint8_t out[32])
{
@@ -473,14 +498,19 @@ static int db_prev_chain_hash(struct DB_SYNC_INSTANCE* si, uint64_t ts, const ui
return -1; }
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_blob(stmt, 2, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
if (sqlite3_step(stmt) == SQLITE_ROW) {
int rc = sqlite3_step(stmt);
if (rc == SQLITE_ROW) {
const void* b = sqlite3_column_blob(stmt, 0);
if (b && sqlite3_column_bytes(stmt, 0) >= 32) memcpy(out, b, 32);
else memset(out, 0, 32);
} else {
if (b && sqlite3_column_bytes(stmt, 0) == 32) memcpy(out, b, 32);
else rc = SQLITE_CORRUPT;
} else if (rc == SQLITE_DONE) {
memset(out, 0, 32);
}
sqlite3_finalize(stmt);
if (rc != SQLITE_ROW && rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history previous hash tbl=%s rc=%d: %s",
SI_TBL(si), rc, sqlite3_errmsg(SI_DB(si))); return -1;
}
return 0;
}
@@ -502,94 +532,91 @@ static void db_recalc_tick(void* arg);
// Планирует асинхронный пересчёт цепных хешей начиная с позиции from_pos (батчами по 50)
static void db_sync_schedule_recalc(struct DB_SYNC_INSTANCE* si, uint32_t from_pos)
{
if (!si->recalc.timer) {
si->recalc.pos = from_pos;
si->recalc.timer = uasync_set_timeout(si->db_sync->inst->ua, 0, si, db_recalc_tick, "db_recalc");
return;
}
if (from_pos < si->recalc.pos)
si->recalc.pos = from_pos;
if (!si->recalc.pending || from_pos < si->recalc.pos) si->recalc.pos = from_pos;
si->recalc.pending = 1;
if (!si->recalc.timer)
si->recalc.timer = uasync_set_timeout(si->db_sync->inst->ua, 100, si, db_recalc_tick, "db_recalc");
if (!si->recalc.timer) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history recalc timer failed tbl=%s; work retained", SI_TBL(si));
}
// Обрабатывает один батч (до 50 записей) пересчёта цепных хешей, вызывается по таймеру
static void db_recalc_tick(void* arg)
{
struct DB_SYNC_INSTANCE* si = (struct DB_SYNC_INSTANCE*)arg;
/* Один атомарный батч. Только COMMIT продвигает позицию; ошибка оставляет работу pending. */
static int db_recalc_batch(struct DB_SYNC_INSTANCE* si) {
sqlite3* db = SI_DB(si);
uint32_t total = db_count(si);
if (si->recalc.pos >= total) { si->recalc.timer = NULL; return; }
uint8_t prev_ch[32];
if (si->recalc.pos > 0) {
if (db_chain_hash_at(si, si->recalc.pos - 1, prev_ch) != 0)
memset(prev_ch, 0, 32);
} else {
memset(prev_ch, 0, 32);
}
uint32_t batch = (total - si->recalc.pos > 50) ? 50 : total - si->recalc.pos;
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "recalc BEGIN: %s", sqlite3_errmsg(db)); return; }
sqlite3_stmt* sel, *upd;
if (si_prep(si, &sel,
"SELECT id,timestamp,node_id,author_signature FROM \"%s\""
" ORDER BY timestamp, author_signature"
" LIMIT -1 OFFSET ?") != SQLITE_OK)
{ DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "recalc SELECT prep: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return; }
sqlite3_bind_int64(sel, 1, (sqlite3_int64)si->recalc.pos);
if (si_prep(si, &upd,
"UPDATE \"%s\" SET chain_hash=?"
" WHERE timestamp=? AND author_signature=?") != SQLITE_OK)
{ DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "recalc UPDATE prep: %s", sqlite3_errmsg(db)); sqlite3_finalize(sel); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return; }
uint8_t ch[32]; uint32_t processed = 0;
while (sqlite3_step(sel) == SQLITE_ROW && processed < batch) {
sqlite3_stmt *sel = NULL, *upd = NULL;
const char* stage = "begin";
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) goto error;
uint32_t total;
stage = "count";
if (db_count_checked(si, &total) < 0) goto rollback;
uint8_t prev_ch[32] = {0};
stage = "previous hash";
if (si->recalc.pos && si->recalc.pos < total && db_chain_hash_at(si, si->recalc.pos - 1, prev_ch) < 0) goto rollback;
stage = "prepare";
if (si_prep(si, &sel, "SELECT id,timestamp,node_id,author_signature,rowid FROM \"%s\""
" ORDER BY timestamp,author_signature LIMIT 50 OFFSET ?") != SQLITE_OK) goto rollback;
if (sqlite3_bind_int64(sel, 1, si->recalc.pos) != SQLITE_OK) goto rollback;
if (si_prep(si, &upd, "UPDATE \"%s\" SET chain_hash=? WHERE timestamp=? AND author_signature=?") != SQLITE_OK) goto rollback;
uint32_t processed = 0;
int rc;
stage = "rows";
while ((rc = 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_auth = sqlite3_column_int64(sel, 2);
const void* sig_blob = sqlite3_column_blob(sel, 3);
uint8_t sig[DB_SIG_SIZE];
if (sig_blob) memcpy(sig, sig_blob, DB_SIG_SIZE); else memset(sig, 0, DB_SIG_SIZE);
db_chain_hash_compute(prev_ch, (uint64_t)n_id, (uint64_t)n_ts, (uint64_t)n_auth, sig, ch);
sqlite3_reset(upd);
sqlite3_bind_blob(upd, 1, ch, 32, SQLITE_STATIC);
sqlite3_bind_int64(upd, 2, n_ts);
sqlite3_bind_blob(upd, 3, sig, DB_SIG_SIZE, SQLITE_STATIC);
sqlite3_step(upd);
const uint8_t* sig = sqlite3_column_blob(sel, 3);
uint8_t ch[32];
if (!sig || sqlite3_column_bytes(sel, 3) != DB_SIG_SIZE) {
sqlite3_int64 rowid = sqlite3_column_int64(sel, 4);
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history recalc malformed stored signature tbl=%s ts=%lld bytes=%d",
SI_TBL(si), (long long)n_ts, sqlite3_column_bytes(sel, 3));
sqlite3_finalize(upd); upd = NULL; sqlite3_finalize(sel); sel = NULL;
stage = "delete malformed signature";
if (db_delete_bad_row(si, rowid) < 0 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
db_history_deleted(si); return 0;
}
if (db_chain_hash_compute(prev_ch, (uint64_t)n_id, (uint64_t)n_ts, (uint64_t)n_auth, sig, ch) < 0) goto rollback;
if (sqlite3_bind_blob(upd, 1, ch, 32, SQLITE_TRANSIENT) != SQLITE_OK ||
sqlite3_bind_int64(upd, 2, n_ts) != SQLITE_OK ||
sqlite3_bind_blob(upd, 3, sig, DB_SIG_SIZE, SQLITE_TRANSIENT) != SQLITE_OK ||
sqlite3_step(upd) != SQLITE_DONE || sqlite3_changes(db) != 1 || sqlite3_reset(upd) != SQLITE_OK) goto rollback;
memcpy(prev_ch, ch, 32);
processed++;
}
sqlite3_finalize(upd);
sqlite3_finalize(sel);
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "recalc COMMIT: %s", sqlite3_errmsg(db));
if (rc != SQLITE_DONE) goto rollback;
sqlite3_finalize(upd); upd = NULL; sqlite3_finalize(sel); sel = NULL;
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
si->recalc.pos += processed;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "recalc tick tbl=%s pos=%u batch=%u/%u total=%u",
SI_TBL(si), si->recalc.pos - processed, processed, batch, total);
if (si->recalc.pos >= total) {
si->recalc.timer = NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "recalc complete tbl=%s pos=%u(total)", SI_TBL(si), si->recalc.pos);
} else {
si->recalc.timer = uasync_set_timeout(si->db_sync->inst->ua, 100, si, db_recalc_tick, "db_recalc");
}
si->recalc.pending = si->recalc.pos < total;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history recalc committed tbl=%s pos=%u processed=%u total=%u pending=%u",
SI_TBL(si), si->recalc.pos, processed, total, si->recalc.pending);
return 0;
rollback:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history recalc tbl=%s stage=%s: %s; retaining position=%u",
SI_TBL(si), stage, sqlite3_errmsg(db), si->recalc.pos);
sqlite3_finalize(upd); sqlite3_finalize(sel);
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history recalc rollback tbl=%s: %s", SI_TBL(si), sqlite3_errmsg(db));
return -1;
error:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history recalc tbl=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(db)); return -1;
}
// Синхронно завершает все отложенные пересчёты хешей — вызывается перед операциями, требующими актуальных хешей
static void db_sync_flush_recalc(struct DB_SYNC_INSTANCE* si)
{
struct UASYNC* ua = si->db_sync->inst->ua;
while (si->recalc.timer) {
uasync_cancel_timeout(ua, si->recalc.timer);
si->recalc.timer = NULL;
db_recalc_tick(si);
static void db_recalc_tick(void* arg) {
struct DB_SYNC_INSTANCE* si = arg;
si->recalc.timer = NULL;
db_recalc_batch(si);
if (si->recalc.pending) db_sync_schedule_recalc(si, si->recalc.pos);
}
/* Перед публикацией хешей ошибка прекращает обмен, сохраняя отложенный пересчёт. */
static int db_sync_flush_recalc(struct DB_SYNC_INSTANCE* si) {
while (si->recalc.pending) {
if (db_recalc_batch(si) < 0) { db_sync_schedule_recalc(si, si->recalc.pos); return -1; }
}
if (si->recalc.timer) uasync_cancel_timeout(si->db_sync->inst->ua, si->recalc.timer);
si->recalc.timer = NULL;
return 0;
}
// ── Ed25519 verification ──
@@ -607,7 +634,9 @@ static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t o
}
}
if (db->inst->topo_sqlite_db) {
if (topo_node_sqlite_get_ed25519_pubkey(db->inst->topo_sqlite_db, node_id, out) == 0) return 0;
int rc = topo_node_sqlite_get_ed25519_pubkey(db->inst->topo_sqlite_db, node_id, out);
if (rc == 0) return 0;
if (rc == TOPO_SQLITE_LOOKUP_ERROR) return -1;
}
struct ETCP_CONN* conn = instance_find_conn(db->inst, node_id);
if (conn) {
@@ -615,36 +644,63 @@ static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t o
uint64_t chk; memcpy(&chk, out, 8);
if (chk != 0) return 0;
}
return -1;
return 1; /* Ключ ещё не получен; это не доказательство порчи. */
}
// Проверяет Ed25519 подпись автора: verify(pubkey, ts[8] || json, sig). Возвращает 0 если подпись верна
static int db_verify_author_sig(struct DB_SYNC_INSTANCE* si,
enum db_record_check { DB_CHECK_ERROR = -1, DB_CHECK_VALID, DB_CHECK_BAD_SIGNATURE, DB_CHECK_BAD_FORMAT, DB_CHECK_DEFERRED };
/* Проверяем конкретные байты. Только BAD_* доказывает их непригодность. */
static enum db_record_check db_verify_author_sig(struct DB_SYNC_INSTANCE* si,
uint64_t ts, uint64_t author_node_id,
const char* json, size_t jlen,
const uint8_t* sig)
const uint8_t* sig, size_t sig_len)
{
if (!sig || jlen > DB_SYNC_MESSAGE_MAX || (jlen && !json)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history: invalid signed body tbl=%s bytes=%zu", SI_TBL(si), jlen); return -1;
if (!sig || sig_len != DB_SIG_SIZE || !json || !jlen || jlen > DB_SYNC_MESSAGE_MAX) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed record tbl=%s ts=%llu body=%zu signature=%zu",
SI_TBL(si), (unsigned long long)ts, jlen, sig_len); return DB_CHECK_BAD_FORMAT;
}
uint8_t pubkey[32];
if (db_get_ed25519_pubkey(si->db_sync, author_node_id, pubkey) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC,
"cannot get Ed25519 pubkey for author=%016llx — discarding",
(unsigned long long)author_node_id);
return -1;
int key_rc = db_get_ed25519_pubkey(si->db_sync, author_node_id, pubkey);
if (key_rc != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history verification unavailable tbl=%s author=%016llx key_status=%d; preserving data",
SI_TBL(si), (unsigned long long)author_node_id, key_rc);
return key_rc < 0 ? DB_CHECK_ERROR : DB_CHECK_DEFERRED;
}
uint8_t msg[8192]; size_t off = 0;
memcpy(msg + off, &ts, 8); off += 8;
if (jlen > 0) { memcpy(msg + off, json, jlen); off += jlen; }
if (sc_ed25519_verify(pubkey, msg, off, sig) != SC_OK) {
uint64_t key_prefix, sig_prefix; memcpy(&key_prefix, pubkey, 8); memcpy(&sig_prefix, sig, 8);
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC,
"author_sig VERIFY FAIL author=%016llx ed_pubkey=%016llx... msg_len=%zu sig=%016llx... — discarding as forgery",
(unsigned long long)author_node_id, (unsigned long long)key_prefix, off, (unsigned long long)sig_prefix);
return -1;
int rc = sc_ed25519_verify(pubkey, msg, off, sig);
if (rc != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history signature check tbl=%s ts=%llu author=%016llx rc=%d result=%s",
SI_TBL(si), (unsigned long long)ts, (unsigned long long)author_node_id, rc,
rc == SC_ERR_AUTH_FAILED ? "mismatch" : "operational failure; preserving data");
return rc == SC_ERR_AUTH_FAILED ? DB_CHECK_BAD_SIGNATURE : DB_CHECK_ERROR;
}
return 0;
return DB_CHECK_VALID;
}
/* Вызывается после завершения SELECT и только внутри собственной транзакции. */
static int db_delete_bad_row(struct DB_SYNC_INSTANCE* si, sqlite3_int64 rowid) {
sqlite3_stmt* stmt = NULL;
int rc = si_prep(si, &stmt, "DELETE FROM \"%s\" WHERE rowid=?");
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 1, rowid);
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
if (rc == SQLITE_DONE && final_rc == SQLITE_OK && changed == 1) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history delete tbl=%s rowid=%lld rc=%d finalize=%d changed=%d: %s",
SI_TBL(si), (long long)rowid, rc, final_rc, changed, sqlite3_errmsg(SI_DB(si))); return -1;
}
/* Удаление сдвигает позиции: старый раунд больше не может подтверждать текущую историю. */
static void db_history_deleted(struct DB_SYNC_INSTANCE* si) {
db_sync_schedule_recalc(si, 0);
for (int i = 0; i < si->peer_count; i++) {
struct SI_PEER* p = &si->peers[i];
p->synced_pos = 0; p->verified_pos = 0; p->sync_state = 0; p->sync_start_tb = 0;
}
db_sync_resume_peer_check(si->db_sync);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history corrupt stored row deleted tbl=%s; rebuilding chain and restarting %d peers",
SI_TBL(si), si->peer_count);
}
// Вставляет запись в таблицу: проверяет подпись, дубликаты, вычисляет chain_hash, возвращает 0/1/-1/-2
@@ -656,72 +712,88 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
{
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "id=%llu ts=%llu author=%N len=%zu", (unsigned long long)id, (unsigned long long)ts, (unsigned long long)author_node_id, jlen);
sqlite3* db = SI_DB(si);
sqlite3_stmt* stmt;
int rc;
uint8_t message_id[32];
if (chat_message_id(si->group_id, ts, author_sig, message_id) < 0) return -1;
// Verify author signature (mandatory)
if (db_verify_author_sig(si, ts, author_node_id, json, jlen, author_sig) != 0) {
sqlite3_stmt* del;
if (si_prep(si, &del, "DELETE FROM \"%s\" WHERE timestamp=? AND author_signature=?") == SQLITE_OK) {
sqlite3_bind_int64(del, 1, (sqlite3_int64)ts);
sqlite3_bind_blob(del, 2, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
sqlite3_step(del);
int chg = sqlite3_changes(SI_DB(si));
sqlite3_finalize(del);
if (chg > 0) db_sync_schedule_recalc(si, si_find_pos(si, ts, author_sig));
}
return -2;
sqlite3_stmt* stmt = NULL;
enum db_record_check incoming = db_verify_author_sig(si, ts, author_node_id, json, jlen, author_sig, DB_SIG_SIZE);
if (incoming == DB_CHECK_ERROR || incoming == DB_CHECK_DEFERRED) return -1;
if (!author_sig) return -2;
const char* stage = "begin";
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history insert BEGIN tbl=%s: %s", SI_TBL(si), sqlite3_errmsg(db)); return -1;
}
int ret = incoming == DB_CHECK_VALID ? 0 : -2, deleted = 0, exists = 0;
stage = "read stored copy";
if (si_prep(si, &stmt, "SELECT rowid,node_id,data,author_signature FROM \"%s\" WHERE timestamp=? AND author_signature=?")
!= SQLITE_OK) goto rollback;
if (sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts) != SQLITE_OK ||
sqlite3_bind_blob(stmt, 2, author_sig, DB_SIG_SIZE, SQLITE_STATIC) != SQLITE_OK) goto rollback;
int rc = sqlite3_step(stmt);
sqlite3_int64 rowid = 0;
if (rc == SQLITE_ROW) {
exists = 1; rowid = sqlite3_column_int64(stmt, 0);
uint64_t author = (uint64_t)sqlite3_column_int64(stmt, 1);
const char* data = sqlite3_column_blob(stmt, 2);
size_t len = (size_t)sqlite3_column_bytes(stmt, 2);
const uint8_t* sig = sqlite3_column_blob(stmt, 3);
size_t sig_len = (size_t)sqlite3_column_bytes(stmt, 3);
enum db_record_check stored = db_verify_author_sig(si, ts, author, data, len, sig, sig_len);
if (stored == DB_CHECK_ERROR || stored == DB_CHECK_DEFERRED) goto rollback;
deleted = stored == DB_CHECK_BAD_SIGNATURE || stored == DB_CHECK_BAD_FORMAT;
} else if (rc != SQLITE_DONE) goto rollback;
if (sqlite3_finalize(stmt) != SQLITE_OK) { stmt = NULL; goto rollback; }
stmt = NULL;
if (deleted) {
stage = "delete stored corruption";
if (db_delete_bad_row(si, rowid) < 0) goto rollback;
}
if (incoming != DB_CHECK_VALID || (exists && !deleted)) {
if (incoming == DB_CHECK_VALID) ret = 1;
goto commit;
}
rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "insert BEGIN: %s", sqlite3_errmsg(db)); return -1; }
// Check if record already exists by (timestamp, author_signature)
if (si_prep(si, &stmt, "SELECT 1 FROM \"%s\" WHERE timestamp=? AND author_signature=?") != SQLITE_OK)
{ DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "db_record_insert SELECT prep: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return -1; }
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_blob(stmt, 2, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
int exists = (sqlite3_step(stmt) == SQLITE_ROW);
sqlite3_finalize(stmt);
if (exists) { DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "→ already exists, skip"); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return 1; }
// Compute chain hash
uint8_t prev_ch[32];
db_prev_chain_hash(si, ts, author_sig, prev_ch);
uint8_t ch[32];
db_chain_hash_compute(prev_ch, id, ts, author_node_id, author_sig, ch);
uint8_t prev_ch[32], ch[32], message_id[32];
stage = "identity and chain hash";
if (chat_message_id(si->group_id, ts, author_sig, message_id) < 0 ||
db_prev_chain_hash(si, ts, author_sig, prev_ch) < 0 ||
db_chain_hash_compute(prev_ch, id, ts, author_node_id, author_sig, ch) < 0) goto rollback;
// Insert new record
stage = "insert";
if (si_prep(si, &stmt,
"INSERT INTO \"%s\""
" (timestamp,node_id,id,chain_hash,flags,data,"
" author_signature,local_attrs,delivered_peers,delivery_chain,message_id)"
" VALUES (?,?,?,?,0,?,?,?,0,'',?)") != SQLITE_OK)
{ DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "db_record_insert INSERT prep: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return -1; }
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)author_node_id);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)id);
sqlite3_bind_blob(stmt, 4, ch, 32, SQLITE_STATIC);
if (jlen > 0) sqlite3_bind_blob(stmt, 5, json, (int)jlen, SQLITE_STATIC);
else sqlite3_bind_null(stmt, 5);
sqlite3_bind_blob(stmt, 6, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
if (local_attrs && local_attrs[0]) sqlite3_bind_text(stmt, 7, local_attrs, -1, SQLITE_STATIC);
else sqlite3_bind_text(stmt, 7, "", -1, SQLITE_STATIC);
sqlite3_bind_blob(stmt, 8, message_id, sizeof(message_id), SQLITE_STATIC);
rc = sqlite3_step(stmt);
goto rollback;
if (sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts) != SQLITE_OK ||
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)author_node_id) != SQLITE_OK ||
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)id) != SQLITE_OK ||
sqlite3_bind_blob(stmt, 4, ch, 32, SQLITE_TRANSIENT) != SQLITE_OK ||
sqlite3_bind_blob(stmt, 5, json, (int)jlen, SQLITE_STATIC) != SQLITE_OK ||
sqlite3_bind_blob(stmt, 6, author_sig, DB_SIG_SIZE, SQLITE_STATIC) != SQLITE_OK ||
sqlite3_bind_text(stmt, 7, local_attrs ? local_attrs : "", -1, SQLITE_STATIC) != SQLITE_OK ||
sqlite3_bind_blob(stmt, 8, message_id, sizeof(message_id), SQLITE_TRANSIENT) != SQLITE_OK ||
sqlite3_step(stmt) != SQLITE_DONE) goto rollback;
if (sqlite3_finalize(stmt) != SQLITE_OK) { stmt = NULL; goto rollback; }
stmt = NULL;
commit:
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
if (deleted) db_history_deleted(si);
if (ret == 0) {
if (id >= si->next_id) si->next_id = id + 1;
db_sync_schedule_recalc(si, si_find_pos(si, ts, author_sig));
}
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history insert tbl=%s ts=%llu incoming=%d stored=%d deleted=%d result=%d",
SI_TBL(si), (unsigned long long)ts, incoming, exists, deleted, ret);
return ret;
rollback:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history insert tbl=%s stage=%s: %s; rolling back",
SI_TBL(si), stage, sqlite3_errmsg(db));
sqlite3_finalize(stmt);
if (rc != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "INSERT: %s", sqlite3_errmsg(db)); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return -1; }
si->next_id = id + 1;
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "insert COMMIT: %s", sqlite3_errmsg(db)); return -1; }
uint32_t ins_pos = si_find_pos(si, ts, author_sig);
db_sync_schedule_recalc(si, ins_pos);
return 0;
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history insert rollback tbl=%s: %s", SI_TBL(si), sqlite3_errmsg(db));
return -1;
}
// Вставляет запись + каскад: on_insert + PUSH всем connected пирам (кроме источника)
@@ -734,9 +806,9 @@ static int db_record_insert_cascade(struct DB_SYNC_INSTANCE* si,
int ret = db_record_insert(si, rid, rts, rauthor, rdata, rdlen, rsig, local_attrs);
if (ret != 0) return ret;
uint8_t message_id[32];
if (chat_message_id(si->group_id, rts, rsig, message_id) < 0) return -1;
if (si->on_insert) {
uint8_t message_id[32];
if (chat_message_id(si->group_id, rts, rsig, message_id) < 0) return -1;
si->on_insert(si, message_id, rts, rdata, rdlen, rauthor, initial_sync, si->on_insert_arg);
}
@@ -756,7 +828,7 @@ static int db_record_insert_cascade(struct DB_SYNC_INSTANCE* si,
if (si->peers[i].sync_state >= 1 && pid != source_peer && pid != my_id) {
if (db_sync_peer_throttled(si->db_sync, &si->peers[i])) continue; /* спящий пир в «тихом» чате — не пушим */
if (db_sync_send(si, pid, pbuf, poff) >= 0)
{ si_delivery_update(si, rts, rauthor, pid); push_count++; }
{ push_count++; }
}
}
if (push_count > 0)
@@ -822,40 +894,28 @@ static int db_sync_send(struct DB_SYNC_INSTANCE* si, uint64_t dst_node_id, const
// Delivery chain helpers (local fields)
// ============================================================
// Обновляет delivery_chain (добавляет peer_id hex) и счётчик delivered_peers для записи после отправки пиру
static void si_delivery_update(struct DB_SYNC_INSTANCE* si, uint64_t ts, uint64_t author, uint64_t peer_id)
/* ACK конкретного сообщения: повтор не увеличивает счётчик, очередь отправки не считается доставкой. */
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t peer_id)
{
char hex[17]; snprintf(hex, sizeof(hex), "%016llx", (unsigned long long)peer_id);
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
"SELECT delivery_chain FROM \"%s\""
" WHERE timestamp=? AND node_id=?") == SQLITE_OK)
{
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)author);
if (sqlite3_step(stmt) == SQLITE_ROW) {
const char* dc = (const char*)sqlite3_column_text(stmt, 0);
if (dc && strstr(dc, hex)) { sqlite3_finalize(stmt); return; }
}
sqlite3_finalize(stmt);
}
if (si_prep(si, &stmt,
"UPDATE \"%s\""
" SET delivered_peers = delivered_peers + 1,"
" delivery_chain = CASE"
" WHEN delivery_chain = '' THEN ?1"
" ELSE delivery_chain || ',' || ?1"
" END"
" WHERE timestamp = ?2 AND author = ?3") == SQLITE_OK)
{
sqlite3_bind_text(stmt, 1, hex, -1, SQLITE_STATIC);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)author);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
sqlite3_stmt* stmt = NULL;
int rc = si_prep(si, &stmt,
"UPDATE \"%s\" SET flags=flags | CASE WHEN node_id=?3 THEN 1 ELSE 0 END,"
" delivered_peers=delivered_peers + (instr(',' || delivery_chain || ',', ',' || ?1 || ',')=0),"
" delivery_chain=CASE WHEN instr(',' || delivery_chain || ',', ',' || ?1 || ',')>0 THEN delivery_chain"
" WHEN delivery_chain='' THEN ?1 ELSE delivery_chain || ',' || ?1 END WHERE message_id=?2");
if (rc == SQLITE_OK) rc = sqlite3_bind_text(stmt, 1, hex, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(stmt, 2, message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 3, (sqlite3_int64)si->db_sync->inst->node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history ACK update tbl=%s peer=%s rc=%d finalize=%d: %s",
SI_TBL(si), hex, rc, final_rc, sqlite3_errmsg(SI_DB(si))); return -1;
}
if (!changed) DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history ACK unknown message tbl=%s peer=%s", SI_TBL(si), hex);
else DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK recorded tbl=%s peer=%s", SI_TBL(si), hex);
return 0;
}
// ============================================================
@@ -900,9 +960,10 @@ static void db_handle_error(struct DB_SYNC_INSTANCE* si, uint64_t src, const uin
static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 4) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "INIT_SYNC too short %zu from %016llx", len, (unsigned long long)src); return; }
db_sync_flush_recalc(si);
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild failed"); return; }
uint32_t pc; memcpy(&pc, p, 4);
uint32_t mc = db_count(si);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
uint32_t tp = (pc < mc ? pc : mc);
if (tp > 0) tp--;
uint64_t my_ch8_at_tp = 0;
@@ -953,7 +1014,7 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const
static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 14) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "INIT_RESP too short %zu from %016llx", len, (unsigned long long)src); return; }
db_sync_flush_recalc(si);
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild failed"); return; }
uint32_t tp; memcpy(&tp, p, 4);
uint64_t peer_ch8; memcpy(&peer_ch8, p + 4, 8);
uint8_t has_tail = p[12];
@@ -961,7 +1022,8 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const
const uint8_t* spr = p + 14;
uint64_t my_ch8 = 0;
uint32_t mc = db_count(si);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
if (mc > 0 && tp < mc) db_chain_hash8_at(si, tp, &my_ch8);
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← INIT_RESP: tp=%u my=%u my_ch8=%016llX peer_ch8=%016llX has_tail=%u sc=%d",
SI_SHRT(si), (unsigned long long)(src >> 16), tp, mc, my_ch8, peer_ch8, has_tail, sc);
@@ -1048,15 +1110,20 @@ static int si_send_data_batch(struct DB_SYNC_INSTANCE* si, uint64_t dst, uint32_
memcpy(buf + off, &want_from, 4); off += 4; // NEW: request peer's data from this position, DB_WANT_FROM_NONE=none
if (count > 0) {
sqlite3_stmt* stmt;
sqlite3_stmt* stmt = NULL;
sqlite3* db = SI_DB(si);
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history batch BEGIN tbl=%s: %s", SI_TBL(si), sqlite3_errmsg(db)); return -1;
}
sqlite3_int64 bad_rowid = 0;
int bad = 0, step_rc = SQLITE_DONE;
if (si_prep(si, &stmt,
"SELECT id,timestamp,node_id,data,author_signature"
"SELECT id,timestamp,node_id,data,author_signature,rowid"
" FROM \"%s\" ORDER BY timestamp, author_signature"
" LIMIT ? OFFSET ?") != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] SEND_DATA: si_prep failed tbl=%s", SI_SHRT(si), (unsigned long long)(dst >> 16), SI_TBL(si)); return -1; }
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)count);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)from);
int bad_del = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
" LIMIT ? OFFSET ?") != SQLITE_OK) goto batch_error;
if (sqlite3_bind_int64(stmt, 1, (sqlite3_int64)count) != SQLITE_OK ||
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)from) != SQLITE_OK) goto batch_error;
while ((step_rc = 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 rauth = (uint64_t)sqlite3_column_int64(stmt, 2);
@@ -1064,16 +1131,10 @@ static int si_send_data_batch(struct DB_SYNC_INSTANCE* si, uint64_t dst, uint32_
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;
if (rsl == DB_SIG_SIZE && db_verify_author_sig(si, rts, rauth, (const char*)rd, rdl, rsig) != 0) {
uint32_t del_pos = si_find_pos(si, rts, rsig);
sqlite3_stmt* del;
if (si_prep(si, &del, "DELETE FROM \"%s\" WHERE timestamp=? AND author_signature=?") == SQLITE_OK) {
sqlite3_bind_int64(del, 1, (sqlite3_int64)rts);
sqlite3_bind_blob(del, 2, rsig, DB_SIG_SIZE, SQLITE_STATIC);
sqlite3_step(del); sqlite3_finalize(del);
db_sync_schedule_recalc(si, del_pos); bad_del++;
}
continue;
enum db_record_check check = db_verify_author_sig(si, rts, rauth, (const char*)rd, rdl, rsig, (size_t)rsl);
if (check == DB_CHECK_ERROR || check == DB_CHECK_DEFERRED) goto batch_error;
if (check != DB_CHECK_VALID) {
bad = 1; bad_rowid = sqlite3_column_int64(stmt, 5); break;
}
int rec_sz = 28 + rdl + 1 + rsl;
if (off + rec_sz > 8000) break;
@@ -1086,12 +1147,28 @@ static int si_send_data_batch(struct DB_SYNC_INSTANCE* si, uint64_t dst, uint32_
if (rsl > 0) { memcpy(buf + off, rsig, rsl); off += rsl; }
rc++;
}
if (step_rc != SQLITE_ROW && step_rc != SQLITE_DONE) goto batch_error;
if (sqlite3_finalize(stmt) != SQLITE_OK) { stmt = NULL; goto batch_error; }
stmt = NULL;
if (bad && db_delete_bad_row(si, bad_rowid) < 0) goto batch_error;
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto batch_error;
if (bad) {
db_history_deleted(si);
return -1; /* Позиции изменились: не отправляем пачку старого раунда. */
}
goto batch_ready;
batch_error:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history batch aborted tbl=%s from=%u: %s; preserving data",
SI_TBL(si), from, sqlite3_errmsg(db));
sqlite3_finalize(stmt);
if (rc == 0 && bad_del == 0)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] SEND_DATA: 0 records from=%u count=%u",
SI_SHRT(si), (unsigned long long)(dst >> 16), from, count);
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history batch rollback tbl=%s: %s", SI_TBL(si), sqlite3_errmsg(db));
struct SI_PEER* peer = si_peer_find(si, dst);
if (peer) { peer->sync_state = 0; peer->sync_start_tb = 0; }
db_sync_resume_peer_check(si->db_sync);
return -1;
}
batch_ready:
memcpy(rcp, &rc, 2);
if (db_sync_send(si, dst, buf, off) != 0) return -1;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] → SEND_DATA: %u recs from=%u want=%u vp=%u",
@@ -1108,25 +1185,25 @@ static int si_parse_record(const uint8_t** pp, const uint8_t* end,
uint32_t* rdlen, const uint8_t** rdata,
const uint8_t** rsig, int* rsiglen)
{
if (*pp + 28 > end) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: truncated header, need 28 have %td", end - *pp); return -1; }
if (end - *pp < 28) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: truncated header, need 28 have %td", end - *pp); return -1; }
memcpy(rid, *pp, 8); *pp += 8;
memcpy(rts, *pp, 8); *pp += 8;
memcpy(rauthor, *pp, 8); *pp += 8;
memcpy(rdlen, *pp, 4); *pp += 4;
if (*pp + *rdlen > end) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: data overrun dlen=%u have=%td", *rdlen, end - *pp); return -1; }
if (!*rdlen || *rdlen > DB_SYNC_MESSAGE_MAX || *rdlen > (size_t)(end - *pp)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: invalid body dlen=%u have=%td", *rdlen, end - *pp); return -1;
}
*rdata = (*rdlen > 0) ? *pp : NULL;
*pp += *rdlen;
if (*pp + 1 > end) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: sig_len overrun"); return -1; }
if (*pp == end) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: sig_len overrun"); return -1; }
*rsiglen = (int)(*(*pp)++);
if (*rsiglen > 0) {
if (*pp + *rsiglen > end) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: sig data overrun slen=%d have=%td", *rsiglen, end - *pp); return -1; }
*rsig = *pp;
*pp += *rsiglen;
} else {
*rsig = NULL;
if (*rsiglen != DB_SIG_SIZE || *rsiglen > end - *pp) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "si_parse_record: invalid signature slen=%d have=%td", *rsiglen, end - *pp); return -1;
}
*rsig = *pp;
*pp += *rsiglen;
return 0;
}
@@ -1136,24 +1213,31 @@ static int si_parse_record(const uint8_t** pp, const uint8_t* end,
static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 14) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← SEND_DATA: truncated len=%zu (need >=14)", SI_SHRT(si), (unsigned long long)(src >> 16), len); return; }
db_sync_flush_recalc(si);
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild failed"); return; }
uint32_t from; memcpy(&from, p, 4);
uint16_t count; memcpy(&count, p + 4, 2);
uint32_t vp; memcpy(&vp, p + 6, 4);
uint32_t want_from; memcpy(&want_from, p + 10, 4);
if (count > DB_SEND_DATA_MAX) { si_abort_sync(si, src, "invalid batch count"); return; }
struct SI_PEER* sp = si_peer_find(si, src);
const uint8_t* ptr = p + 14;
uint16_t received = 0, duplicates = 0, bad_sigs = 0;
for (uint16_t i = 0; i < count && i < 32; i++) {
uint64_t rid, rts, rauthor; uint32_t rdlen; const uint8_t* rdata, *rsig; int rsiglen;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) break;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) {
si_abort_sync(si, src, "malformed incoming record"); return;
}
int ret = db_record_insert_cascade(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, src, NULL, 1);
if (ret < 0) { si_abort_sync(si, src, "incoming verification or insertion failed"); return; }
if (ret >= 0) received++;
if (ret == 1) duplicates++;
if (ret == -2) bad_sigs++;
}
if (ptr != p + len) { si_abort_sync(si, src, "trailing batch bytes"); return; }
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild after insertion failed"); return; }
uint32_t old_spos = sp ? sp->synced_pos : 0;
if (received > 0) {
@@ -1172,12 +1256,13 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
// Respond with our data if peer requested (want_from != DB_WANT_FROM_NONE)
if (want_from != DB_WANT_FROM_NONE) {
uint32_t mc = db_count(si);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
if (want_from < mc) {
uint32_t scnt = mc - want_from; if (scnt > DB_SEND_DATA_MAX) scnt = DB_SEND_DATA_MAX;
si_send_data_batch(si, src, want_from, scnt, vp, DB_WANT_FROM_NONE);
if (si_send_data_batch(si, src, want_from, scnt, vp, DB_WANT_FROM_NONE) < 0) return;
} else {
si_send_data_batch(si, src, want_from, 0, vp, DB_WANT_FROM_NONE);
if (si_send_data_batch(si, src, want_from, 0, vp, DB_WANT_FROM_NONE) < 0) return;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← DATA: responded with %u recs from=%u",
SI_SHRT(si), (unsigned long long)(src >> 16), want_from < mc ? (mc - want_from > DB_SEND_DATA_MAX ? DB_SEND_DATA_MAX : mc - want_from) : 0, want_from);
@@ -1185,7 +1270,8 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
// If slave responded with want=DB_WANT_FROM_NONE and we are in active sync — complete the exchange with SYNC_DONE
if (want_from == DB_WANT_FROM_NONE && sp && sp->sync_state == 1) {
uint32_t mc = db_count(si);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
uint64_t mch8 = 0; if (mc > 0) db_chain_hash8_at(si, mc - 1, &mch8);
uint8_t sd[13]; sd[0] = DB_MSG_SYNC_DONE;
memcpy(sd + 1, &mc, 4); memcpy(sd + 5, &mch8, 8);
@@ -1202,10 +1288,11 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 12) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "SYNC_DONE too short %zu from %016llx", len, (unsigned long long)src); return; }
db_sync_flush_recalc(si);
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild failed"); return; }
uint32_t pc; memcpy(&pc, p, 4);
uint64_t pch8; memcpy(&pch8, p + 4, 8);
uint32_t mc = db_count(si);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
uint64_t mch8 = 0; if (mc > 0) db_chain_hash8_at(si, mc - 1, &mch8);
struct SI_PEER* sp = si_peer_find(si, src);
@@ -1244,28 +1331,10 @@ static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si, uint64_t src, const
// Обрабатывает ACK_PUSH: помечает запись флагом DB_REC_FLAG_WAS_SENT и обновляет delivery_chain
static void db_handle_ack_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 16) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← PUSH ACK: truncated len=%zu (need >=16)", SI_SHRT(si), (unsigned long long)(src >> 16), len); return; }
uint64_t ts; memcpy(&ts, p, 8);
uint64_t author; memcpy(&author, p + 8, 8);
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "from=%016llx ts=%llu author=%016llx", (unsigned long long)src, (unsigned long long)ts, (unsigned long long)author);
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
"UPDATE \"%s\" SET flags = flags | 1"
" WHERE timestamp=? AND node_id=?"
" AND node_id=? AND (flags & 1) = 0") == SQLITE_OK)
{
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)author);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)si->db_sync->inst->node_id);
sqlite3_step(stmt);
int changed = sqlite3_changes(SI_DB(si));
sqlite3_finalize(stmt);
if (changed > 0)
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← PUSH ACK: ts=%llu — delivery confirmed by peer",
SI_SHRT(si), (unsigned long long)(src >> 16), (unsigned long long)ts);
if (len != 32) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history ACK invalid tbl=%s bytes=%zu expected=32", SI_TBL(si), len); return;
}
si_delivery_update(si, ts, author, src);
si_delivery_update(si, p, src);
}
// Обрабатывает PUSH: вставляет новую запись от пира, шлёт ACK_PUSH, корректирует synced_pos остальных пиров
@@ -1278,10 +1347,16 @@ static void db_handle_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint
uint32_t rdlen;
const uint8_t* rdata, *rsig;
int rsiglen;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) return;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0 || ptr != p + len) {
si_abort_sync(si, src, "malformed PUSH"); return;
}
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "from=%016llx id=%llu author=%016llx ts=%llu len=%u", (unsigned long long)src, (unsigned long long)rid, (unsigned long long)rauthor, (unsigned long long)rts, rdlen);
int ret = db_record_insert_cascade(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, src, NULL, 0);
if (ret < 0) { si_abort_sync(si, src, "PUSH verification or insertion failed"); return; }
uint8_t ack[33]; ack[0] = DB_MSG_ACK_PUSH;
if (chat_message_id(si->group_id, rts, rsig, ack + 1) < 0) return;
if (db_sync_send(si, src, ack, sizeof(ack)) != 0) return;
if (ret == 0) {
uint32_t ins_pos = si_find_pos(si, rts, rsig);
uint32_t total = db_count(si);
@@ -1289,11 +1364,6 @@ static void db_handle_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint
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; adj_count++; }
}
uint8_t ack[17];
ack[0] = DB_MSG_ACK_PUSH;
memcpy(ack + 1, &rts, 8);
memcpy(ack + 9, &rauthor, 8);
db_sync_send(si, src, ack, 17);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← PUSH: msg id=%llu author=%016llX ts=%llu → inserted at pos=%u (of %u total)",
SI_SHRT(si), (unsigned long long)(src >> 16),
(unsigned long long)rid, (unsigned long long)rauthor, (unsigned long long)rts, ins_pos, total);
@@ -1421,7 +1491,8 @@ static int db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si, uint64_t pid)
(unsigned long long)si->group_id, (unsigned long long)pid);
return 0;
}
uint32_t mc = db_count(si);
uint32_t mc;
if (db_sync_flush_recalc(si) < 0 || db_count_checked(si, &mc) < 0) { si_abort_sync(si, pid, "local history unavailable"); return 0; }
uint64_t my_id = si->db_sync->inst->node_id;
struct SI_PEER* p = si_peer_find(si, pid);
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history start without peer group=%016llx", si->group_id); return 0; }
@@ -1451,40 +1522,10 @@ static int db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si, uint64_t pid)
// Проверяет целостность цепочки хешей всей таблицы, при несовпадении запускает пересчёт с позиции ошибки
static void db_verify_chain(struct DB_SYNC_INSTANCE* si)
{
uint32_t mc = db_count(si);
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "tbl=%s mc=%u", SI_TBL(si), mc);
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,node_id,author_signature,chain_hash FROM \"%s\""
" ORDER BY timestamp, author_signature") != 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 rauth = (uint64_t)sqlite3_column_int64(stmt, 2);
const void* sig_blob = sqlite3_column_blob(stmt, 3);
uint8_t sig[DB_SIG_SIZE];
if (sig_blob) memcpy(sig, sig_blob, DB_SIG_SIZE); else memset(sig, 0, DB_SIG_SIZE);
const void* b = sqlite3_column_blob(stmt, 4);
if (b && sqlite3_column_bytes(stmt, 4) >= 32) memcpy(stored_ch, b, 32);
else memset(stored_ch, 0, 32);
db_chain_hash_compute(prev_ch, rid, rts, rauth, sig, 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_CHAT_SYNC, "chain_hash mismatch at pos %d/%u in %s, recalculating",
bad_pos, mc, SI_TBL(si));
db_sync_schedule_recalc(si, (uint32_t)bad_pos);
int rc = db_sync_chain_verify(si);
if (rc == 1) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history chain mismatch tbl=%s; rebuilding", SI_TBL(si));
db_sync_schedule_recalc(si, 0);
}
}
@@ -1492,9 +1533,7 @@ static void db_verify_chain(struct DB_SYNC_INSTANCE* si)
int db_sync_chain_verify(struct DB_SYNC_INSTANCE* si)
{
if (!si || !si->enabled) return 0;
db_sync_flush_recalc(si);
uint32_t mc = db_count(si);
if (mc == 0) return 0;
if (db_sync_flush_recalc(si) < 0) return -1;
uint8_t prev_ch[32]; memset(prev_ch, 0, 32);
uint8_t exp_ch[32], stored_ch[32];
@@ -1506,22 +1545,30 @@ int db_sync_chain_verify(struct DB_SYNC_INSTANCE* si)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "verify_chain_hash: si_prep failed tbl=%s", SI_TBL(si));
return -1;
}
for (uint32_t pos = 0; sqlite3_step(stmt) == SQLITE_ROW; pos++) {
int rc;
for (uint32_t pos = 0; (rc = 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 rauth = (uint64_t)sqlite3_column_int64(stmt, 2);
const void* sig_blob = sqlite3_column_blob(stmt, 3);
uint8_t sig[DB_SIG_SIZE];
if (sig_blob) memcpy(sig, sig_blob, DB_SIG_SIZE); else memset(sig, 0, DB_SIG_SIZE);
const void* b = sqlite3_column_blob(stmt, 4);
if (b && sqlite3_column_bytes(stmt, 4) >= 32) memcpy(stored_ch, b, 32);
else memset(stored_ch, 0, 32);
db_chain_hash_compute(prev_ch, rid, rts, rauth, sig, exp_ch);
if (memcmp(exp_ch, stored_ch, 32) != 0) { sqlite3_finalize(stmt); return 1; }
if (!sig_blob || sqlite3_column_bytes(stmt, 3) != DB_SIG_SIZE || !b || sqlite3_column_bytes(stmt, 4) != 32) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history chain malformed fields tbl=%s pos=%u", SI_TBL(si), pos);
sqlite3_finalize(stmt); return 1;
}
memcpy(stored_ch, b, 32);
if (db_chain_hash_compute(prev_ch, rid, rts, rauth, sig_blob, exp_ch) < 0) { sqlite3_finalize(stmt); return -1; }
if (memcmp(exp_ch, stored_ch, 32) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history chain mismatch tbl=%s pos=%u", SI_TBL(si), pos);
sqlite3_finalize(stmt); return 1;
}
memcpy(prev_ch, exp_ch, 32);
}
sqlite3_finalize(stmt);
int final_rc = sqlite3_finalize(stmt);
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history chain verify tbl=%s rc=%d finalize=%d: %s",
SI_TBL(si), rc, final_rc, sqlite3_errmsg(SI_DB(si))); return -1;
}
return 0;
}
@@ -1588,8 +1635,9 @@ void db_sync_reinitiate(struct DB_SYNC_INSTANCE* si, uint64_t node_id)
int db_sync_last_chain_hash8(struct DB_SYNC_INSTANCE* si, uint64_t* out_hash8)
{
if (!si || !out_hash8) return -1;
db_sync_flush_recalc(si);
uint32_t mc = db_count(si);
if (db_sync_flush_recalc(si) < 0) return -1;
uint32_t mc;
if (db_count_checked(si, &mc) < 0) return -1;
if (mc == 0) { *out_hash8 = 0; return 0; }
return db_chain_hash8_at(si, mc - 1, out_hash8);
}
@@ -1886,6 +1934,7 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, si
{
if (!si || !si->enabled || !json_data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "insert invalid args"); return -1; }
if (!sig || sig_len != DB_SIG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "insert_signed: signature required (64 bytes Ed25519)"); return -1; }
if (len > UINT32_MAX) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history insert body too large bytes=%zu", len); return -2; }
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "tbl=%s id=%llu ts=%llu len=%zu", SI_TBL(si), (unsigned long long)si->next_id, (unsigned long long)ts, len);
uint64_t id = si->next_id;
src/chat/db_sync.h
+5 -5
View File
@@ -40,6 +40,7 @@ struct DB_SYNC_INSTANCE;
// Record: [id:8][ts:8][node_id:8][dlen:4][data][sig_len:1=64][sig:64]
// SYNC_DONE wire: [count:4][chain_hash8:8]
// ACK_PUSH wire: [message_id:32]. Подтверждает конкретную запись; повторы идемпотентны.
// Error codes for DB_MSG_ERROR
#define DB_ERR_NOT_FOUND 0x01 // instance not found
@@ -82,10 +83,9 @@ uint64_t db_sync_instance_group_id(struct DB_SYNC_INSTANCE* si);
// db_sync_insert_signed: sig MUST be non-NULL, 64 bytes.
// sig = Ed25519(ts[8] || json); ts записан в порядке байтов хоста.
// ts — call db_sync_next_timestamp(si) before signing to reserve monotonically increasing timestamp.
// Возвращает 0 (вставлено), 1 (повтор), -1 (ошибка), -2 (отказ проверки).
// Известный дефект: операционный отказ проверки либо битый входящий дубликат могут удалить
// корректную сохранённую строку с тем же ключом. Удалять допустимо только доказанно битые
// сохранённые данные (формат/подпись); см. doc/member_merkle_audit_findings.txt, раздел10.
// Возвращает 0 (вставлено), 1 (повтор), -1 (операционная ошибка/ключ ещё неизвестен), -2 (неверные данные).
// Корректная сохранённая копия переживает неверный входящий повтор. Удаляем только строку,
// чьи собственные байты имеют битый формат либо подпись не совпала при завершённой проверке.
int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, size_t len,
const uint8_t* sig, size_t sig_len, uint64_t ts,
const char* local_attrs);
@@ -115,7 +115,7 @@ typedef void (*db_sync_insert_cb)(struct DB_SYNC_INSTANCE* si,
uint64_t author_node_id, int initial_sync, void* arg);
void db_sync_set_insert_cb(struct DB_SYNC_INSTANCE* si, db_sync_insert_cb cb, void* arg);
// Verify chain integrity: returns 0 if all chain_hashes are correct, 1 if any mismatch found
// Проверить цепочку: 0 — корректна, 1 — несовпадение/битые поля, -1 — операционная ошибка.
int db_sync_chain_verify(struct DB_SYNC_INSTANCE* si);
// Peer state control (for testing — disable PUSH to isolate sync protocol)