fix(member-files): retry busy mutations within replay guarantees
This commit is contained in:
1 parent
20b4975d57
commit
0c8e9eef3c
3 files changed
+21
-6
No files matched your search
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user