From 0444a4bfd9b866a0404e853eb569d8a9fe75774c Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 17:40:01 +0100 Subject: [PATCH] Resume admitted history peers and gate automatic Merkle work by policy --- src/chat/db_sync.c | 14 ++++++++++++-- src/chat/merkle_sync.c | 15 ++++++++------- tests/test_merkle_protocol.c | 2 +- 3 files changed, 21 insertions(+), 10 deletions(-) diff --git a/src/chat/db_sync.c b/src/chat/db_sync.c index 293418e6..6b52aa98 100644 --- a/src/chat/db_sync.c +++ b/src/chat/db_sync.c @@ -511,7 +511,9 @@ static int db_index_batch(struct DB_SYNC_INSTANCE* si) { if (now == DB_CHECK_VALID && item->check != DB_CHECK_BAD_FORMAT) matches = db_same_record(&item->record, ¤t); if (sqlite3_finalize(st) != SQLITE_OK) { st = NULL; goto verify_rollback; } st = NULL; - if (!matches) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history preflight source changed tbl=%s; retry", SI_TBL(si)); goto verify_rollback; } + if (!matches) { + DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history preflight source changed tbl=%s; retry", SI_TBL(si)); goto verify_rollback; + } if (item->check == DB_CHECK_BAD_FORMAT || item->check == DB_CHECK_BAD_SIGNATURE) { if (db_delete_bad_row(si, item->record.rowid) < 0) goto verify_rollback; removed++; @@ -806,7 +808,10 @@ static int db_send(struct DB_SYNC_INSTANCE* si, uint64_t peer, uint8_t type, con if (len) memcpy(e->dgram + 10, data, len); e->len = (uint16_t)(10 + len); int rc = etcp_send(conn, e); - if (rc != 0) { queue_dgram_free(e); queue_entry_free(e); DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history send rejected peer=%016llx rc=%d", peer, rc); } + if (rc != 0) { + queue_dgram_free(e); queue_entry_free(e); + DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history send rejected peer=%016llx rc=%d", peer, rc); + } return rc; } @@ -1154,6 +1159,11 @@ void db_sync_set_peer_sound(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t void db_sync_members_changed(struct UTUN_INSTANCE* inst, const char* ns) { struct DB_SYNC* db = inst ? inst->db_sync : NULL; if (!db) return; merkle_sync_revalidate_peers(inst, MT_DATA_MESSAGES, ns); + struct DB_SYNC_INSTANCE* channel = db_sync_instance_find(inst, strtoull(ns, NULL, 10)); + if (channel) { + for (int i = 0; i < channel->peer_count; i++) + if (!channel->peers[i].sync_state) db_start_peer(channel, channel->peers[i].node_id); + } for (int i = 0; i < db->instance_count; i++) { struct DB_SYNC_INSTANCE* si = db->instances[i]; if (si->index.pending && si->index.waiting_key) { diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index 30da6222..c9294416 100644 --- a/src/chat/merkle_sync.c +++ b/src/chat/merkle_sync.c @@ -207,7 +207,8 @@ static int ms_root(struct ms_session* s, uint8_t root[MT_HASH_SIZE]) { return merkle_sync_read_hash(s->ms->inst, s->dataset->id, s->ns, 0, 0, root); } -static int ms_can_run(struct merkle_sync* ms, struct ms_dataset* model, struct ETCP_CONN* conn, const char* ns) { +static int ms_can_run(struct ms_dataset* model, struct ETCP_CONN* conn, const char* ns) { + if (model->ops->validate_peer && model->ops->validate_peer(model->data_ctx, ns, conn->peer_node_id) != 1) return 0; return model->ops->can_run ? model->ops->can_run(model->data_ctx, ns, conn->peer_node_id) == 1 : !conn->peer_sleep_phase; } @@ -219,7 +220,7 @@ static void ms_sent(struct ll_queue* q, void* arg) { int type = entry->dgram[9]; size_t len = entry->len; if (!s->conn->links_up || s->conn->state == 2 || s->conn->close_requested || q != s->conn->send_input_q || - !ms_can_run(s->ms, s->dataset, s->conn, s->ns)) { + !ms_can_run(s->dataset, s->conn, s->ns)) { queue_dgram_free(entry); queue_entry_free(entry); ms_fail(s, MT_ERR_DISCONNECTED, "connection unavailable", 0); return; @@ -371,7 +372,7 @@ static void ms_work(void* arg) { struct ms_session* s = arg; s->work = NULL; if (s->state != MS_IDLE || !s->dirty || s->tx) return; - if (!ms_can_run(s->ms, s->dataset, s->conn, s->ns)) { + if (!ms_can_run(s->dataset, s->conn, s->ns)) { ms_fail(s, MT_ERR_DISCONNECTED, "model work deferred", 0); return; } int rc; @@ -387,7 +388,7 @@ void merkle_sync_changed(struct UTUN_INSTANCE* inst, uint8_t dataset, const char if (s->dataset->id != dataset || strcmp(s->ns, ns)) continue; s->revision++; s->dirty = 1; if (s->state == MS_FAILED) { s->state = MS_IDLE; s->previous_valid = 0; } - if (s->state == MS_IDLE && ms_schedule(s) < 0) { + if (s->state == MS_IDLE && ms_can_run(s->dataset, s->conn, s->ns) && ms_schedule(s) < 0) { ms_fail(s, MT_ERR_IO, "change scheduling", 1); return; } } @@ -535,7 +536,7 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: unauthorized ns=%s peer=%016llx", ns, (unsigned long long)conn->peer_node_id); queue_dgram_free(entry); queue_entry_free(entry); return; } - if (!ms_can_run(ms, model, conn, ns)) { + if (!ms_can_run(model, conn, ns)) { DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: model receive deferred dataset=%u ns=%s peer=%016llx", dataset, ns, (unsigned long long)conn->peer_node_id); queue_dgram_free(entry); queue_entry_free(entry); return; @@ -725,7 +726,7 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer uint64_t ns_id; struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL; if (!ms || !ms->datasets[dataset].ops || ms_namespace(ns, &ns_id) < 0 || !conn || !conn->initialized || !conn->links_up || conn->state == 2 - || peer == inst->node_id || conn == ms->closing_conn || !ms_can_run(ms, &ms->datasets[dataset], conn, ns)) { + || peer == inst->node_id || conn == ms->closing_conn || !ms_can_run(&ms->datasets[dataset], conn, ns)) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: start unavailable ns=%s peer=%016llx", ns ? ns : "-", (unsigned long long)peer); return -1; } @@ -800,7 +801,7 @@ void merkle_sync_recheck_peer(struct UTUN_INSTANCE* inst, uint64_t peer) { if (!ms) return; struct ms_session* selected = NULL; for (struct ms_session* s = ms->sessions; s; s = s->next) - if (s->peer == peer && s->state != MS_FAILED && !ms_can_run(ms, s->dataset, s->conn, s->ns)) { selected = s; break; } + if (s->peer == peer && s->state != MS_FAILED && !ms_can_run(s->dataset, s->conn, s->ns)) { selected = s; break; } if (!selected) return; ms_fail(selected, MT_ERR_DISCONNECTED, "model policy changed", 0); } diff --git a/tests/test_merkle_protocol.c b/tests/test_merkle_protocol.c index 33675c8f..c90b3225 100644 --- a/tests/test_merkle_protocol.c +++ b/tests/test_merkle_protocol.c @@ -151,7 +151,7 @@ static void test_handshake_failures(void) { t_destroy(&f); } struct fixture f; t_init(&f); struct ms_session* s = t_session(&f); - s->state = MS_BEGIN_WAIT; memset(s->accept_id, 0, sizeof(s->accept_id)); + s->state = MS_BEGIN_WAIT; s->request_id = 0; memset(s->accept_id, 0, sizeof(s->accept_id)); uint8_t accepted[MS_TOKEN_SIZE]; memset(accepted, 0xbc, sizeof(accepted)); t_receive_tokens(&f, MS_READY, 7, 0, s->round_id, accepted, NULL, 0); assert(s->state == MS_PULL && s->waiting && s->tx);