jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Bind Merkle rounds to fresh initiator and responder tokens

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 18:27:26 +01:00
1 parent 137e30ec82
commit 45c282e0b7
2 files changed
+45 -11
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/merkle_sync.c
+43 -11
View File
@@ -11,10 +11,13 @@
#include <stdlib.h>
#include <errno.h>
/* Wire v3: svc:1, namespace:8, type:1, round:8, request:4, dataset:1, payload.
/* Wire v4: svc:1, namespace:8, type:1, round:8, request:4, dataset:1,
* round_id:16, accept_id:16, payload. round — диагностический счётчик;
* пара случайных идентификаторов связывает запрос с текущим handshake обеих сторон.
* Все целые заголовка big-endian. Один запрос на проход и один исходящий пакет на сессию.
* Меньший node_id ведёт раунд: pull -> TURN -> reverse pull -> CHECK -> DONE. */
#define MS_HEADER 23
#define MS_HEADER 55
#define MS_TOKEN_SIZE 16
#define MS_QUERY_SIZE (10 + MT_KEY_SIZE)
#define MS_PAGE_HEADER (1 + MT_KEY_SIZE)
#define MS_PROGRESS_TIMEOUT_TB (30u * 10000u)
@@ -53,6 +56,7 @@ struct ms_session {
int failure_error;
struct ms_request* requests;
uint64_t ns_id, peer, round, revision, verified_revision;
uint8_t round_id[MS_TOKEN_SIZE], accept_id[MS_TOKEN_SIZE];
char ns[21], tree_ns[25];
int state, leader, dirty, depth, after_valid, waiting;
uint32_t request_id, remote_request_id;
@@ -77,6 +81,16 @@ static void ms_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8
static uint32_t ms_read32(const uint8_t* p) { uint32_t n; memcpy(&n, p, 4); return ntohl(n); }
static void ms_write32(uint8_t* p, uint32_t n) { n = htonl(n); memcpy(p, &n, 4); }
static int ms_token_empty(const uint8_t token[MS_TOKEN_SIZE]) {
static const uint8_t zero[MS_TOKEN_SIZE];
return !memcmp(token, zero, sizeof(zero));
}
static int ms_new_token(uint8_t token[MS_TOKEN_SIZE]) {
if (random_bytes(token, MS_TOKEN_SIZE) == 0 && !ms_token_empty(token)) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: handshake RNG failed"); return MT_ERR_IO;
}
static int ms_namespace(const char* ns, uint64_t* id) {
if (!ns || !ns[0] || strlen(ns) > 20) return -1;
for (const char* p = ns; *p; p++) if (*p < '0' || *p > '9') return -1;
@@ -245,6 +259,9 @@ static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8
ms_write64(e->dgram + 10, type == MS_WAKE ? 0 : s->round);
ms_write32(e->dgram + 18, request);
e->dgram[22] = s->dataset->id;
memset(e->dgram + 23, 0, 2 * MS_TOKEN_SIZE);
if (type != MS_WAKE) memcpy(e->dgram + 23, s->round_id, MS_TOKEN_SIZE);
if (type != MS_WAKE && type != MS_BEGIN) memcpy(e->dgram + 39, s->accept_id, MS_TOKEN_SIZE);
if (len) memcpy(e->dgram + MS_HEADER, payload, len);
s->tx = e;
/* Transport normalizer configures deferred waiters. Не вызываем callback рекурсивно. */
@@ -328,9 +345,13 @@ static int ms_walk_next(struct ms_session* s) {
}
static int ms_begin(struct ms_session* s) {
uint8_t token[MS_TOKEN_SIZE];
if (ms_new_token(token) < 0) return MT_ERR_IO;
ms_abort_pull(s);
s->round = ++s->ms->round_seq;
if (!s->round) return -1;
memcpy(s->round_id, token, sizeof(token));
memset(s->accept_id, 0, sizeof(s->accept_id));
s->request_id = s->remote_request_id = 0;
s->state = MS_BEGIN_WAIT; s->dirty = 0; s->waiting = 0;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: begin dataset=%u ns=%s peer=%016llx round=%llu",
@@ -512,13 +533,15 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
dataset, ns, (unsigned long long)s->peer, (unsigned long long)round, type, request, len, s->state);
int rc = 0, notify = 0;
if (s->state == MS_FAILING) goto done;
if (type != MS_WAKE && type != MS_BEGIN && round != s->round) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard old round=%llu current=%llu", (unsigned long long)round,
(unsigned long long)s->round);
if (type != MS_WAKE && type != MS_BEGIN &&
(round != s->round || memcmp(d + 23, s->round_id, MS_TOKEN_SIZE) || ms_token_empty(s->round_id) ||
(type != MS_READY && (memcmp(d + 39, s->accept_id, MS_TOKEN_SIZE) || ms_token_empty(s->accept_id))))) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard stale handshake ns=%s round=%llu current=%llu type=%u",
ns, (unsigned long long)round, (unsigned long long)s->round, type);
goto done;
}
if (type == MS_WAKE) {
if (!s->leader || round || request || len) rc = MT_ERR_PROTOCOL;
if (!s->leader || round || request || len || !ms_token_empty(d + 23) || !ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL;
else {
/* Follower мог отменить и заново подключить namespace. Старый запрос уже некому обслужить. */
ms_discard_tx(s);
@@ -526,19 +549,28 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
rc = ms_begin(s);
}
} else if (type == MS_BEGIN) {
if (s->leader || !round || request || len) rc = MT_ERR_PROTOCOL;
else if (round <= s->round) {
if (s->leader || !round || request || len || ms_token_empty(d + 23) || !ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL;
else if (!memcmp(d + 23, s->round_id, MS_TOKEN_SIZE)) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard duplicate BEGIN ns=%s", ns);
} else {
uint8_t token[MS_TOKEN_SIZE];
if (ms_new_token(token) < 0) { rc = MT_ERR_IO; goto done; }
ms_discard_tx(s);
ms_abort_pull(s);
memcpy(s->round_id, d + 23, MS_TOKEN_SIZE);
memcpy(s->accept_id, token, sizeof(token));
s->round = round; s->state = MS_SERVE; s->waiting = 0;
s->request_id = s->remote_request_id = 0; s->dirty = 0;
rc = ms_send(s, MS_READY, 0, NULL, 0) < 0 ? MT_ERR_IO : 0;
}
} else if (type == MS_READY) {
if (!s->leader || s->state != MS_BEGIN_WAIT || request || len) rc = MT_ERR_PROTOCOL;
else rc = ms_pull_start(s);
if (!s->leader || request || len || ms_token_empty(d + 39)) rc = MT_ERR_PROTOCOL;
else if (s->state == MS_BEGIN_WAIT) {
memcpy(s->accept_id, d + 39, MS_TOKEN_SIZE);
rc = ms_pull_start(s);
} else if (!memcmp(d + 39, s->accept_id, MS_TOKEN_SIZE)) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard duplicate READY ns=%s", ns);
} else rc = MT_ERR_PROTOCOL;
} else if (type == MS_QUERY) rc = ms_serve_query(s, request, p, len);
else if (type == MS_HASHES) rc = ms_receive_hashes(s, request, p, len);
else if (type == MS_PAGE) rc = ms_receive_page(s, request, p, len);
@@ -629,7 +661,7 @@ int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id, const struct me
if (merkle_sync_register(inst, MT_DATA_MEMBERS, ops, data_ctx) < 0) { inst->msync = NULL; u_free(ms); return -1; }
etcp_bind(inst, svc_id, ms_receive);
etcp_add_conn_status_cbk(inst, ms_conn_status, ms);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: initialized service=%u protocol=3", svc_id);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: initialized service=%u protocol=4", svc_id);
return 0;
}
src/chat/merkle_sync.h
+2
View File
@@ -2,6 +2,8 @@
* Движок сверяет хеши и передаёт страницы изменившихся листьев; модель данных
* через data_ops читает записи, проверяет подписи/права и сохраняет изменения.
* Сессия относится к (пир, dataset, namespace), отправка учитывает backpressure.
* Wire v4 привязывает запросы к свежим round_id и accept_id обеих сторон;
* отменённая сессия не может подтвердить новый запрос прежними пакетами.
* Все операции — в потоке uasync; один движок на UTUN_INSTANCE. */
#ifndef MERKLE_SYNC_H
#define MERKLE_SYNC_H