Cover stable history cursors, deleted authors and bounded TTL cleanup

src/chat/db_sync.c

@@ -897,6 +897,7 @@ static void db_ttl_tick(void* arg) {
     uint64_t now = db_now_ms(si);
     uint64_t ttl = (uint64_t)si->db_sync->inst->config->global.db_sync_ttl * 1000;
     uint64_t cut = now > ttl ? now - ttl : 0;
+    int removed = 0, committed = 0;
     if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) { db_error(si, "TTL begin"); goto schedule; }
     char query[512];
     snprintf(query, sizeof(query), "DELETE FROM \"%s\" WHERE rowid IN (SELECT rowid FROM \"%s\""
@@ -905,19 +906,22 @@ static void db_ttl_tick(void* arg) {
     if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)si->db_sync->inst->node_id);
     if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)cut);
     if (rc == SQLITE_OK) rc = sqlite3_step(st);
-    int removed = sqlite3_changes(db), end_rc = sqlite3_finalize(st); st = NULL;
+    removed = sqlite3_changes(db);
+    int end_rc = sqlite3_finalize(st); st = NULL;
     if (rc != SQLITE_DONE || end_rc != SQLITE_OK) { db_rollback(si, "TTL delete"); goto schedule; }
     if (removed && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) && db_index_clear(si) < 0) {
         db_rollback(si, "TTL index"); goto schedule;
     }
     if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "TTL commit"); goto schedule; }
+    committed = 1;
     if (removed) {
         db_index_request(si);
         DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history TTL committed tbl=%s removed=%d cutoff_ms=%llu",
             SI_TBL(si), removed, (unsigned long long)cut);
     }
 schedule:
-    si->ttl_timer = uasync_set_timeout(si->db_sync->inst->ua, DB_SYNC_TTL_INTERVAL * 10000ULL, si, db_ttl_tick, "history_ttl");
+    uint64_t delay = committed && removed == DB_BATCH_RECORDS ? 0 : DB_SYNC_TTL_INTERVAL * 10000ULL;
+    si->ttl_timer = uasync_set_timeout(si->db_sync->inst->ua, delay, si, db_ttl_tick, "history_ttl");
     if (!si->ttl_timer) { db_error(si, "TTL timer"); db_sync_resume_peer_check(si->db_sync); }
 }
 

tests/test_db_sync_integrity.c

@@ -7,6 +7,14 @@ static int fail_verify, fail_commit, fail_timer, fail_blob, blob_nomem;
 static unsigned fault_hits;
 static sqlite3* verification_db;
 static int assert_unlocked;
+static int concentrate_bucket;
+extern int __real_chat_message_id(uint64_t, uint64_t, const uint8_t[64], uint8_t[32]);
+int __wrap_chat_message_id(uint64_t group, uint64_t ts, const uint8_t sig[64], uint8_t key[32]) {
+    int rc = __real_chat_message_id(group, ts, sig, key);
+    /* Только для плотного тестового листа: подписи настоящие, оставшиеся223 бита ключа сохраняются. */
+    if (rc == 0 && concentrate_bucket) { key[0] = key[1] = key[2] = 0; key[3] &= 0x7f; }
+    return rc;
+}
 extern int __real_EVP_DigestVerifyInit(EVP_MD_CTX*, EVP_PKEY_CTX**, const EVP_MD*, ENGINE*, EVP_PKEY*);
 int __wrap_EVP_DigestVerifyInit(EVP_MD_CTX* ctx, EVP_PKEY_CTX** pctx, const EVP_MD* md, ENGINE* engine, EVP_PKEY* key) {
     if (assert_unlocked) assert(sqlite3_get_autocommit(verification_db));
@@ -214,10 +222,77 @@ static void test_callbacks(void) {
     destroy(&f);
     DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS completion can remove itself, next callback and channel safely");
 }
+
+static void copy_pages(struct fixture* source, struct fixture* target, int insert_before_cursor) {
+    merkle_key_t after = {{0}}, next; int av = 0, more, pages = 0;
+    do {
+        uint8_t data[16384]; size_t len = sizeof(data);
+        assert(db_message_page(&source->inst, "10", 0, av, &after, data, &len, &next, &more) == 0);
+        assert(db_message_apply(&target->inst, "10", 2, 0, av, &after, &next, more, data, len) == 0);
+        after = next; av = 1; pages++;
+        if (pages == 1 && insert_before_cursor) {
+            uint8_t sig[64], key[32]; int found = 0;
+            for (uint64_t ts = 1000; ts < 2000; ts++) {
+                sign(source, ts, "inserted before cursor", sig);
+                assert(chat_message_id(10, ts, sig, key) == 0);
+                if (memcmp(key, after.bytes, 32) < 0) {
+                    assert(insert(source, ts, "inserted before cursor", sig) == 0); found = 1; break;
+                }
+            }
+            assert(found);
+        }
+    } while (more);
+    assert(pages >= 3);
+}
+static void test_pages(void) {
+    concentrate_bucket = 1;
+    struct fixture a, b; init(&a); init(&b); uint8_t sig[64], ra[32], rb[32];
+    b.si->next_id = 900;
+    for (uint64_t ts = 1; ts <= 40; ts++) { sign(&a, ts, "page record", sig); assert(insert(&a, ts, "page record", sig) == 0); }
+    assert(db_sync_verify(a.si) == 0 && db_sync_verify(b.si) == 0);
+    copy_pages(&a, &b, 1);
+    assert(db_sync_count(a.si) == 41 && db_sync_count(b.si) == 40);
+    assert(db_sync_read_root(a.si, ra) == 0 && db_sync_read_root(b.si, rb) == 0 && memcmp(ra, rb, 32));
+    copy_pages(&a, &b, 0);
+    assert(db_sync_count(b.si) == 41 && db_sync_read_root(b.si, rb) == 0 && !memcmp(ra, rb, 32));
+    copy_pages(&a, &b, 0); assert(db_sync_count(b.si) == 41);
+    destroy(&a); destroy(&b); concentrate_bucket = 0;
+    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS multi-page leaf, different local ids, duplicate replay and insertion before cursor need a new root check");
+}
+static void test_deleted_author(void) {
+    struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "historic", sig);
+    struct db_record r = {.timestamp = 100, .author = 99, .data = "historic", .data_len = 8, .sig_len = 64};
+    memcpy(r.signature, sig, 64);
+    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);
+    sql(&f, "CREATE TABLE peers_10(node_id INTEGER PRIMARY KEY,deleted INTEGER);INSERT INTO peers_10 VALUES(99,1)");
+    assert(db_insert_one(f.si, &r, NULL) == 0 && db_sync_verify(f.si) == 0 && page(&f, sig) == 0 && db_sync_count(f.si) == 1);
+    destroy(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS valid history of a deleted author remains in the root and outgoing page");
+}
+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;
+    config->global.db_sync_ttl = 3600;
+    uint64_t now = db_now_ms(f.si), old = now - 3610000;
+    sign(&f, now, "fresh", sig); assert(insert(&f, now, "fresh", sig) == 0);
+    for (unsigned i = 0; i < 20; i++) { sign(&f, old + i, "expired", sig); assert(insert(&f, old + i, "expired", sig) == 0); }
+    sql(&f, "UPDATE history SET flags=1 WHERE rowid=2");
+    assert(db_sync_verify(f.si) == 0);
+    db_ttl_tick(f.si);
+    assert(db_sync_count(f.si) == 5 && f.si->index.pending && sqlite3_get_autocommit(f.db.db));
+    uasync_poll(f.inst.ua, 0);
+    assert(db_sync_count(f.si) == 2 && db_sync_verify(f.si) == 0);
+    destroy(&f); u_free(config);
+    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS TTL uses milliseconds and short batches, preserving fresh and individually acknowledged records");
+}
 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();
     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);
     return 0;