From db8f3622d98fa325fdd38e66aea1af21b45e7fcb Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 12:23:09 +0100 Subject: [PATCH] Add atomic DM schema migration and durable author sequence --- src/dm/dm_schema.c | 132 +++++++++++++++++++++++++++++++++++++++++++++ src/dm/dm_schema.h | 16 ++++++ 2 files changed, 148 insertions(+) create mode 100644 src/dm/dm_schema.c create mode 100644 src/dm/dm_schema.h diff --git a/src/dm/dm_schema.c b/src/dm/dm_schema.c new file mode 100644 index 000000000..fc96f7767 --- /dev/null +++ b/src/dm/dm_schema.c @@ -0,0 +1,132 @@ +#include "dm_schema.h" +#include "../utun_instance.h" +#include "../../lib/debug_config.h" +#include +#include + +static int schema_error(sqlite3* db, const char* stage) { + DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM schema: %s sql=%s", stage, db ? sqlite3_errmsg(db) : "unavailable"); + return -1; +} + +static int schema_exec(sqlite3* db, const char* sql) { + if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) return schema_error(db, sql); + return 0; +} + +/* Не подменять ошибку чтения схемы отсутствием колонки. */ +static int schema_has_column(sqlite3* db, const char* table, const char* column) { + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "SELECT 1 FROM pragma_table_info(?) WHERE name=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, table, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, column, -1, SQLITE_STATIC); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int found = rc == SQLITE_ROW; + int final = sqlite3_finalize(st); + return (rc == SQLITE_ROW || rc == SQLITE_DONE) && final == SQLITE_OK ? found : schema_error(db, "column lookup"); +} + +int dm_schema_init(struct UTUN_INSTANCE* inst) { + sqlite3* db = inst ? inst->topo_sqlite_db : NULL; + if (!db) return schema_error(db, "missing database"); + if (schema_exec(db, "SAVEPOINT dm_schema")) return -1; + const char* conversations = + "CREATE TABLE IF NOT EXISTS dm_conversations(conv_id TEXT PRIMARY KEY,peer_node_id INTEGER NOT NULL," + "peer_x25519 BLOB NOT NULL,peer_ed25519 BLOB NOT NULL,peer_name TEXT DEFAULT '',group_id INTEGER DEFAULT 0," + "last_out_seq INTEGER DEFAULT 0,last_in_seq INTEGER DEFAULT 0,created_at INTEGER DEFAULT 0,last_ts INTEGER DEFAULT 0)"; + const char* messages = + "CREATE TABLE IF NOT EXISTS dm_messages(conv_id TEXT NOT NULL,dir INTEGER NOT NULL,seq INTEGER NOT NULL," + "ts INTEGER NOT NULL,author INTEGER NOT NULL,ct TEXT DEFAULT '',data BLOB,sig BLOB,body BLOB,PRIMARY KEY(conv_id,author,seq))"; + const char* deleted = + "CREATE TABLE IF NOT EXISTS dm_deleted_messages(conv_id TEXT NOT NULL,dir INTEGER NOT NULL,seq INTEGER NOT NULL," + "author INTEGER NOT NULL,body_hash BLOB NOT NULL,media_id BLOB,media_receipt BLOB,PRIMARY KEY(conv_id,author,seq))"; + if (schema_exec(db, conversations)) goto fail; + const char* names[] = {"mode", "local_party", "peer_party", "local_key", "closed"}; + const char* declarations[] = {"INTEGER NOT NULL DEFAULT 0", "BLOB", "BLOB", "BLOB", "INTEGER NOT NULL DEFAULT 0"}; + for (unsigned i = 0; i < sizeof(names) / sizeof(names[0]); i++) { + int found = schema_has_column(db, "dm_conversations", names[i]); + if (found < 0) goto fail; + if (!found) { + char sql[160]; snprintf(sql, sizeof(sql), "ALTER TABLE dm_conversations ADD COLUMN %s %s", names[i], declarations[i]); + if (schema_exec(db, sql)) goto fail; + } + } + if (schema_exec(db, messages) || schema_exec(db, deleted)) goto fail; + int found = schema_has_column(db, "dm_messages", "body"); + if (found < 0) goto fail; + if (!found) { + if (schema_exec(db, "ALTER TABLE dm_messages RENAME TO dm_messages_previous") || schema_exec(db, messages) || + schema_exec(db, "INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig)" + " SELECT conv_id,dir,seq,ts,author,ct,data,sig FROM dm_messages_previous") || + schema_exec(db, "DROP TABLE dm_messages_previous")) goto fail; + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: retained DEVICE history; message identity now includes author"); + } + found = schema_has_column(db, "dm_deleted_messages", "author"); + if (found < 0) goto fail; + if (!found) { + if (schema_exec(db, "ALTER TABLE dm_deleted_messages RENAME TO dm_deleted_previous") || schema_exec(db, deleted)) goto fail; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "INSERT INTO dm_deleted_messages(conv_id,dir,seq,author,body_hash,media_id,media_receipt)" + " SELECT d.conv_id,d.dir,d.seq,CASE WHEN d.dir=1 THEN ? ELSE c.peer_node_id END,d.body_hash,d.media_id,d.media_receipt" + " FROM dm_deleted_previous d LEFT JOIN dm_conversations c USING(conv_id)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int final = sqlite3_finalize(st); + if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain deleted message fingerprints"); goto fail; } + if (schema_exec(db, "DROP TABLE dm_deleted_previous")) goto fail; + } + if (schema_exec(db, + "CREATE INDEX IF NOT EXISTS dm_messages_time ON dm_messages(conv_id,ts,author,seq);" + "CREATE TABLE IF NOT EXISTS dm_outbox(conv_id TEXT NOT NULL,seq INTEGER NOT NULL,body BLOB NOT NULL," + "retry_at INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(conv_id,seq));" + "CREATE TABLE IF NOT EXISTS dm_delivery(conv_id TEXT NOT NULL,author INTEGER NOT NULL,seq INTEGER NOT NULL," + "recipient INTEGER NOT NULL,receipt BLOB,first_attempt INTEGER NOT NULL DEFAULT 0,direct_until INTEGER NOT NULL DEFAULT 0," + "retry_at INTEGER NOT NULL DEFAULT 0,supernode INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(conv_id,author,seq,recipient));" + "CREATE TABLE IF NOT EXISTS dm_sequence(node_id INTEGER PRIMARY KEY,value INTEGER NOT NULL);")) goto fail; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "INSERT INTO dm_sequence(node_id,value) SELECT ?,COALESCE(MAX(last_out_seq),0) FROM dm_conversations" + " WHERE 1 ON CONFLICT(node_id) DO UPDATE SET value=MAX(value,excluded.value)", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int final = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "initialize author sequence"); goto fail; } + rc = sqlite3_prepare_v2(db, "INSERT INTO dm_delivery(conv_id,author,seq,recipient,retry_at)" + " SELECT o.conv_id,?,o.seq,c.peer_node_id,o.retry_at FROM dm_outbox o JOIN dm_conversations c USING(conv_id)" + " WHERE c.mode=0 ON CONFLICT(conv_id,author,seq,recipient) DO NOTHING", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + final = sqlite3_finalize(st); + if (rc != SQLITE_DONE || final != SQLITE_OK) { schema_error(db, "retain DEVICE outbox recipients"); goto fail; } + if (schema_exec(db, "RELEASE dm_schema")) goto fail; + DEBUG_INFO(DEBUG_CATEGORY_DM, "DM schema: ready; author sequence, frozen parties and per-device delivery"); + return 0; +fail: + schema_exec(db, "ROLLBACK TO dm_schema"); schema_exec(db, "RELEASE dm_schema"); + return -1; +} + +int dm_sequence_next(struct UTUN_INSTANCE* inst, uint64_t* seq) { + sqlite3* db = inst ? inst->topo_sqlite_db : NULL; sqlite3_stmt* st = NULL; + if (!db || !seq) return schema_error(db, "invalid sequence arguments"); + int rc = sqlite3_prepare_v2(db, "SELECT value FROM dm_sequence WHERE node_id=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + sqlite3_int64 value = rc == SQLITE_ROW ? sqlite3_column_int64(st, 0) : -1; + int final = sqlite3_finalize(st); + if (rc != SQLITE_ROW || final != SQLITE_OK || value < 0 || value == INT64_MAX) return schema_error(db, "sequence unavailable/exhausted"); + *seq = (uint64_t)value + 1; + return 0; +} + +int dm_sequence_take(struct UTUN_INSTANCE* inst, uint64_t seq) { + sqlite3* db = inst ? inst->topo_sqlite_db : NULL; sqlite3_stmt* st = NULL; + if (!db || !seq || seq > INT64_MAX || sqlite3_get_autocommit(db)) return schema_error(db, "sequence outside send transaction"); + int rc = sqlite3_prepare_v2(db, "UPDATE dm_sequence SET value=? WHERE node_id=? AND value=?", -1, &st, NULL); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)seq); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)inst->node_id); + if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)(seq - 1)); + if (rc == SQLITE_OK) rc = sqlite3_step(st); + int changed = sqlite3_changes(db), final = sqlite3_finalize(st); + if (rc != SQLITE_DONE || final != SQLITE_OK || changed != 1) return schema_error(db, "sequence reservation conflict"); + return 0; +} diff --git a/src/dm/dm_schema.h b/src/dm/dm_schema.h new file mode 100644 index 000000000..214a24955 --- /dev/null +++ b/src/dm/dm_schema.h @@ -0,0 +1,16 @@ +/* dm_schema — атомарное создание/обновление локальной схемы личных бесед. + * Сохраняет историю, счётчики, точные тела outbox и отпечатки удалённых сообщений. + * Изменения выполняются внутри savepoint в uasync-потоке; ошибка откатывает схему. */ +#ifndef UTUN_DM_SCHEMA_H +#define UTUN_DM_SCHEMA_H +#include +struct UTUN_INSTANCE; + +/* 0/-1: создать схему и перенести существующие DEVICE-записи без переподписи. */ +int dm_schema_init(struct UTUN_INSTANCE* inst); +/* 0/-1: прочитать следующий seq автора. Резервирование выполняет take в той же + * транзакции, что запись сообщения; до commit тело нельзя передавать в сеть. */ +int dm_sequence_next(struct UTUN_INSTANCE* inst, uint64_t* seq); +/* 0/-1: увеличить счётчик с seq-1 до seq. Вызывать внутри транзакции отправки. */ +int dm_sequence_take(struct UTUN_INSTANCE* inst, uint64_t seq); +#endif