diff --git a/src/media_delivery/media_delivery.c b/src/media_delivery/media_delivery.c index c3f152c5..b673995a 100644 --- a/src/media_delivery/media_delivery.c +++ b/src/media_delivery/media_delivery.c @@ -90,6 +90,11 @@ static int md_sql_begin(sqlite3* db, const char* sql) { return md_sql_exec(db, sql); } +/* SQLite может сама откатить транзакцию при отказе commit/IO; чужую или уже завершённую не трогаем. */ +static void md_sql_rollback(sqlite3* db) { + if (!sqlite3_get_autocommit(db)) md_sql_exec(db, "ROLLBACK"); +} + /* Вызывается внутри собственной транзакции. Максимум (status,timestamp) идемпотентен; * REPLACE выполняется только при изменении, поэтому повтор не расходует AUTOINCREMENT ID. */ static int md_ba_merge(sqlite3* db, const uint8_t block_id[16], uint64_t group_id, @@ -469,7 +474,7 @@ static size_t md_repl_page(struct media_super_peer* peer, uint8_t packet[MEDIA_S (unsigned long long)from, (unsigned long long)(more ? last : high), count, more); return sizeof(*r) + count * sizeof(*rows); failed: - md_sql_exec(md->db, "ROLLBACK"); + md_sql_rollback(md->db); return sizeof(*r); } @@ -648,7 +653,7 @@ static void md_handle_announcement(struct media_delivery_ctx* md, uint64_t group if (!md_sql_begin(md->db, "BEGIN IMMEDIATE")) { changed = md_ba_merge(md->db, hb->block_id, group_id, from_node, hb->chunk, hb->timestamp, status); if (changed < 0 || md_sql_exec(md->db, "COMMIT")) { - md_sql_exec(md->db, "ROLLBACK"); changed = -1; + md_sql_rollback(md->db); changed = -1; } } if (status) { @@ -710,7 +715,7 @@ static void md_handle_sync_resp(struct media_delivery_ctx* md, uint64_t group_id if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)from_node); if (rc == SQLITE_OK) rc = sqlite3_step(st); if (md_sql_finish(md->db, st, rc, "reset ahead cursor") || md_sql_exec(md->db, "COMMIT")) { - md_sql_exec(md->db, "ROLLBACK"); return; + md_sql_rollback(md->db); return; } peer->request_pending = peer->request_send = 0; peer->timeout_tb = MEDIA_REPL_TIMEOUT_MAX_TB; @@ -741,7 +746,7 @@ static void md_handle_sync_resp(struct media_delivery_ctx* md, uint64_t group_id int applied = md_ba_merge(md->db, rows[i].block_id, group_id, md_read_le(&rows[i].node_id, 8), (uint32_t)md_read_le(&rows[i].chunk, 4), (int64_t)md_read_le(&rows[i].timestamp, 8), rows[i].status); - if (applied < 0) { md_sql_exec(md->db, "ROLLBACK"); return; } + if (applied < 0) { md_sql_rollback(md->db); return; } changed += applied; } sqlite3_stmt* st = NULL; @@ -753,7 +758,7 @@ static void md_handle_sync_resp(struct media_delivery_ctx* md, uint64_t group_id if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)next); if (rc == SQLITE_OK) rc = sqlite3_step(st); if (md_sql_finish(md->db, st, rc, "save RX cursor") || md_sql_exec(md->db, "COMMIT")) { - md_sql_exec(md->db, "ROLLBACK"); return; + md_sql_rollback(md->db); return; } DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: SYNC committed group=%llx peer=%llx uuid=%02x%02x epoch=%02x%02x from=%llu next=%llu n=%u changed=%d more=%u dirty=%u", MD_ID, (unsigned long long)group_id, (unsigned long long)from_node, r->request_id[0], r->request_id[1], @@ -1144,7 +1149,7 @@ static int md_purge_group(struct media_delivery_ctx* md, uint64_t group_id, int MD_ID, (unsigned long long)group_id, removed, all); return 0; failed: - md_sql_exec(md->db, "ROLLBACK"); return -1; + md_sql_rollback(md->db); return -1; } /* Единственный суперузел тоже должен реплицировать собственные зарегистрированные блоки. @@ -1166,7 +1171,7 @@ static int md_seed_local(struct media_delivery_ctx* md, uint64_t group_id) { changed += applied; } if (md_sql_finish(md->db, st, rc, "seed own blocks") || md_sql_exec(md->db, "COMMIT")) { - md_sql_exec(md->db, "ROLLBACK"); return -1; + md_sql_rollback(md->db); return -1; } if (changed) { DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: own blocks seeded group=%llx changed=%d", MD_ID, @@ -1398,7 +1403,7 @@ static int media_delivery_create_tables(struct UTUN_INSTANCE* inst) { DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: catalog initialized reset=%d epoch=%02x%02x", MD_ID, reset, epoch[0], epoch[1]); return 0; failed: - md_sql_exec(db, "ROLLBACK"); return -1; + md_sql_rollback(db); return -1; } /* ── public API ── */ diff --git a/tests/test_media_delivery_integration.c b/tests/test_media_delivery_integration.c index 4365148c..dc37a409 100644 --- a/tests/test_media_delivery_integration.c +++ b/tests/test_media_delivery_integration.c @@ -20,6 +20,7 @@ #include "../routing_layer/topo_node.h" #include "../routing_layer/topo_node_sqlite.h" #include "../chat/member_sync.h" +#include "../chat/chat_core.h" #include "../transport_layer/secure_channel.h" #include "../lib/u_async.h" #include "../lib/debug_config.h" @@ -346,7 +347,11 @@ int main(void) { if (!g_a || !g_b) { printf("FAIL: instance create\n"); goto done; } test_etcp_set_peer(g_a, "to_b", g_b, "s1"); utun_instance_init(g_a); utun_instance_init(g_b); - media_delivery_init(g_a); media_delivery_init(g_b); + char chat_db_a[512], chat_db_b[512]; + snprintf(chat_db_a, sizeof(chat_db_a), "%s/chats.db", db_a); + snprintf(chat_db_b, sizeof(chat_db_b), "%s/chats.db", db_b); + if (chat_core_init(g_a, chat_db_a) || chat_core_init(g_b, chat_db_b) + || media_delivery_init(g_a) || media_delivery_init(g_b)) { FAIL("chat/media initialization"); goto done; } if (setup_chat()) { FAIL("CHAT setup failed"); goto done; }