From 809f00979622c4bb0c31e78edb3159315e2de833 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 22:35:34 +0100 Subject: [PATCH] Replicate group message revisions atomically with versioned acknowledgements --- src/chat/db_sync.c | 297 ++++++++++++++++++++++++++------------------- src/chat/db_sync.h | 29 +++-- 2 files changed, 190 insertions(+), 136 deletions(-) diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index ffc34cba..5d52307a 100644 --- a/src/chat/db_sync.c +++ b/src/chat/db_sync.c @@ -20,7 +20,7 @@ #include #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, ¬ice) < 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; diff --git a/src/chat/db_sync.h b/src/chat/db_sync.h index 611279ab..f58b5691 100644 --- a/src/chat/db_sync.h +++ b/src/chat/db_sync.h @@ -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); // Диагностика: синхронно перепроверить/очистить сохранённые записи и восстановить индекс.