Own feedback version selection and page publication transactions

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) {

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);