test(member-files): verify timed replay and bulk removal

src/member_files/member_files.c

@@ -26,6 +26,7 @@ struct mf_pending {
     mf_response_fn callback;
     void* arg;
     void* pump;
+    void* retry;
     struct CONN_MGR_HANDLE* cm;
     struct queue_waiter_handle waiter;
 };
@@ -170,6 +171,7 @@ int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct
 static void mf_pending_finish(struct mf_pending* p, const struct mf_message* response, int status) {
     struct member_files* f = p->files;
     if (p->pump) uasync_call_soon_cancel(f->inst->ua, p->pump);
+    if (p->retry) uasync_cancel_timeout(f->inst->ua, p->retry);
     etcp_router_cancel_send_ready(f->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES, &p->waiter);
     if (p->cm) conn_mgr_close(p->cm);
     struct mf_pending** link = &f->pending;
@@ -188,6 +190,7 @@ static void mf_pending_finish(struct mf_pending* p, const struct mf_message* res
 static void mf_pending_pump(void* arg) {
     struct mf_pending* p = arg; p->pump = NULL;
     if (p->files->closing) return;
+    if (p->retry) { uasync_cancel_timeout(p->files->inst->ua, p->retry); p->retry = NULL; }
     if (get_time_tb() - p->started >= MF_REPLAY_TB) { mf_pending_finish(p, NULL, MF_TIMEOUT); return; }
     int rc = mf_send(p->files, p->group, p->server, &p->request);
     if (!rc) { p->sent = get_time_tb(); p->retry_at = 0; }
@@ -203,6 +206,12 @@ static void mf_pending_wake(struct mf_pending* p) {
 
 static void mf_client_ready(struct ll_queue* q, void* arg) { (void)q; mf_pending_wake(arg); }
 
+/* Отдельный таймер не округляет срок освобождения квитанции до секундного sweep. */
+static void mf_pending_retry(void* arg) {
+    struct mf_pending* p = arg; p->retry = NULL; p->retry_at = 0;
+    mf_pending_wake(p);
+}
+
 static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t group, enum conn_mgr_event event, void* arg) {
     (void)h; (void)peer; (void)group;
     struct mf_pending* p = arg;
@@ -433,6 +442,10 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c
                 p->retry_at = now + delay;
                 p->sent = now;
                 if (p->pump) { uasync_call_soon_cancel(f->inst->ua, p->pump); p->pump = NULL; }
+                if (p->retry) uasync_cancel_timeout(f->inst->ua, p->retry);
+                p->retry = uasync_set_timeout(f->inst->ua, (int)delay, p, mf_pending_retry, "member_files_retry");
+                if (!p->retry) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: busy retry timer failed id=%llu; tick will retry",
+                    (unsigned long long)m.request_id);
                 etcp_router_cancel_send_ready(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES, &p->waiter);
                 if (!p->busy) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: mutation busy; retaining request id=%llu cmd=%s limit=120s",
                     (unsigned long long)m.request_id, mf_command_text(m.command));

src/member_files/protocol.h

@@ -4,6 +4,8 @@
  * Нулевая version в BEGIN означает создание; замена требует version из STAT.
  * LIST: cursor — число уже выданных записей; version фиксирует снимок каталога.
  * READ/WRITE: offset и size задают диапазон; WRITE.data_len обязан равняться size.
+ * COMMIT/MKDIR/RENAME/REMOVE: RESPONSE BUSY.cursor — срок до освобождения квитанции в0.1мс;
+ * 0 означает обычный повтор через5с. Общий клиентский срок120с не продлевается.
  * Данные LIST — последовательность entry: name_len:2,type:1,size:8,mtime:8,version:16,name.
  * Токены меняются при перезапуске сервера. Никаких ABI-зависимых структур в wire. */
 #ifndef MEMBER_FILES_PROTOCOL_H

tests/test_member_files_network.c

@@ -183,7 +183,8 @@ static void mutation_receive(struct ETCP_CONN* conn, struct ll_entry* 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 };
+    struct mf_message q = { .command = t->phase == 26 ? MF_STAT : MF_MKDIR, .owner = t->ids[READER],
+        .request_id = t->phase == 26 ? 2000 + 2 * t->mutation_index : 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");
 }
@@ -378,6 +379,22 @@ static void tick(void* arg) {
         }
         DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: PASS76 mutations, lost response replayed once, busy_responses=%u directories=76",
             t->busy_responses);
+        t->phase = 26; t->mutation_index = 0; mutation_start(t);
+    } else if (t->phase == 26 && t->answered) {
+        if (t->response.status) { fail(t, "bulk removal did not survive busy"); return; }
+        if (t->response.command == MF_STAT) {
+            struct mf_message q = t->response; q.command = MF_REMOVE_PATH; q.request_id++; t->answered = 0;
+            if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) fail(t, "bulk remove start");
+            goto again;
+        }
+        if (++t->mutation_index < 76) { mutation_start(t); goto again; }
+        t->phase = 27; request(t, MF_QUOTA, t->ids[READER]);
+    } else if (t->phase == 27 && 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)) {
+            fail(t, "bulk removal left directories or quota"); return;
+        }
+        DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: PASS76 removals, directories=0, used=0, reserved=0");
         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; }