Preserve interrupted Merkle draft before member versioning

src/chat/merkle_sync.c

@@ -21,8 +21,12 @@
 #define MS_QUERY_SIZE (10 + MT_KEY_SIZE)
 #define MS_PAGE_HEADER (1 + MT_KEY_SIZE)
 #define MS_PROGRESS_TIMEOUT_TB (30u * 10000u)
+/* В type два флага отделяют pull от обычного обмена и встречного pull того же пира. */
+#define MS_READ_ONLY 0x40
+#define MS_RESPONSE 0x80
 enum { MS_WAKE = 0x10, MS_BEGIN, MS_READY, MS_QUERY, MS_HASHES, MS_PAGE, MS_TURN, MS_CHECK, MS_DONE, MS_ERROR };
-enum { MS_IDLE, MS_BEGIN_WAIT, MS_PULL, MS_SERVE, MS_DONE_WAIT, MS_FAILED, MS_FAILING };
+enum { MS_IDLE, MS_BEGIN_WAIT, MS_PULL, MS_SERVE, MS_DONE_WAIT, MS_FAILED, MS_FAILING, MS_COMPLETE };
+enum { MS_EXCHANGE, MS_READER, MS_WRITER };
 
 struct ms_request {
     struct ms_request* next;
@@ -64,6 +68,7 @@ struct ms_session {
     struct ms_frame stack[MT_MAX_LEVEL + 1];
     uint8_t previous_roots[2 * MT_HASH_SIZE];
     int previous_valid;
+    uint8_t mode; /* EXCHANGE либо локальный READER/WRITER независимого однонаправленного запроса. */
     uint64_t sent_bytes, received_bytes;
 };
 
@@ -103,9 +108,9 @@ static int ms_namespace(const char* ns, uint64_t* id) {
     return 0;
 }
 
-static struct ms_session* ms_find(struct merkle_sync* ms, struct ETCP_CONN* conn, uint8_t dataset, uint64_t ns_id) {
+static struct ms_session* ms_find(struct merkle_sync* ms, struct ETCP_CONN* conn, uint8_t dataset, uint64_t ns_id, uint8_t mode) {
     for (struct ms_session* s = ms->sessions; s; s = s->next)
-        if (s->conn == conn && s->dataset->id == dataset && s->ns_id == ns_id) return s;
+        if (s->conn == conn && s->dataset->id == dataset && s->ns_id == ns_id && s->mode == mode) return s;
     return NULL;
 }
 
@@ -114,10 +119,12 @@ static void ms_notify(struct ms_session* s, int result) {
     struct ms_request* r = s->requests;
     s->requests = NULL;
     uint64_t peer = s->peer;
+    struct UTUN_INSTANCE* inst = s->ms->inst;
+    uint8_t dataset = s->dataset->id, mode = s->mode;
     char ns[sizeof(s->ns)]; memcpy(ns, s->ns, sizeof(ns));
     void (*on_result)(void*, const char*, uint64_t, int) = s->dataset->ops->on_result;
     void* ctx = s->dataset->data_ctx;
-    if (on_result) on_result(ctx, ns, peer, result);
+    if (on_result && mode != MS_WRITER) on_result(ctx, ns, peer, result);
     while (r) {
         struct ms_request* next = r->next;
         merkle_sync_done_cb cb = r->cb; void* arg = r->arg;
@@ -125,6 +132,15 @@ static void ms_notify(struct ms_session* s, int result) {
         cb(peer, ns, result, arg);
         r = next;
     }
+    /* on_result публикует допуск; callbacks могли отменить запрос или удалить канал. */
+    struct merkle_sync* ms = inst->msync;
+    if (mode == MS_READER && result == MT_OK && ms && ms->datasets[dataset].ops &&
+        ms->datasets[dataset].ops->validate_peer && ms->datasets[dataset].ops->validate_peer(ctx, ns, peer) == 1) {
+        DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: pull restored exchange dataset=%u ns=%s peer=%016llx", dataset, ns,
+            (unsigned long long)peer);
+        if (merkle_sync_start(inst, dataset, peer, ns, NULL, NULL) < 0)
+            DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: restored exchange unavailable dataset=%u ns=%s", dataset, ns);
+    }
 }
 
 static int ms_schedule(struct ms_session* s);
@@ -179,7 +195,7 @@ static void ms_free(struct ms_session* s) {
     u_free(s);
 }
 
-static struct ms_session* ms_create(struct merkle_sync* ms, struct ETCP_CONN* conn, uint8_t dataset, uint64_t ns_id) {
+static struct ms_session* ms_create(struct merkle_sync* ms, struct ETCP_CONN* conn, uint8_t dataset, uint64_t ns_id, uint8_t mode) {
     if (!conn || !conn->send_input_q || !conn->initialized || !conn->links_up || conn->state == 2) return NULL;
     struct ms_session* s = u_calloc(1, sizeof(*s));
     if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: session allocation failed"); return NULL; }
@@ -188,10 +204,11 @@ static struct ms_session* ms_create(struct merkle_sync* ms, struct ETCP_CONN* co
     s->peer = conn->peer_node_id; s->ns_id = ns_id;
     snprintf(s->ns, sizeof(s->ns), "%llu", (unsigned long long)ns_id);
     snprintf(s->tree_ns, sizeof(s->tree_ns), "%u:%s", dataset, s->ns);
-    s->leader = ms->inst->node_id < s->peer;
+    s->mode = mode;
+    s->leader = mode == MS_READER || (mode == MS_EXCHANGE && ms->inst->node_id < s->peer);
     s->next = ms->sessions; ms->sessions = s;
-    DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: attach dataset=%u ns=%s peer=%016llx conn=%p leader=%d",
-        s->dataset->id, s->ns, (unsigned long long)s->peer, (void*)conn, s->leader);
+    DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: attach dataset=%u ns=%s peer=%016llx conn=%p leader=%d mode=%u",
+        s->dataset->id, s->ns, (unsigned long long)s->peer, (void*)conn, s->leader, mode);
     return s;
 }
 
@@ -207,8 +224,14 @@ 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 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;
+static int ms_allowed(struct ms_dataset* model, struct ETCP_CONN* conn, const char* ns, uint8_t mode) {
+    if (mode == MS_READER)
+        return model->ops->validate_request && model->ops->validate_request(model->data_ctx, ns, conn->peer_node_id) == 1;
+    return !model->ops->validate_peer || model->ops->validate_peer(model->data_ctx, ns, conn->peer_node_id) == 1;
+}
+
+static int ms_can_run(struct ms_dataset* model, struct ETCP_CONN* conn, const char* ns, uint8_t mode) {
+    if (!ms_allowed(model, conn, ns, mode)) return 0;
     return model->ops->can_run ? model->ops->can_run(model->data_ctx, ns, conn->peer_node_id) == 1 : !conn->peer_sleep_phase;
 }
 
@@ -217,11 +240,11 @@ static void ms_sent(struct ll_queue* q, void* arg) {
     struct ll_entry* entry = s->tx;
     s->tx = NULL;
     if (!entry) return;
-    int type = entry->dgram[9];
+    int type = entry->dgram[9] & ~(MS_READ_ONLY | MS_RESPONSE);
     size_t len = entry->len;
     const char* unavailable = !s->conn->links_up || s->conn->state == 2 || s->conn->close_requested || q != s->conn->send_input_q
         ? "connection unavailable" : NULL;
-    if (!unavailable && !ms_can_run(s->dataset, s->conn, s->ns)) unavailable = "model send deferred";
+    if (!unavailable && !ms_can_run(s->dataset, s->conn, s->ns, s->mode)) unavailable = "model send deferred";
     if (unavailable) {
         queue_dgram_free(entry); queue_entry_free(entry);
         ms_fail(s, MT_ERR_DISCONNECTED, unavailable, 0);
@@ -243,8 +266,9 @@ static void ms_sent(struct ll_queue* q, void* arg) {
         return;
     }
     if (type == MS_DONE) {
-        s->state = MS_IDLE;
+        s->state = s->mode == MS_WRITER ? MS_COMPLETE : MS_IDLE;
         ms_stop_deadline(s);
+        if (s->mode == MS_WRITER) return; /* Выдача данных не подтверждает синхронизацию собственного состава. */
         if (s->revision == s->verified_revision) {
             s->dirty = 0;
             DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: converged dataset=%u ns=%s peer=%016llx round=%llu tx=%llu rx=%llu",
@@ -267,7 +291,7 @@ static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8
     e->len = (uint16_t)(MS_HEADER + len);
     e->dgram[0] = s->ms->svc_id;
     ms_write64(e->dgram + 1, s->ns_id);
-    e->dgram[9] = (uint8_t)type;
+    e->dgram[9] = (uint8_t)type | (s->mode ? MS_READ_ONLY : 0) | (s->mode == MS_WRITER ? MS_RESPONSE : 0);
     ms_write64(e->dgram + 10, type == MS_WAKE ? 0 : s->round);
     ms_write32(e->dgram + 18, request);
     e->dgram[22] = s->dataset->id;
@@ -344,6 +368,10 @@ static int ms_walk_next(struct ms_session* s) {
         s->depth--;
     }
     s->waiting = 0;
+    if (s->mode == MS_READER) {
+        s->state = MS_DONE_WAIT;
+        return ms_send(s, MS_CHECK, 0, NULL, 0);
+    }
     if (s->pull) {
         int rc = s->dataset->ops->commit_pull(s->dataset->data_ctx, s->pull);
         ms_abort_pull(s);
@@ -376,7 +404,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->dataset, s->conn, s->ns)) {
+    if (!ms_can_run(s->dataset, s->conn, s->ns, s->mode)) {
         ms_fail(s, MT_ERR_DISCONNECTED, "model work deferred", 0); return;
     }
     int rc;
@@ -390,9 +418,11 @@ void merkle_sync_changed(struct UTUN_INSTANCE* inst, uint8_t dataset, const char
     if (!ms || !ns) return;
     for (struct ms_session* s = ms->sessions; s; s = s->next) {
         if (s->dataset->id != dataset || strcmp(s->ns, ns)) continue;
-        s->revision++; s->dirty = 1;
+        s->revision++;
+        if (s->mode != MS_EXCHANGE) continue; /* Pull — один запрос; изменение источника проверяется перед DONE. */
+        s->dirty = 1;
         if (s->state == MS_FAILED) { s->state = MS_IDLE; s->previous_valid = 0; }
-        if (s->state == MS_IDLE && ms_can_run(s->dataset, s->conn, s->ns) && ms_schedule(s) < 0) {
+        if (s->state == MS_IDLE && ms_can_run(s->dataset, s->conn, s->ns, s->mode) && ms_schedule(s) < 0) {
             ms_fail(s, MT_ERR_IO, "change scheduling", 1); return;
         }
     }
@@ -404,18 +434,19 @@ void merkle_sync_revalidate_peers(struct UTUN_INSTANCE* inst, uint8_t dataset, c
         DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid permission namespace"); return;
     }
     char owned_ns[21]; strcpy(owned_ns, ns);
-    uint64_t after = 0; int after_valid = 0;
+    uint64_t after = 0; uint8_t after_mode = 0; int after_valid = 0;
     for (;;) {
         struct merkle_sync* ms = inst ? inst->msync : NULL;
         if (!ms || !ms->datasets[dataset].ops || !ms->datasets[dataset].ops->validate_peer) return;
         struct ms_session* selected = NULL;
         for (struct ms_session* s = ms->sessions; s; s = s->next)
-            if (s->dataset->id == dataset && !strcmp(s->ns, owned_ns) && (!after_valid || s->peer > after)
-                && (!selected || s->peer < selected->peer)) selected = s;
+            if (s->dataset->id == dataset && !strcmp(s->ns, owned_ns) &&
+                (!after_valid || s->peer > after || (s->peer == after && s->mode > after_mode)) &&
+                (!selected || s->peer < selected->peer || (s->peer == selected->peer && s->mode < selected->mode))) selected = s;
         if (!selected) return;
-        after = selected->peer; after_valid = 1;
-        if (selected->state != MS_FAILED
-            && selected->dataset->ops->validate_peer(selected->dataset->data_ctx, owned_ns, after) != 1)
+        after = selected->peer; after_mode = selected->mode; after_valid = 1;
+        if (selected->state != MS_FAILED && selected->state != MS_COMPLETE &&
+            !ms_allowed(selected->dataset, selected->conn, owned_ns, selected->mode))
             ms_fail(selected, MT_ERR_DATA, "model permission revoked", 0);
         /* Callback мог удалить selected и соседние сессии: следующий поиск начинается заново. */
     }
@@ -458,7 +489,8 @@ int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, uint8_t dataset, cons
 }
 
 static int ms_serve_query(struct ms_session* s, uint32_t request, const uint8_t* p, size_t len) {
-    if (s->state != MS_SERVE || len != MS_QUERY_SIZE || request != s->remote_request_id + 1) return MT_ERR_PROTOCOL;
+    if (s->mode == MS_READER || s->state != MS_SERVE || len != MS_QUERY_SIZE || request != s->remote_request_id + 1)
+        return MT_ERR_PROTOCOL;
     uint8_t level = p[0];
     uint64_t prefix = ms_read64(p + 1);
     merkle_key_t after; memcpy(after.bytes, p + 10, MT_KEY_SIZE);

src/chat/merkle_sync.h

@@ -1,7 +1,7 @@
 /* merkle_sync — обмен различиями между наборами записей по Merkle-дереву.
  * Движок сверяет хеши и передаёт страницы изменившихся листьев; модель данных
  * через data_ops читает записи, проверяет подписи/права и сохраняет изменения.
- * Сессия относится к (пир, dataset, namespace), отправка учитывает backpressure.
+ * Сессия относится к (пир, dataset, namespace, направление), отправка учитывает backpressure.
  * Wire v4 привязывает запросы к свежим round_id и accept_id обеих сторон;
  * отменённая сессия не может подтвердить новый запрос прежними пакетами.
  * Все операции — в потоке uasync; один движок на UTUN_INSTANCE. */
@@ -47,6 +47,11 @@ struct merkle_sync_data_ops {
     int (*apply_page)(void* ctx, const char* ns, uint64_t peer, uint64_t prefix, int after_valid,
             const merkle_key_t* after, const merkle_key_t* next, int more, const uint8_t* data, size_t len);
     int (*validate_peer)(void* ctx, const char* ns, uint64_t peer); // допуск пира: только 1 разрешает обмен
+    /* Необязательное основание запросить данные, когда двусторонний обмен ещё запрещён.
+     * Только 1 разрешает локальный pull; удалённая сторона всё равно проверяет validate_peer.
+     * Такой получатель не отдаёт ни записей, ни хешей своего дерева. Ответы принимаются
+     * только в созданную локально сессию. Модели без callback этого режима не поддерживают. */
+    int (*validate_request)(void* ctx, const char* ns, uint64_t peer);
     /* Политика работы (например sleep/sound), 1 разрешает обмен; без callback спящий пир отложен. */
     int (*can_run)(void* ctx, const char* ns, uint64_t peer);
     /* Результат каждой контрольной точки/ошибки, включая автоматические раунды.
@@ -85,10 +90,14 @@ int merkle_sync_registered(struct UTUN_INSTANCE* inst, uint8_t dataset);
 void merkle_sync_destroy(struct UTUN_INSTANCE* inst);
 
 /* Подключить namespace к текущему ETCP-соединению и запросить контрольную точку.
- * Успех — подтверждение одинакового корня конкретного раунда, не вечный статус.
+ * При validate_peer=1 успех подтверждает одинаковый корень конкретного раунда.
+ * Иначе validate_request=1 разрешает только получение различий из неизменившегося
+ * за проход источника. Его локальные лишние записи источнику не передаются.
+ * После успешного pull и on_result движок проверяет validate_peer заново: если допуск
+ * восстановлен, запускает обычную синхронизацию без нового пользовательского callback.
  * Повторные запросы объединяются; каждый callback вызывается один раз, обход не сбрасывается.
  * После первого start изменения дерева автоматически синхронизируются с этим пиром.
- * cb может быть NULL. Нужен UP-пир, не self, с допуском validate_peer и can_run.
+ * cb может быть NULL. Нужен UP-пир, не self, с основанием запроса и can_run.
  * Без can_run обмен со спящим пиром откладывается.
  * <0: запрос не принят, callback не вызывается. DOWN/DELETE/REINIT завершают запросы ошибкой.
  * При блокировке исходящего пакета или ожидании ответа действует таймаут 30 секунд.