Resume admitted history peers and gate automatic Merkle work by policy

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, &current);
             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) {

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);
     }

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);