From e22e66bc4e24ba78441c8758577a78a65cef58a9 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 3 Nov 2026 20:11:31 +0100 Subject: [PATCH] Cancel revoked file streams and separate publication from cleanup --- src/member_files/private.h | 2 + src/member_files/transfer.c | 99 ++++++++++++++++++++++++------------- 2 files changed, 68 insertions(+), 33 deletions(-) diff --git a/src/member_files/private.h b/src/member_files/private.h index 5529e6952..209eee3fd 100644 --- a/src/member_files/private.h +++ b/src/member_files/private.h @@ -37,6 +37,8 @@ int mf_transfer_receive_ready(struct member_files* files, uint64_t group, uint64 void mf_transfer_dispatch(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); /* Завершить потоки пары с IO_ERROR при reset/close router; прочие пары не затрагиваются. */ void mf_transfers_peer_down(struct member_files* files, uint64_t group, uint64_t peer); +/* Отменить передачи группы с отозванным членством/ролью/owner; уже начатый I/O завершает только очистку. */ +void mf_transfers_revalidate(struct member_files* files, uint64_t group); /* Планировать готовые pumps / обработать прогресс и таймауты; все функции вызываются в uasync. */ void mf_transfers_wake(struct member_files* files); void mf_transfers_tick(struct member_files* files, uint64_t now); diff --git a/src/member_files/transfer.c b/src/member_files/transfer.c index 6524bb7d5..81670dc82 100644 --- a/src/member_files/transfer.c +++ b/src/member_files/transfer.c @@ -18,7 +18,7 @@ #define TRANSFER_MAX 16 #define TRANSFER_TIMEOUT_TB 1200000ULL -enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_VERIFY, TRANSFER_FINISH }; +enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_VERIFY, TRANSFER_PUBLISH, TRANSFER_FINISH }; struct mf_chunk { uint32_t bytes; uint8_t data[MF_BLOCK_SIZE]; }; struct mf_transfer { struct mf_transfer* next; @@ -37,6 +37,7 @@ struct mf_transfer { int fd, server, sending, starting, active, opened, busy, stopping, finished_io, peer_down; int status, work_status, reply_ready, reply_final, verified, committed, end_sent, end_received; int temp_owned; /* Только worker: локальный download-temp успешно создан через O_EXCL. */ + int cleanup_abort; /* Неизменяемый снимок для обязательной очистки; worker не читает живой status. */ enum transfer_work work; uint64_t size, offset, received, read_offset, io_offset, io_bytes, io_size, progress_tb, last_event, last_log, logged_offset; unsigned head, count, io_head, io_count; @@ -84,7 +85,8 @@ static struct mf_message transfer_storage_request(struct mf_transfer* t, uint8_t static int transfer_allowed(struct mf_transfer* t) { struct UTUN_INSTANCE* inst = t->files->inst; if (!t->server) return member_files_member(inst, t->request.group, inst->node_id, 0) && - member_files_member(inst, t->request.group, t->request.server, 1); + member_files_member(inst, t->request.group, t->request.server, 1) && + member_files_member(inst, t->request.group, t->request.owner, 0); struct mf_message q = t->open; return mf_server_allowed(t->files, t->request.group, t->request.server, &q); } @@ -190,28 +192,30 @@ static void transfer_work(void* arg) { if (!rc && memcmp(&after, &t->source, sizeof(after))) rc = MF_STALE; } } + } else if (t->work == TRANSFER_PUBLISH) { + rc = mf_disk_sync(t->fd); + int closed = mf_disk_close(t->fd); t->fd = -1; + if (!rc) rc = closed; + if (!rc) rc = mf_disk_local_publish(t->temp, t->request.local); + if (!rc) t->temp_owned = 0; + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download publish id=%llu status=%s path=%s", + (unsigned long long)t->request.id, mf_status_text(rc), t->request.local); } else { - if (t->server && !t->sending && t->status && !t->committed) { + if (t->cleanup_abort) { struct mf_message q = transfer_storage_request(t, MF_ABORT); mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &t->result); if (t->result.status) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream cleanup failed id=%llu status=%s", (unsigned long long)t->request.id, mf_status_text(t->result.status)); } if (t->fd >= 0) { - if (!t->status && !t->sending) rc = mf_disk_sync(t->fd); - int closed = mf_disk_close(t->fd); t->fd = -1; - if (!rc) rc = closed; + rc = mf_disk_close(t->fd); t->fd = -1; } if (!t->server && !t->sending && t->temp_owned) { - if (!t->status && !rc) rc = mf_disk_local_publish(t->temp, t->request.local); - if (t->status || rc) { - int removed = mf_disk_local_remove(t->temp); - if (removed) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: owned temp cleanup failed id=%llu path=%s status=%s", - (unsigned long long)t->request.id, t->temp, mf_status_text(removed)); - else t->temp_owned = 0; - } else t->temp_owned = 0; - DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download temp finish id=%llu owned=%d status=%s io_status=%s path=%s", - (unsigned long long)t->request.id, t->temp_owned, mf_status_text(t->status), mf_status_text(rc), t->temp); + int removed = mf_disk_local_remove(t->temp); + if (removed) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: owned temp cleanup failed id=%llu path=%s status=%s", + (unsigned long long)t->request.id, t->temp, mf_status_text(removed)); + else t->temp_owned = 0; + if (!rc) rc = removed; } EVP_MD_CTX_free(t->digest); t->digest = NULL; } @@ -221,6 +225,16 @@ static void transfer_work(void* arg) { static void transfer_done(void* arg, int error) { struct mf_transfer* t = arg; t->busy = 0; int status = error ? MF_CANCELLED : t->work_status; + if (!status && !transfer_allowed(t)) { + status = MF_ACCESS_DENIED; t->peer_down = 1; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: disk result revoked group=%llx peer=%llx id=%llu work=%d", + (unsigned long long)t->request.group, (unsigned long long)t->request.server, + (unsigned long long)t->request.id, t->work); + } + if (t->work == TRANSFER_PUBLISH) { + if (!t->status) t->status = status; + transfer_submit(t, TRANSFER_FINISH); return; + } if (t->work == TRANSFER_FINISH) { t->finished_io = 1; if (!t->status) t->status = status; @@ -255,33 +269,36 @@ static void transfer_done(void* arg, int error) { } static void transfer_submit(struct mf_transfer* t, enum transfer_work work) { + if (work != TRANSFER_FINISH && !transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; } t->work = work; t->busy = 1; if (work == TRANSFER_FINISH) { + t->cleanup_abort = t->server && !t->sending && t->status && !t->committed; struct mf_work* cleanup = t->cleanup; t->cleanup = NULL; mf_worker_submit_reserved(t->files->disk, cleanup, transfer_work, t, transfer_done, t); } else mf_worker_submit(t->files->disk, transfer_work, t, transfer_done, t); } static void transfer_stop(struct mf_transfer* t, int status) { - if (t->stopping) { - if ((t->files->closing || status) && t->finished_io && !t->busy) transfer_release(t); - else if (!t->busy && !t->finished_io) transfer_submit(t, TRANSFER_FINISH); - return; - } if (status && !t->status) { t->status = status; DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream stopped id=%llu peer=%llx reason=%s", (unsigned long long)t->request.id, (unsigned long long)t->request.server, mf_status_text(status)); } + if (t->stopping) { + if (t->busy && t->work != TRANSFER_FINISH) mf_worker_cancel(t->files->disk, t); + else if ((t->files->closing || status) && t->finished_io && !t->busy) transfer_release(t); + else if (!t->busy && !t->finished_io) transfer_submit(t, TRANSFER_FINISH); + return; + } DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: stream stopping id=%llu starting=%d active=%d io=%d pending=%d", (unsigned long long)t->request.id, t->starting, t->active, t->busy, t->pending != NULL); t->stopping = 1; t->starting = 0; if (t->pump) { uasync_call_soon_cancel(t->files->inst->ua, t->pump); t->pump = NULL; } etcp_router_cancel_send_ready(t->files->inst, t->request.group, t->request.server, ETCP_RT_ID_MEMBER_FILES, &t->waiter); if (t->pending) { struct mf_pending* p = t->pending; t->pending = NULL; member_files_cancel(p); } - if (t->busy) return; + if (t->busy) { if (t->work != TRANSFER_FINISH) mf_worker_cancel(t->files->disk, t); return; } if (t->finished_io) { transfer_release(t); return; } - transfer_submit(t, TRANSFER_FINISH); + transfer_submit(t, !t->status && !t->server && !t->sending ? TRANSFER_PUBLISH : TRANSFER_FINISH); } static int transfer_enqueue(struct mf_transfer* t, struct mf_message* q) { @@ -295,6 +312,7 @@ static int transfer_enqueue(struct mf_transfer* t, struct mf_message* q) { static void transfer_pump(void* arg) { struct mf_transfer* t = arg; t->pump = NULL; if (t->files->closing) return; + if (!t->stopping && !transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; } if (t->starting) { t->starting = 0; if (t->sending || t->server) transfer_submit(t, TRANSFER_PREPARE); @@ -402,7 +420,7 @@ struct mf_transfer* member_files_transfer_start(const struct mf_transfer_request if (!f || f->closing || !callback || f->transfer_count >= TRANSFER_MAX || !q->group || !q->server || !q->owner || !mf_path_valid(q->remote, 0) || !*q->local || strnlen(q->local, sizeof(q->local)) > MF_PATH_MAX || (q->upload && q->owner != q->inst->node_id) || !member_files_member(q->inst, q->group, q->inst->node_id, 0) || - !member_files_member(q->inst, q->group, q->server, 1)) { + !member_files_member(q->inst, q->group, q->server, 1) || !member_files_member(q->inst, q->group, q->owner, 0)) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream start denied/limit reached"); return NULL; } struct mf_transfer_request request = *q; @@ -526,20 +544,35 @@ void mf_transfers_tick(struct member_files* f, uint64_t now) { void member_files_transfer_cancel(struct mf_transfer* t) { if (t && !t->stopping) transfer_stop(t, MF_CANCELLED); } void mf_transfers_quiesce(struct member_files* f, uint64_t group) { - struct mf_transfer* t = f->transfers; - while (t) { - struct mf_transfer* next = t->next; - if (!group || t->request.group == group) { t->peer_down = 1; transfer_stop(t, MF_CANCELLED); } - t = next; + for (;;) { + struct mf_transfer* found = NULL; + for (struct mf_transfer* t = f->transfers; t; t = t->next) + if (!t->peer_down && (!group || t->request.group == group)) { found = t; break; } + if (!found) break; + found->peer_down = 1; transfer_stop(found, MF_CANCELLED); } } void mf_transfers_peer_down(struct member_files* f, uint64_t group, uint64_t peer) { - struct mf_transfer* t = f->transfers; - while (t) { - struct mf_transfer* next = t->next; - if (t->request.group == group && t->request.server == peer) { t->peer_down = 1; transfer_stop(t, MF_IO_ERROR); } - t = next; + for (;;) { + struct mf_transfer* found = NULL; + for (struct mf_transfer* t = f->transfers; t; t = t->next) + if (!t->peer_down && t->request.group == group && t->request.server == peer) { found = t; break; } + if (!found) break; + found->peer_down = 1; transfer_stop(found, MF_IO_ERROR); + } +} + +void mf_transfers_revalidate(struct member_files* f, uint64_t group) { + for (;;) { + struct mf_transfer* denied = NULL; + for (struct mf_transfer* t = f->transfers; t; t = t->next) + if (!t->peer_down && t->request.group == group && !transfer_allowed(t)) { denied = t; break; } + if (!denied) break; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: revoke stream group=%llx peer=%llx id=%llu io=%d", + (unsigned long long)group, (unsigned long long)denied->request.server, + (unsigned long long)denied->request.id, denied->busy); + denied->peer_down = 1; transfer_stop(denied, MF_ACCESS_DENIED); } }