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

Extend Merkle sessions with datasets and full record cursors

master
evgeny 5 days ago
parent
27fd9ee62d
commit
539bfd6db9
2 changed files with 135 additions and 72 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 174
      src/chat/merkle_sync.c
  2. 33
      src/chat/merkle_sync.h

174
src/chat/merkle_sync.c
Unescape View File

@ -11,11 +11,12 @@
#include <stdlib.h>
#include <errno.h>
/* Wire v2: svc:1, namespace:8, type:1, round:8, request:4, payload.
/* Wire v3: svc:1, namespace:8, type:1, round:8, request:4, dataset:1, payload.
* Все целые заголовка big-endian. Один запрос на проход и один исходящий пакет на сессию.
* Меньший node_id ведёт раунд: pull -> TURN -> reverse pull -> CHECK -> DONE. */
#define MS_HEADER 22
#define MS_QUERY_SIZE 18
#define MS_HEADER 23
#define MS_QUERY_SIZE (10 + MT_KEY_SIZE)
#define MS_PAGE_HEADER (1 + MT_KEY_SIZE)
#define MS_PROGRESS_TIMEOUT_TB (30u * 10000u)
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 };
@ -32,9 +33,16 @@ struct ms_frame {
uint8_t level;
};
struct ms_dataset {
const struct merkle_sync_data_ops* ops;
void* data_ctx;
uint8_t id;
};
struct ms_session {
struct ms_session* next;
struct merkle_sync* ms;
struct ms_dataset* dataset;
struct ETCP_CONN* conn;
struct ll_queue* send_q;
struct queue_waiter_handle waiter;
@ -45,10 +53,10 @@ struct ms_session {
int failure_error;
struct ms_request* requests;
uint64_t ns_id, peer, round, revision, verified_revision;
char ns[21];
char ns[21], tree_ns[25];
int state, leader, dirty, depth, after_valid, waiting;
uint32_t request_id, remote_request_id;
uint64_t after;
merkle_key_t after;
struct ms_frame stack[MT_MAX_LEVEL + 1];
uint8_t previous_roots[2 * MT_HASH_SIZE];
int previous_valid;
@ -58,8 +66,7 @@ struct ms_session {
struct merkle_sync {
struct UTUN_INSTANCE* inst;
uint8_t svc_id;
const struct merkle_sync_data_ops* ops;
void* data_ctx;
struct ms_dataset datasets[256];
struct ms_session* sessions;
uint64_t round_seq;
struct ETCP_CONN* closing_conn;
@ -82,9 +89,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, 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) {
for (struct ms_session* s = ms->sessions; s; s = s->next)
if (s->conn == conn && s->ns_id == ns_id) return s;
if (s->conn == conn && s->dataset->id == dataset && s->ns_id == ns_id) return s;
return NULL;
}
@ -110,7 +117,7 @@ static void ms_fail(struct ms_session* s, int error, const char* reason, int rep
static void ms_abort_pull(struct ms_session* s) {
if (!s->pull) return;
void* pull = s->pull; s->pull = NULL;
s->ms->ops->abort_pull(s->ms->data_ctx, pull);
s->dataset->ops->abort_pull(s->dataset->data_ctx, pull);
}
static void ms_stop_deadline(struct ms_session* s) {
@ -155,14 +162,15 @@ 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, 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) {
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; }
if (etcp_conn_ref_take(conn) != 0) { u_free(s); return NULL; }
s->ms = ms; s->conn = conn; s->send_q = conn->send_input_q;
s->ms = ms; s->dataset = &ms->datasets[dataset]; s->conn = conn; s->send_q = conn->send_input_q;
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->next = ms->sessions; ms->sessions = s;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: attach ns=%s peer=%016llx conn=%p leader=%d",
@ -170,12 +178,16 @@ static struct ms_session* ms_create(struct merkle_sync* ms, struct ETCP_CONN* co
return s;
}
int merkle_sync_read_hash(struct UTUN_INSTANCE* inst, const char* ns, uint8_t level, uint64_t prefix, uint8_t hash[MT_HASH_SIZE]) {
return merkle_tree_get(inst ? inst->topo_sqlite_db : NULL, ns, level, prefix, hash);
int merkle_sync_read_hash(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns,
uint8_t level, uint64_t prefix, uint8_t hash[MT_HASH_SIZE]) {
char tree_ns[25];
if (!ns || strlen(ns) > 20) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid hash namespace"); return -1; }
snprintf(tree_ns, sizeof(tree_ns), "%u:%s", dataset, ns);
return merkle_tree_get(inst ? inst->topo_sqlite_db : NULL, tree_ns, level, prefix, hash);
}
static int ms_root(struct ms_session* s, uint8_t root[MT_HASH_SIZE]) {
return merkle_sync_read_hash(s->ms->inst, s->ns, 0, 0, root);
return merkle_sync_read_hash(s->ms->inst, s->dataset->id, s->ns, 0, 0, root);
}
static void ms_sent(struct ll_queue* q, void* arg) {
@ -232,6 +244,7 @@ static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8
e->dgram[9] = (uint8_t)type;
ms_write64(e->dgram + 10, type == MS_WAKE ? 0 : s->round);
ms_write32(e->dgram + 18, request);
e->dgram[22] = s->dataset->id;
if (len) memcpy(e->dgram + MS_HEADER, payload, len);
s->tx = e;
/* Transport normalizer configures deferred waiters. Не вызываем callback рекурсивно. */
@ -271,7 +284,7 @@ static int ms_query(struct ms_session* s) {
struct ms_frame* f = &s->stack[s->depth];
uint8_t body[MS_QUERY_SIZE];
body[0] = f->level; ms_write64(body + 1, f->prefix);
body[9] = (uint8_t)s->after_valid; ms_write64(body + 10, s->after);
body[9] = (uint8_t)s->after_valid; memcpy(body + 10, s->after.bytes, MT_KEY_SIZE);
if (++s->request_id == 0) return -1;
s->waiting = 1;
return ms_send(s, MS_QUERY, s->request_id, body, sizeof(body));
@ -279,9 +292,9 @@ static int ms_query(struct ms_session* s) {
static int ms_pull_start(struct ms_session* s) {
ms_abort_pull(s);
if (s->ms->ops->begin_pull && s->ms->ops->begin_pull(s->ms->data_ctx, s->ns, s->peer, s->round, &s->pull) < 0)
if (s->dataset->ops->begin_pull && s->dataset->ops->begin_pull(s->dataset->data_ctx, s->ns, s->peer, s->round, &s->pull) < 0)
return MT_ERR_DATA;
s->state = MS_PULL; s->depth = 0; s->after_valid = 0; s->after = 0;
s->state = MS_PULL; s->depth = 0; s->after_valid = 0; memset(&s->after, 0, sizeof(s->after));
memset(s->stack, 0, sizeof(s->stack));
return ms_query(s);
}
@ -295,19 +308,19 @@ static int ms_walk_next(struct ms_session* s) {
uint8_t level = f->level + 1;
uint64_t prefix = f->prefix | ((uint64_t)i << (64 - level * 5));
s->stack[++s->depth] = (struct ms_frame){ .prefix = prefix, .level = level };
s->after_valid = 0; s->after = 0;
s->after_valid = 0; memset(&s->after, 0, sizeof(s->after));
return ms_query(s);
}
s->depth--;
}
s->waiting = 0;
if (s->pull) {
int rc = s->ms->ops->commit_pull(s->ms->data_ctx, s->pull);
int rc = s->dataset->ops->commit_pull(s->dataset->data_ctx, s->pull);
ms_abort_pull(s);
if (rc < 0) return rc;
}
if (s->leader) { s->state = MS_SERVE; return ms_send(s, MS_TURN, 0, NULL, 0); }
if (s->ms->ops->finish && s->ms->ops->finish(s->ms->data_ctx, s->ns) < 0) return MT_ERR_DATA;
if (s->dataset->ops->finish && s->dataset->ops->finish(s->dataset->data_ctx, s->ns) < 0) return MT_ERR_DATA;
uint8_t root[MT_HASH_SIZE];
if (ms_root(s, root) < 0) return -1;
s->state = MS_DONE_WAIT;
@ -335,11 +348,11 @@ static void ms_work(void* arg) {
if (rc < 0) ms_fail(s, MT_ERR_IO, "start send", 1);
}
void merkle_sync_changed(struct UTUN_INSTANCE* inst, const char* ns) {
void merkle_sync_changed(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ns) return;
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (strcmp(s->ns, ns)) continue;
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) {
@ -348,15 +361,27 @@ void merkle_sync_changed(struct UTUN_INSTANCE* inst, const char* ns) {
}
}
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key) {
/* Дерево индексируется по dataset:channel, модель получает исходный channel. */
struct ms_leaf_context { struct ms_dataset* dataset; const char* ns; };
static int ms_leaf_hash(void* ctx, const char* tree_ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
struct ms_leaf_context* leaf = ctx;
(void)tree_ns;
return leaf->dataset->ops->update_bucket_hash(leaf->dataset->data_ctx, leaf->ns, level, prefix, hash);
}
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns, uint64_t key) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
sqlite3* db = inst ? inst->topo_sqlite_db : NULL;
if (!ms || !db || !ns) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid recompute"); return -1; }
if (!ms || !db || !ns || strlen(ns) > 20 || !ms->datasets[dataset].ops) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid recompute dataset=%u", dataset); return -1;
}
char tree_ns[25]; snprintf(tree_ns, sizeof(tree_ns), "%u:%s", dataset, ns);
struct ms_leaf_context leaf = { &ms->datasets[dataset], ns };
int own = sqlite3_get_autocommit(db);
if (own && sqlite3_exec(db, "SAVEPOINT merkle_update", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: begin update ns=%s: %s", ns, sqlite3_errmsg(db)); return -1;
}
int rc = merkle_tree_update(db, ns, key, ms->ops->update_bucket_hash, ms->data_ctx);
int rc = merkle_tree_update(db, tree_ns, key, ms_leaf_hash, &leaf);
if (own) {
if (rc < 0 && sqlite3_exec(db, "ROLLBACK TO merkle_update", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: rollback ns=%s: %s", ns, sqlite3_errmsg(db));
@ -366,7 +391,7 @@ int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint6
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: commit cleanup ns=%s: %s", ns, sqlite3_errmsg(db));
rc = -1;
}
if (rc > 0) merkle_sync_changed(inst, ns);
if (rc > 0) merkle_sync_changed(inst, dataset, ns);
}
if (rc < 0) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: recompute failed ns=%s key=%016llx", ns, (unsigned long long)key);
return rc;
@ -375,25 +400,30 @@ int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint6
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;
uint8_t level = p[0];
uint64_t prefix = ms_read64(p + 1), after = ms_read64(p + 10);
uint64_t prefix = ms_read64(p + 1);
merkle_key_t after; memcpy(after.bytes, p + 10, MT_KEY_SIZE);
static const merkle_key_t empty;
if (level > MT_MAX_LEVEL || prefix != merkle_sync_level_prefix(prefix, level) || p[9] > 1) return MT_ERR_PROTOCOL;
if (level < MT_MAX_LEVEL && (p[9] || after)) return MT_ERR_PROTOCOL;
if (p[9] && merkle_sync_level_prefix(after, MT_MAX_LEVEL) != prefix) return MT_ERR_PROTOCOL;
if ((!p[9] || level < MT_MAX_LEVEL) && memcmp(&after, &empty, sizeof(after))) return MT_ERR_PROTOCOL;
if (level < MT_MAX_LEVEL && p[9]) return MT_ERR_PROTOCOL;
if (p[9] && merkle_sync_level_prefix(ms_read64(after.bytes), MT_MAX_LEVEL) != prefix) return MT_ERR_PROTOCOL;
s->remote_request_id = request;
if (level < MT_MAX_LEVEL) {
uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE];
if (merkle_tree_children(s->ms->inst->topo_sqlite_db, s->ns, level, prefix, hashes) < 0) return MT_ERR_IO;
if (merkle_tree_children(s->ms->inst->topo_sqlite_db, s->tree_ns, level, prefix, hashes) < 0) return MT_ERR_IO;
return ms_send(s, MS_HASHES, request, &hashes[0][0], sizeof(hashes)) < 0 ? MT_ERR_IO : 0;
}
uint8_t body[MT_PAGE_SIZE - MS_HEADER];
size_t size = sizeof(body) - 9;
uint64_t next = 0; int more = 0;
int rc = s->ms->ops->get_page(s->ms->data_ctx, s->ns, prefix, p[9], after, body + 9, &size, &next, &more);
size_t size = sizeof(body) - MS_PAGE_HEADER;
merkle_key_t next = {{0}}; int more = 0;
int rc = s->dataset->ops->get_page(s->dataset->data_ctx, s->ns, prefix, p[9], &after,
body + MS_PAGE_HEADER, &size, &next, &more);
if (rc < 0) return MT_ERR_DATA;
if (size > sizeof(body) - 9 || (more != 0 && more != 1)
|| (more && (merkle_sync_level_prefix(next, MT_MAX_LEVEL) != prefix || (p[9] && next <= after)))) return MT_ERR_DATA;
body[0] = (uint8_t)more; ms_write64(body + 1, next);
return ms_send(s, MS_PAGE, request, body, size + 9) < 0 ? MT_ERR_IO : 0;
if (size > sizeof(body) - MS_PAGE_HEADER || (more != 0 && more != 1)
|| (more && (!size || merkle_sync_level_prefix(ms_read64(next.bytes), MT_MAX_LEVEL) != prefix
|| (p[9] && memcmp(next.bytes, after.bytes, MT_KEY_SIZE) <= 0)))) return MT_ERR_DATA;
body[0] = (uint8_t)more; memcpy(body + 1, next.bytes, MT_KEY_SIZE);
return ms_send(s, MS_PAGE, request, body, size + MS_PAGE_HEADER) < 0 ? MT_ERR_IO : 0;
}
static int ms_receive_hashes(struct ms_session* s, uint32_t request, const uint8_t* p, size_t len) {
@ -401,7 +431,7 @@ static int ms_receive_hashes(struct ms_session* s, uint32_t request, const uint8
struct ms_frame* f = &s->stack[s->depth];
if (f->level >= MT_MAX_LEVEL) return MT_ERR_PROTOCOL;
uint8_t local[MT_BUCKETS][MT_HASH_SIZE];
if (merkle_tree_children(s->ms->inst->topo_sqlite_db, s->ns, f->level, f->prefix, local) < 0) return MT_ERR_IO;
if (merkle_tree_children(s->ms->inst->topo_sqlite_db, s->tree_ns, f->level, f->prefix, local) < 0) return MT_ERR_IO;
f->children = 0;
for (unsigned i = 0; i < MT_BUCKETS; i++)
if (memcmp(local[i], p + i * MT_HASH_SIZE, MT_HASH_SIZE)) f->children |= UINT32_C(1) << i;
@ -410,13 +440,15 @@ static int ms_receive_hashes(struct ms_session* s, uint32_t request, const uint8
}
static int ms_receive_page(struct ms_session* s, uint32_t request, const uint8_t* p, size_t len) {
if (s->state != MS_PULL || !s->waiting || request != s->request_id || len < 9 || p[0] > 1) return MT_ERR_PROTOCOL;
if (s->state != MS_PULL || !s->waiting || request != s->request_id || len < MS_PAGE_HEADER || p[0] > 1) return MT_ERR_PROTOCOL;
struct ms_frame* f = &s->stack[s->depth];
uint64_t next = ms_read64(p + 1);
if (f->level != MT_MAX_LEVEL || (p[0] && (len == 9 || merkle_sync_level_prefix(next, MT_MAX_LEVEL) != f->prefix
|| (s->after_valid && next <= s->after)))) return MT_ERR_PROTOCOL;
int rc = s->pull ? s->ms->ops->stage_page(s->ms->data_ctx, s->pull, f->prefix, s->after_valid, s->after,
next, p[0], p + 9, len - 9) : s->ms->ops->apply_items(s->ms->data_ctx, s->ns, s->peer, p + 9, len - 9);
merkle_key_t next; memcpy(next.bytes, p + 1, MT_KEY_SIZE);
if (f->level != MT_MAX_LEVEL || (p[0] && (len == MS_PAGE_HEADER
|| merkle_sync_level_prefix(ms_read64(next.bytes), MT_MAX_LEVEL) != f->prefix
|| (s->after_valid && memcmp(next.bytes, s->after.bytes, MT_KEY_SIZE) <= 0)))) return MT_ERR_PROTOCOL;
int rc = s->pull ? s->dataset->ops->stage_page(s->dataset->data_ctx, s->pull, f->prefix, s->after_valid, &s->after,
&next, p[0], p + MS_PAGE_HEADER, len - MS_PAGE_HEADER)
: s->dataset->ops->apply_items(s->dataset->data_ctx, s->ns, s->peer, p + MS_PAGE_HEADER, len - MS_PAGE_HEADER);
if (rc < 0) return rc == MT_ERR_CONFLICT ? rc : MT_ERR_DATA;
s->waiting = 0;
if (p[0]) { s->after = next; s->after_valid = 1; return ms_query(s) < 0 ? MT_ERR_IO : 0; }
@ -433,14 +465,19 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
const uint8_t* d = entry->dgram;
uint64_t ns_id = ms_read64(d + 1), round = ms_read64(d + 10);
uint32_t request = ms_read32(d + 18);
uint8_t type = d[9];
uint8_t type = d[9], dataset = d[22];
struct ms_dataset* model = &ms->datasets[dataset];
if (!model->ops) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: unknown dataset=%u", dataset);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
char ns[21]; snprintf(ns, sizeof(ns), "%llu", (unsigned long long)ns_id);
if (ms->ops->validate_peer && ms->ops->validate_peer(ms->data_ctx, ns, conn->peer_node_id) != 1) {
if (model->ops->validate_peer && model->ops->validate_peer(model->data_ctx, ns, conn->peer_node_id) != 1) {
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;
}
struct ms_session* s = ms_find(ms, conn, ns_id);
if (!s && (type == MS_WAKE || type == MS_BEGIN)) s = ms_create(ms, conn, ns_id);
struct ms_session* s = ms_find(ms, conn, dataset, ns_id);
if (!s && (type == MS_WAKE || type == MS_BEGIN)) s = ms_create(ms, conn, dataset, ns_id);
if (!s) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: no session ns=%s type=%u", ns, type);
queue_dgram_free(entry); queue_entry_free(entry); return;
@ -490,7 +527,7 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!s->leader || s->state != MS_SERVE || request || len != MT_HASH_SIZE) rc = MT_ERR_PROTOCOL;
else if (ms_root(s, roots) < 0) rc = MT_ERR_IO;
else if (!memcmp(roots, p, MT_HASH_SIZE)) {
if (ms->ops->finish && ms->ops->finish(ms->data_ctx, ns) < 0) rc = MT_ERR_DATA;
if (model->ops->finish && model->ops->finish(model->data_ctx, ns) < 0) rc = MT_ERR_DATA;
else {
s->previous_valid = 0; s->verified_revision = s->revision;
rc = ms_send(s, MS_DONE, 0, roots, MT_HASH_SIZE) < 0 ? MT_ERR_IO : 0;
@ -556,22 +593,34 @@ static void ms_conn_status(struct ETCP_CONN* conn, int event, void* arg) {
}
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id, const struct merkle_sync_data_ops* ops, void* data_ctx) {
if (!inst || inst->msync || !inst->ua || !inst->topo_sqlite_db || !ops || !ops->update_bucket_hash || !ops->get_page
|| (!ops->apply_items && !ops->stage_page)
|| ((ops->begin_pull || ops->commit_pull || ops->abort_pull || ops->stage_page)
&& !(ops->begin_pull && ops->commit_pull && ops->abort_pull && ops->stage_page))) {
if (!inst || inst->msync || !inst->ua || !inst->topo_sqlite_db) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid initialization"); return -1;
}
if (merkle_tree_init(inst->topo_sqlite_db) < 0) return -1;
struct merkle_sync* ms = u_calloc(1, sizeof(*ms));
if (!ms) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: allocation failed"); return -1; }
ms->inst = inst; ms->svc_id = svc_id; ms->ops = ops; ms->data_ctx = data_ctx;
ms->inst = inst; ms->svc_id = svc_id;
/* Round counter is local to a connection lifetime; monotonic across session cancellation. */
ms->round_seq = get_time_tb();
inst->msync = ms;
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=2", svc_id);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: initialized service=%u protocol=3", svc_id);
return 0;
}
/* Регистрирует независимую модель в общем движке. */
int merkle_sync_register(struct UTUN_INSTANCE* inst, uint8_t dataset, const struct merkle_sync_data_ops* ops, void* data_ctx) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || ms->datasets[dataset].ops || !ops || !ops->update_bucket_hash || !ops->get_page
|| (!ops->apply_items && !ops->stage_page)
|| ((ops->begin_pull || ops->commit_pull || ops->abort_pull || ops->stage_page)
&& !(ops->begin_pull && ops->commit_pull && ops->abort_pull && ops->stage_page))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid dataset registration dataset=%u", dataset); return -1;
}
ms->datasets[dataset] = (struct ms_dataset){ ops, data_ctx, dataset };
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: registered dataset=%u", dataset);
return 0;
}
@ -584,11 +633,12 @@ void merkle_sync_destroy(struct UTUN_INSTANCE* inst) {
inst->msync = NULL; u_free(ms);
}
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns, merkle_sync_done_cb cb, void* arg) {
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer, const char* ns,
merkle_sync_done_cb cb, void* arg) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
uint64_t ns_id;
struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL;
if (!ms || ms_namespace(ns, &ns_id) < 0 || !conn || !conn->initialized || !conn->links_up || conn->state == 2
if (!ms || !ms->datasets[dataset].ops || ms_namespace(ns, &ns_id) < 0 || !conn || !conn->initialized || !conn->links_up || conn->state == 2
|| conn->peer_sleep_phase || peer == inst->node_id || conn == ms->closing_conn) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: start unavailable ns=%s peer=%016llx", ns ? ns : "-", (unsigned long long)peer);
return -1;
@ -599,8 +649,8 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns,
if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: request allocation failed"); return -1; }
r->cb = cb; r->arg = arg;
}
struct ms_session* s = ms_find(ms, conn, ns_id);
if (!s) s = ms_create(ms, conn, ns_id);
struct ms_session* s = ms_find(ms, conn, dataset, ns_id);
if (!s) s = ms_create(ms, conn, dataset, ns_id);
if (!s) { u_free(r); return -1; }
if (s->state == MS_FAILING) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: start during terminal handoff ns=%s", ns);
@ -615,11 +665,11 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns,
return 0;
}
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns) {
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ns) return;
struct ms_session* s = ms->sessions;
while (s) { struct ms_session* next = s->next; if (s->peer == peer && !strcmp(s->ns, ns)) ms_free(s); s = next; }
while (s) { struct ms_session* next = s->next; if (s->dataset->id == dataset && s->peer == peer && !strcmp(s->ns, ns)) ms_free(s); s = next; }
}
void merkle_sync_cancel_ns(struct UTUN_INSTANCE* inst, const char* ns) {

33
src/chat/merkle_sync.h
Unescape View File

@ -1,7 +1,7 @@
/* merkle_sync — обмен различиями между наборами записей по Merkle-дереву.
* Движок сверяет хеши и передаёт страницы изменившихся листьев; модель данных
* через data_ops читает записи, проверяет подписи/права и сохраняет изменения.
* Сессия относится к (пир, namespace), отправка учитывает backpressure.
* Сессия относится к (пир, dataset, namespace), отправка учитывает backpressure.
* Все операции — в потоке uasync; один движок на UTUN_INSTANCE. */
#ifndef MERKLE_SYNC_H
#define MERKLE_SYNC_H
@ -19,6 +19,13 @@ struct UTUN_INSTANCE;
#define MT_ERR_DISCONNECTED -5
#define MT_ERR_TIMEOUT -6
#define MT_PAGE_SIZE 16384
#define MT_KEY_SIZE 32
#define MT_DATA_MEMBERS 0
#define MT_DATA_FEEDBACK 1
/* Полный ключ курсора, big-endian; первые 8 байт задают ветвь дерева.
* Для node_id участника: BE64(node_id), затем 24 нулевых байта. */
typedef struct { uint8_t bytes[MT_KEY_SIZE]; } merkle_key_t;
/* Namespace — каноническая десятичная строка uint64_t. Движок не знает CHAT-групп.
* SHA256 листа обновляется моделью в порядке unsigned key; родителей считает merkle_tree.
@ -28,8 +35,8 @@ struct merkle_sync_data_ops {
/* Одна страница листа: after_valid=0 начинает обход, иначе только key > after.
* len: ёмкость -> размер; next — последний ключ; more — есть продолжение.
* Пустой результат разрешён только с more=0. Ошибка не кодируется пустой страницей. */
int (*get_page)(void* ctx, const char* ns, uint64_t prefix, int after_valid, uint64_t after,
uint8_t* buf, size_t* len, uint64_t* next, int* more);
int (*get_page)(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after,
uint8_t* buf, size_t* len, merkle_key_t* next, int* more);
/* Проверить весь формат, применить записи и обновить дерево в одной DB-транзакции.
* <0 — ошибка. Никаких broadcast/send-back из этого callback. */
int (*apply_items)(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len);
@ -42,8 +49,8 @@ struct merkle_sync_data_ops {
int (*begin_pull)(void* ctx, const char* ns, uint64_t peer, uint64_t round, void** pull);
int (*commit_pull)(void* ctx, void* pull);
void (*abort_pull)(void* ctx, void* pull);
int (*stage_page)(void* ctx, void* pull, uint64_t prefix, int after_valid, uint64_t after,
uint64_t next, int more, const uint8_t* data, size_t len);
int (*stage_page)(void* ctx, void* pull, 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);
};
/* result — MT_OK или MT_ERR_*; ns заимствован на время callback. */
@ -52,6 +59,10 @@ typedef void (*merkle_sync_done_cb)(uint64_t peer, const char* ns, int result, v
/* Зарегистрировать сервис; нужны ua и общая SQLite. ops/data_ctx живут до destroy. 0/-1. */
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
const struct merkle_sync_data_ops* ops, void* data_ctx);
/* Init регистрирует MEMBERS. Добавить другой набор до запуска его сессий; 0/-1.
* ops/data_ctx живут до уничтожения движка. Каждый набор имеет свой Merkle-корень. */
int merkle_sync_register(struct UTUN_INSTANCE* inst, uint8_t dataset,
const struct merkle_sync_data_ops* ops, void* data_ctx);
/* Отменить сессии без done callbacks, снять обработчики и освободить движок. */
void merkle_sync_destroy(struct UTUN_INSTANCE* inst);
@ -62,8 +73,9 @@ void merkle_sync_destroy(struct UTUN_INSTANCE* inst);
* <0: запрос не принят, callback не вызывается. DOWN/DELETE завершают запросы ошибкой.
* Явные cancel/destroy освобождают запросы без callback.
* Callback может отменить сессию; уничтожение всего экземпляра должно быть отложенным. */
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns, merkle_sync_done_cb cb, void* arg);
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns);
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer, const char* ns,
merkle_sync_done_cb cb, void* arg);
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer, const char* ns);
/* Отменить все сессии namespace / пира без callbacks. */
void merkle_sync_cancel_ns(struct UTUN_INSTANCE* inst, const char* ns);
void merkle_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer);
@ -72,10 +84,11 @@ void merkle_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer);
* При внешней транзакции caller вызывает changed ПОСЛЕ commit.
* Без внешней транзакции пересчёт сам создаёт savepoint и уведомляет после commit.
* Возвращает 1=изменилось, 0=no-op, <0=ошибка. */
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key);
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns, uint64_t key);
/* После commit уведомить сессии namespace о локальном изменении дерева. */
void merkle_sync_changed(struct UTUN_INSTANCE* inst, const char* ns);
void merkle_sync_changed(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns);
/* Прочитать хеш узла дерева в hash. 0=успех, <0=ошибка. */
int merkle_sync_read_hash(struct UTUN_INSTANCE* inst, const char* ns, uint8_t level, uint64_t prefix, uint8_t hash[MT_HASH_SIZE]);
int merkle_sync_read_hash(struct UTUN_INSTANCE* inst, uint8_t dataset, const char* ns,
uint8_t level, uint64_t prefix, uint8_t hash[MT_HASH_SIZE]);
#endif

Write Preview
Loading…
Cancel
Save