Initialize per-user file quota accounts from the storage tree

src/member_files/storage.c

@@ -12,10 +12,13 @@
 #define STORE_UPLOADS 16
 #define STORE_RECEIPTS 64
 #define RECEIPT_TB 1200000ULL
+#define STORE_USERS 256
 
 struct store_file { char path[MF_PATH_MAX + 80]; int fd; uint64_t touched; };
+struct store_user { struct store_user* next; uint64_t owner, used, reserved; int valid; };
 struct store_upload {
-    uint64_t group, peer, owner, size, touched;
+    uint64_t group, peer, owner, size, previous, reservation, touched;
+    struct store_user* user;
     uint8_t id[16], hash[32], expected[16];
     char path[MF_PATH_MAX + 80], temp[64];
     int active;
@@ -28,14 +31,58 @@ struct store_receipt {
 struct mf_storage {
     char path[2048];
     struct mf_disk_root root;
-    int ready;
+    int ready, init_status;
     uint8_t epoch[16];
-    uint64_t generation, quota, used, reserved, upload_ttl;
+    uint64_t generation, quota, upload_ttl;
+    unsigned user_count;
+    struct store_user* users[STORE_USERS];
     struct store_file files[STORE_FILES];
     struct store_upload uploads[STORE_UPLOADS];
     struct store_receipt receipts[STORE_RECEIPTS];
 };
 
+static struct store_user* store_user_get(struct mf_storage* s, uint64_t owner, int create) {
+    unsigned bucket = (unsigned)(owner ^ (owner >> 32)) % STORE_USERS;
+    for (struct store_user* u = s->users[bucket]; u; u = u->next) if (u->owner == owner) return u;
+    if (!create) return NULL;
+    struct store_user* u = u_calloc(1, sizeof(*u));
+    if (!u) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: quota account allocation failed owner=%llu", (unsigned long long)owner); return NULL; }
+    u->owner = owner; u->valid = 1; u->next = s->users[bucket]; s->users[bucket] = u; s->user_count++;
+    return u;
+}
+
+static void store_users_free(struct mf_storage* s) {
+    for (unsigned i = 0; i < STORE_USERS; i++) {
+        while (s->users[i]) { struct store_user* next = s->users[i]->next; u_free(s->users[i]); s->users[i] = next; }
+    }
+    s->user_count = 0;
+}
+
+/* Один worker изменяет размер и резерв вместе. Повторы квитанций сюда не попадают. */
+static int store_user_update(struct store_user* u, uint64_t previous, uint64_t size, uint64_t release, const char* operation) {
+    if (!u->valid || previous > u->used || release > u->reserved || size > UINT64_MAX - (u->used - previous)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_MEDIA,
+            "member_files: quota accounting invalid owner=%llu operation=%s used=%llu previous=%llu size=%llu reserved=%llu release=%llu",
+            (unsigned long long)u->owner, operation, (unsigned long long)u->used, (unsigned long long)previous,
+            (unsigned long long)size, (unsigned long long)u->reserved, (unsigned long long)release);
+        u->valid = 0; return MF_IO_ERROR;
+    }
+    u->used = u->used - previous + size; u->reserved -= release;
+    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: quota updated owner=%llu operation=%s used=%llu reserved=%llu",
+        (unsigned long long)u->owner, operation, (unsigned long long)u->used, (unsigned long long)u->reserved);
+    return MF_OK;
+}
+
+static int store_scan_id(const char* name, uint64_t* id) {
+    if (!*name || *name == '0') return 0;
+    *id = 0;
+    for (const char* p = name; *p; p++) {
+        if (*p < '0' || *p > '9' || *id > (UINT64_MAX - (unsigned)(*p - '0')) / 10) return 0;
+        *id = *id * 10 + (unsigned)(*p - '0');
+    }
+    return 1;
+}
+
 /* Версия переживает закрытие fd, но не restart. Метаданные снимка — непрозрачные байты. */
 static void store_version(struct mf_storage* s, const struct mf_disk_info* i, uint8_t out[16]) {
     uint8_t bytes[81], hash[32];
@@ -70,9 +117,15 @@ static int store_fd(struct mf_storage* s, const char* path, int write, int* fd)
     return MF_OK;
 }
 
-struct scan_context { struct mf_storage* s; char path[MF_PATH_MAX + 80]; unsigned depth; int cleanup; };
+struct scan_context {
+    struct mf_storage* s;
+    struct store_user* user;
+    char path[MF_PATH_MAX + 80];
+    unsigned depth;
+    int cleanup;
+};
 
-/* Обход не следует ссылкам. Резервации равны конечному размеру sparse temp-файла. */
+/* Размеры UID суммируются по всем группам. Каталоги и .uploads не занимают пользовательскую квоту. */
 static int store_scan_entry(void* arg, const char* name, const struct mf_disk_info* info) {
     struct scan_context* c = arg;
     char child[MF_PATH_MAX + 80];
@@ -84,37 +137,62 @@ static int store_scan_entry(void* arg, const char* name, const struct mf_disk_in
         return rc ? -rc : 0;
     }
     if (!strcmp(name, ".uploads") && !*c->path) return 0;
+    struct store_user* user = c->user;
+    if (c->depth < 2) {
+        uint64_t id;
+        if (info->type != MF_DIRECTORY || !store_scan_id(name, &id)) {
+            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: unexpected storage namespace path=%s", child);
+            return -MF_INVALID_PATH;
+        }
+        if (c->depth == 1 && !(user = store_user_get(c->s, id, 1))) return -MF_IO_ERROR;
+    }
     if (info->type == MF_FILE) {
-        if (info->size > UINT64_MAX - c->s->used) return -MF_NO_SPACE;
+        if (info->size > UINT64_MAX - user->used) {
-        c->s->used += info->size; return 0;
+            DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: scan size overflow owner=%llu", (unsigned long long)user->owner);
+            return -MF_IO_ERROR;
+        }
+        user->used += info->size; return 0;
     }
     if (c->depth >= 64) return -MF_INVALID_PATH;
-    struct scan_context next = { .s = c->s, .depth = c->depth + 1 };
+    struct scan_context next = { .s = c->s, .user = user, .depth = c->depth + 1 };
     snprintf(next.path, sizeof(next.path), "%s", child);
     int rc = mf_disk_list(&c->s->root, child, store_scan_entry, &next);
     return rc ? -rc : 0;
 }
 
-static int store_initialize(struct mf_storage* s) {
+int mf_storage_initialize(struct mf_storage* s) {
     if (s->ready) return MF_OK;
+    if (s->init_status != MF_BUSY) return s->init_status;
+    uint64_t started = get_time_tb();
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: quota scan started root=%s user_quota=%llu", s->path, (unsigned long long)s->quota);
     int rc = mf_disk_root_open(&s->root, s->path);
-    if (rc) return rc;
+    if (rc) goto failed;
     rc = mf_disk_mkdir(&s->root, ".uploads");
     if (rc == MF_EXISTS) rc = MF_OK;
     struct scan_context c = { .s = s, .cleanup = 1 };
     snprintf(c.path, sizeof(c.path), ".uploads");
     if (!rc) rc = mf_disk_list(&s->root, c.path, store_scan_entry, &c);
-    c.cleanup = 0; c.path[0] = 0; s->used = 0;
+    c.cleanup = 0; c.path[0] = 0;
     if (!rc) rc = mf_disk_list(&s->root, "", store_scan_entry, &c);
-    if (rc) { mf_disk_root_close(&s->root); return rc; }
+    if (rc) { mf_disk_root_close(&s->root); goto failed; }
-    s->ready = 1;
+    s->ready = 1; s->init_status = MF_OK;
-    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: storage ready root=%s used=%llu quota=%llu", s->path,
+    for (unsigned i = 0; i < STORE_USERS; i++) for (struct store_user* u = s->users[i]; u; u = u->next) {
-               (unsigned long long)s->used, (unsigned long long)s->quota);
+        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: quota scan owner=%llu used=%llu limit=%llu",
+            (unsigned long long)u->owner, (unsigned long long)u->used, (unsigned long long)s->quota);
+        if (u->used > s->quota) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: user exceeds quota owner=%llu used=%llu limit=%llu",
+            (unsigned long long)u->owner, (unsigned long long)u->used, (unsigned long long)s->quota);
+    }
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: quota scan ready root=%s users=%u elapsed_ms=%llu", s->path, s->user_count,
+        (unsigned long long)((get_time_tb() - started) / 10));
     return MF_OK;
+failed:
+    store_users_free(s); s->init_status = rc;
+    DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: quota scan failed root=%s status=%s", s->path, mf_status_text(rc));
+    return rc;
 }
 
 struct mf_storage* mf_storage_create(const char* root, uint64_t quota, uint64_t ttl) {
-    if (!root || !*root || strlen(root) >= 2048 || !ttl) {
+    if (!root || !*root || strlen(root) >= 2048 || !quota || !ttl) {
         DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: invalid storage settings root=%s quota=%llu ttl=%llu",
                     root ? root : "(null)", (unsigned long long)quota, (unsigned long long)ttl); return NULL;
     }
@@ -122,7 +200,7 @@ struct mf_storage* mf_storage_create(const char* root, uint64_t quota, uint64_t
     if (!s || RAND_bytes(s->epoch, 16) != 1) {
         DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: storage allocation/random failed"); u_free(s); return NULL;
     }
-    snprintf(s->path, sizeof(s->path), "%s", root); s->quota = quota; s->upload_ttl = ttl;
+    snprintf(s->path, sizeof(s->path), "%s", root); s->quota = quota; s->upload_ttl = ttl; s->init_status = MF_BUSY;
     for (unsigned i = 0; i < STORE_FILES; i++) s->files[i].fd = -1;
     return s;
 }
@@ -137,8 +215,8 @@ 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;
-    s->reserved -= u->size; u->active = 0;
+    int updated = store_user_update(u->user, 0, 0, u->reservation, "abort"); u->active = 0;
-    return MF_OK;
+    return rc ? rc : updated;
 }
 
 void mf_storage_sweep(struct mf_storage* s, uint64_t now) {
@@ -167,8 +245,8 @@ void mf_storage_destroy(struct mf_storage* s) {
         for (unsigned i = 0; i < STORE_UPLOADS; i++) if (s->uploads[i].active) store_upload_discard(s, &s->uploads[i]);
         mf_disk_root_close(&s->root);
     }
-    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: storage stopped used=%llu reserved=%llu",
+    DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: storage stopped users=%u", s->user_count);
-               (unsigned long long)s->used, (unsigned long long)s->reserved);
+    store_users_free(s);
     u_free(s);
 }
 

src/member_files/storage.h

@@ -1,6 +1,6 @@
 /* Серверное хранилище. Вызывать execute/sweep/destroy последовательно, только без
- * одновременно работающего worker. Создание не обращается к диску; корень открывается
+ * одновременно работающего worker. create не обращается к диску; initialize выполняет
- * лениво. ACL проверяет member_files.c перед каждой работой. Файлы временных загрузок
+ * стартовый обход в worker до запросов. ACL проверяет member_files.c перед каждой работой. Файлы временных загрузок
  * закрываются после 10 секунд простоя, удаляются после upload_ttl; restart отменяет их.
  * Открытые regular-файлы кешируются, максимум 64. Родительские пути не следуют ссылкам. */
 #ifndef MEMBER_FILES_STORAGE_H
@@ -8,6 +8,8 @@
 #include "protocol.h"
 struct mf_storage;
 struct mf_storage* mf_storage_create(const char* root, uint64_t quota, uint64_t upload_ttl_tb);
+/* Однократный стартовый обход. До его завершения execute возвращает BUSY, после ошибки — её код. */
+int mf_storage_initialize(struct mf_storage* storage);
 void mf_storage_destroy(struct mf_storage* storage);
 void mf_storage_sweep(struct mf_storage* storage, uint64_t now);
 void mf_storage_execute(struct mf_storage* storage, uint64_t group, uint64_t peer,