diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 2377a090..2f969d8e 100644 --- a/src/member_files/member_files.c +++ b/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; diff --git a/src/member_files/member_files.h b/src/member_files/member_files.h index 961b5f56..ab150b79 100644 --- a/src/member_files/member_files.h +++ b/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); diff --git a/src/member_files/storage.c b/src/member_files/storage.c index 528f932b..115c3a33 100644 --- a/src/member_files/storage.c +++ b/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. */