Test atomic page rollback, late author keys and model run policy

tests/test_db_sync_integrity.c

@@ -1,5 +1,5 @@
 /* Собственная SQLite и настоящие подписи: сохранность истории, очистка порчи,
- * транзакционные отказы, восстановление цепочки и точная идентичность ACK. */
+ * транзакционные отказы, восстановление Merkle и точная идентичность ACK. */
 #include "../src/chat/db_sync.c"
 #include <assert.h>
 
@@ -272,6 +272,54 @@ static void test_deleted_author(void) {
     destroy(&f);
     DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS valid history of a deleted author remains in the root and outgoing page");
 }
+static void page_inserted(struct DB_SYNC_INSTANCE* si, const uint8_t id[32], uint64_t ts,
+        const char* data, size_t len, uint64_t author, int initial, void* arg) {
+    (void)id; (void)ts; (void)data; (void)len; (void)author;
+    assert(initial && sqlite3_get_autocommit(SI_DB(si)) && db_sync_count(si) == 3);
+    (*(unsigned*)arg)++;
+}
+static void test_page_rollback(void) {
+    concentrate_bucket = 1;
+    struct fixture a, b; init(&a); init(&b); uint8_t sig[64], before[32], after_root[32], data[16384];
+    for (uint64_t ts = 1; ts <= 3; ts++) { sign(&a, ts, "atomic page", sig); assert(insert(&a, ts, "atomic page", sig) == 0); }
+    assert(db_sync_verify(a.si) == 0 && db_sync_verify(b.si) == 0 && db_sync_read_root(b.si, before) == 0);
+    merkle_key_t after = {{0}}, next; int more; size_t len = sizeof(data); unsigned notices = 0;
+    assert(db_message_page(&a.inst, "10", 0, 0, &after, data, &len, &next, &more) == 0 && !more);
+    db_sync_set_insert_cb(b.si, page_inserted, &notices);
+    sql(&b, "CREATE TRIGGER deny_second BEFORE INSERT ON history WHEN NEW.id=2 BEGIN SELECT RAISE(FAIL,'second denied');END");
+    assert(db_message_apply(&b.inst, "10", 2, 0, 0, &after, &next, more, data, len) == -1);
+    assert(!db_sync_count(b.si) && b.si->next_id == 1 && !b.db.notices && sqlite3_get_autocommit(b.db.db));
+    assert(db_sync_read_root(b.si, after_root) == 0 && !memcmp(before, after_root, 32));
+    sql(&b, "DROP TRIGGER deny_second"); fail_commit = 1;
+    assert(db_message_apply(&b.inst, "10", 2, 0, 0, &after, &next, more, data, len) == -1);
+    assert(!db_sync_count(b.si) && b.si->next_id == 1 && !b.db.notices && !notices);
+    assert(db_sync_read_root(b.si, after_root) == 0 && !memcmp(before, after_root, 32));
+    assert(db_message_apply(&b.inst, "10", 2, 0, 0, &after, &next, more, data, len) == 0);
+    assert(db_sync_count(b.si) == 3 && b.db.notices && !notices);
+    uint64_t deadline = get_time_tb() + 10000;
+    while (notices != 3 && get_time_tb() < deadline) uasync_poll(b.inst.ua, 1);
+    assert(notices == 3 && !b.db.notices);
+    assert(db_message_apply(&b.inst, "10", 2, 0, 0, &after, &next, more, data, len) == 0 && !b.db.notices && notices == 3);
+    destroy(&a); destroy(&b); concentrate_bucket = 0;
+    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS page SQL/commit failures roll back rows and root; notifications observe only committed records");
+}
+static void test_late_author_key(void) {
+    struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "waiting key", sig);
+    assert(insert(&f, 100, "waiting key", sig) == 0);
+    sql(&f, "UPDATE history SET node_id=99");
+    assert(db_sync_verify(f.si) == 1 && f.si->index.pending && f.si->index.waiting_key && db_sync_count(f.si) == 1);
+    sqlite3_stmt* st = NULL;
+    assert(sqlite3_prepare_v2(f.db.db, "INSERT INTO nodes VALUES(99,?,1)", -1, &st, NULL) == SQLITE_OK);
+    assert(sqlite3_bind_blob(st, 1, f.inst.my_ed25519_pubkey, 32, SQLITE_STATIC) == SQLITE_OK && sqlite3_step(st) == SQLITE_DONE);
+    assert(sqlite3_finalize(st) == SQLITE_OK);
+    db_sync_members_changed(&f.inst, "11"); /* Ключи общие: событие другого канала также разблокирует историю. */
+    uint64_t deadline = get_time_tb() + 10000;
+    while (f.si->index.pending && get_time_tb() < deadline) uasync_poll(f.inst.ua, 1);
+    uint8_t root[32];
+    assert(!f.si->index.pending && db_sync_count(f.si) == 1 && db_sync_read_root(f.si, root) == 0 && page(&f, sig) == 0);
+    destroy(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS late historical author key resumes preflight across channels without deleting or restarting");
+}
 static void test_ttl(void) {
     struct fixture f; init(&f); uint8_t sig[64];
     struct utun_config* config = u_calloc(1, sizeof(*config)); assert(config); f.inst.config = config;
@@ -293,8 +341,8 @@ int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
     size_t baseline = u_get_allocated_count();
     test_preservation(); test_deletion(); test_index(); test_delivery(); test_callbacks();
-    test_pages(); test_deleted_author(); test_ttl();
+    test_pages(); test_deleted_author(); test_page_rollback(); test_late_author_key(); test_ttl();
     DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history injected failures=%u allocations=%zu baseline=%zu", fault_hits, u_get_allocated_count(), baseline);
-    assert(fault_hits == 10 && u_get_allocated_count() == baseline);
+    assert(fault_hits == 11 && u_get_allocated_count() == baseline);
     return 0;
 }

tests/test_merkle_protocol.c

@@ -16,7 +16,7 @@ static int t_random_bytes(uint8_t* data, size_t len) {
 struct fixture {
     struct UTUN_INSTANCE inst;
     struct ETCP_CONN conn;
-    int callbacks, result, apply_error, authorized, applied;
+    int callbacks, result, apply_error, authorized, applied, running;
 };
 
 static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
@@ -378,6 +378,29 @@ static void test_dataset_isolation(void) {
 static int t_authorized(void* ctx, const char* ns, uint64_t peer) {
     (void)ns; (void)peer; return ((struct fixture*)ctx)->authorized;
 }
+static int t_can_run(void* ctx, const char* ns, uint64_t peer) {
+    (void)ns; (void)peer; return ((struct fixture*)ctx)->running;
+}
+static void test_model_policy(void) {
+    struct fixture f; t_init(&f); f.conn.peer_sleep_phase = 1;
+    assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) < 0 && !f.callbacks);
+    struct merkle_sync_data_ops ops = t_ops; ops.validate_peer = t_authorized; ops.can_run = t_can_run;
+    f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops; f.running = 1;
+    assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) < 0 && !f.callbacks);
+    f.authorized = 1;
+    assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0);
+    struct ms_session* s = ms_find(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001); assert(s && s->work);
+    f.running = 0; merkle_sync_recheck_peer(&f.inst, 2);
+    assert(s->state == MS_FAILED && !s->work && f.callbacks == 1 && f.result == MT_ERR_DISCONNECTED);
+    merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
+    assert(s->state == MS_IDLE && !s->work && s->dirty);
+    f.running = 1;
+    assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0 && s->work);
+    f.authorized = 0; merkle_sync_recheck_peer(&f.inst, 2);
+    assert(s->state == MS_FAILED && !s->work && f.callbacks == 2);
+    t_destroy(&f);
+    puts("PASS: model sleep policy allows admitted peers; suspension stops work and restoration reuses the live connection");
+}
 
 /* Отзыв закрывает даже заблокированный ERROR; возврат начинает новый раунд той же подписки. */
 static void test_permission_rounds(void) {
@@ -457,7 +480,7 @@ int main(void) {
     test_page_size_and_cursor(); test_readonly_tree(); test_backpressure_and_disconnect(); test_peer_restart();
     test_send_rejection();
     test_error_handoff(); test_blocked_error_timeout(); test_pull_lifecycle();
-    test_dataset_isolation(); test_full_cursor(); test_permission_rounds(); test_reattach_tokens(); test_handshake_failures();
+    test_dataset_isolation(); test_full_cursor(); test_permission_rounds(); test_model_policy(); test_reattach_tokens(); test_handshake_failures();
     assert(u_get_allocated_count() == baseline);
     puts("ALL PASS: protocol regressions, no tracked allocations leaked");
     return 0;