diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index 47b61c081..171051943 100644 --- a/src/chat/merkle_sync.c +++ b/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); diff --git a/src/chat/merkle_sync.h b/src/chat/merkle_sync.h index 622b8a195..f06e89f6e 100644 --- a/src/chat/merkle_sync.h +++ b/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 секунд.