Commit attachment download intent and cache restoration atomically
This commit is contained in:
evgeny committed
1 parent
45d5aaa7a1
commit
fc60ca295a
2 files changed
+117
-29
No files matched your search
+110
-25
@@ -933,35 +933,120 @@ void chat_attachment_decision_trampoline(void* arg) {
|
||||
}
|
||||
|
||||
/* Пользовательский интерес хранится отдельно от фоновой storage-копии. */
|
||||
void chat_attachment_want(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]) {
|
||||
sqlite3_stmt* check = ak_prepare(inst, "SELECT plain_path FROM attachment_keys WHERE channel=? AND message_id=? AND ready=1");
|
||||
if (!check) return;
|
||||
sqlite3_bind_text(check, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(check, 2, message, 32, SQLITE_STATIC);
|
||||
int result = sqlite3_step(check), missing = result == SQLITE_ROW && ma_file_size((const char*)sqlite3_column_text(check, 0)) < 0;
|
||||
sqlite3_finalize(check);
|
||||
if (result != SQLITE_ROW && result != SQLITE_DONE) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: cache lookup failed rc=%d", result); return; }
|
||||
if (missing) {
|
||||
sqlite3_stmt* reset = ak_prepare(inst, "UPDATE attachment_keys SET ready=0 WHERE channel=? AND message_id=? AND decrypting=0");
|
||||
if (!reset) return;
|
||||
sqlite3_bind_text(reset, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(reset, 2, message, 32, SQLITE_STATIC);
|
||||
if (ak_step(inst, reset)) return;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: decrypted cache missing, restoring from retained key channel=%s", channel);
|
||||
int chat_attachment_want(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]) {
|
||||
if (!inst || !CC(inst) || CC(inst)->attachment_closing || !ak_channel(channel) || !message) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid wanted state/arguments"); return -1;
|
||||
}
|
||||
sqlite3_stmt* st = ak_prepare(inst, "UPDATE attachment_keys SET wanted=1,error=0 WHERE channel=? AND message_id=?");
|
||||
if (!st) return;
|
||||
sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC); ak_step(inst, st);
|
||||
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
|
||||
if (!sqlite3_get_autocommit(db)) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: wanted rejected inside external transaction channel=%s", channel); return -1;
|
||||
}
|
||||
const char* stage = "begin wanted";
|
||||
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE /* attachment wanted */", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "prepare cache";
|
||||
rc = sqlite3_prepare_v2(db, "SELECT ready,decrypting,plain_path FROM attachment_keys WHERE channel=? AND message_id=?", -1, &st, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind channel"; rc = sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind message"; rc = sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "read cache"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_ROW) goto fail;
|
||||
int ready = sqlite3_column_int(st, 0), decrypting = sqlite3_column_int(st, 1);
|
||||
const char* path = (const char*)sqlite3_column_text(st, 2);
|
||||
if (!path || (ready != 0 && ready != 1) || (decrypting != 0 && decrypting != 1)) { stage = "cache state"; rc = SQLITE_MISMATCH; goto fail; }
|
||||
int missing = ready && !decrypting && ma_file_size(path) < 0;
|
||||
stage = "finalize cache"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "prepare wanted";
|
||||
rc = sqlite3_prepare_v2(db, "UPDATE attachment_keys SET wanted=1,error=0,ready=? WHERE channel=? AND message_id=?", -1, &st, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind ready"; rc = sqlite3_bind_int(st, 1, missing ? 0 : ready);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind channel"; rc = sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind message"; rc = sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "write wanted"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_DONE) goto fail;
|
||||
int changed = sqlite3_changes(db);
|
||||
stage = "finalize wanted"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed != 1) { stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail; }
|
||||
stage = "commit wanted"; rc = sqlite3_exec(db, "COMMIT /* attachment wanted */", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (missing) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: missing cache reset committed, restoring from retained key channel=%s", channel);
|
||||
else DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "attachment keys: wanted committed channel=%s ready=%d decrypting=%d", channel, ready, decrypting);
|
||||
return 0;
|
||||
fail:
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: wanted channel=%s stage=%s rc=%d (%s) db=%s",
|
||||
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: wanted cleanup finalize rc=%d", final_rc);
|
||||
ak_rollback(inst, "wanted", channel); return -1;
|
||||
}
|
||||
|
||||
/* Автозагрузка обычного участника запрашивает доступ; storage-реплика хранит только ciphertext. */
|
||||
void chat_attachment_downloaded(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]) {
|
||||
char sql[384]; snprintf(sql, sizeof(sql), "UPDATE attachment_keys SET downloaded=1,wanted=max(wanted,"
|
||||
"NOT EXISTS(SELECT 1 FROM peers_%s WHERE node_id=? AND source=0 AND deleted=0 AND length(adm_tags_sig)=64"
|
||||
" AND json_extract(adm_tags,'$.storage')='yes')) WHERE channel=? AND message_id=?", channel);
|
||||
sqlite3_stmt* st = ak_prepare(inst, sql);
|
||||
if (!st) return;
|
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id); sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC);
|
||||
if (!ak_step(inst, st)) { DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: ciphertext downloaded channel=%s", channel); ak_changed(inst, channel); }
|
||||
int chat_attachment_downloaded(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]) {
|
||||
if (!inst || !CC(inst) || CC(inst)->attachment_closing || !ak_channel(channel) || !message) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid downloaded state/arguments"); return -1;
|
||||
}
|
||||
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
|
||||
if (!sqlite3_get_autocommit(db)) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: downloaded rejected inside external transaction channel=%s", channel); return -1;
|
||||
}
|
||||
const char* stage = "begin downloaded";
|
||||
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE /* attachment downloaded */", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
char sql[512]; snprintf(sql, sizeof(sql), "SELECT downloaded,wanted,NOT EXISTS(SELECT 1 FROM peers_%s"
|
||||
" WHERE node_id=? AND source=0 AND deleted=0 AND length(adm_tags_sig)=64 AND json_extract(adm_tags,'$.storage')='yes')"
|
||||
" FROM attachment_keys WHERE channel=? AND message_id=?", channel);
|
||||
stage = "prepare download state"; rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind node"; rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)inst->node_id);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind channel"; rc = sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind message"; rc = sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "read download state"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_ROW) goto fail;
|
||||
int downloaded = sqlite3_column_int(st, 0), wanted = sqlite3_column_int(st, 1), automatic = sqlite3_column_int(st, 2);
|
||||
if ((downloaded != 0 && downloaded != 1) || (wanted != 0 && wanted != 1)) { stage = "download state"; rc = SQLITE_MISMATCH; goto fail; }
|
||||
int desired = wanted || automatic, changed_state = !downloaded || wanted != desired;
|
||||
stage = "finalize download state"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed_state) {
|
||||
stage = "prepare downloaded";
|
||||
rc = sqlite3_prepare_v2(db, "UPDATE attachment_keys SET downloaded=1,wanted=? WHERE channel=? AND message_id=?", -1, &st, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind wanted"; rc = sqlite3_bind_int(st, 1, desired);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind channel"; rc = sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind message"; rc = sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "write downloaded"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_DONE) goto fail;
|
||||
int changed = sqlite3_changes(db);
|
||||
stage = "finalize downloaded"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed != 1) { stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail; }
|
||||
}
|
||||
stage = "commit downloaded"; rc = sqlite3_exec(db, "COMMIT /* attachment downloaded */", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed_state) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: download state committed channel=%s wanted=%d automatic=%d", channel, desired, automatic);
|
||||
ak_changed(inst, channel);
|
||||
}
|
||||
return 0;
|
||||
fail:
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: downloaded channel=%s stage=%s rc=%d (%s) db=%s",
|
||||
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: downloaded cleanup finalize rc=%d", final_rc);
|
||||
ak_rollback(inst, "downloaded", channel); return -1;
|
||||
}
|
||||
|
||||
/* Один проход ограничен; retry_at даёт всем задачам очередь без голодания. */
|
||||
|
||||
@@ -66,11 +66,14 @@ int chat_attachment_observe(struct UTUN_INSTANCE* inst, const char* channel, con
|
||||
|
||||
// ── Получение файла и выдача доступа ──
|
||||
|
||||
// После проверенной загрузки: сохранить состояние, запросить ключ либо расшифровать.
|
||||
void chat_attachment_downloaded(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]);
|
||||
/* После проверенной загрузки/наличия сохранённого контейнера: своей транзакцией сохранить downloaded и policy wanted.
|
||||
* 0/-1; запись observe обязательна. Повтор неизменного состояния не даёт события. Криптографию проверит worker.
|
||||
* Внешняя транзакция запрещена, SQL-отказ сохраняет прежние поля, событие — после commit. */
|
||||
int chat_attachment_downloaded(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]);
|
||||
|
||||
// Ручная загрузка на storage-узле также означает запрос пользовательского доступа.
|
||||
void chat_attachment_want(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]);
|
||||
/* Ручная загрузка, включая storage: одной транзакцией установить wanted, сбросить error и ready отсутствующего cache.
|
||||
* 0/-1; запись observe обязательна. Внешняя транзакция запрещена, отказ сохраняет прежние поля. */
|
||||
int chat_attachment_want(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]);
|
||||
|
||||
/* Сохранить решение автора по ожидающему запросу peer: approve!=0 — выдать ключ, 0 — отказать.
|
||||
* 0 — собственная транзакция решения+outbox завершена, затем послано одно уведомление изменения;
|
||||
|
||||
Reference in new issue
Block a user