diff --git a/tests/test_member_files.c b/tests/test_member_files.c index dc1799a9..22aae627 100644 --- a/tests/test_member_files.c +++ b/tests/test_member_files.c @@ -38,10 +38,13 @@ static void protocol_test(void) { static void storage_test(const char* directory) { struct mf_storage* s = mf_storage_create(directory, 40000, 200000); EXPECT(s != NULL, "create storage"); if (!s) return; + memset(&q, 0, sizeof(q)); q.owner = 77; + run(s, 55, 77, MF_QUOTA, MF_BUSY); + EXPECT(!mf_storage_initialize(s), "startup scan"); uint8_t bytes[10001], version[16], upload[16]; for (unsigned i = 0; i < sizeof(bytes); i++) bytes[i] = (uint8_t)(i * 73 + 19); memset(&q, 0, sizeof(q)); q.owner = 77; - run(s, 55, 77, MF_LIST, MF_OK); EXPECT(!r.data_len, "empty lazy root"); + run(s, 55, 77, MF_LIST, MF_OK); EXPECT(!r.data_len, "empty initialized root"); snprintf(q.path, sizeof(q.path), "пример.webp"); q.size = sizeof(bytes); memset(q.upload, 9, 16); memcpy(upload, q.upload, 16); SHA256(bytes, sizeof(bytes), q.hash); run(s, 55, 88, MF_BEGIN_WRITE, MF_ACCESS_DENIED); @@ -102,6 +105,7 @@ static void storage_test(const char* directory) { run(s, 55, 77, MF_WRITE, MF_NOT_FOUND); mf_storage_destroy(s); s = mf_storage_create(directory, 40000, 200000); + EXPECT(s && !mf_storage_initialize(s), "restart scan"); snprintf(q.path, sizeof(q.path), "empty"); q.data_len = 0; run(s, 55, 88, MF_STAT, MF_OK); EXPECT(r.size == 0, "restart retains published files"); memset(&q, 0, sizeof(q)); q.owner = 77; for (unsigned i = 0; i < 63; i++) { @@ -150,6 +154,92 @@ static void worker_work(void* arg) { static void worker_done(void* arg, int error) { struct worker_test* t = arg; t->finished = !error && pthread_equal(t->caller, pthread_self()); uasync_stop(t->ua); } +static void check_quota(struct mf_storage* s, uint64_t group, uint64_t owner, uint64_t used, uint64_t reserved, uint64_t limit) { + struct mf_message request = { .command = MF_QUOTA, .kind = MF_REQUEST, .owner = owner, .request_id = ++sequence }, response; + mf_storage_execute(s, group, owner, &request, &response); + EXPECT(response.status == MF_OK && response.data_len == MF_QUOTA_BYTES, "quota response"); + EXPECT(mf_get64(response.data) == used && mf_get64(response.data + 8) == limit && mf_get64(response.data + 16) == reserved, + "used/limit/reservation snapshot"); +} + +static void quota_begin(struct mf_storage* s, uint64_t group, uint64_t owner, const char* name, + unsigned size, unsigned token, const uint8_t* version, int status) { + memset(&q, 0, sizeof(q)); q.owner = owner; q.size = size; + memset(q.upload, token, 16); snprintf(q.path, sizeof(q.path), "%s", name); + if (version) memcpy(q.version, version, 16); + uint8_t bytes[32] = {0}; SHA256(bytes, size, q.hash); + run(s, group, owner, MF_BEGIN_WRITE, status); +} + +static void quota_commit(struct mf_storage* s, uint64_t group) { + q.data_len = (uint32_t)q.size; memset(q.data, 0, q.data_len); + run(s, group, q.owner, MF_WRITE, MF_OK); + q.data_len = 0; run(s, group, q.owner, MF_COMMIT, MF_OK); +} + +static void quota_test(const char* directory) { + char root[1200]; snprintf(root, sizeof(root), "%s/quota", directory); + struct mf_storage* s = mf_storage_create(root, 40, 200000); + EXPECT(s && !mf_storage_initialize(s), "quota storage startup"); if (!s) return; + check_quota(s, 10, 101, 0, 0, 40); + quota_begin(s, 10, 101, "a", 30, 31, NULL, MF_OK); quota_commit(s, 10); + uint8_t version[16]; memcpy(version, r.version, 16); + quota_begin(s, 10, 202, "b", 30, 32, NULL, MF_OK); quota_commit(s, 10); + check_quota(s, 10, 101, 30, 0, 40); check_quota(s, 10, 202, 30, 0, 40); + quota_begin(s, 20, 101, "pending-a", 8, 33, NULL, MF_OK); + struct mf_message pending = q; + check_quota(s, 10, 101, 30, 8, 40); + quota_begin(s, 10, 101, "overflow", 8, 34, NULL, MF_NO_SPACE); + quota_begin(s, 20, 202, "pending-b", 8, 34, NULL, MF_OK); + check_quota(s, 10, 202, 30, 8, 40); + run(s, 20, 202, MF_ABORT, MF_OK); run(s, 20, 202, MF_ABORT, MF_OK); + check_quota(s, 20, 202, 30, 0, 40); + q = pending; run(s, 20, 101, MF_BEGIN_WRITE, MF_OK); + check_quota(s, 20, 101, 30, 8, 40); quota_commit(s, 20); + check_quota(s, 10, 101, 38, 0, 40); + quota_begin(s, 10, 101, "a", 20, 35, version, MF_OK); + check_quota(s, 10, 101, 38, 0, 40); quota_commit(s, 10); + memcpy(version, r.version, 16); + mf_storage_execute(s, 10, 101, &q, &r); EXPECT(r.status == MF_OK, "commit replay retains quota"); + check_quota(s, 20, 101, 28, 0, 40); + quota_begin(s, 10, 101, "a", 30, 36, version, MF_OK); + check_quota(s, 20, 101, 28, 10, 40); + pending = q; quota_begin(s, 20, 101, "overflow", 3, 37, NULL, MF_NO_SPACE); + q = pending; run(s, 10, 101, MF_ABORT, MF_OK); + check_quota(s, 20, 101, 28, 0, 40); + memset(&q, 0, sizeof(q)); q.owner = 101; memcpy(q.version, version, 16); + snprintf(q.path, sizeof(q.path), "a"); snprintf(q.target, sizeof(q.target), "renamed"); + run(s, 10, 101, MF_RENAME, MF_OK); check_quota(s, 10, 101, 28, 0, 40); + snprintf(q.path, sizeof(q.path), "renamed"); memset(q.version, 0, 16); + run(s, 10, 101, MF_STAT, MF_OK); memcpy(q.version, r.version, 16); + run(s, 10, 101, MF_REMOVE, MF_OK); + mf_storage_execute(s, 10, 101, &q, &r); EXPECT(r.status == MF_OK, "remove replay retains quota"); + check_quota(s, 20, 101, 8, 0, 40); + memset(&q, 0, sizeof(q)); q.owner = 101; snprintf(q.path, sizeof(q.path), "folder"); + run(s, 10, 101, MF_MKDIR, MF_OK); run(s, 10, 101, MF_STAT, MF_OK); memcpy(q.version, r.version, 16); + run(s, 10, 101, MF_REMOVE, MF_OK); check_quota(s, 10, 101, 8, 0, 40); + quota_begin(s, 10, 101, "expired", 6, 38, NULL, MF_OK); + check_quota(s, 20, 101, 8, 6, 40); mf_storage_sweep(s, get_time_tb() + 200001); + check_quota(s, 10, 101, 8, 0, 40); run(s, 10, 101, MF_ABORT, MF_OK); + quota_begin(s, 10, 101, "persist", 12, 39, NULL, MF_OK); quota_commit(s, 10); + for (unsigned i = 0; i < 80; i++) check_quota(s, 20, 101, 20, 0, 40); + mf_storage_destroy(s); + char abandoned[1400]; snprintf(abandoned, sizeof(abandoned), "%s/.uploads/abcdef0123456789abcdef0123456789", root); + FILE* file = fopen(abandoned, "wb"); EXPECT(file != NULL, "abandoned upload fixture"); + if (file) { fputs("unfinished", file); fclose(file); } + s = mf_storage_create(root, 10, 200000); + EXPECT(s && !mf_storage_initialize(s), "rescan with smaller per-user quota"); if (!s) return; + check_quota(s, 10, 101, 20, 0, 10); check_quota(s, 20, 202, 30, 0, 10); + file = fopen(abandoned, "rb"); EXPECT(!file, "startup removes abandoned temp"); if (file) fclose(file); + quota_begin(s, 20, 101, "over-limit", 1, 40, NULL, MF_NO_SPACE); + memset(&q, 0, sizeof(q)); q.owner = 101; snprintf(q.path, sizeof(q.path), "persist"); + run(s, 10, 101, MF_STAT, MF_OK); memcpy(version, r.version, 16); + quota_begin(s, 10, 101, "persist", 2, 41, version, MF_OK); quota_commit(s, 10); + check_quota(s, 20, 101, 10, 0, 10); + quota_begin(s, 20, 101, "at-limit", 1, 42, NULL, MF_NO_SPACE); + check_quota(s, 10, 202, 30, 0, 10); mf_storage_destroy(s); +} + static void worker_test(void) { struct worker_test t = {0}; t.ua = uasync_create(); t.caller = pthread_self(); t.worker = mf_worker_create(t.ua); EXPECT(t.worker != NULL, "worker start"); @@ -166,7 +256,7 @@ int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG); char directory[1024] = "/tmp/utun_member_files_XXXXXX"; if (test_mkdtemp(directory)) return 1; - protocol_test(); storage_test(directory); worker_test(); + protocol_test(); storage_test(directory); quota_test(directory); worker_test(); #ifndef _WIN32 large_offset_test(directory); #endif diff --git a/tests/test_member_files_network.c b/tests/test_member_files_network.c index b02a56c2..0c1c5753 100644 --- a/tests/test_member_files_network.c +++ b/tests/test_member_files_network.c @@ -142,7 +142,7 @@ static int start(struct network_test* t, int index, int port) { FILE* f = fopen(path, "w"); if (!f) return -1; fprintf(f, "[global]\nmy_private_key=%s\nmy_public_key=%s\ndb_path=%s\n" "[server:s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n" - "[chatserver]\nstorage_autoload=0\nmember_files_quota_mb=8\n", priv, pub, db, port + index); + "[chatserver]\nstorage_autoload=0\nmember_files_user_quota_mb=8\n", priv, pub, db, port + index); if (index != SERVER) fprintf(f, "[client:to_m]\npeer_public_key=%s\nlink=s1:127.0.0.1:%d\n", server, port + SERVER); fclose(f); t->nodes[index] = utun_instance_create(t->ua, path); @@ -184,7 +184,7 @@ static void transfer(struct network_test* t, int upload, uint64_t id) { static void request(struct network_test* t, uint8_t command, uint64_t owner) { struct mf_message q = { .command = command, .owner = owner }; - if (owner) snprintf(q.path, sizeof(q.path), "file.bin"); + if (owner && command != MF_QUOTA) snprintf(q.path, sizeof(q.path), "file.bin"); t->answered = 0; if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) fail(t, "request start"); } @@ -201,12 +201,18 @@ static void tick(void* arg) { if (t->phase == 0) { int a = etcp_router_get_path(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; int b = etcp_router_get_path(t->nodes[READER], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; - if (a && b) { t->phase = 1; transfer(t, 1, 1); } + if (a && b && t->nodes[SERVER]->member_files->storage_status == MF_OK) { t->phase = 1; transfer(t, 1, 1); } } else if (t->phase == 1 && t->finished) { if (t->transfer_status) { fail(t, mf_status_text(t->transfer_status)); return; } t->phase = 2; request(t, MF_STAT, t->ids[AUTHOR]); } else if (t->phase == 2 && t->answered) { if (t->response.status || t->response.size != BYTES) { fail(t, "STAT after upload"); return; } + t->phase = 20; request(t, MF_QUOTA, t->ids[AUTHOR]); + } else if (t->phase == 20 && t->answered) { + if (t->response.status || t->response.data_len != MF_QUOTA_BYTES || mf_get64(t->response.data) != BYTES || + mf_get64(t->response.data + 8) != 8 * 1024 * 1024 || mf_get64(t->response.data + 16)) { + fail(t, "signed/encrypted quota response"); return; + } t->phase = 3; request(t, MF_BEGIN_WRITE, t->ids[AUTHOR]); } else if (t->phase == 3 && t->answered) { if (t->response.status != MF_ACCESS_DENIED) { fail(t, "foreign write accepted"); return; }