fix(member-files): bound busy retries by receipt lifetime

src/member_files/member_files.c

@@ -20,8 +20,8 @@
 struct mf_pending {
     struct mf_pending* next;
     struct member_files* files;
-    uint64_t group, server, started, sent, deadline;
-    int busy; // Получен явный отказ мутации: допустимо продлить ожидание, сохранив окно квитанции.
+    uint64_t group, server, started, sent, retry_at;
+    int busy; // Уже залогировано ожидание квитанции; общий срок запроса не продлевается.
     struct mf_message request;
     mf_response_fn callback;
     void* arg;
@@ -188,14 +188,15 @@ 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 (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();
+    if (!rc) { p->sent = get_time_tb(); p->retry_at = 0; }
     else if (rc > 0) etcp_router_on_send_ready(p->files->inst, p->group, p->server, ETCP_RT_ID_MEMBER_FILES,
                                               &p->waiter, mf_client_ready, p);
 }
 
 static void mf_pending_wake(struct mf_pending* p) {
-    if (p->pump || p->files->closing) return;
+    if (p->pump || p->files->closing || get_time_tb() < p->retry_at) return;
     p->pump = uasync_call_soon(p->files->inst->ua, p, mf_pending_pump);
     if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: client pump allocation failed");
 }
@@ -228,7 +229,6 @@ struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t gro
     if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: client allocation failed"); return NULL; }
     p->files = f; p->group = group; p->server = server; p->request = *request; p->request.kind = MF_REQUEST;
     p->callback = cb; p->arg = arg; p->started = get_time_tb();
-    p->deadline = p->started + MF_REPLAY_TB;
     if (!p->request.request_id && (RAND_bytes((uint8_t*)&p->request.request_id, 8) != 1 || !p->request.request_id)) {
         DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: request random failed"); u_free(p); return NULL;
     }
@@ -386,14 +386,14 @@ static void mf_tick(void* arg) {
     mf_receive_resume(f);
     for (;;) {
         struct mf_pending* expired = NULL;
-        for (struct mf_pending* p = f->pending; p; p = p->next) if (now >= p->deadline) { expired = p; break; }
+        for (struct mf_pending* p = f->pending; p; p = p->next) if (now - p->started >= MF_REPLAY_TB) { expired = p; break; }
         if (!expired) break;
         DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: timeout id=%llu cmd=%u",
                    (unsigned long long)expired->request.request_id, expired->request.command);
         mf_pending_finish(expired, NULL, MF_TIMEOUT);
     }
     for (struct mf_pending* p = f->pending; p; p = p->next)
-        if (!p->sent || now - p->sent >= 50000) mf_pending_wake(p);
+        if (p->retry_at ? now >= p->retry_at : !p->sent || now - p->sent >= 50000) mf_pending_wake(p);
     struct mf_job* j = f->jobs;
     while (j) {
         struct mf_job* next = j->next;
@@ -429,17 +429,17 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c
             }
             if (!member_files_member(f->inst, group, peer, 1)) { m.status = MF_ACCESS_DENIED; m.data_len = 0; }
             if (m.status == MF_BUSY && (m.command == MF_COMMIT || (m.command >= MF_MKDIR && m.command <= MF_REMOVE_PATH))) {
-                uint64_t now = get_time_tb(), limit = p->started + 2 * MF_REPLAY_TB;
-                uint64_t deadline = now + MF_REPLAY_TB;
-                p->deadline = deadline < limit ? deadline : limit;
+                uint64_t now = get_time_tb(), delay = m.cursor && m.cursor <= MF_REPLAY_TB ? m.cursor : 50000;
+                p->retry_at = now + delay;
                 p->sent = now;
                 if (p->pump) { uasync_call_soon_cancel(f->inst->ua, p->pump); p->pump = NULL; }
                 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 retry=5s limit=240s",
+                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));
                 p->busy = 1;
-                DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: busy retry id=%llu elapsed_ms=%llu deadline_tb=%llu",
-                    (unsigned long long)m.request_id, (unsigned long long)((now - p->started) / 10), (unsigned long long)p->deadline);
+                DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: busy retry id=%llu elapsed_ms=%llu delay_tb=%llu retry_at=%llu",
+                    (unsigned long long)m.request_id, (unsigned long long)((now - p->started) / 10),
+                    (unsigned long long)delay, (unsigned long long)p->retry_at);
                 return;
             }
             mf_pending_finish(p, &m, 0); break;

src/member_files/member_files.h

@@ -33,8 +33,8 @@ void member_files_cancel_group(struct UTUN_INSTANCE* inst, uint64_t group);
 
 /* Отправить одну команду; request_id=0 выбирает случайный ID. NULL — отказ запуска, callback не вызывается.
  * При успехе callback ровно один, включая отмену; handle освобождён до callback, arg принадлежит вызывающему.
- * Без ответа — timeout120с. BUSY для COMMIT/MKDIR/RENAME/REMOVE сохраняет ID и повторяет запрос каждые5с:
- * явный отказ продлевает окно ожидания до120с от него, суммарно до240с. Остальные BUSY передаются callback. */
+ * Общий timeout120с. BUSY для COMMIT/MKDIR/RENAME/REMOVE сохраняет ID и ждёт подсказанный сервером срок
+ * освобождения квитанции (иначе5с), не продлевая общий timeout. Остальные 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);
 

src/member_files/storage.c

@@ -577,6 +577,8 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
         return;
     }
     int mutates = q->command == MF_COMMIT || q->command == MF_MKDIR || q->command == MF_RENAME || q->command == MF_REMOVE_PATH;
+    uint64_t first_expiry = UINT64_MAX;
+    if (mutates) r->cursor = 0;
     struct store_receipt* slot = NULL;
     uint8_t wire[MF_WIRE_MAX], fingerprint[32];
     int n = mf_encode(q, wire, sizeof(wire));
@@ -585,6 +587,7 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
     for (unsigned i = 0; i < STORE_RECEIPTS; i++) {
         struct store_receipt* p = &s->receipts[i];
         if (p->until <= now) { if (!slot) slot = p; continue; }
+        if (p->until < first_expiry) first_expiry = p->until;
         if (p->group != group || p->peer != peer) continue;
         if (p->response.request_id == q->request_id) {
             if (memcmp(p->fingerprint, fingerprint, 32)) r->status = MF_BAD_PACKET;
@@ -596,7 +599,11 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
             *r = p->response; r->request_id = q->request_id; return;
         }
     }
-    if (mutates && !slot) r->status = MF_BUSY;
+    if (mutates && !slot) {
+        r->status = MF_BUSY; r->cursor = first_expiry - now;
+        DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: receipts full id=%llu retry_after_tb=%llu",
+            (unsigned long long)q->request_id, (unsigned long long)r->cursor);
+    }
     else {
         mf_storage_sweep(s, now); r->status = (uint16_t)store_command(s, group, peer, q, r);
         /* Idle отсчитывается от завершения I/O, а не от начала медленного read/write. */