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

Preserve signed history through schema conversion and resume both peers

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 18:27:26 +01:00
1 parent 1402413ab9
commit d76eb067bb
2 files changed
+36 -16
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_db_sync.c
+1
View File
@@ -313,6 +313,7 @@ int main(void) {
reset_done_flags();
db_sync_peer_set_paused(si_a, inst_b->node_id, 1);
if (insert_many(si_a, inst_a, 5, 3) != 0) { test_phase = 2; goto done; }
db_sync_reinitiate(si_a, inst_b->node_id);
db_sync_reinitiate(si_b, inst_a->node_id);
cb_target = 8;
if (!wait_for("B=8", _cond_cb, PHASE_TIMEOUT_TB)) { test_phase = 2; goto done; }
tests/test_db_sync_schema.c
+35 -16
View File
@@ -12,7 +12,7 @@
#include <stdlib.h>
#include <string.h>
#define OLD_COLUMNS "timestamp,node_id,id,chain_hash,flags,data,author_signature,local_attrs,delivered_peers,delivery_chain"
#define SAVED_COLUMNS "timestamp,node_id,id,flags,data,author_signature,local_attrs,delivered_peers,delivery_chain"
static void execute(sqlite3* db, const char* sql) {
int rc = sqlite3_exec(db, sql, NULL, NULL, NULL);
@@ -38,9 +38,9 @@ static void legacy_table(sqlite3* db, const char* name) {
static void check_history(sqlite3* db, const char* table, uint64_t group) {
char sql[768];
snprintf(sql, sizeof(sql), "SELECT count(*) FROM (SELECT " OLD_COLUMNS " FROM \"%s\" EXCEPT SELECT * FROM expected)", table);
snprintf(sql, sizeof(sql), "SELECT count(*) FROM (SELECT " SAVED_COLUMNS " FROM \"%s\" EXCEPT SELECT * FROM expected)", table);
assert(number(db, sql) == 0);
snprintf(sql, sizeof(sql), "SELECT count(*) FROM (SELECT * FROM expected EXCEPT SELECT " OLD_COLUMNS " FROM \"%s\")", table);
snprintf(sql, sizeof(sql), "SELECT count(*) FROM (SELECT * FROM expected EXCEPT SELECT " SAVED_COLUMNS " FROM \"%s\")", table);
assert(number(db, sql) == 0);
snprintf(sql, sizeof(sql), "SELECT timestamp,author_signature,message_id FROM \"%s\"", table);
sqlite3_stmt* st = NULL; assert(sqlite3_prepare_v2(db, sql, -1, &st, NULL) == SQLITE_OK);
@@ -56,6 +56,11 @@ static void check_history(sqlite3* db, const char* table, uint64_t group) {
assert(sqlite3_get_autocommit(db));
}
static int deny_drop(void* ctx, int action, const char* name, const char* arg, const char* db, const char* source) {
(void)ctx; (void)arg; (void)db; (void)source;
return action == SQLITE_DROP_TABLE && name && !strcmp(name, "msg_9001") ? SQLITE_DENY : SQLITE_OK;
}
static struct chat_msg_submit send_text(struct UTUN_INSTANCE* inst, const char* channel, const char* text) {
struct chat_msg_submit req = { .inst = inst, .data = (const uint8_t*)text, .data_len = strlen(text) };
snprintf(req.channel_id, sizeof(req.channel_id), "%s", channel); strcpy(req.content_type, "text");
@@ -71,7 +76,7 @@ static void check_copy(struct UTUN_INSTANCE* inst) {
for (int i = 0; i < rows; i++) {
const char* table = tables[i + 1]; uint64_t gid = strtoull(table + 4, NULL, 10); assert(gid);
assert(topo_groups_create_group(inst->topo_groups, gid, TOPO_GROUP_TYPE_CHAT, table + 4));
char sql[768]; snprintf(sql, sizeof(sql), "CREATE TEMP TABLE expected AS SELECT " OLD_COLUMNS " FROM \"%s\"", table);
char sql[768]; snprintf(sql, sizeof(sql), "CREATE TEMP TABLE expected AS SELECT " SAVED_COLUMNS " FROM \"%s\"", table);
execute(db, sql); struct DB_SYNC_INSTANCE* si = db_sync_instance_add(inst, table, gid); assert(si);
check_history(db, table, gid); db_sync_instance_remove(si);
si = db_sync_instance_add(inst, table, gid); assert(si); check_history(db, table, gid); db_sync_instance_remove(si);
@@ -104,7 +109,7 @@ int main(int argc, char** argv) {
assert(db_sync_insert_signed(db_sync_instance_find(inst, gid), json, strlen(json), signature, 64, first.timestamp, NULL) == 0);
uasync_poll(ua, 0);
snprintf(sql, sizeof(sql), "UPDATE \"%s\" SET flags=1,local_attrs=NULL,delivered_peers=7,delivery_chain='saved chain';"
"CREATE TEMP TABLE expected AS SELECT " OLD_COLUMNS " FROM \"%s\"", table, table); execute(db, sql);
"CREATE TEMP TABLE expected AS SELECT " SAVED_COLUMNS " FROM \"%s\"", table, table); execute(db, sql);
chat_core_create_channel_auto(inst, "Empty history"); chat_service_stop(inst);
/* Воспроизводим старую схему всех каналов, чтобы тест покрывал DDL при загрузке списка каналов. */
char** tables = NULL; int rows, columns;
@@ -113,37 +118,51 @@ int main(int argc, char** argv) {
for (int i = 0; i < rows; i++) {
const char* name = tables[i + 1];
snprintf(sql, sizeof(sql), "DROP INDEX \"idx_%s_ttl\";ALTER TABLE \"%s\" RENAME TO saved", name, name); execute(db, sql);
legacy_table(db, name); snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" SELECT " OLD_COLUMNS " FROM saved;DROP TABLE saved", name);
legacy_table(db, name); snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" SELECT timestamp,node_id,id,zeroblob(32),flags,data,"
"author_signature,local_attrs,delivered_peers,delivery_chain FROM saved;DROP TABLE saved", name);
execute(db, sql);
}
sqlite3_free_table(tables);
assert(chat_service_start(inst) == 0); check_history(db, table, gid); assert(chat_core_count(inst, channel) == 3);
char listing[8192]; size_t len;
assert(chat_core_get_messages_json(inst, channel, 10, 0, listing, sizeof(listing), &len) == 0 && len > 0);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" (" OLD_COLUMNS ") SELECT * FROM expected", table);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" (" SAVED_COLUMNS ") SELECT * FROM expected", table);
assert(sqlite3_exec(db, sql, NULL, NULL, NULL) == SQLITE_CONSTRAINT);
assert(sqlite3_extended_errcode(db) == SQLITE_CONSTRAINT_NOTNULL);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" (" OLD_COLUMNS ",message_id) SELECT *,x'01' FROM expected", table);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" (" SAVED_COLUMNS ",message_id,bucket) SELECT *,x'01',0 FROM expected", table);
assert(sqlite3_exec(db, sql, NULL, NULL, NULL) == SQLITE_CONSTRAINT);
assert(sqlite3_extended_errcode(db) == SQLITE_CONSTRAINT_CHECK);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" SELECT timestamp+1000000,node_id,id,chain_hash,flags,data,"
"author_signature,message_id,local_attrs,delivered_peers,delivery_chain FROM \"%s\" LIMIT 1", table, table);
snprintf(sql, sizeof(sql), "INSERT INTO \"%s\" SELECT timestamp+1000000,node_id,id,flags,data,"
"author_signature,message_id,bucket,local_attrs,delivered_peers,delivery_chain FROM \"%s\" LIMIT 1", table, table);
assert(sqlite3_exec(db, sql, NULL, NULL, NULL) == SQLITE_CONSTRAINT);
assert(sqlite3_extended_errcode(db) == SQLITE_CONSTRAINT_UNIQUE);
chat_service_stop(inst); assert(chat_service_start(inst) == 0); check_history(db, table, gid);
send_text(inst, channel, "Отправка после обновления 👍"); assert(chat_core_count(inst, channel) == 4);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS: restart, empty table, preserved fields, canonical IDs and new message submission");
/* Ошибка после копирования первой строки должна откатить всю перестройку. */
/* Отказ DDL после копирования строк откатывает всю перестройку, не удаляя допустимую историю. */
assert(topo_groups_create_group(inst->topo_groups, 9001, TOPO_GROUP_TYPE_CHAT, "9001")); legacy_table(db, "msg_9001");
execute(db, "INSERT INTO msg_9001 VALUES(1,1,1,zeroblob(32),0,x'01',zeroblob(64),'kept',0,'');"
"INSERT INTO msg_9001 VALUES(2,1,2,zeroblob(32),0,x'02',zeroblob(3),'kept',0,'')");
snprintf(sql, sizeof(sql), "INSERT INTO msg_9001 SELECT timestamp,node_id,id,zeroblob(32),flags,data,"
"author_signature,local_attrs,delivered_peers,delivery_chain FROM \"%s\" LIMIT 2", table); execute(db, sql);
assert(sqlite3_set_authorizer(db, deny_drop, NULL) == SQLITE_OK);
assert(!db_sync_instance_add(inst, "msg_9001", 9001));
assert(sqlite3_set_authorizer(db, NULL, NULL) == SQLITE_OK);
assert(sqlite3_get_autocommit(db) && number(db, "SELECT count(*) FROM msg_9001") == 2);
assert(number(db, "SELECT count(*) FROM pragma_table_info('msg_9001') WHERE name='message_id'") == 0);
assert(number(db, "SELECT count(*) FROM sqlite_master WHERE name='new_msg_9001'") == 0);
execute(db, "UPDATE msg_9001 SET author_signature=zeroblob(64) WHERE timestamp=2");
struct DB_SYNC_INSTANCE* si = db_sync_instance_add(inst, "msg_9001", 9001); assert(si && db_sync_count(si) == 2);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS: partial-copy failure rolls back, registration can be retried");
struct DB_SYNC_INSTANCE* si = db_sync_instance_add(inst, "msg_9001", 9001);
assert(si && db_sync_count(si) == 2 && db_sync_verify(si) == 0);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS schema DDL failure rolls back; valid records survive retry");
/* Доказанный битый формат удаляется при перестройке, неверная подпись — при preflight. */
assert(topo_groups_create_group(inst->topo_groups, 9002, TOPO_GROUP_TYPE_CHAT, "9002")); legacy_table(db, "msg_9002");
snprintf(sql, sizeof(sql), "INSERT INTO msg_9002 SELECT timestamp,node_id,id,zeroblob(32),flags,data,"
"author_signature,local_attrs,delivered_peers,delivery_chain FROM \"%s\" LIMIT 1;"
"INSERT INTO msg_9002 SELECT 2,node_id,200,zeroblob(32),0,x'01',zeroblob(3),'bad format',0,'' FROM \"%s\" LIMIT 1;"
"INSERT INTO msg_9002 SELECT 3,node_id,201,zeroblob(32),0,x'01',zeroblob(64),'bad signature',0,'' FROM \"%s\" LIMIT 1",
table, table, table); execute(db, sql);
si = db_sync_instance_add(inst, "msg_9002", 9002);
assert(si && db_sync_count(si) == 2 && db_sync_verify(si) == 0 && db_sync_count(si) == 1);
assert(number(db, "SELECT count(*) FROM pragma_table_info('msg_9002') WHERE name='chain_hash'") == 0);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS proven invalid legacy rows removed; valid signed data preserved");
done:
utun_instance_destroy(inst); uasync_poll(ua, 0);
assert(ua->timer_alloc_count == ua->timer_free_count && ua->socket_alloc_count == ua->socket_free_count);