diff --git a/doc/encrypted_attachments.md b/doc/encrypted_attachments.md index 4704c757..9da51e67 100644 --- a/doc/encrypted_attachments.md +++ b/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 и очистка после их завершения удаляют оставшиеся `.uenc.tmp.` / `.file.tmp.` в `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, с пустой/существующей очередью. Подписанный мембер и сообщение создаются diff --git a/doc/service_lifecycle.md b/doc/service_lifecycle.md index 33bb81db..3bb67045 100644 --- a/doc/service_lifecycle.md +++ b/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 diff --git a/tests/test_chat_attachment.c b/tests/test_chat_attachment.c index 72910c25..22e9c7a4 100644 --- a/tests/test_chat_attachment.c +++ b/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;