diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index 0ee29c24..06e8246f 100644 --- a/src/chat/db_sync.c +++ b/src/chat/db_sync.c @@ -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) { diff --git a/tests/test_db_sync.c b/tests/test_db_sync.c index b435b0c2..2fc48346 100644 --- a/tests/test_db_sync.c +++ b/tests/test_db_sync.c @@ -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");