Verify completion ownership through cancellation, retry and task limits

doc/encrypted_attachments.md

@@ -83,6 +83,18 @@ Nonce метаданных использует `part=UINT64_MAX`; файл —
 Completion не удаляет файлы по заранее вычисленному имени: очистка учитывает,
 какие артефакты задача действительно создала или опубликовала.
 
+Перед расшифровкой собственная SQL-транзакция фиксирует `decrypting=1` ровно для
+одного подходящего файла с выбранным ключом. Worker запускается только после commit.
+Завершение также владеет транзакцией: отличает удалённую строку от ошибки чтения,
+проверяет все bind/step/changes/finalize и сохраняет метаданные/ready либо ошибку.
+До commit оно остаётся в списке задач instance; SQL-отказ повторяется таймером
+с готовым результатом, без повторного worker. Внешняя транзакция остаётся нетронутой.
+Лимит16 включает workers и такие завершения; отказ новой отправки сохраняет source
+у вызывающего. Событие/публикация сообщения следуют после commit результата.
+Quiesce останавливает повторы; после завершения workers destroy освобождает остаток
+списка и принадлежащие ему результаты. При restart незавершённая расшифровка может
+выполниться заново по сохранённому ключу и контейнеру.
+
 Реестр `file_io` защищает активные temp от очистки кеша, включая время после close.
 Восстановление до запуска workers и очистка после их завершения удаляют оставшиеся
 `<UUID>.uenc.tmp.<UUID>` / `<UUID>.file.tmp.<UUID>` в `media/decrypted`.
@@ -209,6 +221,13 @@ backpressure и повтором через 5 секунд; restart продол
 Отдельное readonly-соединение проверяет точные ключ, автора, UUID, метаданные и пути до worker.
 Дополнительно проверяются настоящий commit veto, INSERT ABORT/IGNORE, внешняя транзакция,
 синхронный отказ запуска, restart с опубликованным ключом и отказ rollback с закрытием БД владельцем.
+Дополнительно:9 SQL-отказов захвата расшифровки,15 записи успешного результата и8
+записи ошибки настоящего worker. Проверяются ABORT/IGNORE, commit veto, внешняя
+транзакция, точные ключ/метаданные/флаги через reader перед worker и в событии,
+один worker при повторе completion, отмена ожидающего результата при delete/stop,
+restart сервиса и rollback fault без повторного таймера. 16 готовых авторских задач
+удерживаются при отказе чтения; семнадцатая сохраняет source, снятие отказа публикует
+все16 ровно по одному разу. Баланс выделений восстанавливается.
 
 `test_attachment_requests` проверяет247 SQL-отказов/повторов новой записи и дубликатов
 0/2/3/4, с пустой/существующей очередью. Подписанный мембер и сообщение создаются

doc/service_lifecycle.md

@@ -211,6 +211,12 @@ completion. Отказ запуска/неверные аргументы даю
 `encrypted_file_write/read` только готовят закрытый durable temp; worker публикует
 его одним rename, completion подтверждает результат в БД/UI. Отмена до completion
 не публикует сообщение; cleanup учитывает реально созданные/заменённые файлы.
+Модуль канальных вложений отдельно владеет списком до16 файлов: worker либо
+завершение, ожидающее SQL commit. Захват decrypting фиксируется до worker;
+ошибка записи результата сохраняет задачу для следующего tick без новой криптографии.
+Quiesce останавливает эти повторы. После media_async_destroy chat_attachment_destroy
+освобождает оставшиеся завершения до удаления chat_core; внешняя транзакция
+откладывает запись, незавершённый rollback запрашивает штатную остановку владельцем.
 Ошибка синхронизации каталога после rename означает установленное имя без подтверждённой
 durability; отложенный discard уже не удаляет результат. Общей транзакции ФС/БД нет.
 До запуска workers и после их завершения recovery очищает собственные stale temp

tests/test_chat_attachment.c

@@ -23,6 +23,10 @@ static const char* insert_prefix = "INSERT INTO attachment_keys(channel,crypto_i
 static const char* fault_prefix;
 static int target_written;
 static timeout_callback_t attachment_tick;
+static unsigned completion_calls;
+static int block_presence;
+static unsigned blocked_completions;
+static int completion_transaction;
 static const char content[] = "attachment source must survive SQL failure";
 
 struct fixture {
@@ -65,11 +69,17 @@ int __wrap_sqlite3_exec(sqlite3* db, const char* sql, int (*cb)(void*, int, char
         fail_rollback = 0; rollback_fault++;
         DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment test: injected rollback failure"); return SQLITE_IOERR;
     }
-    return __real_sqlite3_exec(db, sql, cb, arg, error);
+    int rc = __real_sqlite3_exec(db, sql, cb, arg, error);
+    if (!strcmp(sql, "COMMIT") || !strcmp(sql, "ROLLBACK")) completion_transaction = 0;
+    return rc;
 }
 
 extern int __real_sqlite3_prepare_v2(sqlite3*, const char*, int, sqlite3_stmt**, const char**);
 int __wrap_sqlite3_prepare_v2(sqlite3* db, const char* sql, int length, sqlite3_stmt** st, const char** tail) {
+    if (!strncmp(sql, "SELECT 1 FROM attachment_keys WHERE", strlen("SELECT 1 FROM attachment_keys WHERE"))) {
+        completion_calls++; completion_transaction = 1;
+        if (block_presence) { blocked_completions++; *st = NULL; return SQLITE_BUSY; }
+    }
     const char* prefix = fault_prefix ? fault_prefix : insert_prefix;
     if (!strncmp(sql, prefix, strlen(prefix)) && take(PREPARE)) { *st = NULL; return SQLITE_NOMEM; }
     return __real_sqlite3_prepare_v2(db, sql, length, st, tail);
@@ -164,6 +174,9 @@ void __wrap_media_async_submit(struct media_async* ma, struct UASYNC* ua, ma_wor
         if (fail_submit) { fail_submit = 0; done(arg, -1); return; }
     }
     if (active && active->decrypt_mode) {
+        if (!active->allow_worker || !sqlite3_get_autocommit(active->inst->topo_sqlite_db))
+            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "decryption test: worker started allowed=%d autocommit=%d previous_workers=%u",
+                active->allow_worker, sqlite3_get_autocommit(active->inst->topo_sqlite_db), active->decrypt_submissions);
         assert(active->allow_worker && sqlite3_get_autocommit(active->inst->topo_sqlite_db));
         decryption_state(active, 0, 1, 0); active->decrypt_submissions++;
     }
@@ -351,6 +364,9 @@ static void tick(struct fixture* f) {
 
 static void decryption_done(struct fixture* f) {
     for (unsigned poll = 0; poll < 2000 && !f->key_events; poll++) uasync_poll(f->ua, 10);
+    if (f->key_events != 1 || f->decrypt_submissions != 1 || CC(f->inst)->attachment_file_count || CC(f->inst)->attachment_files)
+        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "decryption test: completion events=%u workers=%u retained=%u list=%d",
+            f->key_events, f->decrypt_submissions, CC(f->inst)->attachment_file_count, CC(f->inst)->attachment_files != NULL);
     assert(f->key_events == 1 && f->decrypt_submissions == 1 && !CC(f->inst)->attachment_file_count && !CC(f->inst)->attachment_files);
     decryption_state(f, !f->result_error, 0, f->result_error);
     if (!f->result_error) {
@@ -360,6 +376,11 @@ static void decryption_done(struct fixture* f) {
     }
 }
 
+static void wait_completion(struct fixture* f, unsigned previous) {
+    for (unsigned poll = 0; poll < 2000 && completion_calls == previous; poll++) uasync_poll(f->ua, 10);
+    assert(completion_calls > previous);
+}
+
 static void test_decryption_claim(void) {
     struct fixture f; setup(&f); source(&f); publish(&f);
     const enum fault points[] = {BEGIN, PREPARE, BIND_1, BIND_2, BIND_3, STEP, FINALIZE, CHANGES, COMMIT};
@@ -406,13 +427,101 @@ static void test_decryption_completion(void) {
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS decryption completion:%u SQL faults retain result/key, retry commits without second worker; foreign transaction untouched", cases);
 }
 
+/* Hash-обслуживание истории может завершить свою транзакцию в том же poll; veto относится только к completion. */
+static int veto_completion(void* arg) {
+    unsigned* count = arg;
+    if (!completion_transaction || *count) return 0;
+    *count = 1; return 1;
+}
+
+static void test_completion_controls(void) {
+    struct fixture f; setup(&f); source(&f); publish(&f);
+    const char* triggers[] = {
+        "CREATE TEMP TRIGGER reject_result BEFORE UPDATE ON attachment_keys WHEN NEW.ready=1 AND OLD.ready=0 BEGIN SELECT RAISE(IGNORE); END",
+        "CREATE TEMP TRIGGER reject_result BEFORE UPDATE ON attachment_keys WHEN NEW.ready=1 AND OLD.ready=0 BEGIN SELECT RAISE(ABORT,'result'); END"
+    };
+    for (unsigned i = 0; i < 2; i++) {
+        decryption_reset(&f); sql(f.inst->topo_sqlite_db, triggers[i]); f.allow_worker = 1;
+        unsigned previous = completion_calls; tick(&f); wait_completion(&f, previous);
+        assert(!f.key_events && CC(f.inst)->attachment_file_count == 1); decryption_state(&f, 0, 1, 0);
+        sql(f.inst->topo_sqlite_db, "DROP TRIGGER reject_result"); tick(&f); decryption_done(&f);
+    }
+    decryption_reset(&f); f.allow_worker = 1; tick(&f);
+    unsigned vetoes = 0, previous = completion_calls; sqlite3_commit_hook(f.inst->topo_sqlite_db, veto_completion, &vetoes);
+    wait_completion(&f, previous); assert(vetoes == 1 && !f.key_events && sqlite3_get_autocommit(f.inst->topo_sqlite_db));
+    decryption_state(&f, 0, 1, 0); sqlite3_commit_hook(f.inst->topo_sqlite_db, NULL, NULL); tick(&f); decryption_done(&f);
+    /* Файл уже проверен и опубликован; удаление канала отменяет только принадлежащий задаче результат. */
+    decryption_reset(&f); f.allow_worker = 1; tick(&f); previous = completion_calls;
+    fault_prefix = "UPDATE attachment_keys SET info="; fault = STEP; wait_completion(&f, previous);
+    assert(fault == NONE && !f.key_events && CC(f.inst)->attachment_file_count == 1);
+    char path[512]; plaintext_path(&f, path); assert(ma_file_size(path) == sizeof(content));
+    chat_core_delete_channel(f.inst, f.channel); tick(&f);
+    assert(!CC(f.inst)->attachment_files && !CC(f.inst)->attachment_file_count && !f.key_events && ma_file_size(path) < 0);
+    fault_prefix = NULL; cleanup(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS completion controls: ABORT/IGNORE, real commit veto and deletion while result waits for commit");
+}
+
+static void test_completion_stop(void) {
+    struct fixture f; setup(&f); source(&f); publish(&f); decryption_reset(&f); f.allow_worker = 1;
+    unsigned previous = completion_calls; tick(&f); fault_prefix = "UPDATE attachment_keys SET info="; fault = STEP;
+    wait_completion(&f, previous); assert(fault == NONE && !f.key_events && CC(f.inst)->attachment_file_count == 1);
+    char path[512]; plaintext_path(&f, path); assert(ma_file_size(path) == sizeof(content));
+    chat_service_stop(f.inst); assert(!f.inst->chat_core && ma_file_size(path) < 0 && !f.key_events);
+    assert(!chat_service_start(f.inst)); chat_event_set_handler(f.inst, event);
+    f.decrypt_submissions = 0; // Новый lifecycle запускает новый worker после удаления незавершённого plaintext.
+    decryption_state(&f, 0, 0, 0); tick(&f); decryption_done(&f);
+    decryption_reset(&f); f.allow_worker = 1; previous = completion_calls; tick(&f); fault = COMMIT; fail_rollback = 1;
+    wait_completion(&f, previous);
+    assert(fault == NONE && !fail_rollback && !f.inst->running && !f.key_events && CC(f.inst)->attachment_file_count == 1);
+    assert(!CC(f.inst)->attachment_timer && !sqlite3_get_autocommit(f.inst->topo_sqlite_db));
+    chat_event_set_handler(f.inst, NULL); utun_instance_destroy(f.inst); f.inst = NULL;
+    assert(ma_file_size(path) < 0); decryption_state(&f, 0, 1, 0);
+    assert(sqlite3_close(f.reader) == SQLITE_OK); uasync_destroy(f.ua, 0); active = NULL; fault_prefix = NULL;
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS completion stop: deferred result freed after workers, service restart retries; rollback fault stops without event or timer");
+}
+
+static void test_worker_error_completion(void) {
+    struct fixture f; setup(&f); source(&f); publish(&f);
+    char hex[33], cipher[512], backup[540];
+    for (unsigned i = 0; i < 16; i++) snprintf(hex + i * 2, 3, "%02x", f.id[i]);
+    snprintf(cipher, sizeof(cipher), "%s/media/%s/%s.uenc", f.directory, f.channel, hex);
+    snprintf(backup, sizeof(backup), "%s.test-backup", cipher);
+    const enum fault points[] = {BEGIN, PREPARE, BIND_1, BIND_2, STEP, FINALIZE, CHANGES, COMMIT};
+    for (unsigned i = 0; i < sizeof(points) / sizeof(points[0]); i++) {
+        decryption_reset(&f); assert(!rename(cipher, backup)); f.result_error = f.allow_worker = 1;
+        tick(&f); fault_prefix = "UPDATE attachment_keys SET decrypting=0,error=1 WHERE"; fault = points[i];
+        for (unsigned poll = 0; poll < 2000 && fault != NONE; poll++) uasync_poll(f.ua, 10);
+        assert(fault == NONE && !f.key_events && CC(f.inst)->attachment_file_count == 1); decryption_state(&f, 0, 1, 0);
+        tick(&f); decryption_done(&f); assert(!rename(backup, cipher));
+    }
+    fault_prefix = NULL; cleanup(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS failed-worker completion:8 SQL faults retain key/result, error persisted on retry without new worker");
+}
+
+static void test_completion_limit(void) {
+    struct fixture f; setup(&f); block_presence = 1; blocked_completions = 0;
+    for (unsigned i = 0; i < MEDIA_ASYNC_MAX_TASKS; i++) {
+        snprintf(f.source, sizeof(f.source), "%s/prepared-%u.bin", f.directory, i); source(&f);
+        assert(!send_attachment(&f)); f.rows++;
+    }
+    for (unsigned poll = 0; poll < 5000 && blocked_completions < MEDIA_ASYNC_MAX_TASKS; poll++) uasync_poll(f.ua, 10);
+    assert(blocked_completions == MEDIA_ASYNC_MAX_TASKS && CC(f.inst)->attachment_file_count == MEDIA_ASYNC_MAX_TASKS && !f.events);
+    snprintf(f.source, sizeof(f.source), "%s/rejected.bin", f.directory); source(&f);
+    assert(send_attachment(&f) == -1); unchanged(&f); f.inspect_submit = 0;
+    block_presence = 0; tick(&f); assert(!CC(f.inst)->attachment_file_count && !CC(f.inst)->attachment_files);
+    for (unsigned poll = 0; poll < 5000 && f.events < MEDIA_ASYNC_MAX_TASKS; poll++) uasync_poll(f.ua, 10);
+    assert(f.events == MEDIA_ASYNC_MAX_TASKS && f.success == 1); source_preserved(&f); cleanup(&f);
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS completion limit:16 retained results, seventeenth source preserved, retry publishes all exactly once");
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_INFO);
     utun_instance_set_tun_init_enabled(0); size_t baseline = u_get_allocated_count();
     test_sql_faults(); test_sql_controls(); test_inline_failure(); test_rollback_failure();
     assert(fault == NONE && !fail_rollback && !fail_submit && rollback_fault == 1);
     for (unsigned i = BEGIN; i < FAULT_COUNT; i++) assert(hits[i] == (i == COMMIT ? 2 : 1));
-    test_decryption_claim(); test_decryption_completion();
+    test_decryption_claim(); test_decryption_completion(); test_completion_controls(); test_completion_stop();
+    test_worker_error_completion(); test_completion_limit();
     assert(u_get_allocated_count() == baseline);
     DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "PASS attachment: independent committed key/metadata before worker, allocations restored");
     return 0;