Register history after shared Merkle startup and validate page cursors

This commit is contained in:
evgeny committed 2026-10-26 18:27:26 +01:00
1 parent e822f9fef7
commit 012cbdcd49
2 files changed
+18 -3

No files matched your search

+6 -1
View File
@@ -573,11 +573,12 @@ static void cs_on_peer_sleep(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
if (sleeping) {
cs_cancel_sync(cs, peer_node_id, 0, "peer_sleep");
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer sleep peer=0x%016llx — cancel member_sync sessions", CS_ID, (unsigned long long)peer_node_id);
merkle_sync_cancel_peer(inst, peer_node_id);
merkle_sync_recheck_peer(inst, peer_node_id);
} else {
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer awake peer=0x%016llx — schedule sync", CS_ID, (unsigned long long)peer_node_id);
cs_schedule_peer_sync(cs, peer_node_id, "peer_awake");
}
db_sync_peer_policy_changed(inst, peer_node_id);
}
/* ── Per-chat «play sound on message» (троттлинг db_sync на сервере) ── */
@@ -867,6 +868,10 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: feedback initialization failed", CS_ID);
member_sync_destroy(inst); chat_join_port_close(cs->port); chat_join_port_release(cs->port); inst->chat_sync = NULL; u_free(cs); return -1;
}
if (db_sync_register_model(inst) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: history model initialization failed", CS_ID);
member_sync_destroy(inst); chat_join_port_close(cs->port); chat_join_port_release(cs->port); inst->chat_sync = NULL; u_free(cs); return -1;
}
if (group_avatar_init(inst) < 0) {
member_sync_destroy(inst); chat_join_port_close(cs->port); chat_join_port_release(cs->port); inst->chat_sync = NULL; u_free(cs); return -1;
}
+12 -2
View File
@@ -737,7 +737,8 @@ static int db_notice_stage(struct DB_SYNC_INSTANCE* si, struct db_record* record
}
/* Полная проверка входящей страницы — до BEGIN; write-lock охватывает только небольшой commit данных и индекса. */
static int db_message_apply(void* ctx, const char* ns, uint64_t peer, const uint8_t* data, size_t len) {
static int db_message_apply(void* ctx, const char* ns, uint64_t peer, uint64_t prefix, int av,
const merkle_key_t* after, const merkle_key_t* next_key, int more, const uint8_t* data, size_t len) {
(void)peer;
struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
if (!si || si->index.pending) return -1;
@@ -746,12 +747,21 @@ static int db_message_apply(void* ctx, const char* ns, uint64_t peer, const uint
while (p < data + len) {
if (count == DB_BATCH_RECORDS || db_parse_record(si, &p, data + len, &records[count]) < 0) return -1;
struct db_record* r = &records[count];
if (merkle_sync_level_prefix(db_read64(r->message_id), MT_MAX_LEVEL) != prefix ||
(av && memcmp(r->message_id, after->bytes, 32) <= 0)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page key outside requested range ns=%s", ns); return -1;
}
if (db_verify_author_sig(si, r->timestamp, r->author, r->data, r->data_len, r->signature, r->sig_len) != DB_CHECK_VALID) return -1;
if (count && memcmp(records[count - 1].message_id, r->message_id, 32) >= 0) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history incoming page not strictly ordered ns=%s", ns); return -1;
}
count++;
}
static const uint8_t empty[32];
if ((!count && (more || memcmp(next_key->bytes, empty, 32))) ||
(count && memcmp(next_key->bytes, records[count - 1].message_id, 32))) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page cursor does not match records ns=%s", ns); return -1;
}
if (!count) return 0;
sqlite3* db = SI_DB(si);
if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "page begin");
@@ -773,7 +783,7 @@ rollback:
}
static const struct merkle_sync_data_ops db_message_ops = {
.update_bucket_hash = db_message_hash, .get_page = db_message_page, .apply_items = db_message_apply,
.update_bucket_hash = db_message_hash, .get_page = db_message_page, .apply_page = db_message_apply,
.validate_peer = db_message_peer, .can_run = db_message_can_run, .on_result = db_message_result
};