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

Check feedback lookups and writes and retain rollback ownership

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 03:33:34 +01:00
1 parent 1f19720cb8
commit a591469770
1 file changed
+40 -21
Download Patch File Download Diff File Expand all files Collapse all files
src/chat/chat_feedback.c
+40 -21
View File
@@ -171,12 +171,19 @@ int chat_feedback_get(struct UTUN_INSTANCE* inst, const char* channel, const uin
snprintf(sql, sizeof(sql), "SELECT message_id,node_id,kind,ver,data,signature FROM %s"
" WHERE message_id=? AND node_id=? AND kind=?", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "get");
sqlite3_bind_blob(st, 1, message_id, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id);
sqlite3_bind_int(st, 3, kind);
int rc = sqlite3_step(st), result = rc == SQLITE_ROW ? (fb_read_row(st, out) < 0 ? -1 : 1) : 0;
sqlite3_finalize(st);
if (rc != SQLITE_ROW && rc != SQLITE_DONE) return fb_db_error(inst, channel, "get step");
int rc = sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 3, kind);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
struct chat_feedback_record record;
int result = rc == SQLITE_ROW ? (fb_read_row(st, &record) < 0 ? -1 : 1) : 0;
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: lookup failed ch=%s node=%016llx kind=%u step=%d finalize=%d db=%s",
channel, (unsigned long long)node_id, kind, rc, final, sqlite3_errmsg(inst->topo_sqlite_db)); return -1;
}
if (result > 0) *out = record;
return result;
}
@@ -217,14 +224,22 @@ static int fb_apply_row(struct UTUN_INSTANCE* inst, const char* channel, const s
snprintf(sql, sizeof(sql), "INSERT INTO %s(record_key,bucket,message_id,node_id,kind,ver,data,signature)"
" VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(record_key) DO UPDATE SET ver=excluded.ver,data=excluded.data,signature=excluded.signature", table);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) != SQLITE_OK) return fb_db_error(inst, channel, "put");
sqlite3_bind_blob(st, 1, record_key.bytes, 32, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)(fb_read64(record_key.bytes) >> 39));
sqlite3_bind_blob(st, 3, r->message_id, 32, SQLITE_STATIC); sqlite3_bind_int64(st, 4, (sqlite3_int64)r->node_id);
sqlite3_bind_int(st, 5, r->kind); sqlite3_bind_int64(st, 6, (sqlite3_int64)r->ver);
sqlite3_bind_blob(st, 7, r->data, r->data_len, SQLITE_STATIC); sqlite3_bind_blob(st, 8, r->signature, 64, SQLITE_STATIC);
int rc = sqlite3_step(st); sqlite3_finalize(st);
if (rc != SQLITE_DONE) return fb_db_error(inst, channel, "put step");
int rc = sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, record_key.bytes, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)(fb_read64(record_key.bytes) >> 39));
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 3, r->message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)r->node_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 5, r->kind);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 6, (sqlite3_int64)r->ver);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 7, r->data, r->data_len, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 8, r->signature, 64, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int changed = rc == SQLITE_DONE ? sqlite3_changes(inst->topo_sqlite_db) : 0;
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || changed != 1 || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: write failed ch=%s node=%016llx kind=%u step=%d changes=%d finalize=%d db=%s",
channel, (unsigned long long)r->node_id, r->kind, rc, changed, final, sqlite3_errmsg(inst->topo_sqlite_db)); return -1;
}
if (merkle_sync_recompute_path(inst, MT_DATA_FEEDBACK, channel, fb_read64(record_key.bytes)) < 0) return -1;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "feedback: applied ch=%s message=%s node=%016llx kind=%u ver=%llu previous=%llu bytes=%u",
channel, message_hex, (unsigned long long)r->node_id, r->kind, (unsigned long long)r->ver,
@@ -239,15 +254,19 @@ static void fb_committed(struct UTUN_INSTANCE* inst, const char* channel, const
chat_event_post(inst, CHAT_EVT_FEEDBACK_CHANGED, event, (int)len + 34);
}
/* Завершает savepoint, при ошибке не публикует частичные изменения. */
/* Завершает только свою транзакцию; неудачный rollback никогда не переходит в commit. */
static int fb_finish_transaction(struct UTUN_INSTANCE* inst, const char* channel, int result) {
sqlite3* db = inst->topo_sqlite_db;
if (result < 0 && sqlite3_exec(db, "ROLLBACK TO feedback_update", NULL, NULL, NULL) != SQLITE_OK)
if (result >= 0) {
if (sqlite3_exec(db, "COMMIT /* feedback update */", NULL, NULL, NULL) == SQLITE_OK) return result;
fb_db_error(inst, channel, "commit");
}
if (!sqlite3_get_autocommit(db) && sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK)
fb_db_error(inst, channel, "rollback");
if (sqlite3_exec(db, "RELEASE feedback_update", NULL, NULL, NULL) == SQLITE_OK) return result;
fb_db_error(inst, channel, "commit");
if (sqlite3_exec(db, "ROLLBACK TO feedback_update; RELEASE feedback_update", NULL, NULL, NULL) != SQLITE_OK)
fb_db_error(inst, channel, "commit cleanup");
if (!sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: transaction remains open ch=%s; owner must stop core and close DB after callback", channel);
utun_instance_stop(inst);
}
return -1;
}
@@ -259,7 +278,7 @@ int chat_feedback_apply(struct UTUN_INSTANCE* inst, const char* channel, const s
if (!sqlite3_get_autocommit(inst->topo_sqlite_db)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "feedback: public apply inside external transaction ch=%s", channel); 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 update */", NULL, NULL, NULL) != SQLITE_OK)
return fb_db_error(inst, channel, "begin");
int result = fb_finish_transaction(inst, channel, fb_apply_row(inst, channel, r));
if (result > 0) { merkle_sync_changed(inst, MT_DATA_FEEDBACK, channel); fb_committed(inst, channel, r); }