Claim decryption before workers and retry retained completions

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

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