From 3a4143183de8698acc2064f0ca8aac066ee4e10a Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 17:28:08 +0100 Subject: [PATCH] Cover stable history cursors, deleted authors and bounded TTL cleanup --- src/chat/db_sync.c | 8 +++- tests/test_db_sync_integrity.c | 75 ++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 2 deletions(-) diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index fa0e7703..abcdf610 100644 --- a/src/chat/db_sync.c +++ b/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); } } diff --git a/tests/test_db_sync_integrity.c b/tests/test_db_sync_integrity.c index 1355bad5..5585279f 100644 --- a/tests/test_db_sync_integrity.c +++ b/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;