Commit author attachment decisions and delivery queue atomically
This commit is contained in:
evgeny committed
1 parent
77d6e8cfd0
commit
8a7f4469dc
2 files changed
+117
-35
No files matched your search
+112
-30
@@ -366,13 +366,40 @@ static int ak_send(struct UTUN_INSTANCE* inst, const char* channel, uint64_t pee
|
||||
return etcp_router_conn_send(conn, body, length, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED);
|
||||
}
|
||||
|
||||
/* Очередь содержит только команду; ключ читается при отправке из отдельной таблицы. */
|
||||
/* Записать ровно одну команду; транзакцией владеет вызывающий, ключ читается при отправке. */
|
||||
static int ak_queue(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32], uint64_t peer, int cmd) {
|
||||
sqlite3_stmt* st = ak_prepare(inst, "INSERT INTO attachment_key_outbox(channel,message_id,peer,command) VALUES(?,?,?,?)"
|
||||
" ON CONFLICT(channel,message_id,peer) DO UPDATE SET command=excluded.command,retry_at=0");
|
||||
if (!st) return -1;
|
||||
sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC);
|
||||
sqlite3_bind_int64(st, 3, (sqlite3_int64)peer); sqlite3_bind_int(st, 4, cmd); return ak_step(inst, st);
|
||||
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
|
||||
const char* stage = "prepare";
|
||||
int rc = sqlite3_prepare_v2(db, "INSERT INTO attachment_key_outbox(channel,message_id,peer,command) VALUES(?,?,?,?)"
|
||||
" ON CONFLICT(channel,message_id,peer) DO UPDATE SET command=excluded.command,retry_at=0", -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 = "bind peer"; rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)peer);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind command"; rc = sqlite3_bind_int(st, 4, cmd);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "step"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_DONE) goto fail;
|
||||
int changed = sqlite3_changes(db);
|
||||
stage = "finalize"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed != 1) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: queue rows=%d expected=1 channel=%s peer=%llu command=%d",
|
||||
changed, channel, (unsigned long long)peer, cmd);
|
||||
stage = "row count"; rc = SQLITE_CONSTRAINT; goto fail;
|
||||
}
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "attachment keys: queue staged channel=%s peer=%llu command=%d rows=%d",
|
||||
channel, (unsigned long long)peer, cmd, changed);
|
||||
return 0;
|
||||
fail:
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: queue channel=%s peer=%llu command=%d stage=%s rc=%d (%s) db=%s",
|
||||
channel, (unsigned long long)peer, cmd, 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: queue cleanup finalize rc=%d", final_rc);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* Разрешение проверяет полное подписанное членство и immutable историю. */
|
||||
@@ -381,14 +408,29 @@ static int ak_authorized(struct UTUN_INSTANCE* inst, const char* channel, const
|
||||
if (member_sync_verified_ed25519(inst, channel, peer, key)) return 0;
|
||||
char sql[256]; snprintf(sql, sizeof(sql), "SELECT 1 FROM msg_%s m JOIN attachment_keys a ON a.message_id=m.message_id"
|
||||
" AND a.channel=? WHERE m.message_id=? AND m.node_id=? AND a.author=? AND length(a.key)=32", channel);
|
||||
sqlite3_stmt* st = ak_prepare(inst, sql);
|
||||
if (!st) return 0;
|
||||
sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); sqlite3_bind_blob(st, 2, message, 32, SQLITE_STATIC);
|
||||
sqlite3_bind_int64(st, 3, (sqlite3_int64)inst->node_id); sqlite3_bind_int64(st, 4, (sqlite3_int64)inst->node_id);
|
||||
int rc = sqlite3_step(st); sqlite3_finalize(st);
|
||||
if (rc != SQLITE_ROW) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: request has no local author key/message channel=%s peer=%llu rc=%d",
|
||||
channel, (unsigned long long)peer, rc);
|
||||
return rc == SQLITE_ROW;
|
||||
sqlite3* db = CC(inst)->db; sqlite3_stmt* st = NULL;
|
||||
const char* stage = "prepare";
|
||||
int rc = sqlite3_prepare_v2(db, sql, -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 = "bind node"; rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)inst->node_id);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "bind author"; rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)inst->node_id);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "step"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_ROW) goto fail;
|
||||
stage = "finalize"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
return 1;
|
||||
fail:
|
||||
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: author authorization channel=%s peer=%llu stage=%s rc=%d (%s) db=%s",
|
||||
channel, (unsigned long long)peer, 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: authorization cleanup finalize rc=%d", final_rc);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Подтверждение протокола посылается только после durable записи. */
|
||||
@@ -569,30 +611,70 @@ done:
|
||||
|
||||
/* Только явное решение автора создаёт GRANT в durable outbox. */
|
||||
int chat_attachment_decide(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32], uint64_t peer, int approve) {
|
||||
if (!inst || !CC(inst) || CC(inst)->attachment_closing || !ak_channel(channel) || !ak_authorized(inst, channel, message, peer)) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: decision rejected by authorization"); return -1;
|
||||
if (!inst || !CC(inst) || !CC(inst)->attachment_initialized || CC(inst)->attachment_closing ||
|
||||
!ak_channel(channel) || !message || !peer || peer == inst->node_id) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: invalid decision arguments/state"); return -1;
|
||||
}
|
||||
if (ak_exec(inst, "SAVEPOINT attachment_decision")) return -1;
|
||||
sqlite3_stmt* st = ak_prepare(inst, "UPDATE attachment_key_requests SET state=? WHERE channel=? AND message_id=? AND peer=? AND incoming=1 AND state=0");
|
||||
if (!st) goto rollback;
|
||||
sqlite3_bind_int(st, 1, approve ? 2 : 3); sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 4, (sqlite3_int64)peer);
|
||||
if (ak_step(inst, st) || sqlite3_changes(CC(inst)->db) != 1 || ak_queue(inst, channel, message, peer, approve ? AK_GRANT : AK_DENY) ||
|
||||
ak_exec(inst, "RELEASE attachment_decision")) goto rollback;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: author decision channel=%s peer=%llu approve=%d", channel, (unsigned long long)peer, approve);
|
||||
sqlite3* db = CC(inst)->db;
|
||||
if (!sqlite3_get_autocommit(db)) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: decision rejected inside external transaction channel=%s peer=%llu",
|
||||
channel, (unsigned long long)peer); return -1;
|
||||
}
|
||||
sqlite3_stmt* st = NULL;
|
||||
int next = approve ? 2 : 3, command = approve ? AK_GRANT : AK_DENY;
|
||||
const char* stage = "begin";
|
||||
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "authorization";
|
||||
if (!ak_authorized(inst, channel, message, peer)) { rc = SQLITE_AUTH; goto fail; }
|
||||
stage = "state prepare";
|
||||
rc = sqlite3_prepare_v2(db, "UPDATE attachment_key_requests SET state=?"
|
||||
" WHERE channel=? AND message_id=? AND peer=? AND incoming=1 AND state=0", -1, &st, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "state bind state"; rc = sqlite3_bind_int(st, 1, next);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "state bind channel"; rc = sqlite3_bind_text(st, 2, channel, -1, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "state bind message"; rc = sqlite3_bind_blob(st, 3, message, 32, SQLITE_STATIC);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "state bind peer"; rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)peer);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
stage = "state step"; rc = sqlite3_step(st);
|
||||
if (rc != SQLITE_DONE) goto fail;
|
||||
int changed = sqlite3_changes(db);
|
||||
stage = "state finalize"; rc = sqlite3_finalize(st); st = NULL;
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
if (changed != 1) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "attachment keys: decision pending rows=%d expected=1 channel=%s peer=%llu",
|
||||
changed, channel, (unsigned long long)peer);
|
||||
stage = "state row count"; rc = SQLITE_CONSTRAINT; goto fail;
|
||||
}
|
||||
stage = "outbox";
|
||||
if (ak_queue(inst, channel, message, peer, command)) { rc = SQLITE_ERROR; goto fail; }
|
||||
stage = "commit"; rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
|
||||
if (rc != SQLITE_OK) goto fail;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "attachment keys: decision committed channel=%s peer=%llu message=%02x%02x%02x%02x state=%d command=%d",
|
||||
channel, (unsigned long long)peer, message[0], message[1], message[2], message[3], next, command);
|
||||
ak_changed(inst, channel); return 0;
|
||||
rollback:
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: decision failed channel=%s peer=%llu", channel, (unsigned long long)peer);
|
||||
ak_exec(inst, "ROLLBACK TO attachment_decision"); ak_exec(inst, "RELEASE attachment_decision"); return -1;
|
||||
fail:
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "attachment keys: decision channel=%s peer=%llu command=%d stage=%s rc=%d (%s) db=%s",
|
||||
channel, (unsigned long long)peer, command, 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: decision cleanup finalize rc=%d", final_rc);
|
||||
ak_rollback(inst, "decision", channel);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* GUI передаёт зафиксированные ID, результат возвращается после записи решения. */
|
||||
void chat_attachment_decision_trampoline(void* arg) {
|
||||
struct chat_attachment_decision* request = arg;
|
||||
int rc = chat_attachment_decide(request->inst, request->channel, request->message, request->peer, request->approve);
|
||||
uint8_t event[34]; size_t n = strlen(request->channel);
|
||||
event[0] = (uint8_t)n; memcpy(event + 1, request->channel, n); event[n + 1] = rc ? 0 : 1;
|
||||
chat_event_post(request->inst, CHAT_EVT_ATTACHMENT_KEYS_CHANGED, event, (int)n + 2); u_free(request);
|
||||
if (rc) {
|
||||
uint8_t event[34]; size_t n = strlen(request->channel);
|
||||
event[0] = (uint8_t)n; memcpy(event + 1, request->channel, n); event[n + 1] = 0;
|
||||
chat_event_post(request->inst, CHAT_EVT_ATTACHMENT_KEYS_CHANGED, event, (int)n + 2);
|
||||
}
|
||||
u_free(request);
|
||||
}
|
||||
|
||||
/* Пользовательский интерес хранится отдельно от фоновой storage-копии. */
|
||||
|
||||
@@ -61,10 +61,10 @@ void chat_attachment_downloaded(struct UTUN_INSTANCE* inst, const char* channel,
|
||||
void chat_attachment_want(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32]);
|
||||
|
||||
/* Сохранить решение автора по ожидающему запросу peer: approve!=0 — выдать ключ, 0 — отказать.
|
||||
* 0 — операция сообщила успех, -1 — ошибка/нет прав или ожидающего запроса; доставка идёт позже.
|
||||
* Известный дефект: SAVEPOINT внутри внешней транзакции не фиксирует решение до события UI;
|
||||
* пропуск INSERT в outbox также может дать 0 с решением без ответа. См. doc/chat_audit_findings.txt,
|
||||
* «Аудит решения автора»: проверенный commit решения+очереди здесь пока не гарантирован. */
|
||||
* 0 — собственная транзакция решения+outbox завершена, затем послано одно уведомление изменения;
|
||||
* -1 — ошибка/нет прав или ожидающего запроса, прежнее состояние сохранено. Доставка идёт позже.
|
||||
* Внешняя транзакция отклоняется без изменений; проверка прав выполняется внутри собственной.
|
||||
* Открытая после отказа rollback транзакция требует остановки/закрытия БД владельцем после poll. */
|
||||
int chat_attachment_decide(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message[32], uint64_t peer, int approve);
|
||||
|
||||
// ── Команда интерфейса ──
|
||||
@@ -76,7 +76,7 @@ struct chat_attachment_decision {
|
||||
uint64_t peer;
|
||||
int approve;
|
||||
};
|
||||
// arg — u_malloc/u_calloc chat_attachment_decision, освобождается здесь; результат — CHAT_EVT_ATTACHMENT_KEYS_CHANGED.
|
||||
// arg — u_malloc/u_calloc chat_attachment_decision, освобождается здесь; успех уже уведомлён decide, ошибка — событие с success=0.
|
||||
void chat_attachment_decision_trampoline(void* arg);
|
||||
|
||||
// ── Локальное удаление канала ──
|
||||
|
||||
Reference in new issue
Block a user