From 2d8384830d6f5409bd23a00b3d8be14316aa5749 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 17:54:13 +0200 Subject: [PATCH] test(member-files): cover bulk mutations and lost result replay --- src/member_files/member_files.h | 4 ++- tests/test_member_files_network.c | 44 ++++++++++++++++++++++++++++++- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/src/member_files/member_files.h b/src/member_files/member_files.h index 8576acbf..961b5f56 100644 --- a/src/member_files/member_files.h +++ b/src/member_files/member_files.h @@ -32,7 +32,9 @@ void member_files_cancel_group(struct UTUN_INSTANCE* inst, uint64_t group); // ── Команды файлового сервера ── /* Отправить одну команду; request_id=0 выбирает случайный ID. NULL — отказ запуска, callback не вызывается. - * При успехе callback ровно один, включая отмену; handle освобождён до callback, arg принадлежит вызывающему. */ + * При успехе callback ровно один, включая отмену; handle освобождён до callback, arg принадлежит вызывающему. + * Без ответа — timeout120с. BUSY для COMMIT/MKDIR/RENAME/REMOVE сохраняет ID и повторяет запрос каждые5с: + * явный отказ продлевает окно ожидания до120с от него, суммарно до240с. Остальные BUSY передаются callback. */ struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t server, const struct mf_message* request, mf_response_fn callback, void* arg); diff --git a/tests/test_member_files_network.c b/tests/test_member_files_network.c index fc0bf35a..263a17b1 100644 --- a/tests/test_member_files_network.c +++ b/tests/test_member_files_network.c @@ -51,6 +51,8 @@ struct network_test { unsigned restart_tick; unsigned preservation_case; char preserved_temp[864]; + unsigned mutation_index, busy_responses, dropped_response; + etcp_recv_fn reader_receive; struct api_case api[8]; void* timer; }; @@ -163,6 +165,29 @@ static void response(void* arg, const struct mf_message* r) { struct network_test* t = arg; t->response = *r; t->answered = 1; } +/* Потеря первого результата проверяет повтор с тем же ID без повторного mkdir. */ +static void mutation_receive(struct ETCP_CONN* conn, struct ll_entry* e) { + struct network_test* t = active; + struct mf_message m; + if (e && e->dgram && e->len > ROUTER_SVC_PAYLOAD_OFF && + !mf_decode(e->dgram + ROUTER_SVC_PAYLOAD_OFF, e->len - ROUTER_SVC_PAYLOAD_OFF, &m) && + m.kind == MF_RESPONSE && m.command == MF_MKDIR && m.request_id >= 1000 && m.request_id < 1076) { + if (m.status == MF_BUSY) t->busy_responses++; + if (!m.status && m.request_id == 1000 && !t->dropped_response) { + t->dropped_response = 1; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: dropped successful mutation response id=1000; waiting for replay"); + queue_dgram_free(e); queue_entry_free(e); return; + } + } + t->reader_receive(conn, e); +} + +static void mutation_start(struct network_test* t) { + struct mf_message q = { .command = MF_MKDIR, .owner = t->ids[READER], .request_id = 1000 + t->mutation_index }; + snprintf(q.path, sizeof(q.path), "bulk_%02u", t->mutation_index); t->answered = 0; + if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) fail(t, "bulk mutation start"); +} + static int put_member(struct network_test* t, int destination, int member, int deleted) { uint8_t msg[512], join[64], tags_sig[64], tree[64]; uint64_t ts = 100000; @@ -238,7 +263,7 @@ static int downloaded(struct network_test* t) { static void tick(void* arg) { struct network_test* t = arg; t->timer = NULL; - if (++t->ticks > 6000) { fail(t, "60s deadline"); return; } + if (++t->ticks > 30000) { fail(t, "300s deadline including receipt expiry"); return; } // Отсчитываем полные 6с от фактической блокировки worker, не от постановки задания. pthread_mutex_lock(&t->disk_mutex); if (t->disk_started_tb && !t->disk_released && get_time_tb() - t->disk_started_tb >= 60000) { @@ -336,6 +361,23 @@ static void tick(void* arg) { fail(t, "download refusal/cancel removed unowned temp or changed target"); return; } if (++t->preservation_case < 3) { preservation_start(t); goto again; } + t->phase = 24; + t->reader_receive = t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES]; + t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES] = mutation_receive; + mutation_start(t); + } else if (t->phase == 24 && t->answered) { + if (t->response.status) { fail(t, "bulk mutation did not survive busy/replay"); return; } + if (++t->mutation_index < 76) { mutation_start(t); goto again; } + t->nodes[READER]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES] = t->reader_receive; + if (!t->dropped_response || !t->busy_responses) { fail(t, "bulk mutation did not exercise lost result and busy"); return; } + t->phase = 25; request(t, MF_QUOTA, t->ids[READER]); + } else if (t->phase == 25 && t->answered) { + if (t->response.status || t->response.data_len != MF_QUOTA_BYTES || mf_get64(t->response.data) || + mf_get64(t->response.data + 16) || mf_get64(t->response.data + 24) || mf_get64(t->response.data + 32) != 76) { + fail(t, "bulk mutation duplicated or lost directories"); return; + } + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: PASS76 mutations, lost response replayed once, busy_responses=%u directories=76", + t->busy_responses); t->phase = 10; t->answered = 0; struct mf_message q = { .command = MF_STAT, .owner = t->ids[AUTHOR] }; snprintf(q.path, sizeof(q.path), "cancelled.bin"); if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "cancelled STAT"); return; }