jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
33f8dacbfc2827da882dc926eecf974f1e877d5d
utun2/tests/test_member_files_network.c
T

370 lines
21 KiB
C
Raw Blame History

/* Три настоящих ядра A/B/M: подписанный storage, wire-запросы, загрузка A,
* скачивание B, запрет чужой записи, удалённого участника и restart сервиса. */
#include "../src/member_files/member_files.h"
#include "../src/member_files/private.h"
#include "../src/utun_instance.h"
#include "../src/chat/chat_core.h"
#include "../src/chat/chat_event.h"
#include "../src/chat/member_sync.h"
#include "../src/routing_layer/topo_node_sqlite.h"
#include "../src/routing_layer/etcp_router.h"
#include "../src/transport_layer/etcp.h"
#include "../src/transport_layer/secure_channel.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "test_utils.h"
#include "test_clock.h"
#include "test_net.h"
#include <openssl/sha.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#define GROUP 5858585858585858ULL
#define CHANNEL "5858585858585858"
#define BYTES (128 * MF_BLOCK_SIZE + 37)
enum { AUTHOR, READER, SERVER, NODE_COUNT };
struct api_case {
struct mf_transfer* handle;
uint64_t id, done, total;
int finished, status, cancel_after_progress;
char local[768];
};
struct network_test {
struct UASYNC* ua;
struct UTUN_INSTANCE* nodes[NODE_COUNT];
struct SC_MYKEYS keys[NODE_COUNT];
uint64_t ids[NODE_COUNT];
uint8_t ed[NODE_COUNT][32], channel_x[32], channel_ed[32], channel_private[32];
char dir[512], source[768], destination[768];
struct mf_message response;
unsigned phase, ticks, progress;
int answered, finished, transfer_status, failed;
int slow_queued, slow_done, receive_pauses;
pthread_mutex_t disk_mutex;
pthread_cond_t disk_condition;
uint64_t disk_started_tb;
int disk_released, disk_error;
unsigned peak_inflight;
int (*saved_receive_ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t);
unsigned restart_tick;
struct api_case api[8];
void* timer;
};
static struct network_test* active;
static void fail(struct network_test* t, const char* why) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files network: FAIL phase=%u reason=%s", t->phase, why);
t->failed = 1; uasync_stop(t->ua);
}
static void hex(const uint8_t* input, char* out) { for (unsigned i = 0; i < 32; i++) snprintf(out + 2 * i, 3, "%02x", input[i]); }
static void slow_disk(void* arg) {
struct network_test* t = arg;
pthread_mutex_lock(&t->disk_mutex);
t->disk_started_tb = get_time_tb();
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: disk worker blocked started_tb=%llu",
(unsigned long long)t->disk_started_tb);
while (!t->disk_released) {
int rc = pthread_cond_wait(&t->disk_condition, &t->disk_mutex);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files test: disk wait failed rc=%d", rc); t->disk_error = rc; break; }
}
pthread_mutex_unlock(&t->disk_mutex);
}
static void slow_done(void* arg, int error) {
struct network_test* t = arg;
if (error || t->disk_error) fail(t, "slow disk task rejected or wait failed");
t->slow_done = 1;
}
static void capture_log(int level, const char* category, const char* message) {
(void)level; (void)category;
struct network_test* t = active;
if (strstr(message, "router receive paused:")) t->receive_pauses++;
if (!t->slow_queued && strstr(message, "member_files: rx") && strstr(message, "kind=1 cmd=DATA")) {
t->slow_queued = 1;
mf_worker_submit(t->nodes[SERVER]->member_files->disk, slow_disk, t, slow_done, t);
}
}
static void api_event(void* arg, const struct mf_transfer_progress* p) {
struct api_case* c = arg;
if (p->id != c->id || p->done < c->done || p->done > p->total || c->finished) { fail(active, "API progress identity/order"); return; }
c->done = p->done; c->total = p->total;
if (p->finished) { c->handle = NULL; c->finished++; c->status = p->status; }
else if (c->cancel_after_progress && p->done) {
c->cancel_after_progress = 0; member_files_transfer_cancel(c->handle);
}
}
static void api_start(struct network_test* t, unsigned index, int node, int upload, const char* remote, const char* source) {
struct api_case* c = &t->api[index]; memset(c, 0, sizeof(*c)); c->id = 100 + index;
snprintf(c->local, sizeof(c->local), "%s/api%u.bin", t->dir, index);
struct mf_transfer_request q = { .inst = t->nodes[node], .id = c->id, .group = GROUP,
.server = t->ids[SERVER], .owner = t->ids[upload ? node : AUTHOR], .upload = upload };
snprintf(q.remote, sizeof(q.remote), "%s", remote);
snprintf(q.local, sizeof(q.local), "%s", upload ? source : c->local);
if (!upload && source) memcpy(q.version, source, 16);
c->handle = member_files_transfer_start(&q, api_event, c);
if (!c->handle) fail(t, "parallel API start");
}
static int check_file(const char* path, unsigned size) {
FILE* file = fopen(path, "rb"); if (!file) return 0;
int ok = 1;
for (unsigned i = 0; i < size; i++) if (fgetc(file) != (int)((i * 73 + 19) & 255)) { ok = 0; break; }
if (fgetc(file) != EOF) ok = 0;
fclose(file); return ok;
}
static int hold_receive(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) {
(void)inst; (void)group; (void)peer; (void)payload; (void)bytes; return 0;
}
static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) {
(void)inst;
struct network_test* t = active;
if (type != CHAT_EVT_MEMBER_FILES_TRANSFER || len != 51) return;
t->progress++;
if (data[50] == 2) { t->finished = 1; t->transfer_status = data[48] | ((int)data[49] << 8); }
}
static void response(void* arg, const struct mf_message* r) {
struct network_test* t = arg; t->response = *r; t->answered = 1;
}
static int put_member(struct network_test* t, int destination, int member, int deleted) {
uint8_t msg[512], join[64], tags_sig[64], tree[64];
uint64_t ts = 100000;
int n = member_sync_build_join_msg(t->channel_x, t->channel_ed, t->ids[member], t->keys[member].public_key, ts, msg, sizeof(msg));
if (n < 0 || sc_ed25519_sign(t->nodes[member]->my_ed25519_privkey, msg, n, join) != SC_OK) return -1;
const char* tags = deleted ? "{\"deleted\":\"yes\",\"ver\":\"2\"}" :
member == SERVER ? "{\"storage\":\"yes\",\"ver\":\"1\"}" : "{\"ver\":\"1\"}";
n = member_sync_build_owner_msg(t->ids[member], tags, msg, sizeof(msg));
if (n < 0 || sc_ed25519_sign(t->channel_private, msg, n, tags_sig) != SC_OK ||
member_sync_sign_pubkey(t->channel_private, t->keys[member].public_key, tree)) return -1;
char name[64]; snprintf(name, sizeof(name), "{\"name\":\"node%d\"}", member);
return member_sync_put(t->nodes[destination], CHANNEL, t->ids[member], t->keys[member].public_key, t->ed[member],
join, ts, NULL, 0, name, tags, tags_sig, member == SERVER, 0, tree) < 0 ? -1 : 0;
}
static int create(struct network_test* t, int index) {
char pub[65], priv[65], server[65], path[768], db[768];
hex(t->keys[index].public_key, pub); hex(t->keys[index].private_key, priv); hex(t->keys[SERVER].public_key, server);
snprintf(path, sizeof(path), "%s/node%d.conf", t->dir, index); snprintf(db, sizeof(db), "%s/db%d", t->dir, index);
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:0\ntype=public\n[allowed_keys]\nallow_all=1\n"
"[chatserver]\nstorage_autoload=0\nmember_files_user_quota_mb=8\n", priv, pub, db);
if (index != SERVER) fprintf(f, "[client:to_m]\npeer_public_key=%s\nlink=s1:127.0.0.1:0\n", server);
fclose(f);
t->nodes[index] = utun_instance_create(t->ua, path);
if (!t->nodes[index]) return -1;
memcpy(t->ed[index], t->nodes[index]->my_ed25519_pubkey, 32); t->ids[index] = t->nodes[index]->node_id;
return 0;
}
static int setup(struct network_test* t) {
struct SC_MYKEYS keys; sc_generate_keypair(&keys); memcpy(t->channel_x, keys.public_key, 32);
uint8_t seed[SHA512_DIGEST_LENGTH]; SHA512(keys.private_key, 32, seed); memcpy(t->channel_private, seed, 32);
if (sc_derive_ed25519_pubkey(keys.private_key, t->channel_ed)) return -1;
uint8_t signature[64], message[256]; size_t at = sizeof(CHANNEL);
memcpy(message, CHANNEL, at); const char name[] = "member-files";
memcpy(message + at, name, sizeof(name)); at += sizeof(name); memcpy(message + at, &t->ids[AUTHOR], 8); at += 8;
memcpy(message + at, t->channel_x, 32); at += 32; memcpy(message + at, t->channel_ed, 32); at += 32;
if (sc_ed25519_sign(t->channel_private, message, at, signature)) return -1;
for (unsigned i = 0; i < NODE_COUNT; i++) {
if (topo_node_sqlite_channel_put(t->nodes[i]->topo_sqlite_db, CHANNEL, "member-files", t->ids[AUTHOR],
t->channel_x, NULL, t->channel_ed, i == AUTHOR ? t->channel_private : NULL, signature)) return -1;
chat_core_ensure_channel_ready(t->nodes[i], CHANNEL);
for (unsigned j = 0; j < NODE_COUNT; j++) if (put_member(t, i, j, 0)) return -1;
}
snprintf(t->source, sizeof(t->source), "%s/source.bin", t->dir);
snprintf(t->destination, sizeof(t->destination), "%s/download.bin", t->dir);
FILE* file = fopen(t->source, "wb"); if (!file) return -1;
for (unsigned i = 0; i < BYTES; i++) if (fputc((i * 73 + 19) & 255, file) == EOF) { fclose(file); return -1; }
return fclose(file);
}
static void transfer(struct network_test* t, int upload, uint64_t id) {
struct mf_transfer_request* q = u_calloc(1, sizeof(*q));
if (!q) { fail(t, "transfer allocation"); return; }
q->inst = t->nodes[upload ? AUTHOR : READER]; q->group = GROUP; q->server = t->ids[SERVER]; q->owner = t->ids[AUTHOR];
q->id = id; q->upload = upload;
snprintf(q->remote, sizeof(q->remote), "file.bin"); snprintf(q->local, sizeof(q->local), "%s", upload ? t->source : t->destination);
t->finished = 0; member_files_gui_transfer(q);
}
static void request(struct network_test* t, uint8_t command, uint64_t owner) {
struct mf_message q = { .command = command, .owner = owner };
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");
}
static int downloaded(struct network_test* t) {
return check_file(t->destination, BYTES);
}
static void tick(void* arg) {
struct network_test* t = arg; t->timer = NULL;
if (++t->ticks > 6000) { fail(t, "60s deadline"); return; }
// Отсчитываем полные 6с от фактической блокировки worker, не от постановки задания.
pthread_mutex_lock(&t->disk_mutex);
if (t->disk_started_tb && !t->disk_released && get_time_tb() - t->disk_started_tb >= 60000) {
t->disk_released = 1;
pthread_cond_signal(&t->disk_condition);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files test: disk released elapsed_tb=%llu receive_pauses=%d peak_inflight=%u",
(unsigned long long)(get_time_tb() - t->disk_started_tb), t->receive_pauses, t->peak_inflight);
}
pthread_mutex_unlock(&t->disk_mutex);
struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES);
if (route->inflight_q->count > t->peak_inflight) t->peak_inflight = route->inflight_q->count;
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->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) ||
mf_get64(t->response.data + 24) != 1 || mf_get64(t->response.data + 32)) {
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; }
t->phase = 4; transfer(t, 0, 2);
} else if (t->phase == 4 && t->finished) {
if (t->transfer_status || !downloaded(t)) { fail(t, "download bytes"); return; }
t->phase = 5; request(t, MF_LIST, 0);
} else if (t->phase == 5 && t->answered) {
if (t->response.status || !t->response.data_len) { fail(t, "virtual group root"); return; }
t->phase = 6; t->answered = 0;
struct mf_message q = { .command = MF_STAT, .owner = t->ids[AUTHOR] };
snprintf(q.path, sizeof(q.path), "file.bin");
if (!member_files_request(t->nodes[SERVER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "local server request"); return; }
} else if (t->phase == 6 && t->answered) {
if (t->response.status || t->response.size != BYTES) { fail(t, "local server STAT"); return; }
if (!t->slow_done || !t->receive_pauses || t->peak_inflight < 2) { fail(t, "stream did not use receive backpressure/pipelining"); return; }
// Переполненная очередь запросов не должна блокировать ответы в обратном направлении.
struct UTUN_INSTANCE* inst = t->nodes[AUTHOR]; struct member_files* files = inst->member_files;
unsigned jobs = files->job_count; files->job_count = 32;
struct mf_message q = { .kind = MF_RESPONSE, .command = MF_STAT, .request_id = 1 };
uint8_t wire[MF_WIRE_MAX]; int bytes = mf_encode(&q, wire, sizeof(wire));
int response_ready = inst->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES](inst, GROUP, t->ids[SERVER], wire, bytes);
q.kind = MF_REQUEST; bytes = mf_encode(&q, wire, sizeof(wire));
int request_ready = inst->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES](inst, GROUP, t->ids[SERVER], wire, bytes);
files->job_count = jobs;
if (!response_ready || request_ready) { fail(t, "full request queue blocked response or admitted request"); return; }
t->phase = 8;
api_start(t, 0, AUTHOR, 1, "parallel-a.bin", t->source);
api_start(t, 1, AUTHOR, 1, "parallel-b.bin", t->source);
api_start(t, 2, AUTHOR, 0, "file.bin", NULL);
api_start(t, 3, AUTHOR, 1, "cancelled.bin", t->source); t->api[3].cancel_after_progress = 1;
api_start(t, 4, SERVER, 0, "file.bin", NULL);
api_start(t, 5, SERVER, 1, "local.bin", t->source);
char empty[768]; snprintf(empty, sizeof(empty), "%s/empty.bin", t->dir);
FILE* file = fopen(empty, "wb"); if (!file) { fail(t, "empty source"); return; } fclose(file);
api_start(t, 6, AUTHOR, 1, "empty.bin", empty);
} else if (t->phase == 8) {
for (unsigned i = 0; i < 7; i++) if (!t->api[i].finished) goto again;
for (unsigned i = 0; i < 7; i++) if (t->api[i].status != (i == 3 ? MF_CANCELLED : MF_OK)) { fail(t, "parallel completion status"); return; }
if (!check_file(t->api[2].local, BYTES) || !check_file(t->api[4].local, BYTES)) { fail(t, "duplex/local download bytes"); return; }
t->phase = 9;
api_start(t, 0, AUTHOR, 0, "parallel-a.bin", NULL);
api_start(t, 1, AUTHOR, 0, "parallel-b.bin", NULL);
api_start(t, 7, AUTHOR, 0, "empty.bin", NULL);
} else if (t->phase == 9) {
if (!t->api[0].finished || !t->api[1].finished || !t->api[7].finished) goto again;
if (t->api[0].status || t->api[1].status || t->api[7].status || !check_file(t->api[0].local, BYTES) ||
!check_file(t->api[1].local, BYTES) || !check_file(t->api[7].local, 0)) { fail(t, "parallel/empty download bytes"); return; }
t->phase = 22;
uint8_t stale[16]; memset(stale, 0xa5, sizeof(stale));
api_start(t, 7, READER, 0, "file.bin", (const char*)stale);
} else if (t->phase == 22 && t->api[7].finished) {
if (t->api[7].status != MF_STALE || !check_file(t->api[7].local, 0)) {
fail(t, "stale download changed existing destination"); return;
}
t->phase = 10; t->answered = 0;
struct mf_message q = { .command = MF_STAT, .owner = t->ids[AUTHOR] }; snprintf(q.path, sizeof(q.path), "cancelled.bin");
if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "cancelled STAT"); return; }
} else if (t->phase == 10 && t->answered) {
if (t->response.status != MF_NOT_FOUND) { fail(t, "cancelled upload published"); return; }
t->saved_receive_ready = t->nodes[AUTHOR]->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES];
etcp_router_set_receive_ready(t->nodes[AUTHOR], ETCP_RT_ID_MEMBER_FILES, hold_receive);
api_start(t, 0, AUTHOR, 1, "restart.bin", t->source);
api_start(t, 1, AUTHOR, 0, "file.bin", NULL);
t->restart_tick = t->ticks; t->phase = 11;
} else if (t->phase == 11 && t->ticks > t->restart_tick + 2) {
etcp_router_conn_restart(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES);
etcp_router_set_receive_ready(t->nodes[AUTHOR], ETCP_RT_ID_MEMBER_FILES, t->saved_receive_ready);
t->phase = 12;
} else if (t->phase == 12) {
if (!t->api[0].finished || !t->api[1].finished || t->nodes[SERVER]->member_files->server_transfer_count) goto again;
if (t->api[0].status != MF_IO_ERROR || t->api[1].status != MF_IO_ERROR) { fail(t, "restart completion status"); return; }
if (put_member(t, SERVER, READER, 1)) { fail(t, "signed deletion"); return; }
t->phase = 7; request(t, MF_STAT, t->ids[AUTHOR]);
} else if (t->phase == 7 && t->answered) {
if (t->response.status != MF_ACCESS_DENIED) { fail(t, "deleted reader accepted"); return; }
t->phase = 21; request(t, MF_QUOTA, t->ids[AUTHOR]);
} else if (t->phase == 21 && t->answered) {
if (t->response.status != MF_ACCESS_DENIED || t->response.data_len) { fail(t, "quota leaked to deleted reader"); return; }
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files network: PASS progress=%u ticks=%u peak_inflight=%u pauses=%d",
t->progress, t->ticks, t->peak_inflight, t->receive_pauses);
uasync_stop(t->ua); return;
}
again:
t->timer = uasync_set_timeout(t->ua, 100, t, tick, "member_files_test");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG);
test_clock_init(1000);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_TRACE);
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG);
utun_instance_set_tun_init_enabled(0);
struct network_test t = { .disk_mutex = PTHREAD_MUTEX_INITIALIZER, .disk_condition = PTHREAD_COND_INITIALIZER }; active = &t;
debug_set_log_hook(capture_log);
snprintf(t.dir, sizeof(t.dir), "/tmp/utun_member_files_net_XXXXXX");
if (test_mkdtemp(t.dir)) return 1;
t.ua = uasync_create();
for (unsigned i = 0; i < NODE_COUNT; i++) sc_generate_keypair(&t.keys[i]);
for (unsigned i = 0; i < NODE_COUNT; i++) if (create(&t, i)) { t.failed = 1; goto stop; }
for (unsigned i = 0; i < NODE_COUNT; i++) {
if (i != SERVER) test_etcp_set_peer(t.nodes[i], "to_m", t.nodes[SERVER], "s1");
if (utun_instance_init(t.nodes[i])) { t.failed = 1; goto stop; }
chat_event_set_handler(t.nodes[i], event);
}
if (setup(&t)) { t.failed = 1; goto stop; }
t.timer = uasync_set_timeout(t.ua, 100, &t, tick, "member_files_test"); test_clock_mainloop(t.ua);
if (!t.failed) {
// Stop/destroy вне callbacks: server cleanup ещё использует storage после текущего I/O.
api_start(&t, 0, AUTHOR, 1, "stop.bin", t.source);
api_start(&t, 1, AUTHOR, 0, "file.bin", NULL);
uint64_t deadline = get_time_tb() + 50000;
while (!t.nodes[SERVER]->member_files->server_transfer_count && get_time_tb() < deadline) {
test_clock_poll(t.ua, 10);
}
if (!t.nodes[SERVER]->member_files->server_transfer_count) fail(&t, "server stream did not start before teardown");
}
stop:
if (t.timer) uasync_cancel_timeout(t.ua, t.timer);
// При ошибке сценария освобождаем worker до destroy, иначе drain будет ждать блокировку.
pthread_mutex_lock(&t.disk_mutex);
t.disk_released = 1; pthread_cond_signal(&t.disk_condition);
pthread_mutex_unlock(&t.disk_mutex);
for (unsigned i = 0; i < NODE_COUNT; i++) if (t.nodes[i]) utun_instance_destroy(t.nodes[i]);
if (!t.failed && (!t.api[0].finished || !t.api[1].finished || t.api[0].status != MF_CANCELLED || t.api[1].status != MF_CANCELLED))
t.failed = 1;
uasync_destroy(t.ua, 0);
pthread_cond_destroy(&t.disk_condition); pthread_mutex_destroy(&t.disk_mutex);
debug_set_log_hook(NULL);
test_clock_finish();
printf("member_files network: %s phase=%u\n", t.failed ? "FAIL" : "PASS", t.phase); return t.failed;
}
Reference in new issue View Git Blame Copy Permalink