diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index 322c2151..0ee29c24 100644 --- a/src/chat/db_sync.c +++ b/src/chat/db_sync.c @@ -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, ×tamp, 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; diff --git a/src/chat/db_sync.h b/src/chat/db_sync.h index c0d3f228..ca953f13 100644 --- a/src/chat/db_sync.h +++ b/src/chat/db_sync.h @@ -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)