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

Keep chain reads and delivery confirmation failures explicit

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 18:26:58 +01:00
1 parent 557c98aeae
commit 7853fb0fc6
2 files changed
+58 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/db_sync.c
+28 -17
View File
@@ -133,7 +133,7 @@ void db_sync_remove_done_cbk(struct UTUN_INSTANCE* inst, db_sync_done_fn fn, voi
#define SI_DB(si) ((si)->db_sync->db)
#define SI_TBL(si) ((si)->table_name)
#define SI_SHRT(si) ({ \
static char _tbuf[28]; \
static char _tbuf[72]; \
snprintf(_tbuf, sizeof(_tbuf), "%s.%04llX", SI_TBL(si), (unsigned long long)((si)->group_id & 0xFFFF)); \
_tbuf; \
})
@@ -188,7 +188,7 @@ static int si_table_init(struct DB_SYNC_INSTANCE* si) {
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history schema: adding message_id table=%s group=%016llx",
SI_TBL(si), (unsigned long long)si->group_id);
char replacement[64], sql[768];
snprintf(replacement, sizeof(replacement), "new_%s", SI_TBL(si));
snprintf(replacement, sizeof(replacement), "new_%.48s", SI_TBL(si));
stage = "create replacement";
if (si_table_create(si, replacement, 0) < 0) goto rollback;
stage = "read history";
@@ -565,6 +565,7 @@ static int db_recalc_batch(struct DB_SYNC_INSTANCE* si) {
sqlite3_int64 n_auth = sqlite3_column_int64(sel, 2);
const uint8_t* sig = sqlite3_column_blob(sel, 3);
uint8_t ch[32];
if (!sig && sqlite3_errcode(db) == SQLITE_NOMEM) goto rollback;
if (!sig || sqlite3_column_bytes(sel, 3) != DB_SIG_SIZE) {
sqlite3_int64 rowid = sqlite3_column_int64(sel, 4);
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history recalc malformed stored signature tbl=%s ts=%lld bytes=%d",
@@ -585,7 +586,10 @@ static int db_recalc_batch(struct DB_SYNC_INSTANCE* si) {
if (rc != SQLITE_DONE) goto rollback;
sqlite3_finalize(upd); upd = NULL; sqlite3_finalize(sel); sel = NULL;
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history recalc COMMIT tbl=%s rc=%d", SI_TBL(si), rc); goto rollback;
}
si->recalc.pos += processed;
si->recalc.pending = si->recalc.pos < total;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history recalc committed tbl=%s pos=%u processed=%u total=%u pending=%u",
@@ -732,8 +736,10 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
exists = 1; rowid = sqlite3_column_int64(stmt, 0);
uint64_t author = (uint64_t)sqlite3_column_int64(stmt, 1);
const char* data = sqlite3_column_blob(stmt, 2);
if (!data && sqlite3_errcode(db) == SQLITE_NOMEM) goto rollback;
size_t len = (size_t)sqlite3_column_bytes(stmt, 2);
const uint8_t* sig = sqlite3_column_blob(stmt, 3);
if (!sig && sqlite3_errcode(db) == SQLITE_NOMEM) goto rollback;
size_t sig_len = (size_t)sqlite3_column_bytes(stmt, 3);
enum db_record_check stored = db_verify_author_sig(si, ts, author, data, len, sig, sig_len);
if (stored == DB_CHECK_ERROR || stored == DB_CHECK_DEFERRED) goto rollback;
@@ -778,7 +784,10 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
stmt = NULL;
commit:
stage = "commit";
if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
rc = sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history insert COMMIT tbl=%s rc=%d", SI_TBL(si), rc); goto rollback;
}
if (deleted) db_history_deleted(si);
if (ret == 0) {
if (id >= si->next_id) si->next_id = id + 1;
@@ -967,7 +976,7 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint32_t tp = (pc < mc ? pc : mc);
if (tp > 0) tp--;
uint64_t my_ch8_at_tp = 0;
if (mc > 0) db_chain_hash8_at(si, tp, &my_ch8_at_tp);
if (mc > 0 && db_chain_hash8_at(si, tp, &my_ch8_at_tp) < 0) { si_abort_sync(si, src, "chain read failed"); return; }
uint8_t has_tail = (mc > tp + 1) ? 1 : 0;
static const uint16_t iv[16] = {1, 1, 1, 2, 2, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4};
uint32_t accum = 0; int sparse_cnt = 0;
@@ -979,8 +988,7 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint32_t off = 0;
resp[off++] = DB_MSG_INIT_RESP;
memcpy(resp + off, &tp, 4); off += 4;
uint64_t my_ch8 = 0;
if (mc > 0) db_chain_hash8_at(si, tp, &my_ch8);
uint64_t my_ch8 = my_ch8_at_tp;
memcpy(resp + off, &my_ch8, 8); off += 8;
resp[off++] = has_tail;
uint32_t sp = off;
@@ -992,7 +1000,7 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const
accum += iv[i];
uint32_t pos = tp - accum;
uint64_t sch8;
if (db_chain_hash8_at(si, pos, &sch8) != 0) break;
if (db_chain_hash8_at(si, pos, &sch8) != 0) { si_abort_sync(si, src, "sparse chain read failed"); return; }
if (off + 12 > sizeof(resp)) break;
memcpy(resp + off, &pos, 4); off += 4;
memcpy(resp + off, &sch8, 8); off += 8;
@@ -1024,7 +1032,7 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint64_t my_ch8 = 0;
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
if (mc > 0 && tp < mc) db_chain_hash8_at(si, tp, &my_ch8);
if (mc > 0 && tp < mc && db_chain_hash8_at(si, tp, &my_ch8) < 0) { si_abort_sync(si, src, "chain read failed"); return; }
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← INIT_RESP: tp=%u my=%u my_ch8=%016llX peer_ch8=%016llX has_tail=%u sc=%d",
SI_SHRT(si), (unsigned long long)(src >> 16), tp, mc, my_ch8, peer_ch8, has_tail, sc);
@@ -1039,8 +1047,9 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const
while (sent < mc) {
uint32_t b = mc - sent; if (b > DB_SEND_DATA_MAX) b = DB_SEND_DATA_MAX;
uint32_t want = (sent + b >= mc) ? mc : DB_WANT_FROM_NONE;
if (si_send_data_batch(si, src, sent, b, vp, want) <= 0) break;
sent += b;
int batch_sent = si_send_data_batch(si, src, sent, b, vp, want);
if (batch_sent <= 0) break;
sent += (uint32_t)batch_sent;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← INIT_RESP: peer empty → sent %u/%u records, awaiting response",
SI_SHRT(si), (unsigned long long)(src >> 16), sent, mc);
@@ -1128,8 +1137,10 @@ static int si_send_data_batch(struct DB_SYNC_INSTANCE* si, uint64_t dst, uint32_
uint64_t rts = (uint64_t)sqlite3_column_int64(stmt, 1);
uint64_t rauth = (uint64_t)sqlite3_column_int64(stmt, 2);
const uint8_t* rd = (const uint8_t*)sqlite3_column_blob(stmt, 3);
if (!rd && sqlite3_errcode(db) == SQLITE_NOMEM) goto batch_error;
uint32_t rdl = (uint32_t)sqlite3_column_bytes(stmt, 3); if (!rd) rdl = 0;
const uint8_t* rsig = (const uint8_t*)sqlite3_column_blob(stmt, 4);
if (!rsig && sqlite3_errcode(db) == SQLITE_NOMEM) goto batch_error;
int rsl = sqlite3_column_bytes(stmt, 4); if (!rsig) rsl = 0;
enum db_record_check check = db_verify_author_sig(si, rts, rauth, (const char*)rd, rdl, rsig, (size_t)rsl);
if (check == DB_CHECK_ERROR || check == DB_CHECK_DEFERRED) goto batch_error;
@@ -1224,7 +1235,7 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
struct SI_PEER* sp = si_peer_find(si, src);
const uint8_t* ptr = p + 14;
uint16_t received = 0, duplicates = 0, bad_sigs = 0;
uint16_t received = 0, duplicates = 0;
for (uint16_t i = 0; i < count && i < 32; i++) {
uint64_t rid, rts, rauthor; uint32_t rdlen; const uint8_t* rdata, *rsig; int rsiglen;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) {
@@ -1234,7 +1245,6 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
if (ret < 0) { si_abort_sync(si, src, "incoming verification or insertion failed"); return; }
if (ret >= 0) received++;
if (ret == 1) duplicates++;
if (ret == -2) bad_sigs++;
}
if (ptr != p + len) { si_abort_sync(si, src, "trailing batch bytes"); return; }
if (db_sync_flush_recalc(si) < 0) { si_abort_sync(si, src, "chain rebuild after insertion failed"); return; }
@@ -1242,14 +1252,13 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint32_t old_spos = sp ? sp->synced_pos : 0;
if (received > 0) {
uint32_t rst = (from < old_spos) ? from : old_spos;
if (sp) sp->synced_pos = from + (count - bad_sigs) - 1;
if (sp) sp->synced_pos = from + received - 1;
for (int j = 0; j < si->peer_count; j++)
if (si->peers[j].synced_pos >= rst && &si->peers[j] != sp)
si->peers[j].synced_pos = rst > 0 ? rst - 1 : 0;
}
char extra[32] = ""; if (duplicates) snprintf(extra, sizeof(extra), " (%u dup)", duplicates);
if (bad_sigs) { size_t el = strlen(extra); snprintf(extra+el, sizeof(extra)-el, "%s%u bad", extra[0]?", ":" (", bad_sigs); }
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "sync [%s:%04llX] ← DATA: from=%u cnt=%u vp=%u want=%u → ins=%u%s mc=%u sp=%u→%u",
SI_SHRT(si), (unsigned long long)(src >> 16), from, count, vp, want_from, received, extra, db_count(si),
old_spos, sp ? sp->synced_pos : 0);
@@ -1272,7 +1281,8 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
if (want_from == DB_WANT_FROM_NONE && sp && sp->sync_state == 1) {
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
uint64_t mch8 = 0; if (mc > 0) db_chain_hash8_at(si, mc - 1, &mch8);
uint64_t mch8 = 0;
if (mc > 0 && db_chain_hash8_at(si, mc - 1, &mch8) < 0) { si_abort_sync(si, src, "chain read failed"); return; }
uint8_t sd[13]; sd[0] = DB_MSG_SYNC_DONE;
memcpy(sd + 1, &mc, 4); memcpy(sd + 5, &mch8, 8);
if (db_sync_send(si, src, sd, 13) != 0) return;
@@ -1293,7 +1303,8 @@ static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint64_t pch8; memcpy(&pch8, p + 4, 8);
uint32_t mc;
if (db_count_checked(si, &mc) < 0) { si_abort_sync(si, src, "count failed"); return; }
uint64_t mch8 = 0; if (mc > 0) db_chain_hash8_at(si, mc - 1, &mch8);
uint64_t mch8 = 0;
if (mc > 0 && db_chain_hash8_at(si, mc - 1, &mch8) < 0) { si_abort_sync(si, src, "chain read failed"); return; }
struct SI_PEER* sp = si_peer_find(si, src);
if (mc == pc && mch8 == pch8) {
tests/test_db_sync.c
+30
View File
@@ -60,6 +60,20 @@ static void reset_done_flags(void) { g_done_a = g_done_b = g_done_c = 0; }
static char temp_dir[] = "/tmp/utun_dbsync_XXXXXX";
static char config_a[256], config_b[256], config_c[256];
static uint32_t ca_target, cb_target, cc_target;
static uint64_t delivery_timestamp;
/* Оба сообщения одного автора в одной ms получают отдельные реальные ACK. */
static int cond_delivery(void) {
sqlite3_stmt* stmt = NULL;
int rc = sqlite3_prepare_v2(inst_a->topo_sqlite_db,
"SELECT count(*) FROM push_t WHERE timestamp=? AND node_id=? AND delivered_peers=1 AND (flags & 1)=1", -1, &stmt, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 1, (sqlite3_int64)delivery_timestamp);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 2, (sqlite3_int64)inst_a->node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
int complete = rc == SQLITE_ROW && sqlite3_column_int(stmt, 0) == 2;
if (rc != SQLITE_ROW) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "delivery test lookup rc=%d: %s", rc, sqlite3_errmsg(inst_a->topo_sqlite_db));
sqlite3_finalize(stmt); return complete;
}
static int write_file(const char* path, const char* fmt, ...) {
va_list ap; FILE* f = fopen(path, "w");
@@ -376,6 +390,22 @@ int main(void) {
}
printf(" signed reply resolves on sender and receiver PASS\n");
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "Phase 5c: same-timestamp messages over PUSH/ACK");
delivery_timestamp = db_sync_next_timestamp(si_a);
for (int i = 0; i < 2; i++) {
char json[64]; snprintf(json, sizeof(json), "{\"delivery\":%d}", i);
uint8_t body[72], signature[64]; size_t len = strlen(json);
memcpy(body, &delivery_timestamp, 8); memcpy(body + 8, json, len);
if (sc_ed25519_sign(inst_a->my_ed25519_privkey, body, len + 8, signature) != SC_OK ||
db_sync_insert_signed(si_a, json, len, signature, 64, delivery_timestamp, NULL) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "delivery test submission failed index=%d", i); test_phase = 2; goto done;
}
}
ca_target = cb_target = 6;
if (cond_delivery() || !wait_for("delivery messages on both peers", _cond_both_count, PHASE_TIMEOUT_TB) ||
!wait_for("two distinct delivery ACKs", cond_delivery, PHASE_TIMEOUT_TB)) { test_phase = 2; goto done; }
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "PASS two same-timestamp PUSH records acknowledged separately");
// Phase 6: triple star — cascade notification.
// ===================================================================
printf("Phase 6: triple star\n");