test(member-files): cover bulk mutations and lost result replay

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);
 

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; }