diff --git a/src/chat/chat_attachment.c b/src/chat/chat_attachment.c index 57ecca14..9ff2251e 100644 --- a/src/chat/chat_attachment.c +++ b/src/chat/chat_attachment.c @@ -544,6 +544,43 @@ static void ak_tick(void* arg) { if (!CC(inst)->attachment_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: timer allocation failed"); } +/* После crash/отмены убрать только собственные подготовки, не имеющие сообщения. + * Вызывается до запуска workers либо после их полного завершения. */ +static void ak_recover_preparations(struct UTUN_INSTANCE* inst) { + char** channels = NULL; int count = 0, columns = 0; + if (sqlite3_get_table(CC(inst)->db, "SELECT channel_id FROM channels", &channels, &count, &columns, NULL) != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: recovery channel scan failed: %s", sqlite3_errmsg(CC(inst)->db)); return; + } + for (int i = 0; i < count; i++) { + const char* channel = channels[(i + 1) * columns]; + if (!ak_channel(channel)) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid channel in recovery"); continue; } + char condition[384], sql[512]; + snprintf(condition, sizeof(condition), "a.channel='%s' AND a.author=%llu AND NOT EXISTS" + "(SELECT 1 FROM msg_%s m WHERE m.message_id=a.message_id)", channel, (unsigned long long)inst->node_id, channel); + snprintf(sql, sizeof(sql), "SELECT cipher_path,plain_path FROM attachment_keys a WHERE %s", condition); + char** paths = NULL; int rows = 0, fields = 0, failed = 0; + if (sqlite3_get_table(CC(inst)->db, sql, &paths, &rows, &fields, NULL) != SQLITE_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: recovery file scan failed channel=%s: %s", channel, sqlite3_errmsg(CC(inst)->db)); + sqlite3_free_table(paths); continue; + } + for (int row = 0; row < rows; row++) { + for (int field = 0; field < fields; field++) { + const char* path = paths[(row + 1) * fields + field]; + char temporary[1040]; snprintf(temporary, sizeof(temporary), "%s.part", path); + if ((remove(path) && errno != ENOENT) || (remove(temporary) && errno != ENOENT)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: unfinished file removal failed path=%s: %s", path, strerror(errno)); failed = 1; + } + } + } + sqlite3_free_table(paths); + if (rows && !failed) { + snprintf(sql, sizeof(sql), "DELETE FROM attachment_keys AS a WHERE %s", condition); + if (!ak_exec(inst, sql)) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: discarded unpublished preparations channel=%s count=%d", channel, rows); + } + } + sqlite3_free_table(channels); +} + /* Отдельные локальные таблицы никогда не регистрируются в db_sync/Merkle. */ int chat_attachment_init(struct UTUN_INSTANCE* inst) { if (!CC(inst) || !inst->media_async) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid initialization"); return -1; } @@ -560,6 +597,7 @@ int chat_attachment_init(struct UTUN_INSTANCE* inst) { "UPDATE attachment_keys SET decrypting=0; UPDATE attachment_key_outbox SET retry_at=0;")) return -1; CC(inst)->attachment_closing = 0; CC(inst)->attachment_initialized = 1; + ak_recover_preparations(inst); if (etcp_router_bind(inst, ETCP_RT_ID_ATTACHMENT_KEYS, ak_receive)) return -1; CC(inst)->attachment_timer = uasync_set_timeout(inst->ua, 10000, inst, ak_tick, "attachment keys"); if (!CC(inst)->attachment_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: initial timer allocation failed"); return -1; } @@ -578,6 +616,7 @@ void chat_attachment_quiesce(struct UTUN_INSTANCE* inst) { void chat_attachment_destroy(struct UTUN_INSTANCE* inst) { chat_attachment_quiesce(inst); + if (CC(inst) && CC(inst)->attachment_initialized) ak_recover_preparations(inst); } /* Удаление ключей отменяет completion активных workers; они удалят свои результаты. */ diff --git a/tests/test_attachment_send.c b/tests/test_attachment_send.c index 701ed393..c764e32c 100644 --- a/tests/test_attachment_send.c +++ b/tests/test_attachment_send.c @@ -7,6 +7,7 @@ #include "../src/utun_instance.h" #include "../src/chat/chat_core.h" #include "../src/chat/chat_event.h" +#include "../src/chat/chat_attachment.h" #include "../src/dm/dm_core.h" #include "../src/dm/dm_crypto.h" #include "../src/media_async/attachment_send.h" @@ -80,6 +81,15 @@ int main(void) { REQUIRE(!dm_list_messages_json(inst, active_target, 20, 0, json, sizeof(json), &length) && !strcmp(json, "[]")); /* Остановка до poll завершает preparation с CANCELLED, без регистрации сообщения. */ + chat_core_create_channel_auto(inst, "encrypted cancel"); + sqlite3_stmt* channel_row = NULL; + REQUIRE(!sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT channel_id FROM channels WHERE name='encrypted cancel'", -1, &channel_row, NULL)); + REQUIRE(sqlite3_step(channel_row) == SQLITE_ROW); + char channel_id[32]; snprintf(channel_id, sizeof(channel_id), "%s", sqlite3_column_text(channel_row, 0)); sqlite3_finalize(channel_row); + char prepared[1024]; snprintf(prepared, sizeof(prepared), "%s/cancel-encrypted.bin", dir); + FILE* prepared_file = fopen(prepared, "wb"); REQUIRE(prepared_file && fputc(42, prepared_file) != EOF && !fclose(prepared_file)); + struct attachment_info encrypted_info = { .kind = ATTACHMENT_FILE, .name = "cancel.bin" }; uint8_t no_reply[32] = {0}; + REQUIRE(!chat_attachment_send(inst, channel_id, prepared, &encrypted_info, "application/octet-stream", no_reply, 0)); req = u_calloc(1, sizeof(*req)); REQUIRE(req); req->inst = inst; req->is_dm = 1; req->info.kind = ATTACHMENT_VOICE; req->preset = 1; snprintf(req->target, sizeof(req->target), "%s", conv); @@ -87,6 +97,9 @@ int main(void) { attachment_send_trampoline(req); chat_service_stop(inst); REQUIRE(test.released == 2 && test.failed == 1 && count(inst) == 1); + sqlite3_stmt* cancelled_key = NULL; + REQUIRE(!sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT count(*) FROM attachment_keys", -1, &cancelled_key, NULL)); + REQUIRE(sqlite3_step(cancelled_key) == SQLITE_ROW && sqlite3_column_int(cancelled_key, 0) == 0); sqlite3_finalize(cancelled_key); uasync_poll(ua, 0); REQUIRE(test.failed == 1); /* Отклонение после stop удаляет только переданный собственный cache. */