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

Adapt member sync and protocol tests to dataset cursors

master
evgeny 6 days ago
parent
539bfd6db9
commit
5e146f20bc
2 changed files with 29 additions and 24 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 23
      src/chat/member_sync.c
  2. 30
      tests/test_merkle_protocol.c

23
src/chat/member_sync.c
Unescape View File

@ -204,12 +204,17 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
return count; return count;
} }
static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, uint64_t after, static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after,
uint8_t* buf, size_t* len, uint64_t* next, int* more) { uint8_t* buf, size_t* len, merkle_key_t* next, int* more) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx; struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db || !buf || !len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items — invalid args db=%p buf=%p len=%p", MS_ID, (void*)db, (void*)buf, (void*)len); return -1; } sqlite3* db = _db(inst); if (!db || !buf || !len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items — invalid args db=%p buf=%p len=%p", MS_ID, (void*)db, (void*)buf, (void*)len); return -1; }
const uint8_t level = MT_MAX_LEVEL; const uint8_t level = MT_MAX_LEVEL;
*next = 0; *more = 0; memset(next, 0, sizeof(*next)); *more = 0;
uint64_t after_id = member_read64(after->bytes);
static const uint8_t zero_tail[MT_KEY_SIZE - 8];
if (after_valid && memcmp(after->bytes + 8, zero_tail, sizeof(zero_tail))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid member cursor ns=%s", MS_ID, ns); return -1;
}
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix); DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl)); char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
@ -227,7 +232,7 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte
sqlite3_stmt* stmt = NULL; sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items prepare failed ns=%s", MS_ID, ns); return -1; } if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items prepare failed ns=%s", MS_ID, ns); return -1; }
sqlite3_bind_int(stmt, 1, after_valid); sqlite3_bind_int(stmt, 1, after_valid);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after); sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after_id);
size_t off = 0; size_t off = 0;
if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; } if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; }
@ -290,7 +295,7 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte
if (sgn && sqlite3_column_bytes(stmt, 11) >= 64) { memcpy(buf + off, sgn, 64); off += 64; } if (sgn && sqlite3_column_bytes(stmt, 11) >= 64) { memcpy(buf + off, sgn, 64); off += 64; }
else { memset(buf + off, 0, 64); off += 64; } else { memset(buf + off, 0, 64); off += 64; }
cnt++; cnt++;
*next = nid; member_write64(next->bytes, nid);
} }
if (step != SQLITE_DONE && !*more) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page read failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); sqlite3_finalize(stmt); return -1; } if (step != SQLITE_DONE && !*more) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page read failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); sqlite3_finalize(stmt); return -1; }
sqlite3_finalize(stmt); sqlite3_finalize(stmt);
@ -694,7 +699,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
/* ── recompute и подтвердить реальное изменение по хешу ── */ /* ── recompute и подтвердить реальное изменение по хешу ── */
if (changed) { if (changed) {
int rc = merkle_sync_recompute_path(inst, ch_id, m->node_id); int rc = merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, MT_DATA_MEMBERS, MT_DATA_MEMBERS, ch_id, m->node_id);
if (rc < 0) return -1; if (rc < 0) return -1;
if (rc == 0) changed = 0; /* запись была no-op (данные идентичны) */ if (rc == 0) changed = 0; /* запись была no-op (данные идентичны) */
} }
@ -705,7 +710,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
} }
static void member_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id) { static void member_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id) {
merkle_sync_changed(inst, ns); merkle_sync_changed(inst, MT_DATA_MEMBERS, ns);
struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)) : NULL; struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)) : NULL;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id); struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) { if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) {
@ -1131,11 +1136,11 @@ void member_sync_destroy(struct UTUN_INSTANCE* inst) {
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg) { const char* ch_id, merkle_sync_done_cb done_cb, void* arg) {
member_sync_verify_and_purge(inst, ch_id); member_sync_verify_and_purge(inst, ch_id);
return merkle_sync_start(inst, peer, ch_id, done_cb, arg); return merkle_sync_start(inst, MT_DATA_MEMBERS, peer, ch_id, done_cb, arg);
} }
void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) { void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) {
merkle_sync_cancel(inst, peer, ch_id); merkle_sync_cancel(inst, MT_DATA_MEMBERS, peer, ch_id);
} }
void member_sync_cancel_channel(struct UTUN_INSTANCE* inst, const char* ch_id) { void member_sync_cancel_channel(struct UTUN_INSTANCE* inst, const char* ch_id) {

30
tests/test_merkle_protocol.c
Unescape View File

@ -14,10 +14,10 @@ static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP
return EVP_DigestUpdate(hash, "record", 6) == 1 ? 1 : -1; return EVP_DigestUpdate(hash, "record", 6) == 1 ? 1 : -1;
} }
static int t_page(void* arg, const char* ns, uint64_t prefix, int has_after, uint64_t after, static int t_page(void* arg, const char* ns, uint64_t prefix, int has_after, const merkle_key_t* after,
uint8_t* data, size_t* len, uint64_t* next, int* more) { uint8_t* data, size_t* len, merkle_key_t* next, int* more) {
(void)arg; (void)ns; (void)prefix; (void)has_after; (void)after; (void)arg; (void)ns; (void)prefix; (void)has_after; (void)after;
memset(data, 0, *len); *next = 1; *more = 1; return 0; memset(data, 0, *len); memset(next, 0, sizeof(*next)); ms_write64(next->bytes, 1); *more = 1; return 0;
} }
static int t_apply(void* arg, const char* ns, uint64_t peer, const uint8_t* data, size_t len) { static int t_apply(void* arg, const char* ns, uint64_t peer, const uint8_t* data, size_t len) {
@ -66,7 +66,7 @@ static void t_destroy(struct fixture* f) {
} }
static struct ms_session* t_session(struct fixture* f) { static struct ms_session* t_session(struct fixture* f) {
struct ms_session* s = ms_create(f->inst.msync, &f->conn, 1001); assert(s); struct ms_session* s = ms_create(f->inst.msync, &f->conn, MT_DATA_MEMBERS, 1001); assert(s);
s->round = 7; s->state = MS_PULL; s->request_id = 1; s->waiting = 1; s->round = 7; s->state = MS_PULL; s->request_id = 1; s->waiting = 1;
s->requests = u_calloc(1, sizeof(*s->requests)); assert(s->requests); s->requests = u_calloc(1, sizeof(*s->requests)); assert(s->requests);
s->requests->cb = t_done; s->requests->arg = f; s->requests->cb = t_done; s->requests->arg = f;
@ -75,7 +75,7 @@ static struct ms_session* t_session(struct fixture* f) {
static void t_receive(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, const uint8_t* data, size_t len) { static void t_receive(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, const uint8_t* data, size_t len) {
struct ll_entry* e = ll_alloc_lldgram((uint16_t)(MS_HEADER + len)); assert(e); struct ll_entry* e = ll_alloc_lldgram((uint16_t)(MS_HEADER + len)); assert(e);
e->len = (uint16_t)(MS_HEADER + len); e->dgram[0] = 0x72; e->len = (uint16_t)(MS_HEADER + len); e->dgram[0] = 0x72; e->dgram[22] = MT_DATA_MEMBERS;
ms_write64(e->dgram + 1, 1001); e->dgram[9] = type; ms_write64(e->dgram + 1, 1001); e->dgram[9] = type;
ms_write64(e->dgram + 10, round); ms_write32(e->dgram + 18, request); ms_write64(e->dgram + 10, round); ms_write32(e->dgram + 18, request);
if (len) memcpy(e->dgram + MS_HEADER, data, len); if (len) memcpy(e->dgram + MS_HEADER, data, len);
@ -100,7 +100,7 @@ static void test_rejected_data(void) {
struct fixture f; t_init(&f); struct fixture f; t_init(&f);
struct ms_session* s = t_session(&f); s->stack[0].level = MT_MAX_LEVEL; struct ms_session* s = t_session(&f); s->stack[0].level = MT_MAX_LEVEL;
f.apply_error = MT_ERR_DATA; f.apply_error = MT_ERR_DATA;
uint8_t page[11] = {0}; uint8_t page[MS_PAGE_HEADER + 2] = {0};
t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page)); t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page));
assert(!f.callbacks && s->state == MS_FAILING); assert(!f.callbacks && s->state == MS_FAILING);
uasync_poll(f.inst.ua, 0); uasync_poll(f.inst.ua, 0);
@ -159,9 +159,9 @@ static void test_page_size_and_cursor(void) {
static void test_readonly_tree(void) { static void test_readonly_tree(void) {
struct fixture f; t_init(&f); struct fixture f; t_init(&f);
assert(sqlite3_exec(f.inst.topo_sqlite_db, "PRAGMA query_only=ON", NULL, NULL, NULL) == SQLITE_OK); assert(sqlite3_exec(f.inst.topo_sqlite_db, "PRAGMA query_only=ON", NULL, NULL, NULL) == SQLITE_OK);
assert(merkle_sync_recompute_path(&f.inst, "1001", 0) < 0); assert(merkle_sync_recompute_path(&f.inst, MT_DATA_MEMBERS, "1001", 0) < 0);
uint8_t hash[32], empty[32] = {0}; uint8_t hash[32], empty[32] = {0};
assert(merkle_sync_read_hash(&f.inst, "1001", 0, 0, hash) == 0 && !memcmp(hash, empty, 32)); assert(merkle_sync_read_hash(&f.inst, MT_DATA_MEMBERS, "1001", 0, 0, hash) == 0 && !memcmp(hash, empty, 32));
t_destroy(&f); t_destroy(&f);
puts("PASS: SQLite failure is reported and cannot publish a root"); puts("PASS: SQLite failure is reported and cannot publish a root");
} }
@ -170,11 +170,11 @@ static void test_backpressure_and_disconnect(void) {
struct fixture f; t_init(&f); struct fixture f; t_init(&f);
struct ll_entry* blocker = queue_entry_new(0); assert(blocker); struct ll_entry* blocker = queue_entry_new(0); assert(blocker);
assert(queue_data_put(f.conn.send_input_q, blocker) == 0); assert(queue_data_put(f.conn.send_input_q, blocker) == 0);
assert(merkle_sync_start(&f.inst, 2, "1001", t_done, &f) == 0); assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0);
uasync_poll(f.inst.ua, 0); uasync_poll(f.inst.ua, 0);
struct ms_session* s = f.inst.msync->sessions; struct ms_session* s = f.inst.msync->sessions;
assert(s->tx && s->waiter.internal); assert(s->tx && s->waiter.internal);
for (int i = 0; i < 1000; i++) merkle_sync_changed(&f.inst, "1001"); for (int i = 0; i < 1000; i++) merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
assert(s->tx && queue_entry_count(f.conn.send_input_q) == 1 && s->dirty); assert(s->tx && queue_entry_count(f.conn.send_input_q) == 1 && s->dirty);
f.conn.links_up = 0; f.conn.links_up = 0;
ms_conn_status(&f.conn, ETCP_CONN_STATUS_DOWN, f.inst.msync); ms_conn_status(&f.conn, ETCP_CONN_STATUS_DOWN, f.inst.msync);
@ -210,14 +210,14 @@ static void test_send_rejection(void) {
static void t_cancel_done(uint64_t peer, const char* ns, int result, void* arg) { static void t_cancel_done(uint64_t peer, const char* ns, int result, void* arg) {
struct fixture* f = arg; struct fixture* f = arg;
t_done(peer, ns, result, arg); t_done(peer, ns, result, arg);
merkle_sync_cancel(&f->inst, peer, ns); merkle_sync_cancel(&f->inst, MT_DATA_MEMBERS, peer, ns);
} }
static void test_error_handoff(void) { static void test_error_handoff(void) {
struct fixture f; t_init(&f); struct fixture f; t_init(&f);
struct ms_session* s = t_session(&f); s->requests->cb = t_cancel_done; struct ms_session* s = t_session(&f); s->requests->cb = t_cancel_done;
s->stack[0].level = MT_MAX_LEVEL; f.apply_error = MT_ERR_DATA; s->stack[0].level = MT_MAX_LEVEL; f.apply_error = MT_ERR_DATA;
uint8_t page[11] = {0}; uint8_t page[MS_PAGE_HEADER + 2] = {0};
t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page)); t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page));
assert(!f.callbacks && s->state == MS_FAILING); assert(!f.callbacks && s->state == MS_FAILING);
uasync_poll(f.inst.ua, 0); uasync_poll(f.inst.ua, 0);
@ -249,15 +249,15 @@ static int t_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round
} }
static int t_commit_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_committed++; return 0; } static int t_commit_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_committed++; return 0; }
static void t_abort_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_released++; u_free(pull); } static void t_abort_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_released++; u_free(pull); }
static int t_stage_page(void* ctx, void* pull, uint64_t prefix, int av, uint64_t after, static int t_stage_page(void* ctx, void* pull, uint64_t prefix, int av, const merkle_key_t* after,
uint64_t next, int more, const uint8_t* data, size_t len) { const merkle_key_t* next, int more, const uint8_t* data, size_t len) {
(void)ctx; (void)pull; (void)prefix; (void)av; (void)after; (void)next; (void)more; (void)data; (void)len; return 0; (void)ctx; (void)pull; (void)prefix; (void)av; (void)after; (void)next; (void)more; (void)data; (void)len; return 0;
} }
static void test_pull_lifecycle(void) { static void test_pull_lifecycle(void) {
struct fixture f; t_init(&f); struct fixture f; t_init(&f);
struct merkle_sync_data_ops ops = t_ops; struct merkle_sync_data_ops ops = t_ops;
ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page; ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page;
f.inst.msync->ops = &ops; f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
struct ms_session* s = t_session(&f); struct ms_session* s = t_session(&f);
assert(ms_pull_start(s) == 0 && s->pull); assert(ms_pull_start(s) == 0 && s->pull);
ms_discard_tx(s); s->stack[0].level = MT_MAX_LEVEL; ms_discard_tx(s); s->stack[0].level = MT_MAX_LEVEL;

Write Preview
Loading…
Cancel
Save