From a87ece370c2f8fe4e2d9b422602848e7e7196cc0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 17:11:39 +0100 Subject: [PATCH] Revalidate history peers after membership and cover preflight integrity --- src/chat/member_sync.c | 2 + tests/test_db_sync_integrity.c | 227 ++++++++++++++++++--------------- 2 files changed, 129 insertions(+), 100 deletions(-) diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 3e2f7ca7..a86a1d0b 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -1,5 +1,6 @@ #include "../routing_layer/topo_group_connect.h" #include "member_sync.h" +#include "db_sync.h" #include "chat_feedback.h" #include "../routing_layer/topo_node_sqlite.h" @@ -777,6 +778,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id, void member_sync_record_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id, unsigned changes) { merkle_sync_changed(inst, MT_DATA_MEMBERS, ns); if (changes & MS_APPLY_FEEDBACK) chat_feedback_member_committed(inst, node_id); + db_sync_members_changed(inst, ns); struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)) : NULL; struct ETCP_CONN* conn = instance_find_conn(inst, node_id); if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) { diff --git a/tests/test_db_sync_integrity.c b/tests/test_db_sync_integrity.c index d142368b..1355bad5 100644 --- a/tests/test_db_sync_integrity.c +++ b/tests/test_db_sync_integrity.c @@ -5,8 +5,11 @@ static int fail_verify, fail_commit, fail_timer, fail_blob, blob_nomem; static unsigned fault_hits; +static sqlite3* verification_db; +static int assert_unlocked; 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)); if (fail_verify && !--fail_verify) { fault_hits++; return 0; } return __real_EVP_DigestVerifyInit(ctx, pctx, md, engine, key); } @@ -30,168 +33,192 @@ int __wrap_sqlite3_errcode(sqlite3* db) { extern void* __real_uasync_set_timeout(struct UASYNC*, uint64_t, void*, void (*)(void*), const char*); void* __wrap_uasync_set_timeout(struct UASYNC* ua, uint64_t delay, void* arg, void (*cb)(void*), const char* name) { - if (fail_timer && !strcmp(name, "db_recalc")) { fail_timer = 0; fault_hits++; return NULL; } + if (fail_timer && !strcmp(name, "history_preflight")) { fail_timer = 0; fault_hits++; return NULL; } return __real_uasync_set_timeout(ua, delay, arg, cb, name); } + struct fixture { struct UTUN_INSTANCE inst; struct DB_SYNC db; - struct DB_SYNC_INSTANCE si, other; + struct DB_SYNC_INSTANCE* si; + struct DB_SYNC_INSTANCE* other; uint8_t seed[32]; }; - -static void sql(struct fixture* f, const char* text) { - assert(sqlite3_exec(f->db.db, text, NULL, NULL, NULL) == SQLITE_OK); +static int empty_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) { + (void)ctx; (void)ns; (void)level; (void)prefix; (void)hash; return 0; +} +static int empty_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after, + uint8_t* buf, size_t* len, merkle_key_t* next, int* more) { + (void)ctx; (void)ns; (void)prefix; (void)av; (void)after; (void)buf; + *len = 0; *more = 0; memset(next, 0, sizeof(*next)); return 0; +} +static int empty_apply(void* ctx, const char* ns, uint64_t peer, const uint8_t* p, size_t len) { + (void)ctx; (void)ns; (void)peer; (void)p; return len ? -1 : 0; +} +static const struct merkle_sync_data_ops empty_ops = { + .update_bucket_hash = empty_hash, .get_page = empty_page, .apply_items = empty_apply +}; +static void sql(struct fixture* f, const char* query) { + int rc = sqlite3_exec(f->db.db, query, NULL, NULL, NULL); + if (rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "integrity SQL %s: %s", query, sqlite3_errmsg(f->db.db)); + assert(rc == SQLITE_OK); +} +static struct DB_SYNC_INSTANCE* new_table(struct fixture* f, const char* name, uint64_t group) { + struct DB_SYNC_INSTANCE* si = u_calloc(1, sizeof(*si)); assert(si); + si->db_sync = &f->db; si->enabled = 1; si->group_id = group; si->next_id = 1; si->sync_gated = 1; + snprintf(si->table_name, sizeof(si->table_name), "%s", name); + snprintf(si->ns, sizeof(si->ns), "%llu", (unsigned long long)group); + assert(si_table_create(si, name, 0) == 0); return si; } - static void init(struct fixture* f) { - memset(f, 0, sizeof(*f)); - f->inst.node_id = 1; f->inst.ua = uasync_create(); assert(f->inst.ua); - f->inst.connections = queue_new(f->inst.ua, 8, 0, 8, "history test connections"); assert(f->inst.connections); + memset(f, 0, sizeof(*f)); f->inst.node_id = 1; f->inst.ua = uasync_create(); assert(f->inst.ua); + f->inst.connections = queue_new(f->inst.ua, 8, 0, 8, "history test"); assert(f->inst.connections); memset(f->seed, 9, sizeof(f->seed)); - EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, f->seed, sizeof(f->seed)); assert(key); + EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, f->seed, 32); assert(key); size_t len = 32; - assert(EVP_PKEY_get_raw_public_key(key, f->inst.my_ed25519_pubkey, &len) == 1 && len == 32); EVP_PKEY_free(key); + assert(EVP_PKEY_get_raw_public_key(key, f->inst.my_ed25519_pubkey, &len) == 1); EVP_PKEY_free(key); assert(sqlite3_open(":memory:", &f->db.db) == SQLITE_OK); - f->db.inst = &f->inst; f->inst.db_sync = &f->db; - f->si.db_sync = &f->db; f->si.enabled = 1; f->si.group_id = 10; f->si.next_id = 1; f->si.sync_gated = 1; - strcpy(f->si.table_name, "history"); assert(si_table_create(&f->si, "history", 0) == 0); - f->other = f->si; f->other.group_id = 11; strcpy(f->other.table_name, "other_history"); - assert(si_table_create(&f->other, "other_history", 0) == 0); + f->db.inst = &f->inst; f->db.enabled = 1; f->inst.db_sync = &f->db; f->inst.topo_sqlite_db = f->db.db; + verification_db = f->db.db; + sql(f, "CREATE TABLE nodes(node_id INTEGER PRIMARY KEY,ed25519_pubkey BLOB,deleted INTEGER DEFAULT 0)"); + f->si = new_table(f, "history", 10); f->other = new_table(f, "other_history", 11); + f->db.instances = u_calloc(2, sizeof(*f->db.instances)); assert(f->db.instances); + f->db.instances[0] = f->si; f->db.instances[1] = f->other; f->db.instance_count = f->db.instance_capacity = 2; + assert(merkle_sync_init(&f->inst, ETCP_ID_MEMBER_SYNC, &empty_ops, f) == 0 && db_sync_register_model(&f->inst) == 0); } - static void destroy(struct fixture* f) { - if (f->si.recalc.timer) uasync_cancel_timeout(f->inst.ua, f->si.recalc.timer); - if (f->other.recalc.timer) uasync_cancel_timeout(f->inst.ua, f->other.recalc.timer); - u_free(f->si.peers); u_free(f->other.peers); - queue_free(f->inst.connections); + f->db.enabled = 0; + if (f->db.peer_check_timer) uasync_cancel_timeout(f->inst.ua, f->db.peer_check_timer); + merkle_sync_destroy(&f->inst); + while (f->db.instance_count) db_sync_instance_remove(f->db.instances[f->db.instance_count - 1]); + while (f->db.done_cbks) { struct db_sync_done_cbk_entry* cb = f->db.done_cbks; f->db.done_cbks = cb->next; u_free(cb); } + u_free(f->db.instances); queue_free(f->inst.connections); assert(sqlite3_get_autocommit(f->db.db) && sqlite3_close(f->db.db) == SQLITE_OK); - uasync_destroy(f->inst.ua, 0); + uasync_destroy(f->inst.ua, 0); verification_db = NULL; } - static void sign(struct fixture* f, uint64_t ts, const char* text, uint8_t sig[64]) { uint8_t body[DB_SYNC_MESSAGE_MAX + 8]; size_t len = strlen(text); assert(len <= DB_SYNC_MESSAGE_MAX); - memcpy(body, &ts, 8); memcpy(body + 8, text, len); - assert(sc_ed25519_sign(f->seed, body, len + 8, sig) == SC_OK); + memcpy(body, &ts, 8); memcpy(body + 8, text, len); assert(sc_ed25519_sign(f->seed, body, len + 8, sig) == SC_OK); } - static int insert(struct fixture* f, uint64_t ts, const char* text, const uint8_t sig[64]) { - return db_sync_insert_signed(&f->si, text, strlen(text), sig, 64, ts, NULL); + return db_sync_insert_signed(f->si, text, strlen(text), sig, 64, ts, NULL); } - static void intact(struct fixture* f, const char* text) { - sqlite3_stmt* stmt = NULL; - assert(sqlite3_prepare_v2(f->db.db, "SELECT data FROM history", -1, &stmt, NULL) == SQLITE_OK); - assert(sqlite3_step(stmt) == SQLITE_ROW); - assert(sqlite3_column_bytes(stmt, 0) == (int)strlen(text)); - assert(!memcmp(sqlite3_column_blob(stmt, 0), text, strlen(text))); - assert(sqlite3_step(stmt) == SQLITE_DONE && sqlite3_finalize(stmt) == SQLITE_OK); - assert(sqlite3_get_autocommit(f->db.db)); + sqlite3_stmt* st = NULL; + assert(sqlite3_prepare_v2(f->db.db, "SELECT data FROM history", -1, &st, NULL) == SQLITE_OK && sqlite3_step(st) == SQLITE_ROW); + assert(sqlite3_column_bytes(st, 0) == (int)strlen(text) && !memcmp(sqlite3_column_blob(st, 0), text, strlen(text))); + assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK && sqlite3_get_autocommit(f->db.db)); +} +static int page(struct fixture* f, const uint8_t sig[64]) { + uint8_t id[32], data[16384]; size_t len = sizeof(data); merkle_key_t after = {{0}}, next; int more; + assert(chat_message_id(10, 100, sig, id) == 0); + return db_message_page(&f->inst, "10", merkle_sync_level_prefix(db_read64(id), MT_MAX_LEVEL), 0, &after, data, &len, &next, &more); } - static void test_preservation(void) { struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "original", sig); - assert(insert(&f, 100, "original", sig) == 0 && insert(&f, 100, "original", sig) == 1); + assert(insert(&f, 100, "original", sig) == 0 && db_sync_verify(f.si) == 0); + assert(insert(&f, 100, "original", sig) == 1); fail_verify = 1; assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original"); - fail_verify = 2; assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original"); + fail_verify = 2; assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "original"); fail_blob = 1; assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original"); - fail_blob = 1; - assert(si_send_data_batch(&f.si, 2, 0, 1, 0, DB_WANT_FROM_NONE) == -1); intact(&f, "original"); + fail_blob = 1; assert(page(&f, sig) == -1); intact(&f, "original"); + fail_verify = 1; assert(page(&f, sig) == -1); intact(&f, "original"); assert(insert(&f, 100, "modified", sig) == -2); intact(&f, "original"); sql(&f, "UPDATE history SET node_id=99"); - assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original"); - assert(si_send_data_batch(&f.si, 2, 0, 1, 0, DB_WANT_FROM_NONE) == -1); intact(&f, "original"); - f.inst.topo_sqlite_db = f.db.db; /* Отказ SQL lookup, в отличие от ещё неизвестного ключа. */ - assert(insert(&f, 100, "original", sig) == -1); intact(&f, "original"); - f.inst.topo_sqlite_db = NULL; - sql(&f, "UPDATE history SET node_id=1"); - fail_verify = 1; - assert(si_send_data_batch(&f.si, 2, 0, 1, 0, DB_WANT_FROM_NONE) == -1); intact(&f, "original"); - assert(db_sync_chain_verify(&f.si) == 0); + assert(insert(&f, 100, "original", sig) == -1 && page(&f, sig) == -1); intact(&f, "original"); + assert(db_sync_verify(f.si) == 1 && f.si->index.pending); + sql(&f, "DROP TABLE nodes"); + assert(db_sync_verify(f.si) == -1); intact(&f, "original"); + sql(&f, "CREATE TABLE nodes(node_id INTEGER PRIMARY KEY,ed25519_pubkey BLOB,deleted INTEGER);UPDATE history SET node_id=1"); + assert(db_sync_verify(f.si) == 0); destroy(&f); - DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history operational failure, missing key and invalid duplicate preserve stored bytes"); + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history valid copy survives invalid input, crypto/SQL/BLOB failures and missing key"); } - static void test_deletion(void) { struct fixture f; init(&f); uint8_t sig[64]; sign(&f, 100, "original", sig); assert(insert(&f, 100, "original", sig) == 0); - assert(db_sync_insert_signed(&f.other, "original", 8, sig, 64, 100, NULL) == 0); - struct SI_PEER* peer = si_peer_add(&f.si, 2); assert(peer); peer->sync_state = 2; peer->synced_pos = 20; - sql(&f, "UPDATE history SET data='modified'"); - sql(&f, "CREATE TRIGGER deny_delete BEFORE DELETE ON history BEGIN SELECT RAISE(FAIL,'delete denied'); END"); + assert(db_sync_insert_signed(f.other, "original", 8, sig, 64, 100, NULL) == 0 && db_sync_verify(f.si) == 0); + sql(&f, "UPDATE history SET data='modified';CREATE TRIGGER deny_delete BEFORE DELETE ON history BEGIN SELECT RAISE(FAIL,'delete denied');END"); assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "modified"); - assert(peer->sync_state == 2 && peer->synced_pos == 20); sql(&f, "DROP TRIGGER deny_delete"); fail_commit = 1; assert(insert(&f, 100, "modified", sig) == -1); intact(&f, "modified"); - assert(peer->sync_state == 2 && peer->synced_pos == 20); - assert(insert(&f, 100, "modified", sig) == -2 && db_sync_count(&f.si) == 0); - assert(!peer->sync_state && !peer->synced_pos && f.si.recalc.pending); - assert(db_sync_chain_verify(&f.si) == 0 && db_sync_count(&f.other) == 1); - /* Корректная входящая копия заменяет независимо проверенную порчу. */ + assert(insert(&f, 100, "modified", sig) == -2 && db_sync_count(f.si) == 0 && f.si->index.pending); + assert(db_sync_verify(f.si) == 0 && db_sync_count(f.other) == 1); assert(insert(&f, 100, "original", sig) == 0); sql(&f, "UPDATE history SET data='modified'"); assert(insert(&f, 100, "original", sig) == 0); intact(&f, "original"); sql(&f, "UPDATE history SET data=zeroblob(4001)"); - assert(si_send_data_batch(&f.si, 2, 0, 1, 0, DB_WANT_FROM_NONE) == -1 && db_sync_count(&f.si) == 0); - const unsigned sizes[] = {0, 1, 63, 65, 255}; + assert(db_sync_verify(f.si) == 0 && db_sync_count(f.si) == 0); + const unsigned sizes[] = {0,1,63,65,255}; for (unsigned i = 0; i < sizeof(sizes) / sizeof(*sizes); i++) { assert(insert(&f, 100, "original", sig) == 0); char query[96]; snprintf(query, sizeof(query), "UPDATE history SET author_signature=zeroblob(%u)", sizes[i]); sql(&f, query); - assert(si_send_data_batch(&f.si, 2, 0, 1, 0, DB_WANT_FROM_NONE) == -1 && db_sync_count(&f.si) == 0); - assert(db_sync_chain_verify(&f.si) == 0); + assert(db_sync_verify(f.si) == 0 && db_sync_count(f.si) == 0); } - assert(insert(&f, 100, "original", sig) == 0); - sql(&f, "UPDATE history SET author_signature=zeroblob(1)"); - assert(db_sync_chain_verify(&f.si) == 0 && db_sync_count(&f.si) == 0); /* pending recalc also cleans malformed signature */ destroy(&f); - DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history stored corruption deleted, SQL failures rolled back, peers reset, other channel retained"); + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS proven malformed/signature-invalid stored records removed; rollback preserves data and other channel"); } - -static void test_recalc(void) { - struct fixture f; init(&f); uint8_t sig[64]; +static void test_index(void) { + struct fixture f; init(&f); uint8_t sig[64], root[32], prior[32]; for (uint64_t ts = 1; ts <= 55; ts++) { sign(&f, ts, "record", sig); assert(insert(&f, ts, "record", sig) == 0); } - uint64_t next = f.si.next_id, hash = 0; - fail_blob = 1; - assert(db_sync_last_chain_hash8(&f.si, &hash) == -1 && db_sync_count(&f.si) == 55 && f.si.recalc.pending); - fail_commit = 1; - assert(db_sync_last_chain_hash8(&f.si, &hash) == -1 && f.si.recalc.pending && f.si.recalc.pos == 0); - assert(sqlite3_get_autocommit(f.db.db)); - sql(&f, "CREATE TRIGGER deny_hash BEFORE UPDATE OF chain_hash ON history BEGIN SELECT RAISE(FAIL,'hash denied'); END"); - assert(db_sync_last_chain_hash8(&f.si, &hash) == -1 && f.si.recalc.pending && f.si.recalc.pos == 0); + assert_unlocked = 1; fail_blob = 1; + assert(db_sync_verify(f.si) == -1 && db_sync_count(f.si) == 55 && f.si->index.pending); + fail_commit = 1; assert(db_sync_verify(f.si) == -1 && f.si->index.pending && sqlite3_get_autocommit(f.db.db)); + sql(&f, "CREATE TRIGGER deny_hash BEFORE INSERT ON merkle_tree_hash WHEN NEW.namespace='2:10' BEGIN SELECT RAISE(FAIL,'hash denied');END"); + assert(db_sync_verify(f.si) == -1 && f.si->index.pending && db_sync_read_root(f.si, root) == 1); sql(&f, "DROP TRIGGER deny_hash"); - assert(db_sync_last_chain_hash8(&f.si, &hash) == 0 && hash && !f.si.recalc.pending && db_sync_chain_verify(&f.si) == 0); + assert(db_sync_verify(f.si) == 0 && db_sync_read_root(f.si, prior) == 0); assert_unlocked = 0; + uint64_t next = f.si->next_id; sign(&f, 56, "record", sig); fail_commit = 1; - assert(insert(&f, 56, "record", sig) == -1 && db_sync_count(&f.si) == 55 && f.si.next_id == next); - fail_timer = 1; assert(insert(&f, 56, "record", sig) == 0 && f.si.recalc.pending && !f.si.recalc.timer); - assert(db_sync_chain_verify(&f.si) == 0 && !f.si.recalc.pending); + assert(insert(&f, 56, "record", sig) == -1 && db_sync_count(f.si) == 55 && f.si->next_id == next); + assert(db_sync_read_root(f.si, root) == 0 && !memcmp(root, prior, 32)); + fail_timer = 1; db_index_request(f.si); + assert(!fail_timer && f.si->index.pending && !f.si->index.timer && db_sync_verify(f.si) == 0); + assert(db_sync_read_root(f.si, prior) == 0); + sql(&f, "UPDATE history SET id=id+900,flags=1,local_attrs='local',delivered_peers=17,delivery_chain='different'"); + assert(db_sync_verify(f.si) == 0 && db_sync_read_root(f.si, root) == 0 && !memcmp(root, prior, 32)); destroy(&f); - DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history recalc retries commit/update/timer failures across multiple batches"); + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS bounded preflight verifies outside write-lock; index failures recover; root ignores local ids and delivery"); } - static void test_delivery(void) { struct fixture f; init(&f); uint8_t a[64], b[64], id_a[32], id_b[32]; sign(&f, 100, "first", a); sign(&f, 100, "second", b); assert(insert(&f, 100, "first", a) == 0 && insert(&f, 100, "second", b) == 0); assert(chat_message_id(10, 100, a, id_a) == 0 && chat_message_id(10, 100, b, id_b) == 0); - db_handle_ack_push(&f.si, 2, id_a, 32); db_handle_ack_push(&f.si, 2, id_a, 32); - sqlite3_stmt* stmt = NULL; - assert(sqlite3_prepare_v2(f.db.db, "SELECT data,delivered_peers,flags FROM history ORDER BY data", -1, &stmt, NULL) == SQLITE_OK); - assert(sqlite3_step(stmt) == SQLITE_ROW && sqlite3_column_int(stmt, 1) == 1 && sqlite3_column_int(stmt, 2) == 1); - assert(sqlite3_step(stmt) == SQLITE_ROW && sqlite3_column_int(stmt, 1) == 0 && sqlite3_column_int(stmt, 2) == 0); - assert(sqlite3_step(stmt) == SQLITE_DONE && sqlite3_finalize(stmt) == SQLITE_OK); - sql(&f, "CREATE TRIGGER deny_ack BEFORE UPDATE ON history BEGIN SELECT RAISE(FAIL,'ACK denied'); END"); - assert(si_delivery_update(&f.si, id_b, 3) == -1); - sql(&f, "DROP TRIGGER deny_ack"); - assert(si_delivery_update(&f.si, id_b, 3) == 0); + assert(si_delivery_update(f.si, id_a, 2) == 0 && si_delivery_update(f.si, id_a, 2) == 0); + sqlite3_stmt* st = NULL; + assert(sqlite3_prepare_v2(f.db.db, "SELECT data,delivered_peers,flags FROM history ORDER BY data", -1, &st, NULL) == SQLITE_OK); + assert(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_int(st, 1) == 1 && sqlite3_column_int(st, 2) == 1); + assert(sqlite3_step(st) == SQLITE_ROW && !sqlite3_column_int(st, 1) && !sqlite3_column_int(st, 2)); + assert(sqlite3_step(st) == SQLITE_DONE && sqlite3_finalize(st) == SQLITE_OK); + sql(&f, "CREATE TRIGGER deny_ack BEFORE UPDATE ON history BEGIN SELECT RAISE(FAIL,'ACK denied');END"); + assert(si_delivery_update(f.si, id_b, 3) == -1); sql(&f, "DROP TRIGGER deny_ack"); + assert(si_delivery_update(f.si, id_b, 3) == 0); destroy(&f); - DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history ACK identifies one same-author same-timestamp message and deduplicates confirmations"); + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS exact same-timestamp ACK identity and idempotence"); +} +static int first_calls, second_calls; +static void second_done(struct DB_SYNC_INSTANCE* si, uint64_t peer, void* arg) { + (void)si; (void)peer; (void)arg; second_calls++; +} +static void first_done(struct DB_SYNC_INSTANCE* si, uint64_t peer, void* arg) { + (void)peer; struct fixture* f = arg; first_calls++; + db_sync_remove_done_cbk(&f->inst, first_done, f); db_sync_remove_done_cbk(&f->inst, second_done, f); + db_sync_instance_remove(si); +} +static void test_callbacks(void) { + struct fixture f; init(&f); + db_sync_add_done_cbk(&f.inst, second_done, &f); db_sync_add_done_cbk(&f.inst, first_done, &f); + db_message_result(&f.inst, "10", 2, MT_OK); + assert(first_calls == 1 && !second_calls && !db_instance_find(&f.db, 10) && !f.db.done_cbks); + destroy(&f); + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS completion can remove itself, next callback and channel safely"); } - 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_recalc(); test_delivery(); + test_preservation(); test_deletion(); test_index(); test_delivery(); test_callbacks(); + 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); - DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS history integrity; injected failures=%u; tracked allocations restored", fault_hits); return 0; }