From b71c3040d4dd49b045569ebcd46b38420be44342 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 01:35:30 +0100 Subject: [PATCH] Claim decryption before workers and retry retained completions --- src/chat/chat_attachment.c | 63 +++++++++++++++++++++++++++++++++++--- src/chat/chat_attachment.h | 3 ++ 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/src/chat/chat_attachment.c b/src/chat/chat_attachment.c index 7a08c7ed..488f734b 100644 --- a/src/chat/chat_attachment.c +++ b/src/chat/chat_attachment.c @@ -955,6 +955,46 @@ fail: return -1; } +/* Пометить ровно один файл собственной транзакцией до запуска расшифровки. */ +static int ak_claim_decryption(struct ak_file* job) { + struct UTUN_INSTANCE* inst = job->inst; + sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL; + const char* stage = "begin claim"; + int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "prepare claim"; + rc = sqlite3_prepare_v2(db, "UPDATE attachment_keys SET decrypting=1 WHERE channel=? AND crypto_id=? AND key=?" + " AND downloaded=1 AND wanted=1 AND ready=0 AND decrypting=0 AND error=0", -1, &st, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "bind channel"; rc = sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "bind identity"; rc = sqlite3_bind_blob(st, 2, job->id, 16, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "bind key"; rc = sqlite3_bind_blob(st, 3, job->key, 32, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "write claim"; rc = sqlite3_step(st); + if (rc != SQLITE_DONE) goto fail; + int changed = sqlite3_changes(db); + stage = "finalize claim"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + if (changed != 1) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: decryption claim rows=%d expected=1 channel=%s", changed, job->channel); + stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail; + } + stage = "commit claim"; rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "attachment keys: decryption claimed channel=%s active=%u", + job->channel, CC(inst)->attachment_file_count); + return 0; +fail: + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: decryption claim channel=%s stage=%s rc=%d (%s) db=%s", + job->channel, stage, rc, sqlite3_errstr(rc), sqlite3_errmsg(db)); + int final_rc = sqlite3_finalize(st); + if (final_rc != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: claim cleanup finalize rc=%d", final_rc); + ak_rollback(inst, "decryption claim", job->channel); + return -1; +} + /* Восстанавливаем обработку файлов и запросов из устойчивых данных; внешняя транзакция откладывает весь проход. */ static void ak_tick(void* arg) { struct UTUN_INSTANCE* inst = arg; @@ -964,7 +1004,14 @@ static void ak_tick(void* arg) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: pending files/outbox postponed inside external transaction"); goto schedule; } + for (struct ak_file* job = CC(inst)->attachment_files; job;) { + struct ak_file* next = job->next; + if (job->completed) ak_finish_file(job); + if (CC(inst)->attachment_closing) return; + job = next; + } for (unsigned i = 0; i < 4; i++) { + if (CC(inst)->attachment_file_count >= MEDIA_ASYNC_MAX_TASKS) break; sqlite3_stmt* st = ak_prepare(inst, "SELECT channel,message_id,crypto_id,author,key,cipher_hash,cipher_path,plain_path FROM attachment_keys a" " WHERE downloaded=1 AND wanted=1 AND ready=0 AND decrypting=0 AND error=0 AND" " (length(key)=32 OR NOT EXISTS(SELECT 1 FROM attachment_key_requests r WHERE r.channel=a.channel AND r.message_id=a.message_id)) LIMIT 1"); @@ -1004,9 +1051,8 @@ static void ak_tick(void* arg) { OPENSSL_cleanse(job->key, 32); u_free(job); break; } if (has_key) { - st = ak_prepare(inst, "UPDATE attachment_keys SET decrypting=1 WHERE channel=? AND message_id=?"); - if (st) { sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, job->message, 32, SQLITE_STATIC); } - if (!st || ak_step(inst, st)) { OPENSSL_cleanse(job->key, 32); u_free(job); break; } + if (ak_claim_decryption(job)) { OPENSSL_cleanse(job->key, 32); u_free(job); break; } + job->next = CC(inst)->attachment_files; CC(inst)->attachment_files = job; CC(inst)->attachment_file_count++; media_async_submit(inst->media_async, inst->ua, ak_file_work, job, ak_file_done, job); } else { int failed = ak_request_file(job); @@ -1141,7 +1187,16 @@ 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); + if (!CC(inst) || !CC(inst)->attachment_initialized) return; + while (CC(inst)->attachment_files) { + struct ak_file* job = CC(inst)->attachment_files; + if (!job->completed) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: destroy before worker done channel=%s active=%u", + job->channel, CC(inst)->attachment_file_count); return; + } + ak_finish_file(job); + } + ak_recover_preparations(inst); } /* Удаление ключей отменяет completion активных workers; они удалят свои результаты. */ diff --git a/src/chat/chat_attachment.h b/src/chat/chat_attachment.h index 18753390..fff4ffa0 100644 --- a/src/chat/chat_attachment.h +++ b/src/chat/chat_attachment.h @@ -11,6 +11,9 @@ * state=0 ждёт автора, 2/4 повторяют GRANT, 3 повторяет DENY; чужое направление и другие состояния отклоняются. * Исходящий REQUEST сохраняется собственной транзакцией request+outbox после повторной проверки состояния. * Уведомление/отправка следуют после commit; внешняя транзакция откладывает весь проход файлов и очереди. + * Расшифровка стартует после commit decrypting=1. До commit метаданных завершение остаётся у модуля: + * таймер повторяет SQL без нового worker; не более 16 активных/ожидающих завершения файлов на instance. + * quiesce останавливает повторы, destroy после workers освобождает также отложенные завершения. * Незавершённый rollback отключает модуль и запрашивает остановку ядра владельцем вне callback. */ #ifndef CHAT_ATTACHMENT_H #define CHAT_ATTACHMENT_H