jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Retain attachment completions until their SQL result commits

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 01:34:47 +01:00
1 parent 851c6723ff
commit 4d19de7cce
2 files changed
+110 -26
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_attachment.c
+107 -26
View File
@@ -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:
src/chat/chat_core_priv.h
+3
View File
@@ -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. */