diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 45d8fc76..0d37c138 100644 --- a/src/member_files/member_files.c +++ b/src/member_files/member_files.c @@ -51,7 +51,7 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c static int mf_message_ready(struct member_files* f, uint64_t group, uint64_t peer, int kind, int command, uint64_t id) { if (!f || f->closing) return 0; if (command == MF_STREAM_DATA) return mf_transfer_receive_ready(f, group, peer, kind, id); - return kind != MF_REQUEST || command > MF_STREAM_READ || f->job_count < MF_MAX_JOBS; + return kind != MF_REQUEST || (command > MF_STREAM_READ && command != MF_QUOTA) || f->job_count < MF_MAX_JOBS; } static int mf_receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) { if (bytes < MF_WIRE_HEADER) return 1; // Ошибку формата логирует декодер сервиса. @@ -133,7 +133,8 @@ int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, con int enabled = chat_setting_get_int(f->inst, "member_files_enabled", 1); int allowed = enabled && member_files_member(f->inst, group, f->inst->node_id, 1) && member_files_member(f->inst, group, peer, 0) && (!q->owner || member_files_member(f->inst, group, q->owner, 0)) && - (q->command < MF_BEGIN_WRITE || q->command == MF_STREAM_READ || (q->owner && q->owner == peer)); + (q->command < MF_BEGIN_WRITE || q->command == MF_STREAM_READ || q->command == MF_QUOTA || + (q->owner && q->owner == peer)); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: server access id=%llu cmd=%s enabled=%d allowed=%d owner_is_peer=%d", (unsigned long long)q->request_id, mf_command_text(q->command), enabled, allowed, q->owner == peer); return allowed; @@ -221,8 +222,9 @@ static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t grou struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t server, const struct mf_message* request, mf_response_fn cb, void* arg) { struct member_files* f = inst ? inst->member_files : NULL; - if (request && (request->command < MF_LIST || request->command > MF_STREAM_READ || request->status || - request->data_len > MF_BLOCK_SIZE || !mf_path_valid(request->path, request->command == MF_LIST || request->command == MF_STAT) || + if (request && (request->command < MF_LIST || (request->command > MF_STREAM_READ && request->command != MF_QUOTA) || request->status || + request->data_len > MF_BLOCK_SIZE || !mf_path_valid(request->path, + request->command == MF_LIST || request->command == MF_STAT || request->command == MF_QUOTA) || (request->command == MF_RENAME && !mf_path_valid(request->target, 0)))) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: invalid client command=%u status=%u data=%u", request->command, request->status, request->data_len); return NULL; @@ -429,7 +431,14 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: rx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx jobs=%u worker=%p", (unsigned long long)m.request_id, m.kind, mf_command_text(m.command), (unsigned long long)group, (unsigned long long)peer, (unsigned long long)m.owner, f->job_count, (void*)f->worker); - if (m.command >= MF_STREAM_DATA || (m.kind == MF_REQUEST && m.command >= MF_STREAM_WRITE)) { + if (m.kind == MF_REQUEST && f->storage_status != MF_OK) { + int status = mf_server_allowed(f, group, peer, &m) ? f->storage_status : MF_ACCESS_DENIED; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: request before storage ready id=%llu status=%s", + (unsigned long long)m.request_id, mf_status_text(status)); + mf_reply(f, group, peer, &m, status); return; + } + if (m.command >= MF_STREAM_WRITE && m.command <= MF_STREAM_CANCEL && + (m.kind == MF_REQUEST || m.command >= MF_STREAM_DATA)) { mf_transfer_dispatch(f, group, peer, &m); return; } @@ -483,6 +492,20 @@ out: if (e) { queue_dgram_free(e); queue_entry_free(e); } } +static void mf_storage_start_work(void* arg) { + struct member_files* f = arg; + f->init_result = mf_storage_initialize(f->storage); +} + +static void mf_storage_start_done(void* arg, int error) { + struct member_files* f = arg; + f->storage_status = error ? MF_IO_ERROR : f->init_result; + if (f->storage_status) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: startup storage unavailable status=%s", + mf_status_text(f->storage_status)); + else f->used_storage = 1; + if (!f->closing) { mf_receive_resume(f); mf_server_pump(f); } +} + int member_files_init(struct UTUN_INSTANCE* inst) { if (!inst || inst->member_files) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: invalid init"); return -1; } struct member_files* f = u_calloc(1, sizeof(*f)); @@ -496,15 +519,17 @@ int member_files_init(struct UTUN_INSTANCE* inst) { if (custom && !*custom) custom = NULL; snprintf(path, sizeof(path), "%s%s%s", custom ? custom : inst->config->global.db_path, custom ? "" : "/", custom ? "" : "member_files"); if (!custom && !*inst->config->global.db_path) snprintf(path, sizeof(path), "member_files"); - f->storage = mf_storage_create(path, (uint64_t)chat_setting_get_int(inst, "member_files_quota_mb", 1024) * 1024 * 1024, + f->storage = mf_storage_create(path, (uint64_t)chat_setting_get_int(inst, "member_files_user_quota_mb", 1024) * 1024 * 1024, (uint64_t)chat_setting_get_int(inst, "member_files_upload_ttl_sec", 3600) * 10000); if (!f->storage || etcp_router_bind(inst, ETCP_RT_ID_MEMBER_FILES, mf_recv)) { mf_storage_destroy(f->storage); mf_worker_destroy(f->disk); mf_worker_unreserve(f->cleanup); u_free(f); return -1; } inst->member_files = f; + f->storage_status = MF_BUSY; etcp_router_set_receive_ready(inst, ETCP_RT_ID_MEMBER_FILES, mf_receive_ready); f->tick = uasync_set_timeout(inst->ua, 10000, f, mf_tick, "member_files"); if (!f->tick) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: init timer failed"); member_files_destroy(inst); return -1; } + mf_worker_submit(f->disk, mf_storage_start_work, f, mf_storage_start_done, f); DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: initialized root=%s idle=10s max_requests=%d max_jobs=%d", path, MF_MAX_PENDING, MF_MAX_JOBS); return 0; } diff --git a/src/member_files/private.h b/src/member_files/private.h index 62772656..b1adb372 100644 --- a/src/member_files/private.h +++ b/src/member_files/private.h @@ -24,6 +24,7 @@ struct member_files { unsigned server_transfer_count; unsigned local_count; int closing, used_storage, sweep_due; + int storage_status, init_result; }; void mf_transfers_quiesce(struct member_files* files, uint64_t group); int mf_transfer_receive_ready(struct member_files* files, uint64_t group, uint64_t peer, int kind, uint64_t id); diff --git a/src/member_files/storage.c b/src/member_files/storage.c index fcf7b649..958fa6c7 100644 --- a/src/member_files/storage.c +++ b/src/member_files/storage.c @@ -214,9 +214,12 @@ unsigned mf_storage_open_count(const struct mf_storage* s) { static int store_upload_discard(struct mf_storage* s, struct store_upload* u) { store_invalidate(s); int rc = mf_disk_remove(&s->root, u->temp, 0); - if (rc && rc != MF_NOT_FOUND) return rc; + if (rc && rc != MF_NOT_FOUND) { + struct mf_disk_info after; + if (mf_disk_stat(&s->root, u->temp, &after) != MF_NOT_FOUND) return rc; + } int updated = store_user_update(u->user, 0, 0, u->reservation, "abort"); u->active = 0; - return rc ? rc : updated; + return rc && rc != MF_NOT_FOUND ? rc : updated; } void mf_storage_sweep(struct mf_storage* s, uint64_t now) { @@ -316,8 +319,21 @@ static int store_begin(struct mf_storage* s, uint64_t group, uint64_t peer, cons int rc = store_expected(s, path, q->version, &existing); if (rc) return rc; if (existing.type && existing.type != MF_FILE) return MF_INVALID_PATH; - if (s->quota && (s->used > s->quota || s->reserved > s->quota - s->used || q->size > s->quota - s->used - s->reserved)) + struct store_user* user = store_user_get(s, q->owner, 1); + if (!user) return MF_IO_ERROR; + if (!user->valid || existing.size > user->used) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: quota inconsistent before upload owner=%llu used=%llu old_size=%llu", + (unsigned long long)q->owner, (unsigned long long)user->used, (unsigned long long)existing.size); + user->valid = 0; return MF_IO_ERROR; + } + uint64_t reservation = q->size > existing.size ? q->size - existing.size : 0; + if (reservation && (user->used > s->quota || user->reserved > s->quota - user->used || + reservation > s->quota - user->used - user->reserved)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: user quota exceeded owner=%llu used=%llu reserved=%llu growth=%llu limit=%llu", + (unsigned long long)q->owner, (unsigned long long)user->used, (unsigned long long)user->reserved, + (unsigned long long)reservation, (unsigned long long)s->quota); return MF_NO_SPACE; + } rc = store_base(s, group, q->owner); if (rc) return rc; memset(slot, 0, sizeof(*slot)); @@ -333,20 +349,32 @@ static int store_begin(struct mf_storage* s, uint64_t group, uint64_t peer, cons if (rc) { mf_disk_close(fd); mf_disk_remove(&s->root, slot->temp, 0); return rc; } cached->fd = fd; cached->touched = get_time_tb(); snprintf(cached->path, sizeof(cached->path), "%s", slot->temp); slot->active = 1; slot->group = group; slot->peer = peer; slot->owner = q->owner; slot->size = q->size; + slot->user = user; slot->previous = existing.size; slot->reservation = reservation; memcpy(slot->id, q->upload, 16); memcpy(slot->hash, q->hash, 32); memcpy(slot->expected, q->version, 16); - snprintf(slot->path, sizeof(slot->path), "%s", path); slot->touched = get_time_tb(); s->reserved += q->size; + snprintf(slot->path, sizeof(slot->path), "%s", path); slot->touched = get_time_tb(); user->reserved += reservation; DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: begin group=%llu owner=%llu path=%s size=%llu reserved=%llu", - (unsigned long long)group, (unsigned long long)q->owner, path, (unsigned long long)q->size, (unsigned long long)s->reserved); + (unsigned long long)group, (unsigned long long)q->owner, path, (unsigned long long)q->size, (unsigned long long)user->reserved); return MF_OK; } static int store_command(struct mf_storage* s, uint64_t group, uint64_t peer, const struct mf_message* q, struct mf_message* r) { - if (!mf_path_valid(q->path, q->command == MF_LIST || q->command == MF_STAT) || + if (!mf_path_valid(q->path, q->command == MF_LIST || q->command == MF_STAT || q->command == MF_QUOTA) || (q->command == MF_RENAME && !mf_path_valid(q->target, 0))) return MF_INVALID_PATH; char path[MF_PATH_MAX + 80]; snprintf(path, sizeof(path), "%llu/%llu%s%s", (unsigned long long)group, (unsigned long long)q->owner, *q->path ? "/" : "", q->path); struct mf_disk_info info = {0}; int rc, fd; + if (q->command == MF_QUOTA) { + if (*q->path || *q->target || q->cursor || q->data_len) return MF_BAD_PACKET; + struct store_user* user = store_user_get(s, q->owner, 0); + if (user && !user->valid) return MF_IO_ERROR; + mf_put64(r->data, user ? user->used : 0); mf_put64(r->data + 8, s->quota); + mf_put64(r->data + 16, user ? user->reserved : 0); r->data_len = MF_QUOTA_BYTES; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: quota response owner=%llu used=%llu reserved=%llu limit=%llu", + (unsigned long long)q->owner, (unsigned long long)(user ? user->used : 0), + (unsigned long long)(user ? user->reserved : 0), (unsigned long long)s->quota); + return MF_OK; + } if (q->command == MF_BEGIN_WRITE) return store_begin(s, group, peer, path, q); if (q->command == MF_WRITE || q->command == MF_COMMIT || q->command == MF_ABORT) { struct store_upload* u = store_upload_find(s, group, peer, q->upload); @@ -367,21 +395,33 @@ static int store_command(struct mf_storage* s, uint64_t group, uint64_t peer, co if (rc) return rc; rc = store_expected(s, path, u->expected, &info); if (rc) return rc; + if (!u->user->valid || info.size != u->previous) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: quota/old size changed before commit owner=%llu", (unsigned long long)q->owner); + return MF_IO_ERROR; + } uint64_t previous = info.size; store_invalidate(s); rc = mf_disk_rename(&s->root, u->temp, path); /* Даже при ошибке fsync rename уже мог произойти; сверяем существование temp. */ struct mf_disk_info temp; int temp_status = mf_disk_stat(&s->root, u->temp, &temp); - if (temp_status != MF_NOT_FOUND) return rc ? rc : temp_status ? temp_status : MF_IO_ERROR; - s->used = s->used - previous + u->size; s->reserved -= u->size; u->active = 0; + if (temp_status != MF_NOT_FOUND) { + if (!rc || temp_status) { + u->user->valid = 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: commit outcome uncertain owner=%llu rename=%s temp_stat=%s", + (unsigned long long)q->owner, mf_status_text(rc), mf_status_text(temp_status)); + } + return rc ? rc : temp_status ? temp_status : MF_IO_ERROR; + } + int updated = store_user_update(u->user, previous, u->size, u->reservation, "commit"); u->active = 0; if (rc) return rc; + if (updated) return updated; rc = mf_disk_stat(&s->root, path, &info); if (rc) return rc; r->size = info.size; r->mtime = info.mtime; r->type = info.type; store_version(s, &info, r->version); memcpy(r->hash, hash, 32); DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: published path=%s size=%llu used=%llu", path, - (unsigned long long)r->size, (unsigned long long)s->used); + (unsigned long long)r->size, (unsigned long long)u->user->used); return MF_OK; } if (q->command == MF_MKDIR) { @@ -430,9 +470,22 @@ static int store_command(struct mf_storage* s, uint64_t group, uint64_t peer, co } store_invalidate(s); if (q->command == MF_REMOVE) { + struct store_user* user = info.type == MF_FILE ? store_user_get(s, q->owner, 1) : NULL; + if (info.type == MF_FILE && !user) return MF_IO_ERROR; rc = mf_disk_remove(&s->root, path, info.type == MF_DIRECTORY); - struct mf_disk_info after; - if (mf_disk_stat(&s->root, path, &after) == MF_NOT_FOUND) s->used -= info.size; + int removed = rc == MF_OK; + if (rc) { + struct mf_disk_info after; int checked = mf_disk_stat(&s->root, path, &after); + removed = checked == MF_NOT_FOUND; + if (user && checked && !removed) { + user->valid = 0; + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: remove outcome uncertain owner=%llu path=%s", (unsigned long long)q->owner, path); + } + } + if (removed && user) { + int updated = store_user_update(user, info.size, 0, 0, "remove"); + if (!rc) rc = updated; + } return rc; } if (q->command == MF_RENAME) { @@ -451,17 +504,24 @@ static int store_command(struct mf_storage* s, uint64_t group, uint64_t peer, co void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, const struct mf_message* q, struct mf_message* r) { *r = *q; r->kind = MF_RESPONSE; r->status = MF_OK; r->data_len = 0; r->more = 0; - if (!group || !peer || !q->owner || (q->command >= MF_BEGIN_WRITE && q->owner != peer)) { + if (!group || !peer || !q->owner || (q->command >= MF_BEGIN_WRITE && q->command != MF_QUOTA && q->owner != peer)) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: storage ACL denied group=%llu peer=%llu owner=%llu command=%u", (unsigned long long)group, (unsigned long long)peer, (unsigned long long)q->owner, q->command); r->status = MF_ACCESS_DENIED; return; } uint64_t now = get_time_tb(); + if (!s->ready) { + r->status = (uint16_t)s->init_status; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: storage not ready root=%s status=%s", s->path, mf_status_text(r->status)); + return; + } + struct store_user* user = store_user_get(s, q->owner, 0); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: storage request id=%llu cmd=%s path=%s offset=%llu size=%llu used=%llu reserved=%llu quota=%llu", (unsigned long long)q->request_id, mf_command_text(q->command), q->path, (unsigned long long)q->offset, - (unsigned long long)q->size, (unsigned long long)s->used, (unsigned long long)s->reserved, (unsigned long long)s->quota); - int mutates = q->command == MF_COMMIT || q->command >= MF_MKDIR; + (unsigned long long)q->size, (unsigned long long)(user ? user->used : 0), + (unsigned long long)(user ? user->reserved : 0), (unsigned long long)s->quota); + int mutates = q->command == MF_COMMIT || q->command == MF_MKDIR || q->command == MF_RENAME || q->command == MF_REMOVE; struct store_receipt* slot = NULL; uint8_t wire[MF_WIRE_MAX], fingerprint[32]; int n = mf_encode(q, wire, sizeof(wire)); @@ -483,22 +543,19 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con } if (mutates && !slot) r->status = MF_BUSY; else { - r->status = (uint16_t)store_initialize(s); - if (!r->status) { - mf_storage_sweep(s, now); r->status = (uint16_t)store_command(s, group, peer, q, r); - /* Idle отсчитывается от завершения I/O, а не от начала медленного read/write. */ - if (!r->status && (q->command == MF_READ || q->command == MF_WRITE || q->command == MF_BEGIN_WRITE)) { - char accessed[MF_PATH_MAX + 80]; - if (q->command == MF_READ) snprintf(accessed, sizeof(accessed), "%llu/%llu/%s", - (unsigned long long)group, (unsigned long long)q->owner, q->path); - else { - struct store_upload* u = store_upload_find(s, group, peer, q->upload); - if (u) { snprintf(accessed, sizeof(accessed), "%s", u->temp); u->touched = get_time_tb(); } - else accessed[0] = 0; - } - for (unsigned i = 0; i < STORE_FILES; i++) if (s->files[i].fd >= 0 && !strcmp(s->files[i].path, accessed)) - s->files[i].touched = get_time_tb(); + mf_storage_sweep(s, now); r->status = (uint16_t)store_command(s, group, peer, q, r); + /* Idle отсчитывается от завершения I/O, а не от начала медленного read/write. */ + if (!r->status && (q->command == MF_READ || q->command == MF_WRITE || q->command == MF_BEGIN_WRITE)) { + char accessed[MF_PATH_MAX + 80]; + if (q->command == MF_READ) snprintf(accessed, sizeof(accessed), "%llu/%llu/%s", + (unsigned long long)group, (unsigned long long)q->owner, q->path); + else { + struct store_upload* u = store_upload_find(s, group, peer, q->upload); + if (u) { snprintf(accessed, sizeof(accessed), "%s", u->temp); u->touched = get_time_tb(); } + else accessed[0] = 0; } + for (unsigned i = 0; i < STORE_FILES; i++) if (s->files[i].fd >= 0 && !strcmp(s->files[i].path, accessed)) + s->files[i].touched = get_time_tb(); } } if (!r->status && mutates) {