diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index 5f1b3376..d850f04f 100644 --- a/src/chat/merkle_sync.c +++ b/src/chat/merkle_sync.c @@ -11,11 +11,12 @@ #include #include -/* 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) { diff --git a/src/chat/merkle_sync.h b/src/chat/merkle_sync.h index 7d882c8d..1571aeb0 100644 --- a/src/chat/merkle_sync.h +++ b/src/chat/merkle_sync.h @@ -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