jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Enforce per-user reservations and initialize storage at startup

master
evgeny 1 week ago
parent
3b641a2423
commit
359a327ec4
3 changed files with 118 additions and 35 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 37
      src/member_files/member_files.c
  2. 1
      src/member_files/private.h
  3. 115
      src/member_files/storage.c

37
src/member_files/member_files.c
Unescape View File

@ -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;
}

1
src/member_files/private.h
Unescape View File

@ -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);

115
src/member_files/storage.c
Unescape View File

@ -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) {

Write Preview
Loading…
Cancel
Save