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

Test receive-only Merkle permissions, confirmation and source races

Browse Source
This commit is contained in:
evgeny committed 2026-11-04 00:09:46 +01:00
1 parent 2c9f122d57
commit 306c38ca82
1 file changed
+97 -1
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_merkle_protocol.c
+97 -1
View File
@@ -341,7 +341,7 @@ static int t_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round
(void)ctx; (void)ns; (void)peer; (void)round;
*pull = u_malloc(1); assert(*pull); pull_created++; return 0;
}
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) { assert(pull); pull_committed++; return ((struct fixture*)ctx)->apply_error; }
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, const merkle_key_t* after,
const merkle_key_t* next, int more, const uint8_t* data, size_t len) {
@@ -488,6 +488,101 @@ static void test_full_cursor(void) {
puts("PASS: full cursors preserve colliding 64-bit prefixes and reject backwards progress");
}
/* Проверяем реальный handshake отдельного запроса: ни обычного допуска, ни обратной выдачи данных. */
static void test_receive_only(void) {
struct merkle_sync_data_ops ops = t_ops;
ops.validate_peer = t_authorized; ops.validate_request = t_can_run;
ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page;
for (int rejected = 0; rejected < 2; rejected++) {
struct fixture f; t_init(&f); f.inst.node_id = 3; f.running = 1;
f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
int committed = pull_committed, released = pull_released;
uint8_t id[MS_TOKEN_SIZE] = {8}, accepted[MS_TOKEN_SIZE] = {9};
t_receive_tokens(&f, MS_BEGIN | MS_READ_ONLY, 8, 0, id, NULL, NULL, 0);
assert(!f.inst.msync->sessions); /* Право запросить не разрешает выдавать данные. */
t_receive_tokens(&f, MS_READY | MS_READ_ONLY | MS_RESPONSE, 8, 0, id, accepted, NULL, 0);
assert(!f.inst.msync->sessions); /* Ответ без локального запроса не создаёт контекст. */
assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_cancel_done, &f) == 0);
struct ms_session* s = ms_find(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001, MS_READER); assert(s && s->leader);
uasync_call_soon_cancel(f.inst.ua, s->work); s->work = NULL; ms_work(s);
assert(s->state == MS_BEGIN_WAIT && s->tx->dgram[9] == (MS_BEGIN | MS_READ_ONLY));
ms_discard_tx(s);
t_receive_tokens(&f, MS_READY | MS_READ_ONLY | MS_RESPONSE, s->round, 0, s->round_id, accepted, NULL, 0);
assert(s->state == MS_PULL && s->pull && s->tx->dgram[9] == (MS_QUERY | MS_READ_ONLY));
ms_discard_tx(s);
uint8_t hashes[MT_BUCKETS * MT_HASH_SIZE] = {0};
t_receive_tokens(&f, MS_HASHES | MS_READ_ONLY | MS_RESPONSE, s->round, s->request_id,
s->round_id, s->accept_id, hashes, sizeof(hashes));
assert(s->state == MS_DONE_WAIT && s->pull && pull_committed == committed && !f.callbacks);
assert(s->tx && s->tx->dgram[9] == (MS_CHECK | MS_READ_ONLY) && s->tx->len == MS_HEADER);
ms_discard_tx(s); f.apply_error = rejected ? MT_ERR_DATA : 0;
t_receive_tokens(&f, MS_DONE | MS_READ_ONLY | MS_RESPONSE, s->round, 0, s->round_id, s->accept_id, NULL, 0);
if (rejected) uasync_poll(f.inst.ua, 0);
assert(pull_committed == committed + 1 && pull_released == released + 1);
assert(f.callbacks == 1 && f.result == (rejected ? MT_ERR_DATA : MT_OK) && !f.inst.msync->sessions);
t_destroy(&f);
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS receive-only: no unsolicited state or disclosure; commit waits for DONE and can reject data");
}
/* Изменение источника до CHECK либо перед выдачей DONE отменяет весь полученный снимок. */
static void test_receive_only_source_change(void) {
struct merkle_sync_data_ops ops = t_ops;
ops.validate_peer = t_authorized; ops.validate_request = t_can_run;
ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page;
for (int queued = 0; queued < 2; queued++) {
struct fixture f; t_init(&f); f.authorized = 1;
f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
uint8_t id[MS_TOKEN_SIZE] = {8};
t_receive_tokens(&f, MS_BEGIN | MS_READ_ONLY, 8, 0, id, NULL, NULL, 0);
struct ms_session* s = ms_find(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001, MS_WRITER); assert(s);
assert(s->state == MS_SERVE && !s->leader && s->tx->dgram[9] == (MS_READY | MS_READ_ONLY | MS_RESPONSE));
ms_discard_tx(s);
if (!queued) merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
t_receive_tokens(&f, MS_CHECK | MS_READ_ONLY, 8, 0, s->round_id, s->accept_id, NULL, 0);
if (queued) {
assert(s->tx && s->tx->dgram[9] == (MS_DONE | MS_READ_ONLY | MS_RESPONSE));
merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
}
for (int i = 0; i < 4; i++) uasync_poll(f.inst.ua, 0);
assert(s->state == MS_FAILED && !s->tx && !s->work && !f.callbacks);
struct ll_entry* packet = queue_data_get(f.conn.send_input_q); assert(packet);
assert(packet->dgram[9] == (MS_ERROR | MS_READ_ONLY | MS_RESPONSE));
assert(ms_read32(packet->dgram + MS_HEADER) == -MT_ERR_CONFLICT);
queue_dgram_free(packet); queue_entry_free(packet);
assert(!f.conn.send_input_q->head);
t_destroy(&f);
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS receive-only: source changes invalidate both immediate and queued confirmation");
}
/* Запрос/обычный обмен/встречная выдача не должны смешиваться даже при одинаковых идентификаторах. */
static void test_receive_only_boundaries(void) {
struct fixture f; t_init(&f); f.running = 1;
struct merkle_sync_data_ops ops = t_ops;
ops.validate_peer = t_authorized; ops.validate_request = t_can_run;
assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &ops, &f) < 0); /* Нужна изолированная загрузка. */
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->datasets[MT_DATA_MEMBERS].ops = &ops;
assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0);
struct ms_session* reader = ms_find(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001, MS_READER); assert(reader);
f.authorized = 1;
struct ms_session* exchange = ms_create(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001, MS_EXCHANGE); assert(exchange);
struct ms_session* writer = ms_create(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001, MS_WRITER); assert(writer);
uint8_t query[MS_QUERY_SIZE] = {0}; reader->state = MS_SERVE;
assert(ms_serve_query(reader, 1, query, sizeof(query)) == MT_ERR_PROTOCOL && !reader->tx);
merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
assert(reader->revision == 1 && writer->revision == 1 && exchange->revision == 1);
assert(!writer->work && !writer->dirty && exchange->dirty);
f.running = 0; merkle_sync_revalidate_peers(&f.inst, MT_DATA_MEMBERS, "1001");
assert(reader->state == MS_FAILED && !reader->work && f.callbacks == 1 && f.result == MT_ERR_DATA);
assert(writer->state != MS_FAILED && exchange->state != MS_FAILED);
merkle_sync_cancel(&f.inst, MT_DATA_MEMBERS, 2, "1001");
assert(!f.inst.msync->sessions && !f.conn.ref_count);
t_destroy(&f);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS receive-only: no reverse queries; modes and permissions are independent; cancellation frees all modes");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
size_t baseline = u_get_allocated_count();
@@ -496,6 +591,7 @@ int main(void) {
test_send_rejection(); test_checkpoint_revision();
test_error_handoff(); test_blocked_error_timeout(); test_pull_lifecycle();
test_dataset_isolation(); test_full_cursor(); test_permission_rounds(); test_model_policy(); test_reattach_tokens(); test_handshake_failures();
test_receive_only(); test_receive_only_source_change(); test_receive_only_boundaries();
assert(u_get_allocated_count() == baseline);
puts("ALL PASS: protocol regressions, no tracked allocations leaked");
return 0;