jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
111432aa6405ee5fbaa44ba39c654fd3bde9d113
utun2/src/chat/merkle_sync.c
T

638 lines
29 KiB
C
Raw Blame History

#include "merkle_sync.h"
#include "../utun_instance.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../../lib/ll_queue.h"
#include "../../lib/platform_compat.h"
#include <string.h>
#include <stdlib.h>
#include <errno.h>
/* Wire v2: svc:1, namespace:8, type:1, round:8, request:4, payload.
* Все целые заголовка big-endian. Один запрос на проход и один исходящий пакет на сессию.
* Меньший node_id ведёт раунд: pull -> TURN -> reverse pull -> CHECK -> DONE. */
#define MS_HEADER 22
#define MS_QUERY_SIZE 18
#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 };
struct ms_request {
struct ms_request* next;
merkle_sync_done_cb cb;
void* arg;
};
struct ms_frame {
uint64_t prefix;
uint32_t children;
uint8_t level;
};
struct ms_session {
struct ms_session* next;
struct merkle_sync* ms;
struct ETCP_CONN* conn;
struct ll_queue* send_q;
struct queue_waiter_handle waiter;
struct ll_entry* tx;
void* work;
void* deadline;
void* pull;
int failure_error;
struct ms_request* requests;
uint64_t ns_id, peer, round, revision, verified_revision;
char ns[21];
int state, leader, dirty, depth, after_valid, waiting;
uint32_t request_id, remote_request_id;
uint64_t after;
struct ms_frame stack[MT_MAX_LEVEL + 1];
uint8_t previous_roots[2 * MT_HASH_SIZE];
int previous_valid;
uint64_t sent_bytes, received_bytes;
};
struct merkle_sync {
struct UTUN_INSTANCE* inst;
uint8_t svc_id;
const struct merkle_sync_data_ops* ops;
void* data_ctx;
struct ms_session* sessions;
uint64_t round_seq;
struct ETCP_CONN* closing_conn;
};
static uint64_t ms_read64(const uint8_t* p) { uint64_t n; memcpy(&n, p, 8); return be64toh(n); }
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_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;
if (ns[0] == '0' && ns[1]) return -1;
char* end;
errno = 0;
unsigned long long n = strtoull(ns, &end, 10);
if (errno || *end) return -1;
*id = (uint64_t)n;
return 0;
}
static struct ms_session* ms_find(struct merkle_sync* ms, struct ETCP_CONN* conn, 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;
return NULL;
}
static void ms_notify(struct ms_session* s, int result) {
/* Все данные копируются до callback: callback вправе отменить сессию. */
struct ms_request* r = s->requests;
s->requests = NULL;
uint64_t peer = s->peer;
char ns[sizeof(s->ns)]; memcpy(ns, s->ns, sizeof(ns));
while (r) {
struct ms_request* next = r->next;
merkle_sync_done_cb cb = r->cb; void* arg = r->arg;
u_free(r);
cb(peer, ns, result, arg);
r = next;
}
}
static int ms_schedule(struct ms_session* s);
static void ms_work(void* arg);
static void ms_fail(struct ms_session* s, int error, const char* reason, int report);
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);
}
static void ms_stop_deadline(struct ms_session* s) {
if (s->deadline) uasync_cancel_timeout(s->ms->inst->ua, s->deadline);
s->deadline = NULL;
}
static void ms_timeout(void* arg) {
struct ms_session* s = arg;
s->deadline = NULL;
ms_fail(s, MT_ERR_TIMEOUT, "no protocol progress for 30 seconds", 0);
}
static int ms_progress(struct ms_session* s) {
ms_stop_deadline(s);
s->deadline = uasync_set_timeout(s->ms->inst->ua, MS_PROGRESS_TIMEOUT_TB, s, ms_timeout, "merkle_progress");
if (!s->deadline) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: deadline allocation ns=%s", s->ns); return MT_ERR_IO;
}
return 0;
}
static void ms_discard_tx(struct ms_session* s) {
queue_waiter_cancel(s->send_q, &s->waiter);
if (s->tx) { queue_dgram_free(s->tx); queue_entry_free(s->tx); s->tx = NULL; }
}
static void ms_free(struct ms_session* s) {
struct merkle_sync* ms = s->ms;
struct ms_session** p = &ms->sessions;
while (*p && *p != s) p = &(*p)->next;
if (*p) *p = s->next;
if (s->work) uasync_call_soon_cancel(ms->inst->ua, s->work);
ms_stop_deadline(s);
ms_abort_pull(s);
ms_discard_tx(s);
while (s->requests) {
struct ms_request* r = s->requests;
s->requests = r->next; u_free(r);
}
etcp_conn_ref_free(s->conn);
u_free(s);
}
static struct ms_session* ms_create(struct merkle_sync* ms, struct ETCP_CONN* conn, 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->peer = conn->peer_node_id; s->ns_id = ns_id;
snprintf(s->ns, sizeof(s->ns), "%llu", (unsigned long long)ns_id);
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",
s->ns, (unsigned long long)s->peer, (void*)conn, s->leader);
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);
}
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);
}
static void ms_sent(struct ll_queue* q, void* arg) {
struct ms_session* s = arg;
struct ll_entry* entry = s->tx;
s->tx = NULL;
if (!entry) return;
int type = entry->dgram[9];
size_t len = entry->len;
if (!s->conn->links_up || s->conn->state == 2 || s->conn->close_requested || q != s->conn->send_input_q) {
queue_dgram_free(entry); queue_entry_free(entry);
ms_fail(s, MT_ERR_DISCONNECTED, "connection unavailable", 0);
return;
}
if (etcp_send(s->conn, entry) != 0) {
queue_dgram_free(entry); queue_entry_free(entry);
ms_fail(s, MT_ERR_IO, "transport rejected packet", 0);
return;
}
s->sent_bytes += len;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: TX ns=%s peer=%016llx round=%llu type=%u bytes=%zu",
s->ns, (unsigned long long)s->peer, (unsigned long long)s->round, type, len);
if (type == MS_ERROR) {
s->state = MS_FAILED;
ms_stop_deadline(s);
ms_notify(s, s->failure_error);
return;
}
if (type == MS_DONE) {
s->state = MS_IDLE;
ms_stop_deadline(s);
if (s->revision == s->verified_revision) {
s->dirty = 0;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: converged ns=%s peer=%016llx round=%llu tx=%llu rx=%llu",
s->ns, (unsigned long long)s->peer, (unsigned long long)s->round,
(unsigned long long)s->sent_bytes, (unsigned long long)s->received_bytes);
ms_notify(s, MT_OK);
return;
}
}
if (s->state == MS_IDLE && s->dirty && ms_schedule(s) < 0) ms_fail(s, MT_ERR_IO, "work allocation", 1);
}
static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8_t* payload, size_t len) {
if (s->tx || len > MT_PAGE_SIZE - MS_HEADER) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid TX ns=%s type=%d busy=%d size=%zu", s->ns, type, s->tx != NULL, len);
return -1;
}
struct ll_entry* e = ll_alloc_lldgram((uint16_t)(MS_HEADER + len));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: packet allocation failed ns=%s", s->ns); return -1; }
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;
ms_write64(e->dgram + 10, type == MS_WAKE ? 0 : s->round);
ms_write32(e->dgram + 18, request);
if (len) memcpy(e->dgram + MS_HEADER, payload, len);
s->tx = e;
/* Transport normalizer configures deferred waiters. Не вызываем callback рекурсивно. */
if (queue_waiter_wait(s->send_q, &s->waiter, ms_sent, s) < 0) { ms_discard_tx(s); return -1; }
if (ms_progress(s) < 0) { ms_discard_tx(s); return -1; }
return 0;
}
static void ms_fail(struct ms_session* s, int error, const char* reason, int report) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: failed ns=%s peer=%016llx round=%llu error=%d reason=%s",
s->ns, (unsigned long long)s->peer, (unsigned long long)s->round, error, reason);
s->state = MS_FAILED; s->dirty = 0; s->waiting = 0; s->failure_error = error;
if (s->work) { uasync_call_soon_cancel(s->ms->inst->ua, s->work); s->work = NULL; }
ms_discard_tx(s);
ms_stop_deadline(s);
ms_abort_pull(s);
if (report) {
s->state = MS_FAILING;
uint8_t code[4]; ms_write32(code, (uint32_t)(-error));
if (ms_send(s, MS_ERROR, 0, code, sizeof(code)) == 0) return;
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: could not send error ns=%s", s->ns);
s->state = MS_FAILED;
}
ms_notify(s, error);
}
static int ms_schedule(struct ms_session* s) {
if (s->work || s->state == MS_FAILED) return 0;
s->work = uasync_call_soon(s->ms->inst->ua, s, ms_work);
if (!s->work) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: work allocation ns=%s", s->ns); return MT_ERR_IO;
}
return 0;
}
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);
if (++s->request_id == 0) return -1;
s->waiting = 1;
return ms_send(s, MS_QUERY, s->request_id, body, sizeof(body));
}
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)
return MT_ERR_DATA;
s->state = MS_PULL; s->depth = 0; s->after_valid = 0; s->after = 0;
memset(s->stack, 0, sizeof(s->stack));
return ms_query(s);
}
static int ms_walk_next(struct ms_session* s) {
while (s->depth >= 0) {
struct ms_frame* f = &s->stack[s->depth];
if (f->children) {
unsigned i = (unsigned)__builtin_ctz(f->children);
f->children &= ~(UINT32_C(1) << i);
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;
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);
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;
uint8_t root[MT_HASH_SIZE];
if (ms_root(s, root) < 0) return -1;
s->state = MS_DONE_WAIT;
return ms_send(s, MS_CHECK, 0, root, sizeof(root));
}
static int ms_begin(struct ms_session* s) {
ms_abort_pull(s);
s->round = ++s->ms->round_seq;
if (!s->round) return -1;
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 ns=%s peer=%016llx round=%llu",
s->ns, (unsigned long long)s->peer, (unsigned long long)s->round);
return ms_send(s, MS_BEGIN, 0, NULL, 0);
}
static void ms_work(void* arg) {
struct ms_session* s = arg;
s->work = NULL;
if (s->state != MS_IDLE || !s->dirty || s->tx) return;
int rc;
if (s->leader) rc = ms_begin(s);
else { s->dirty = 0; rc = ms_send(s, MS_WAKE, 0, NULL, 0); }
if (rc < 0) ms_fail(s, MT_ERR_IO, "start send", 1);
}
void merkle_sync_changed(struct UTUN_INSTANCE* inst, 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;
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) {
ms_fail(s, MT_ERR_IO, "change scheduling", 1); return;
}
}
}
int merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, 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; }
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);
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));
if (sqlite3_exec(db, "RELEASE merkle_update", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: commit ns=%s: %s", ns, sqlite3_errmsg(db));
if (sqlite3_exec(db, "ROLLBACK TO merkle_update; RELEASE merkle_update", NULL, NULL, NULL) != SQLITE_OK)
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) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: recompute failed ns=%s key=%016llx", ns, (unsigned long long)key);
return rc;
}
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);
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;
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;
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);
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;
}
static int ms_receive_hashes(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 != MT_BUCKETS * MT_HASH_SIZE) return MT_ERR_PROTOCOL;
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;
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;
s->waiting = 0;
return ms_walk_next(s);
}
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;
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);
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; }
return ms_walk_next(s);
}
static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry) return;
struct merkle_sync* ms = conn && conn->instance ? conn->instance->msync : NULL;
if (!ms || !entry->dgram || entry->len < MS_HEADER || entry->len > MT_PAGE_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: invalid packet size=%u", entry->len);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
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];
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) {
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);
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;
}
s->received_bytes += entry->len;
const uint8_t* p = d + MS_HEADER;
size_t len = entry->len - MS_HEADER;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: RX ns=%s peer=%016llx round=%llu type=%u request=%u bytes=%zu state=%d",
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);
goto done;
}
if (type == MS_WAKE) {
if (!s->leader || round || request || len) rc = MT_ERR_PROTOCOL;
else {
/* Follower мог отменить и заново подключить namespace. Старый запрос уже некому обслужить. */
ms_discard_tx(s);
s->previous_valid = 0;
rc = ms_begin(s);
}
} else if (type == MS_BEGIN) {
if (s->leader || !round || request || len) rc = MT_ERR_PROTOCOL;
else if (round <= s->round) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: discard duplicate BEGIN ns=%s", ns);
} else {
ms_discard_tx(s);
ms_abort_pull(s);
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);
} 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);
else if (type == MS_TURN) {
if (s->leader || s->state != MS_SERVE || request || len) rc = MT_ERR_PROTOCOL;
else rc = ms_pull_start(s);
} else if (type == MS_CHECK) {
uint8_t roots[2 * MT_HASH_SIZE];
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;
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;
}
} else {
memcpy(roots + MT_HASH_SIZE, p, MT_HASH_SIZE);
if (s->previous_valid && !memcmp(s->previous_roots, roots, sizeof(roots))) rc = MT_ERR_CONFLICT;
else {
memcpy(s->previous_roots, roots, sizeof(roots)); s->previous_valid = 1;
s->state = MS_IDLE; s->dirty = 1; rc = ms_schedule(s);
}
}
} else if (type == MS_DONE) {
uint8_t root[MT_HASH_SIZE];
if (s->leader || s->state != MS_DONE_WAIT || request || len != MT_HASH_SIZE) rc = MT_ERR_PROTOCOL;
else if (ms_root(s, root) < 0) rc = MT_ERR_IO;
else {
s->state = MS_IDLE;
ms_stop_deadline(s);
s->dirty = memcmp(root, p, sizeof(root)) != 0;
if (s->dirty) rc = ms_schedule(s);
else {
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: converged ns=%s peer=%016llx round=%llu",
s->ns, (unsigned long long)s->peer, (unsigned long long)s->round);
notify = 1;
}
}
} else if (type == MS_ERROR) {
if (request || len != 4 || ms_read32(p) < 1 || ms_read32(p) > 6) rc = MT_ERR_PROTOCOL;
else {
rc = -(int)ms_read32(p);
queue_dgram_free(entry); queue_entry_free(entry);
ms_fail(s, rc, "remote error", 0); return;
}
} else rc = MT_ERR_PROTOCOL;
done:
queue_dgram_free(entry); queue_entry_free(entry);
if (rc < 0) { ms_fail(s, rc, "receive", 1); return; }
if (notify) ms_notify(s, MT_OK);
}
static void ms_conn_status(struct ETCP_CONN* conn, int event, void* arg) {
struct merkle_sync* ms = arg;
if (event != ETCP_CONN_STATUS_DOWN && event != ETCP_CONN_STATUS_DELETE && event != ETCP_CONN_STATUS_REINIT) return;
ms->closing_conn = conn;
/* DELETE приходит до освобождения normalizer/input. Отмена использует сохранённую очередь. */
for (;;) {
struct ms_session* s = ms->sessions;
while (s && s->conn != conn) s = s->next;
if (!s) break;
/* Удаляем до callback, чтобы повторный start не мог сохранить умирающую сессию. */
struct ms_request* requests = s->requests; s->requests = NULL;
uint64_t peer = s->peer; char ns[21]; memcpy(ns, s->ns, sizeof(ns));
ms_free(s);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: detach ns=%s peer=%016llx event=%d", ns, (unsigned long long)peer, event);
while (requests) {
struct ms_request* next = requests->next;
merkle_sync_done_cb cb = requests->cb; void* cb_arg = requests->arg;
u_free(requests); cb(peer, ns, MT_ERR_DISCONNECTED, cb_arg); requests = next;
}
}
ms->closing_conn = NULL;
}
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))) {
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;
/* Round counter is local to a connection lifetime; monotonic across session cancellation. */
ms->round_seq = get_time_tb();
inst->msync = ms;
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);
return 0;
}
void merkle_sync_destroy(struct UTUN_INSTANCE* inst) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms) return;
etcp_remove_conn_status_cbk(inst, ms_conn_status, ms);
etcp_unbind(inst, ms->svc_id);
while (ms->sessions) ms_free(ms->sessions);
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) {
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
|| 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;
}
struct ms_request* r = NULL;
if (cb) {
r = u_calloc(1, sizeof(*r));
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);
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);
u_free(r); return -1;
}
if (s->state == MS_FAILED) { s->state = MS_IDLE; s->previous_valid = 0; }
s->dirty = 1;
if (s->state == MS_IDLE && ms_schedule(s) < 0) {
u_free(r); ms_fail(s, MT_ERR_IO, "start scheduling", 1); return -1;
}
if (r) { r->next = s->requests; s->requests = r; }
return 0;
}
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, 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; }
}
void merkle_sync_cancel_ns(struct UTUN_INSTANCE* inst, 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 (!strcmp(s->ns, ns)) ms_free(s); s = next; }
}
void merkle_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms) return;
struct ms_session* s = ms->sessions;
while (s) { struct ms_session* next = s->next; if (s->peer == peer) ms_free(s); s = next; }
}
Reference in new issue View Git Blame Copy Permalink