Recover cancelled unpublished encryption jobs without retaining keys

This commit is contained in:
evgeny committed 2026-10-18 10:46:56 +02:00
1 parent a31ade1054
commit 057bdb3112
2 files changed
+52

No files matched your search

+39
View File
@@ -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; они удалят свои результаты. */
+13
View File
@@ -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. */