From 0c8e9eef3c47110a16e58787dd23f10f03130580 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 17:51:32 +0200 Subject: [PATCH] fix(member-files): retry busy mutations within replay guarantees --- src/member_files/member_files.c | 23 +++++++++++++++++++---- src/member_files/protocol.h | 1 + src/member_files/storage.c | 3 +-- 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 18a15074..2377a090 100644 --- a/src/member_files/member_files.c +++ b/src/member_files/member_files.c @@ -16,12 +16,12 @@ #define MF_MAX_PENDING 16 #define MF_MAX_JOBS 32 -#define MF_REQUEST_TB 1200000ULL struct mf_pending { struct mf_pending* next; struct member_files* files; - uint64_t group, server, started, sent; + uint64_t group, server, started, sent, deadline; + int busy; // Получен явный отказ мутации: допустимо продлить ожидание, сохранив окно квитанции. struct mf_message request; mf_response_fn callback; void* arg; @@ -228,6 +228,7 @@ 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; } @@ -385,7 +386,7 @@ 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->started >= MF_REQUEST_TB) { expired = p; break; } + for (struct mf_pending* p = f->pending; p; p = p->next) if (now >= p->deadline) { 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); @@ -396,7 +397,7 @@ static void mf_tick(void* arg) { struct mf_job* j = f->jobs; while (j) { struct mf_job* next = j->next; - if (j != f->worker && j->created && now - j->created >= MF_REQUEST_TB) { + if (j != f->worker && j->created && now - j->created >= MF_REPLAY_TB) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: server reply/queue timeout peer=%llu", (unsigned long long)j->peer); mf_job_free(j); } j = next; @@ -427,6 +428,20 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: mismatched response id=%llu", (unsigned long long)m.request_id); return; } 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; + 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", + (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); + return; + } mf_pending_finish(p, &m, 0); break; } return; diff --git a/src/member_files/protocol.h b/src/member_files/protocol.h index 45ee9a15..789f0a89 100644 --- a/src/member_files/protocol.h +++ b/src/member_files/protocol.h @@ -29,6 +29,7 @@ int mf_avatar_file_valid(const char* file); int mf_avatar_manifest_valid(const char* file); uint64_t mf_avatar_file_limit(const char* file); #define MF_IDLE_TB 100000ULL /* 10 секунд */ +#define MF_REPLAY_TB 1200000ULL /* Окно повторов запроса/успешной квитанции:120с. */ #define MF_REQUEST 1 #define MF_RESPONSE 2 #define MF_FILE 1 diff --git a/src/member_files/storage.c b/src/member_files/storage.c index 053016e9..528f932b 100644 --- a/src/member_files/storage.c +++ b/src/member_files/storage.c @@ -12,7 +12,6 @@ #define STORE_FILES 64 #define STORE_UPLOADS 16 #define STORE_RECEIPTS 64 -#define RECEIPT_TB 1200000ULL #define STORE_USERS 256 // Windows сравнивает имена без регистра; алиас не должен обходить неизменяемость avatars. @@ -615,7 +614,7 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con } } if (!r->status && mutates) { - slot->group = group; slot->peer = peer; slot->until = now + RECEIPT_TB; + slot->group = group; slot->peer = peer; slot->until = now + MF_REPLAY_TB; memcpy(slot->fingerprint, fingerprint, 32); slot->response = *r; } DEBUG_TRACE(DEBUG_CATEGORY_MEDIA, "member_files: storage result id=%llu cmd=%s status=%s bytes=%u more=%u cursor=%llu",