From 4d19de7cce79becc82a5775572e43bd30501a698 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 01:34:47 +0100 Subject: [PATCH] Retain attachment completions until their SQL result commits --- src/chat/chat_attachment.c | 133 +++++++++++++++++++++++++++++-------- src/chat/chat_core_priv.h | 3 + 2 files changed, 110 insertions(+), 26 deletions(-) diff --git a/src/chat/chat_attachment.c b/src/chat/chat_attachment.c index 2e1603d4..7a08c7ed 100644 --- a/src/chat/chat_attachment.c +++ b/src/chat/chat_attachment.c @@ -23,6 +23,8 @@ enum { AK_REQUEST = 1, AK_REQUEST_ACK, AK_GRANT, AK_DENY, AK_RESULT_ACK }; struct ak_file { struct UTUN_INSTANCE* inst; + struct ak_file* next; // Владение uasync до успешного commit результата либо остановки. + int completed, completion_error; char channel[32], source[1024], cipher[1024], plain[1024], type[32]; struct file_io_temp output; // Единственный собственный temp; стабильный адрес до completion. uint8_t id[16], key[32], message[32], hash[32], reply[32]; @@ -136,31 +138,99 @@ static void ak_file_work(void* arg) { job->error = 0; } -/* Проверка существующей строки предотвращает публикацию после удаления канала. */ +static void ak_rollback(struct UTUN_INSTANCE* inst, const char* operation, const char* channel); + +/* 1 — строка существует, 0 — канал/вложение удалены, -1 — SQL-отказ, который нельзя считать удалением. */ static int ak_present(struct ak_file* job) { - sqlite3_stmt* st = ak_prepare(job->inst, "SELECT 1 FROM attachment_keys WHERE channel=? AND crypto_id=?"); - if (!st) return 0; - sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, job->id, 16, SQLITE_STATIC); - int rc = sqlite3_step(st); sqlite3_finalize(st); - if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: presence lookup failed rc=%d", rc); - return rc == SQLITE_ROW; + sqlite3* db = CC(job->inst)->db; sqlite3_stmt* st = NULL; + const char* stage = "prepare presence"; + int rc = sqlite3_prepare_v2(db, "SELECT 1 FROM attachment_keys WHERE channel=? AND crypto_id=?", -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 = "read presence"; rc = sqlite3_step(st); + if (rc != SQLITE_ROW && rc != SQLITE_DONE) goto fail; + int present = rc == SQLITE_ROW; + stage = "finalize presence"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + return present; +fail: + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: presence 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: presence cleanup finalize rc=%d", final_rc); + return -1; } -/* Сохранение metadata и открытой копии не меняет сетевой индекс. */ -static void ak_file_done(void* arg, int error) { - struct ak_file* job = arg; struct UTUN_INSTANCE* inst = job->inst; - int valid = CC(inst) && !CC(inst)->attachment_closing && ak_present(job); - int success = !error && !job->error && valid; - if (success && !job->encrypt) { - uint8_t info[ATTACHMENT_WIRE_MAX]; int n = attachment_encode(&job->info, info, sizeof(info)); - sqlite3_stmt* st = n < 0 ? NULL : ak_prepare(inst, - "UPDATE attachment_keys SET info=?,content_type=?,ready=1,decrypting=0,error=0 WHERE channel=? AND crypto_id=?"); - if (st) { - sqlite3_bind_blob(st, 1, info, n, SQLITE_STATIC); sqlite3_bind_text(st, 2, job->type, -1, SQLITE_STATIC); - sqlite3_bind_text(st, 3, job->channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 4, job->id, 16, SQLITE_STATIC); - } - success = st && !ak_step(inst, st); +/* 1 — результат зафиксирован, 0 — строка удалена, -1 — сохранить задачу для повтора без нового worker. */ +static int ak_store_file_result(struct ak_file* job) { + struct UTUN_INSTANCE* inst = job->inst; + sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL; + if (!sqlite3_get_autocommit(db)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: completion postponed inside external transaction channel=%s", job->channel); + return -1; } + const char* stage = "begin completion"; + int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + stage = "presence"; + int present = ak_present(job); + if (present < 0) { rc = SQLITE_ERROR; goto fail; } + int success = !job->completion_error && !job->error; + if (present && (!job->encrypt || !success)) { + uint8_t info[ATTACHMENT_WIRE_MAX]; int n = 0; + if (success) { + stage = "encode metadata"; n = attachment_encode(&job->info, info, sizeof(info)); + if (n < 0) { rc = SQLITE_MISMATCH; goto fail; } + } + stage = "prepare completion"; + rc = sqlite3_prepare_v2(db, success ? + "UPDATE attachment_keys SET info=?,content_type=?,ready=1,decrypting=0,error=0 WHERE channel=? AND crypto_id=?" : + "UPDATE attachment_keys SET decrypting=0,error=1 WHERE channel=? AND crypto_id=?", -1, &st, NULL); + if (rc != SQLITE_OK) goto fail; + int index = 1; + if (success) { + stage = "bind metadata"; rc = sqlite3_bind_blob(st, index++, info, n, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "bind content type"; rc = sqlite3_bind_text(st, index++, job->type, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + } + stage = "bind channel"; rc = sqlite3_bind_text(st, index++, job->channel, -1, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "bind identity"; rc = sqlite3_bind_blob(st, index, job->id, 16, SQLITE_STATIC); + if (rc != SQLITE_OK) goto fail; + stage = "write completion"; rc = sqlite3_step(st); + if (rc != SQLITE_DONE) goto fail; + int changed = sqlite3_changes(db); + stage = "finalize completion"; rc = sqlite3_finalize(st); st = NULL; + if (rc != SQLITE_OK) goto fail; + if (changed != 1) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: completion rows=%d expected=1 channel=%s", changed, job->channel); + stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail; + } + } + stage = "commit completion"; rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "attachment keys: completion committed channel=%s encrypt=%d present=%d success=%d", + job->channel, job->encrypt, present, success); + return present; +fail: + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: completion channel=%s encrypt=%d stage=%s rc=%d (%s) db=%s", + job->channel, job->encrypt, 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: completion cleanup finalize rc=%d", final_rc); + ak_rollback(inst, "file completion", job->channel); + return -1; +} + +/* Завершить только после commit; при SQL-отказе задача остаётся у модуля до следующего tick. */ +static void ak_finish_file(struct ak_file* job) { + struct UTUN_INSTANCE* inst = job->inst; + int valid = CC(inst)->attachment_closing ? 0 : ak_store_file_result(job); + if (valid < 0) return; + int success = !job->completion_error && !job->error && valid; if (success && job->encrypt) { char name[38], hex[33]; ak_hex(job->id, hex); snprintf(name, sizeof(name), "%s.uenc", hex); struct chat_msg_submit request = { .inst = inst, .data = (uint8_t*)name, .data_len = 37, .request_id = job->request }; @@ -172,11 +242,7 @@ static void ak_file_done(void* arg, int error) { chat_core_submit_media_message(inst, &request); } else if (!success) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: worker failed channel=%s encrypt=%d error=%d worker=%d current=%d", - job->channel, job->encrypt, error, job->error, valid); - if (valid) { - sqlite3_stmt* st = ak_prepare(inst, "UPDATE attachment_keys SET decrypting=0,error=1 WHERE channel=? AND crypto_id=?"); - if (st) { sqlite3_bind_text(st, 1, job->channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, job->id, 16, SQLITE_STATIC); ak_step(inst, st); } - } + job->channel, job->encrypt, job->completion_error, job->error, valid); if (job->encrypt) chat_event_submitted(inst, job->request, 0); } if (job->encrypt && remove(job->source) && errno != ENOENT) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: prepared source cleanup failed"); @@ -191,9 +257,20 @@ static void ak_file_done(void* arg, int error) { if (success && !job->encrypt) chat_media_cache_enforce(inst); ak_changed(inst, job->channel); } + struct ak_file** owner = &CC(inst)->attachment_files; + while (*owner && *owner != job) owner = &(*owner)->next; + if (*owner) { *owner = job->next; CC(inst)->attachment_file_count--; } + else DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: completed job missing from owner channel=%s", job->channel); OPENSSL_cleanse(job->key, 32); u_free(job); } +/* Доставленный done больше не принадлежит media_async; модуль удерживает его до записи результата. */ +static void ak_file_done(void* arg, int error) { + struct ak_file* job = arg; + job->completed = 1; job->completion_error = error; + ak_finish_file(job); +} + /* Вызывается только для собственной транзакции; открытая после отказа БД требует остановки владельцем. */ static void ak_rollback(struct UTUN_INSTANCE* inst, const char* operation, const char* channel) { sqlite3* db = CC(inst)->db; @@ -268,6 +345,9 @@ int chat_attachment_send(struct UTUN_INSTANCE* inst, const char* channel, const !source || !source[0] || strlen(source) >= 1024 || !reply || encrypted_file_validate_type(type) || attachment_validate(info)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid send arguments/state"); return -1; } + if (CC(inst)->attachment_file_count >= MEDIA_ASYNC_MAX_TASKS) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: file task limit reached count=%u", CC(inst)->attachment_file_count); return -1; + } if (!sqlite3_get_autocommit(CC(inst)->db)) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: send rejected inside external transaction channel=%s", channel); return -1; } @@ -279,6 +359,7 @@ int chat_attachment_send(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t encoded[ATTACHMENT_WIRE_MAX]; int n = attachment_encode(info, encoded, sizeof(encoded)); if (RAND_bytes(job->id, 16) != 1 || RAND_bytes(job->key, 32) != 1 || ak_paths(job) || n < 0) goto fail; if (ak_store_preparation(job, encoded, n)) goto fail; + 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); return 0; fail: diff --git a/src/chat/chat_core_priv.h b/src/chat/chat_core_priv.h index 7d8e2e45..609773d8 100644 --- a/src/chat/chat_core_priv.h +++ b/src/chat/chat_core_priv.h @@ -80,6 +80,7 @@ struct media_index_result; struct ms_apply_cbk; /* member_sync.c */ struct group_avatar; struct media_submit_ctx; +struct ak_file; struct chat_core_ctx { struct UTUN_INSTANCE* inst; @@ -104,6 +105,8 @@ struct chat_core_ctx { void* attachment_timer; int attachment_closing; int attachment_initialized; + struct ak_file* attachment_files; /* workers и завершения, ожидающие SQL commit; владеет chat_attachment */ + unsigned attachment_file_count; /* максимум MEDIA_ASYNC_MAX_TASKS, включая отложенные завершения */ }; /* Per-instance chat context. inst->chat_core обязателен после chat_core_init. */