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

Replicate group message revisions atomically with versioned acknowledgements

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 22:35:34 +01:00
1 parent 5c638a60db
commit 809f009796
2 files changed
+190 -136
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/db_sync.c
+170 -127
View File
@@ -20,7 +20,7 @@
#include <string.h>
#define DB_BATCH_RECORDS 16
#define DB_WIRE_FIXED (32 + 8 + 8 + 4 + DB_SIG_SIZE)
#define DB_WIRE_FIXED (32 + 8 + 8 + 16 + 8 + 8 + 4 + DB_SIG_SIZE)
#define DB_WIRE_MAX (DB_WIRE_FIXED + DB_SYNC_MESSAGE_MAX)
#define SI_DB(si) ((si)->db_sync->db)
#define SI_TBL(si) ((si)->table_name)
@@ -80,8 +80,9 @@ struct db_record {
uint64_t id, timestamp, author;
size_t data_len, sig_len;
uint8_t signature[64], message_id[32];
struct db_message_revision revision;
const char* data;
int cache_valid;
int cache_valid, updated;
};
struct db_checked {
struct db_record record;
@@ -105,6 +106,42 @@ static void db_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8
static uint32_t db_read32(const uint8_t* p) { uint32_t n; memcpy(&n, p, 4); return ntohl(n); }
static void db_write32(uint8_t* p, uint32_t n) { n = htonl(n); memcpy(p, &n, 4); }
/* Канонические подписанные байты: домен, канал, постоянный ID, время, автор, nonce, версия, правка, длина и тело. */
static int db_signed_bytes(uint64_t group, const struct db_record* r, uint8_t* out) {
static const char domain[] = "utun-chat-record-v2";
if (!r->data || !r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || !r->timestamp || r->timestamp > INT64_MAX ||
!r->author || !r->revision.ver || r->revision.ver > INT64_MAX || r->revision.edited_at > INT64_MAX ||
(r->revision.ver == 1 ? r->revision.edited_at != 0 : r->revision.edited_at <= r->timestamp)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid envelope group=%llu ts=%llu ver=%llu edited=%llu bytes=%zu",
(unsigned long long)group, (unsigned long long)r->timestamp, (unsigned long long)r->revision.ver,
(unsigned long long)r->revision.edited_at, r->data_len); return -1;
}
size_t off = sizeof(domain) - 1; memcpy(out, domain, off); db_write64(out + off, group); off += 8;
memcpy(out + off, r->message_id, 32); off += 32;
db_write64(out + off, r->timestamp); off += 8; db_write64(out + off, r->author); off += 8;
memcpy(out + off, r->revision.nonce, 16); off += 16;
db_write64(out + off, r->revision.ver); off += 8; db_write64(out + off, r->revision.edited_at); off += 8;
db_write32(out + off, (uint32_t)r->data_len); off += 4; memcpy(out + off, r->data, r->data_len);
return (int)(off + r->data_len);
}
int db_sync_sign_message(uint64_t group, uint64_t timestamp, uint64_t author,
const struct db_message_revision* revision, const char* data, size_t len,
const uint8_t private_key[32], uint8_t signature[64]) {
if (!revision || !private_key || !signature) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid signing arguments"); return -1;
}
struct db_record r = {.timestamp = timestamp, .author = author, .revision = *revision, .data = data, .data_len = len};
uint8_t body[DB_WIRE_MAX + 64];
if (chat_message_id(group, timestamp, author, revision->nonce, r.message_id) < 0) return -1;
int bytes = db_signed_bytes(group, &r, body);
if (bytes < 0 || sc_ed25519_sign(private_key, body, (size_t)bytes, signature) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history signing failed group=%llu ver=%llu",
(unsigned long long)group, (unsigned long long)revision->ver); return -1;
}
return 0;
}
static int db_error(struct DB_SYNC_INSTANCE* si, const char* stage) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history tbl=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(SI_DB(si)));
return -1;
@@ -203,29 +240,25 @@ static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t o
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, size_t sig_len)
static enum db_record_check db_verify_author_sig(struct DB_SYNC_INSTANCE* si, const struct db_record* r)
{
if (!sig || sig_len != DB_SIG_SIZE || !json || !jlen || jlen > DB_SYNC_MESSAGE_MAX) {
uint8_t msg[DB_WIRE_MAX + 64];
int bytes = db_signed_bytes(si->group_id, r, msg);
if (r->sig_len != DB_SIG_SIZE || bytes < 0) {
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;
SI_TBL(si), (unsigned long long)r->timestamp, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
}
uint8_t pubkey[32];
int key_rc = db_get_ed25519_pubkey(si->db_sync, author_node_id, pubkey);
int key_rc = db_get_ed25519_pubkey(si->db_sync, r->author, 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);
SI_TBL(si), (unsigned long long)r->author, 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; }
int rc = sc_ed25519_verify(pubkey, msg, off, sig);
int rc = sc_ed25519_verify(pubkey, msg, (size_t)bytes, r->signature);
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,
SI_TBL(si), (unsigned long long)r->timestamp, (unsigned long long)r->author, rc,
rc == SC_ERR_AUTH_FAILED ? "mismatch" : "operational failure; preserving data");
return rc == SC_ERR_AUTH_FAILED ? DB_CHECK_BAD_SIGNATURE : DB_CHECK_ERROR;
}
@@ -244,107 +277,66 @@ static int db_delete_bad_row(struct DB_SYNC_INSTANCE* si, sqlite3_int64 rowid) {
SI_TBL(si), (long long)rowid, rc, final_rc, changed, sqlite3_errmsg(SI_DB(si))); return -1;
}
// Одна схема для новых таблиц и однократного преобразования сохранённой истории.
// Формат v2: одна строка на постоянный ID, подпись и версия заменяются атомарно.
static int si_table_create(struct DB_SYNC_INSTANCE* si, const char* name, int if_not_exists) {
char sql[768];
char sql[1024];
snprintf(sql, sizeof(sql),
"CREATE TABLE %s\"%s\" ("
" timestamp INTEGER NOT NULL, node_id INTEGER NOT NULL, id INTEGER NOT NULL,"
" flags INTEGER NOT NULL DEFAULT 0, data BLOB,"
" author_signature BLOB NOT NULL, message_id BLOB NOT NULL UNIQUE CHECK(length(message_id)=32),"
" bucket INTEGER NOT NULL, local_attrs TEXT DEFAULT '', delivered_peers INTEGER NOT NULL DEFAULT 0,"
" delivery_chain TEXT NOT NULL DEFAULT '', PRIMARY KEY(timestamp,author_signature))",
" delivery_chain TEXT NOT NULL DEFAULT '', nonce BLOB NOT NULL CHECK(length(nonce)=16),"
" ver INTEGER NOT NULL CHECK(ver>0), edited_at INTEGER NOT NULL CHECK(edited_at>=0))",
if_not_exists ? "IF NOT EXISTS " : "", name);
if (sqlite3_exec(SI_DB(si), sql, NULL, NULL, NULL) == SQLITE_OK) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: create table=%s: %s", name, sqlite3_errmsg(SI_DB(si))); return -1;
}
// Перестройка атомарна: сохраняем подписи, локальные атрибуты и доставку без изменений.
/* Dev-формат без переноса: удаляем только старую групповую историю и зависимые feedback.
* PM, участники, ключи и прочие таблицы общей БД сохраняются. DDL и очистка индекса атомарны. */
static int si_table_init(struct DB_SYNC_INSTANCE* si) {
sqlite3* db = SI_DB(si);
sqlite3* db = SI_DB(si); sqlite3_stmt* st = NULL;
if (si_table_create(si, SI_TBL(si), 1) < 0) return -1;
sqlite3_stmt *rows = NULL, *insert = NULL;
const char* stage = "inspect columns";
int rc = si_prep(si, &rows, "PRAGMA table_info(\"%s\")"), has_id = 0, has_bucket = 0, has_chain = 0;
if (rc != SQLITE_OK) goto error;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
const char* name = (const char*)sqlite3_column_text(rows, 1);
if (name && !strcmp(name, "message_id")) has_id = 1;
if (name && !strcmp(name, "bucket")) has_bucket = 1;
if (name && !strcmp(name, "chain_hash")) has_chain = 1;
int rc = si_prep(si, &st, "PRAGMA table_info(\"%s\")"), versioned = 0;
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
const char* name = (const char*)sqlite3_column_text(st, 1);
if (name && !strcmp(name, "nonce")) versioned = 1;
}
if (rc != SQLITE_DONE) goto error;
sqlite3_finalize(rows); rows = NULL;
if (has_id && has_bucket && !has_chain) return 0;
stage = "begin";
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) goto error;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history schema: preparing Merkle table=%s group=%016llx",
SI_TBL(si), (unsigned long long)si->group_id);
char replacement[64], sql[768];
snprintf(replacement, sizeof(replacement), "new_%.48s", SI_TBL(si));
stage = "create replacement";
if (si_table_create(si, replacement, 0) < 0) goto rollback;
stage = "read history";
if (si_prep(si, &rows, "SELECT timestamp,author_signature FROM \"%s\"") != SQLITE_OK) goto rollback;
snprintf(sql, sizeof(sql),
"INSERT INTO \"%s\" SELECT timestamp,node_id,id,flags,data,author_signature,?1,?4,"
"local_attrs,delivered_peers,delivery_chain FROM \"%s\" WHERE timestamp=?2 AND author_signature=?3",
replacement, SI_TBL(si));
stage = "prepare copy";
if (sqlite3_prepare_v2(db, sql, -1, &insert, NULL) != SQLITE_OK) goto rollback;
sqlite3_int64 count = 0;
while ((rc = sqlite3_step(rows)) == SQLITE_ROW) {
sqlite3_int64 ts = sqlite3_column_int64(rows, 0);
const uint8_t* signature = sqlite3_column_blob(rows, 1);
int size = sqlite3_column_bytes(rows, 1);
uint8_t message_id[32];
stage = "compute message_id";
if (!signature && sqlite3_errcode(db) == SQLITE_NOMEM) goto rollback;
if (!signature || size != DB_SIG_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history schema: removing malformed signature table=%s ts=%lld bytes=%d",
SI_TBL(si), (long long)ts, size);
continue;
}
if (chat_message_id(si->group_id, (uint64_t)ts, signature, message_id) < 0) goto rollback;
stage = "copy record";
rc = sqlite3_bind_blob(insert, 1, message_id, sizeof(message_id), SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(insert, 2, ts);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(insert, 3, signature, DB_SIG_SIZE, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(insert, 4, (sqlite3_int64)(db_read64(message_id) >> 39));
if (rc != SQLITE_OK || sqlite3_step(insert) != SQLITE_DONE) goto rollback;
if (sqlite3_changes(db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: copy table=%s ts=%lld changed=%d expected=1",
SI_TBL(si), (long long)ts, sqlite3_changes(db)); goto rollback;
}
if (sqlite3_reset(insert) != SQLITE_OK) goto rollback;
count++;
int final_rc = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "inspect history format");
if (versioned) return 0;
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "format reset begin");
char sql[512];
snprintf(sql, sizeof(sql), "DROP TABLE \"%s\"", SI_TBL(si));
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK || si_table_create(si, SI_TBL(si), 0) < 0) goto rollback;
/* В тестах произвольные таблицы не имеют feedback; в чате он подготовлен до регистрации истории. */
snprintf(sql, sizeof(sql), "SELECT 1 FROM sqlite_master WHERE type='table' AND name='feedback_%s'", si->ns);
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) goto rollback;
rc = sqlite3_step(st); final_rc = sqlite3_finalize(st); st = NULL;
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) goto rollback;
if (rc == SQLITE_ROW) {
snprintf(sql, sizeof(sql), "DELETE FROM \"feedback_%s\"", si->ns);
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
}
stage = "read history";
if (rc != SQLITE_DONE) goto rollback;
sqlite3_finalize(rows); rows = NULL; sqlite3_finalize(insert); insert = NULL;
stage = "replace table";
snprintf(sql, sizeof(sql), "DROP TABLE \"%s\"; ALTER TABLE \"%s\" RENAME TO \"%s\"",
SI_TBL(si), replacement, SI_TBL(si));
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history schema: message_id ready table=%s records=%lld", SI_TBL(si), (long long)count);
if (sqlite3_prepare_v2(db, "DELETE FROM merkle_tree_hash WHERE namespace IN (?1,?2)", -1, &st, NULL) != SQLITE_OK)
goto rollback;
char messages[25], feedback[25];
snprintf(messages, sizeof(messages), "%u:%s", MT_DATA_MESSAGES, si->ns);
snprintf(feedback, sizeof(feedback), "%u:%s", MT_DATA_FEEDBACK, si->ns);
rc = sqlite3_bind_text(st, 1, messages, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, feedback, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
final_rc = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_DONE || final_rc != SQLITE_OK || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history format v2: reset old group history and feedback tbl=%s group=%s",
SI_TBL(si), si->ns);
return 0;
rollback:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: table=%s stage=%s: %s; rolling back",
SI_TBL(si), stage, sqlite3_errmsg(db));
sqlite3_finalize(rows); sqlite3_finalize(insert);
if (sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: rollback table=%s: %s", SI_TBL(si), sqlite3_errmsg(db));
return -1;
error:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: table=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(db));
sqlite3_finalize(rows); return -1;
sqlite3_finalize(st); db_rollback(si, "history format reset"); return -1;
}
#define DB_COLUMNS "rowid,id,timestamp,node_id,data,author_signature,message_id,bucket"
#define DB_COLUMNS "rowid,id,timestamp,node_id,data,author_signature,message_id,bucket,nonce,ver,edited_at"
/* Чтение конкретной строки. cache_valid — производные поля, их можно восстановить без удаления сообщения. */
static enum db_record_check db_read_record(struct DB_SYNC_INSTANCE* si, sqlite3_stmt* st, struct db_record* r) {
@@ -362,7 +354,16 @@ static enum db_record_check db_read_record(struct DB_SYNC_INSTANCE* si, sqlite3_
SI_TBL(si), (long long)r->rowid, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
}
memcpy(r->signature, sig, 64);
if (chat_message_id(si->group_id, r->timestamp, r->signature, r->message_id) < 0) return DB_CHECK_ERROR;
const void* nonce = sqlite3_column_blob(st, 8);
if (!nonce && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
if (!nonce || sqlite3_column_bytes(st, 8) != 16) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed nonce tbl=%s rowid=%lld", SI_TBL(si), (long long)r->rowid);
return DB_CHECK_BAD_FORMAT;
}
memcpy(r->revision.nonce, nonce, 16);
r->revision.ver = (uint64_t)sqlite3_column_int64(st, 9);
r->revision.edited_at = (uint64_t)sqlite3_column_int64(st, 10);
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return DB_CHECK_ERROR;
const uint8_t* key = sqlite3_column_blob(st, 6);
if (!key && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
r->cache_valid = key && sqlite3_column_bytes(st, 6) == 32 && !memcmp(key, r->message_id, 32) &&
@@ -372,20 +373,22 @@ static enum db_record_check db_read_record(struct DB_SYNC_INSTANCE* si, sqlite3_
static int db_encode_record(const struct db_record* r, uint8_t* buf) {
memcpy(buf, r->message_id, 32); db_write64(buf + 32, r->timestamp); db_write64(buf + 40, r->author);
db_write32(buf + 48, (uint32_t)r->data_len); memcpy(buf + 52, r->data, r->data_len);
memcpy(buf + 52 + r->data_len, r->signature, 64);
memcpy(buf + 48, r->revision.nonce, 16); db_write64(buf + 64, r->revision.ver); db_write64(buf + 72, r->revision.edited_at);
db_write32(buf + 80, (uint32_t)r->data_len); memcpy(buf + 84, r->data, r->data_len);
memcpy(buf + 84 + r->data_len, r->signature, 64);
return (int)(DB_WIRE_FIXED + r->data_len);
}
static int db_parse_record(struct DB_SYNC_INSTANCE* si, const uint8_t** ptr, const uint8_t* end, struct db_record* r) {
if ((size_t)(end - *ptr) < DB_WIRE_FIXED) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history truncated record"); return -1; }
const uint8_t* p = *ptr;
memset(r, 0, sizeof(*r));
r->timestamp = db_read64(p + 32); r->author = db_read64(p + 40); r->data_len = db_read32(p + 48);
r->timestamp = db_read64(p + 32); r->author = db_read64(p + 40); r->data_len = db_read32(p + 80);
memcpy(r->revision.nonce, p + 48, 16); r->revision.ver = db_read64(p + 64); r->revision.edited_at = db_read64(p + 72);
if (!r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || r->data_len > (size_t)(end - p) - DB_WIRE_FIXED) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid wire body=%zu", r->data_len); return -1;
}
r->data = (const char*)p + 52; r->sig_len = 64; memcpy(r->signature, p + 52 + r->data_len, 64);
if (chat_message_id(si->group_id, r->timestamp, r->signature, r->message_id) < 0) return -1;
r->data = (const char*)p + 84; r->sig_len = 64; memcpy(r->signature, p + 84 + r->data_len, 64);
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
if (memcmp(r->message_id, p, 32)) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history wire identity mismatch"); return -1; }
*ptr = p + DB_WIRE_FIXED + r->data_len;
return 0;
@@ -413,6 +416,8 @@ static void db_committed(struct DB_SYNC_INSTANCE* si, int changed) {
static int db_same_record(const struct db_record* a, const struct db_record* b) {
return a->timestamp == b->timestamp && a->author == b->author && a->data_len == b->data_len &&
a->revision.ver == b->revision.ver && a->revision.edited_at == b->revision.edited_at &&
!memcmp(a->revision.nonce, b->revision.nonce, 16) &&
a->sig_len == b->sig_len && !memcmp(a->signature, b->signature, 64) && !memcmp(a->data, b->data, a->data_len);
}
@@ -420,23 +425,23 @@ static int db_same_record(const struct db_record* a, const struct db_record* b)
static int db_put_record(struct DB_SYNC_INSTANCE* si, struct db_record* r, enum db_record_check incoming,
const char* attrs, int* changed) {
sqlite3_stmt* st = NULL;
if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE timestamp=? AND author_signature=?") != SQLITE_OK)
if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE message_id=?") != SQLITE_OK)
return db_error(si, "read duplicate");
int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)r->timestamp);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, r->signature, 64, SQLITE_STATIC);
int rc = sqlite3_bind_blob(st, 1, r->message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
struct db_record stored;
int exists = rc == SQLITE_ROW, remove = 0;
if (exists) {
enum db_record_check check = db_read_record(si, st, &stored);
if (check == DB_CHECK_VALID && !(incoming == DB_CHECK_VALID && db_same_record(&stored, r)))
check = db_verify_author_sig(si, stored.timestamp, stored.author, stored.data, stored.data_len, stored.signature, stored.sig_len);
check = db_verify_author_sig(si, &stored);
if (check == DB_CHECK_ERROR || check == DB_CHECK_DEFERRED) { sqlite3_finalize(st); return -1; }
remove = check == DB_CHECK_BAD_FORMAT || check == DB_CHECK_BAD_SIGNATURE;
if (!remove && !stored.cache_valid) {
sqlite3_finalize(st); db_index_request(si); return db_error(si, "derived identity needs rebuild");
}
} else if (rc != SQLITE_DONE) { sqlite3_finalize(st); return db_error(si, "duplicate step"); }
int same = exists && !remove && db_same_record(&stored, r);
if (sqlite3_finalize(st) != SQLITE_OK) return db_error(si, "duplicate finalize");
if (remove) {
if (db_delete_bad_row(si, stored.rowid) < 0) return -1;
@@ -446,9 +451,37 @@ static int db_put_record(struct DB_SYNC_INSTANCE* si, struct db_record* r, enum
SI_TBL(si), (long long)stored.rowid);
}
if (incoming != DB_CHECK_VALID) return -2;
if (exists && !remove) return 1;
if (si_prep(si, &st, "INSERT INTO \"%s\" (timestamp,node_id,id,flags,data,author_signature,message_id,bucket,local_attrs)"
" VALUES(?,?,?,0,?,?,?,?,?)") != SQLITE_OK) return db_error(si, "insert prepare");
if (exists && !remove) {
if (r->revision.ver < stored.revision.ver || same) {
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history duplicate/stale tbl=%s id=%02x%02x%02x%02x incoming=%llu stored=%llu",
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
(unsigned long long)r->revision.ver, (unsigned long long)stored.revision.ver); return 1;
}
if (r->revision.ver == stored.revision.ver) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history version conflict tbl=%s id=%02x%02x%02x%02x ver=%llu",
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
(unsigned long long)r->revision.ver); return -2;
}
r->updated = 1; r->id = stored.id;
if (si_prep(si, &st, "UPDATE \"%s\" SET data=?,author_signature=?,ver=?,edited_at=?,"
"flags=flags & ~1,delivered_peers=0,delivery_chain='' WHERE message_id=?") != SQLITE_OK) return db_error(si, "update prepare");
rc = sqlite3_bind_blob(st, 1, r->data, (int)r->data_len, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, r->signature, 64, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->revision.ver);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)r->revision.edited_at);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 5, r->message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int end_rc = sqlite3_finalize(st);
if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "update step");
if (db_update_path(si, r->message_id) < 0) return -1;
*changed |= 1;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history updated tbl=%s id=%02x%02x%02x%02x ver=%llu->%llu edited=%llu",
SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
(unsigned long long)stored.revision.ver, (unsigned long long)r->revision.ver,
(unsigned long long)r->revision.edited_at); return 0;
}
if (si_prep(si, &st, "INSERT INTO \"%s\" (timestamp,node_id,id,flags,data,author_signature,message_id,bucket,local_attrs,nonce,ver,edited_at)"
" VALUES(?,?,?,0,?,?,?,?,?,?,?,?)") != SQLITE_OK) return db_error(si, "insert prepare");
rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)r->timestamp);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)r->author);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->id);
@@ -457,6 +490,9 @@ static int db_put_record(struct DB_SYNC_INSTANCE* si, struct db_record* r, enum
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 6, r->message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 7, (sqlite3_int64)(db_read64(r->message_id) >> 39));
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 8, attrs ? attrs : "", -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 9, r->revision.nonce, 16, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 10, (sqlite3_int64)r->revision.ver);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 11, (sqlite3_int64)r->revision.edited_at);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int final_rc = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "insert step");
@@ -491,8 +527,7 @@ static int db_index_batch(struct DB_SYNC_INSTANCE* si) {
item->check = db_read_record(si, st, &item->record);
if (item->check == DB_CHECK_VALID) {
memcpy(item->data, item->record.data, item->record.data_len); item->record.data = item->data;
item->check = db_verify_author_sig(si, item->record.timestamp, item->record.author, item->record.data,
item->record.data_len, item->record.signature, item->record.sig_len);
item->check = db_verify_author_sig(si, &item->record);
}
if (item->check == DB_CHECK_ERROR || item->check == DB_CHECK_DEFERRED) {
si->index.waiting_key = item->check == DB_CHECK_DEFERRED;
@@ -623,7 +658,7 @@ static int db_message_page(void* ctx, const char* ns, uint64_t prefix, int av, c
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
struct db_record r;
enum db_record_check check = db_read_record(si, st, &r);
if (check == DB_CHECK_VALID) check = db_verify_author_sig(si, r.timestamp, r.author, r.data, r.data_len, r.signature, r.sig_len);
if (check == DB_CHECK_VALID) check = db_verify_author_sig(si, &r);
if (check != DB_CHECK_VALID || !r.cache_valid) {
bad = check == DB_CHECK_BAD_SIGNATURE || check == DB_CHECK_BAD_FORMAT || (check == DB_CHECK_VALID && !r.cache_valid);
rc = SQLITE_ERROR; break;
@@ -718,7 +753,7 @@ static void db_notice_run(void* arg) {
struct db_record* records = (struct db_record*)(n + 1);
for (unsigned i = 0; i < n->count && si->enabled; i++)
if (si->on_insert) si->on_insert(si, records[i].message_id, records[i].timestamp, records[i].data,
records[i].data_len, records[i].author, n->initial, si->on_insert_arg);
records[i].data_len, records[i].author, n->initial, records[i].updated, si->on_insert_arg);
db_release(si);
}
u_free(n);
@@ -759,7 +794,7 @@ static int db_message_apply(void* ctx, const char* ns, uint64_t peer, uint64_t p
(av && memcmp(r->message_id, after->bytes, 32) <= 0)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page key outside requested range ns=%s", ns); return -1;
}
if (db_verify_author_sig(si, r->timestamp, r->author, r->data, r->data_len, r->signature, r->sig_len) != DB_CHECK_VALID) return -1;
if (db_verify_author_sig(si, r) != DB_CHECK_VALID) return -1;
if (count && memcmp(records[count - 1].message_id, r->message_id, 32) >= 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history incoming page not strictly ordered ns=%s", ns); return -1;
}
@@ -778,7 +813,7 @@ static int db_message_apply(void* ctx, const char* ns, uint64_t peer, uint64_t p
records[i].id = next;
int result = db_put_record(si, &records[i], DB_CHECK_VALID, NULL, &changed);
if (result < 0) goto rollback;
if (!result) { added[i] = 1; next++; }
if (!result) { added[i] = 1; if (!records[i].updated) next++; }
}
if (db_notice_stage(si, records, count, added, &notice) < 0) goto rollback;
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
@@ -823,7 +858,7 @@ static int db_send(struct DB_SYNC_INSTANCE* si, uint64_t peer, uint8_t type, con
static void db_emit_push(struct DB_SYNC_INSTANCE* si, const struct db_record* r, uint64_t source) {
si->holds++;
if (si->on_insert) si->on_insert(si, r->message_id, r->timestamp, r->data, r->data_len, r->author, 0, si->on_insert_arg);
if (si->on_insert) si->on_insert(si, r->message_id, r->timestamp, r->data, r->data_len, r->author, 0, r->updated, si->on_insert_arg);
if (si->enabled) {
uint8_t wire[DB_WIRE_MAX]; size_t bytes = (size_t)db_encode_record(r, wire);
for (int i = 0; i < si->peer_count; i++)
@@ -834,20 +869,20 @@ static void db_emit_push(struct DB_SYNC_INSTANCE* si, const struct db_record* r,
db_release(si);
}
static int db_insert_one(struct DB_SYNC_INSTANCE* si, struct db_record* r, const char* attrs) {
enum db_record_check incoming = db_verify_author_sig(si, r->timestamp, r->author, r->data, r->data_len, r->signature, r->sig_len);
if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
enum db_record_check incoming = db_verify_author_sig(si, r);
if (incoming == DB_CHECK_ERROR || incoming == DB_CHECK_DEFERRED) return -1;
if (chat_message_id(si->group_id, r->timestamp, r->signature, r->message_id) < 0) return -1;
sqlite3* db = SI_DB(si);
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "insert begin");
int changed = 0; r->id = si->next_id;
int result = db_put_record(si, r, incoming, attrs, &changed);
if (result == -1 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "insert"); return -1; }
if (!result) si->next_id++;
if (!result && !r->updated) si->next_id++;
if (changed) db_committed(si, changed);
return result;
}
/* ACK конкретного сообщения: повтор не увеличивает счётчик, очередь отправки не считается доставкой. */
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t peer_id)
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t ver, uint64_t peer_id)
{
char hex[17]; snprintf(hex, sizeof(hex), "%016llx", (unsigned long long)peer_id);
sqlite3_stmt* stmt = NULL;
@@ -855,17 +890,19 @@ static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message
"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");
" WHEN delivery_chain='' THEN ?1 ELSE delivery_chain || ',' || ?1 END WHERE message_id=?2 AND ver=?4");
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_bind_int64(stmt, 4, (sqlite3_int64)ver);
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);
if (!changed) DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK stale/unknown tbl=%s peer=%s ver=%llu",
SI_TBL(si), hex, (unsigned long long)ver);
else DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK recorded tbl=%s peer=%s", SI_TBL(si), hex);
return 0;
}
@@ -879,14 +916,17 @@ static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* e) {
const uint8_t* data = e->dgram + 10; size_t len = e->len - 10;
si->holds++;
if (e->dgram[9] == DB_MSG_ACK_PUSH) {
if (len == 32) si_delivery_update(si, data, conn->peer_node_id);
if (len == 40) si_delivery_update(si, data, db_read64(data + 32), conn->peer_node_id);
else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid ACK bytes=%zu", len);
} else if (e->dgram[9] == DB_MSG_PUSH) {
const uint8_t* p = data; struct db_record r;
if (db_parse_record(si, &p, data + len, &r) == 0 && p == data + len) {
int result = db_insert_one(si, &r, NULL);
if (!result) db_emit_push(si, &r, conn->peer_node_id);
if (result >= 0 && si->enabled) db_send(si, conn->peer_node_id, DB_MSG_ACK_PUSH, r.message_id, 32);
if (result >= 0 && si->enabled) {
uint8_t ack[40]; memcpy(ack, r.message_id, 32); db_write64(ack + 32, r.revision.ver);
db_send(si, conn->peer_node_id, DB_MSG_ACK_PUSH, ack, sizeof(ack));
}
}
} else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history unknown message type=%u", e->dgram[9]);
db_release(si);
@@ -1088,9 +1128,12 @@ void db_sync_destroy(struct UTUN_INSTANCE* inst) {
}
int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* data, size_t len, const uint8_t* sig, size_t sig_len,
uint64_t timestamp, const char* attrs) {
if (!si || !si->enabled || !sig || sig_len != 64) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid insertion"); return -1; }
uint64_t timestamp, const char* attrs, const struct db_message_revision* revision) {
if (!si || !si->enabled || !sig || sig_len != 64 || !revision) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid insertion"); return -1;
}
struct db_record r = {.timestamp = timestamp, .author = si->db_sync->inst->node_id, .data = data, .data_len = len, .sig_len = sig_len};
r.revision = *revision;
memcpy(r.signature, sig, 64);
int result = db_insert_one(si, &r, attrs);
if (!result) db_emit_push(si, &r, r.author);
@@ -1116,7 +1159,7 @@ int db_sync_select(struct DB_SYNC_INSTANCE* si, uint32_t offset, uint32_t limit,
if (!si || !si->enabled || !cb) return -1;
sqlite3_stmt* st = NULL; si->holds++;
int rc = si_prep(si, &st, "SELECT id,timestamp,node_id,data,author_signature,delivered_peers,delivery_chain"
" FROM \"%s\" ORDER BY timestamp,author_signature LIMIT ? OFFSET ?");
" FROM \"%s\" ORDER BY timestamp,message_id LIMIT ? OFFSET ?");
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, limit ? limit : -1LL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, offset);
int count = 0;
src/chat/db_sync.h
+20 -9
View File
@@ -1,7 +1,7 @@
/* db_sync — репликация подписанных записей SQLite по прямым ETCP-соединениям.
* Каждый DB_SYNC_INSTANCE связывает таблицу с group_id. Различия находятся по
* общему Merkle-дереву; новые записи рассылаются PUSH. Ключ и порядок отображения:
* (timestamp, author_signature). Подпись автора покрывает timestamp || data.
* общему Merkle-дереву; изменения рассылаются PUSH. Постоянный ключ — message_id,
* порядок отображения — (timestamp, message_id). Подпись защищает весь конверт и data.
* Ядро создаёт модуль, чат включает его и регистрирует таблицы каналов.
* Обычно используется общая SQLite ядра; remove отключает таблицу от обмена,
* сохраняя данные. Все вызовы и callbacks выполняются в потоке uasync. */
@@ -27,8 +27,19 @@ struct DB_SYNC_INSTANCE;
/* Максимальная длина подписанного тела. Транспортный буфер также учитывает identity и подпись. */
#define DB_SYNC_MESSAGE_MAX 4000
// PUSH и записи MESSAGES: [message_id:32][ts:8BE][author:8BE][dlen:4BE][data][sig:64].
// ACK_PUSH wire: [message_id:32]. Подтверждает конкретную запись; повторы идемпотентны.
// PUSH/MESSAGES: [id:32][ts:8BE][author:8BE][nonce:16][ver:8BE][edited_at:8BE][dlen:4BE][data][sig:64].
// ACK_PUSH: [id:32][ver:8BE]. Подтверждает конкретную версию; повторы идемпотентны.
struct db_message_revision {
uint8_t nonce[16]; /* Случайность создания, неизменна во всех версиях. */
uint64_t ver; /* 1..INT64_MAX; принимается только более высокая версия. */
uint64_t edited_at; /* Unix ms последней правки; 0 для ver=1. */
};
/* Подписать канонический конверт, включая канал; 0/-1. Не обращается к БД, nonce задаёт вызывающий. */
int db_sync_sign_message(uint64_t group, uint64_t timestamp, uint64_t author,
const struct db_message_revision* revision, const char* data, size_t len,
const uint8_t private_key[32], uint8_t signature[64]);
// Интервалы служебных таймеров в секундах.
#define DB_SYNC_PEER_CHECK_INTERVAL 5
@@ -60,21 +71,21 @@ uint64_t db_sync_instance_group_id(struct DB_SYNC_INSTANCE* si);
// Data operations (per-instance)
// db_sync_insert_signed: sig MUST be non-NULL, 64 bytes.
// sig = Ed25519(ts[8] || json); ts записан в порядке байтов хоста.
// sig создаётся db_sync_sign_message; revision обязателен, автор — локальный узел.
// ts — call db_sync_next_timestamp(si) before signing to reserve monotonically increasing timestamp.
// Возвращает 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);
const char* local_attrs, const struct db_message_revision* revision);
uint32_t db_sync_count(struct DB_SYNC_INSTANCE* si);
// Последний известный/зарезервированный timestamp в миллисекундах Unix; 0 для NULL.
uint64_t db_sync_get_last_timestamp(struct DB_SYNC_INSTANCE* si);
// Зарезервировать следующий timestamp перед подписью; растёт даже в пределах одной ms.
uint64_t db_sync_next_timestamp(struct DB_SYNC_INSTANCE* si);
// Обход по (timestamp, author_signature), начиная с offset; limit=0 — без ограничения.
// Обход по (timestamp, message_id), начиная с offset; limit=0 — без ограничения.
// Возвращает число записей или -1. Данные callback заимствованы на время вызова.
typedef void (*db_sync_select_cb)(void* arg, uint64_t id, uint64_t timestamp,
const char* data, size_t data_len, uint64_t author,
@@ -83,7 +94,7 @@ typedef void (*db_sync_select_cb)(void* arg, uint64_t id, uint64_t timestamp,
int db_sync_select(struct DB_SYNC_INSTANCE* si, uint32_t offset, uint32_t limit,
db_sync_select_cb cb, void* arg);
// Insert callback: fired after local or peer insert succeeds.
// Callback после вставки/обновления. updated=1 заменяет строку без нового unread/звука.
// author_node_id — исходный автор записи, не обязательно переславший её пир.
// initial_sync = 1 для страницы MESSAGES, 0 для live-PUSH и локальной вставки.
// Вызов после commit: для страниц отложен до выхода из обработчика Merkle.
@@ -91,7 +102,7 @@ typedef void (*db_sync_insert_cb)(struct DB_SYNC_INSTANCE* si,
const uint8_t message_id[32],
uint64_t record_timestamp,
const char* json_data, size_t len,
uint64_t author_node_id, int initial_sync, void* arg);
uint64_t author_node_id, int initial_sync, int updated, void* arg);
void db_sync_set_insert_cb(struct DB_SYNC_INSTANCE* si, db_sync_insert_cb cb, void* arg);
// Диагностика: синхронно перепроверить/очистить сохранённые записи и восстановить индекс.