diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 2f969d8e..d9e08fb0 100644 --- a/src/member_files/member_files.c +++ b/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)); diff --git a/src/member_files/protocol.h b/src/member_files/protocol.h index 789f0a89..ec068e7b 100644 --- a/src/member_files/protocol.h +++ b/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 diff --git a/tests/test_member_files_network.c b/tests/test_member_files_network.c index 263a17b1..1273ac30 100644 --- a/tests/test_member_files_network.c +++ b/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; }