diff --git a/tests/test_merkle_protocol.c b/tests/test_merkle_protocol.c index 3c405c925..ffd20fb0e 100644 --- a/tests/test_merkle_protocol.c +++ b/tests/test_merkle_protocol.c @@ -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;