diff --git a/src/chat/chat_feedback.c b/src/chat/chat_feedback.c index 283e7edd..a4e87ea8 100644 --- a/src/chat/chat_feedback.c +++ b/src/chat/chat_feedback.c @@ -291,29 +291,40 @@ int chat_feedback_set(struct UTUN_INSTANCE* inst, const char* channel, const uin if (fb_table(inst, channel, table, &id) < 0 || !message_id || len > CHAT_FEEDBACK_DATA_MAX || (len && !data)) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid local state"); return -1; } + if (!sqlite3_get_autocommit(inst->topo_sqlite_db)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local set inside external transaction ch=%s", channel); return -1; + } + if (sqlite3_exec(inst->topo_sqlite_db, "BEGIN IMMEDIATE /* feedback local set */", NULL, NULL, NULL) != SQLITE_OK) + return fb_db_error(inst, channel, "local begin"); + int result = -1; + struct chat_feedback_record r = { .node_id = inst->node_id, .ver = 1, .kind = kind, .data_len = len }, old; + memcpy(r.message_id, message_id, 32); if (len) memcpy(r.data, data, len); uint8_t author_key[32]; enum ms_author_status status = member_sync_author_key(inst, channel, inst->node_id, author_key); if (status != MS_AUTHOR_ACTIVE) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local action denied ch=%s node=%016llx status=%d", - channel, (unsigned long long)inst->node_id, status); return -1; + channel, (unsigned long long)inst->node_id, status); goto finish; } - struct chat_feedback_record r = { .node_id = inst->node_id, .ver = 1, .kind = kind, .data_len = len }, old; - memcpy(r.message_id, message_id, 32); if (len) memcpy(r.data, data, len); int found = chat_feedback_get(inst, channel, message_id, inst->node_id, kind, &old); - if (found < 0) return -1; - if (found && old.data_len == len && !memcmp(old.data, r.data, len)) { if (out_ver) *out_ver = old.ver; return 0; } + if (found < 0) goto finish; + if (found && old.data_len == len && !memcmp(old.data, r.data, len)) { r.ver = old.ver; result = 0; goto finish; } if (found && old.ver == INT64_MAX) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: version exhausted ch=%s", channel); return -1; + DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: version exhausted ch=%s", channel); goto finish; } if (found) r.ver = old.ver + 1; uint8_t body[FB_WIRE_MAX + 32]; int bytes = fb_signed_bytes(id, &r, body); if (bytes < 0 || sc_ed25519_sign(inst->my_ed25519_privkey, body, bytes, r.signature) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local signing failed ch=%s", channel); return -1; + DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local signing failed ch=%s", channel); goto finish; } - int result = chat_feedback_apply(inst, channel, &r); + result = fb_apply_row(inst, channel, &r); +finish: + result = fb_finish_transaction(inst, channel, result); if (result >= 0 && out_ver) *out_ver = r.ver; - if (result > 0) DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local state saved ch=%s kind=%u ver=%llu bytes=%zu", - channel, kind, (unsigned long long)r.ver, len); + if (result > 0) { + DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "feedback: local state saved ch=%s kind=%u ver=%llu bytes=%zu", + channel, kind, (unsigned long long)r.ver, len); + merkle_sync_changed(inst, MT_DATA_FEEDBACK, channel); fb_committed(inst, channel, &r); + } return result; } @@ -381,7 +392,10 @@ static int fb_hash(void* ctx, const char* channel, uint8_t level, uint64_t prefi /* Вся страница применяется атомарно; события отправляются после её commit. */ static int fb_apply_page(void* ctx, const char* channel, uint64_t peer, const uint8_t* data, size_t len) { struct UTUN_INSTANCE* inst = ctx; (void)peer; - if (len < 2) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: short page ch=%s", channel); return -1; } + if (!inst || !inst->topo_sqlite_db || !sqlite3_get_autocommit(inst->topo_sqlite_db)) { + DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: page requires own transaction ch=%s", channel); return -1; + } + if (!data || len < 2) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: short/missing page ch=%s", channel); return -1; } unsigned count = ((unsigned)data[0] << 8) | data[1]; size_t off = 2; if (count > (len - 2) / FB_FIXED_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: invalid page count=%u bytes=%zu", count, len); return -1; @@ -395,7 +409,7 @@ static int fb_apply_page(void* ctx, const char* channel, uint64_t peer, const ui if (!count) return 0; uint8_t* changed = u_calloc(count, 1); if (!changed) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: page allocation failed"); return -1; } - if (sqlite3_exec(inst->topo_sqlite_db, "SAVEPOINT feedback_update", NULL, NULL, NULL) != SQLITE_OK) { + if (sqlite3_exec(inst->topo_sqlite_db, "BEGIN IMMEDIATE /* feedback page */", NULL, NULL, NULL) != SQLITE_OK) { u_free(changed); return fb_db_error(inst, channel, "page begin"); } off = 2; int result = 0; @@ -526,7 +540,7 @@ int chat_feedback_init(struct UTUN_INSTANCE* inst) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: init requires database outside transaction"); return -1; } if (merkle_sync_register(inst, MT_DATA_FEEDBACK, &fb_ops, inst) < 0) return -1; - if (sqlite3_exec(inst->topo_sqlite_db, "SAVEPOINT feedback_update", NULL, NULL, NULL) != SQLITE_OK) + if (sqlite3_exec(inst->topo_sqlite_db, "BEGIN IMMEDIATE /* feedback init */", NULL, NULL, NULL) != SQLITE_OK) return fb_db_error(inst, "-", "init begin"); char channel[21] = "", next[21]; int scan, result = 0; unsigned buckets = 0; while ((scan = fb_next_channel(inst, channel, next, 0)) > 0) { diff --git a/src/chat/chat_feedback.h b/src/chat/chat_feedback.h index f6d8375b..5a10fa6c 100644 --- a/src/chat/chat_feedback.h +++ b/src/chat/chat_feedback.h @@ -4,7 +4,10 @@ * но не входят в Merkle, выдачу и результаты. Допуск проверяет member_sync; * интерпретацию ответа относительно опроса выполняет вызывающий модуль. * Все операции — в uasync-потоке. init вызывается после member_sync_init; - * общий Merkle-движок освобождается member_sync_destroy. */ + * общий Merkle-движок освобождается member_sync_destroy. + * apply/set/приём страницы/init владеют своей BEGIN IMMEDIATE/COMMIT; внешняя транзакция запрещена. + * Запись и Merkle-индекс фиксируются вместе, уведомление — после commit. Неудачный rollback + * с открытой транзакцией запрашивает stop ядра; владелец закрывает БД вне callback. */ #ifndef CHAT_FEEDBACK_H #define CHAT_FEEDBACK_H @@ -42,14 +45,17 @@ int chat_feedback_member_update(struct UTUN_INSTANCE* inst, uint64_t node_id); void chat_feedback_member_committed(struct UTUN_INSTANCE* inst, uint64_t node_id); /* Подготовить таблицу канала; 0/-1. Не создаёт членство или сетевую группу. */ int chat_feedback_ensure_channel(struct UTUN_INSTANCE* inst, const char* channel); -/* Прочитать сохранённое состояние, включая неактивное: 1 найдено, 0 отсутствует, -1 ошибка. */ +/* Прочитать сохранённое состояние, включая неактивное:1 найдено,0 отсутствует,-1 ошибка. + * out меняется только при успехе1;ошибка bind/step/finalize не считается отсутствием. */ int chat_feedback_get(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message_id[32], uint64_t node_id, uint8_t kind, struct chat_feedback_record* out); /* Проверить и применить подписанную запись: 1 изменилось, 0 повтор/старая/автор недоступен, -1 ошибка. * Одинаковый ver разрешается сравнением data; подпись проверяется до сравнения. */ int chat_feedback_apply(struct UTUN_INSTANCE* inst, const char* channel, const struct chat_feedback_record* record); -/* Подписать локальное полное состояние с ver+1; NULL/0 отменяет. 1/0/-1 как apply. - * out_ver необязателен. Семантику data проверяет вызывающий модуль до этого вызова. */ +/* Подписать локальное полное состояние с ver+1; NULL/0 отменяет.1/0/-1 как apply. + * Допуск/прежняя версия/подпись/запись выполняются под собственной блокировкой,включая no-op. + * out_ver необязателен,меняется только после commit. Канонический формат проверяет codec, + * семантику ответа относительно опроса проверяет вызывающий до этого вызова. */ int chat_feedback_set(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message_id[32], uint8_t kind, const uint8_t* data, size_t len, uint64_t* out_ver);