evgeny
4 weeks ago
6 changed files with 669 additions and 5 deletions
@ -0,0 +1,636 @@
|
||||
// test_media_delivery_chat.c — сквозной тест распространения медиа через db_sync
|
||||
//
|
||||
// 6 узлов в цепочку:
|
||||
// n1(0) — n2(1) — s1(2) — s2(3) — st1(4) — st2(5)
|
||||
// автор клиент супер супер storage storage
|
||||
//
|
||||
// Сценарий:
|
||||
// 1. n1 индексирует медиафайл (media_index_commit) и публикует сообщение с метаданными
|
||||
// через db_sync (реальная репликация по цепочке каскадом).
|
||||
// 2. storage-узлы (st1/st2) по insert-callback'у запускают auto-download
|
||||
// (media_download_start), скачивают блоки у автора, собирают файл.
|
||||
// 3. Проверяем, что суперузлы (s1/s2) узнали о блоках (block_availability от HAVE_BLOCK)
|
||||
// и отреплицировались между собой (super_sync.last_recv_id > 0).
|
||||
// 4. Второй клиент n2 скачивает медиа «по запросу» и собирает файл.
|
||||
//
|
||||
// Используется настоящий db_sync для доставки сообщения и настоящий media-движок
|
||||
// для доставки блоков. chat_core НЕ используется (он — глобальный синглтон на процесс),
|
||||
// поэтому связка «сообщение → auto-download» воспроизведена локальным insert-callback'ом.
|
||||
|
||||
#include "media_delivery.h" |
||||
#include "media_delivery_proto.h" |
||||
#include "media_download.h" |
||||
#include "media_index.h" |
||||
#include "member_sync.h" |
||||
#include "db_sync.h" |
||||
#include "../routing_layer/topo_node_sqlite.h" |
||||
#include "../utun_instance.h" |
||||
#include "../transport_layer/etcp.h" |
||||
#include "../transport_layer/secure_channel.h" |
||||
#include "../config_parser.h" |
||||
#include "../config_updater.h" |
||||
#include "../routing_layer/topo_group.h" |
||||
#include "../routing_layer/topo_node.h" |
||||
#include "../lib/u_async.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
|
||||
#include <sqlite3.h> |
||||
#define OPENSSL_API_COMPAT 0x10100000L |
||||
#include <openssl/evp.h> |
||||
#include <stdio.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <stdarg.h> |
||||
#include <time.h> |
||||
#include <unistd.h> |
||||
|
||||
static int G_PASSED = 0, G_FAILED = 0, G_TOTAL = 0; |
||||
#define TEST(n) do { G_TOTAL++; printf(" %-58s", n); fflush(stdout); } while(0) |
||||
#define OK() do { G_PASSED++; printf("OK\n"); } while(0) |
||||
#define FAIL(f,...) do { G_FAILED++; printf("FAIL: " f "\n", ##__VA_ARGS__); } while(0) |
||||
|
||||
#define TIMEOUT_TB 1200000 /* 120 s общий таймаут */ |
||||
#define POLL_MS 5 |
||||
#define N_NODES 6 |
||||
#define CH_ID "31415926535" |
||||
|
||||
/* роли узлов */ |
||||
#define I_N1 0 /* автор */ |
||||
#define I_N2 1 /* клиент (скачивает позже) */ |
||||
#define I_S1 2 /* суперузел */ |
||||
#define I_S2 3 /* суперузел */ |
||||
#define I_ST1 4 /* storage */ |
||||
#define I_ST2 5 /* storage */ |
||||
|
||||
#define MEDIA_FNAME "photo.bin" |
||||
#define MEDIA_FILE_SIZE 15360 |
||||
#define MEDIA_BLOCK_SIZE 5120 |
||||
#define MEDIA_NUM_BLOCKS 3 |
||||
|
||||
static struct UASYNC* g_ua = NULL; |
||||
static struct UTUN_INSTANCE* g_inst[N_NODES]; |
||||
static uint64_t g_nid[N_NODES]; |
||||
static struct DB_SYNC_INSTANCE* g_si[N_NODES]; |
||||
static uint64_t g_group_id = 0; |
||||
|
||||
static char g_tdir[256] = "/tmp/utun_mdc_XXXXXX"; |
||||
static char g_cfg[N_NODES][256]; |
||||
static char g_db_dir[N_NODES][320]; |
||||
static int g_port[N_NODES]; |
||||
static int g_connected = 0, g_result = 0; |
||||
|
||||
/* опубликованное медиа (автор) — глобально для n2-загрузки и проверок */ |
||||
static uint8_t g_media_id[16]; |
||||
static uint8_t g_content_hash[32]; |
||||
static uint8_t g_block_ids[MEDIA_NUM_BLOCKS][16]; |
||||
static uint8_t g_block_sigs[MEDIA_NUM_BLOCKS][64]; |
||||
static uint8_t g_file_data[MEDIA_FILE_SIZE]; |
||||
|
||||
/* флаги завершения загрузок */ |
||||
static int g_dl_done[N_NODES]; |
||||
static int g_dl_err[N_NODES]; |
||||
|
||||
/* ── helpers ── */ |
||||
|
||||
static int wf(const char* p, const char* f, ...) { |
||||
va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1; |
||||
va_start(ap, f); vfprintf(fp, f, ap); va_end(ap); fclose(fp); return 0; |
||||
} |
||||
static char* gv(const char* p, const char* k) { |
||||
struct utun_config* c = parse_config(p); if (!c) return NULL; |
||||
char* r = (strcmp(k, "pub") == 0) ? u_strdup(c->global.my_public_key_hex) : u_strdup(c->global.my_private_key_hex); |
||||
free_config(c); return r; |
||||
} |
||||
static void t_mkdtemp(char* t) { |
||||
#ifdef _WIN32 |
||||
char tmp_path[512]; GetTempPathA(sizeof(tmp_path), tmp_path); |
||||
snprintf(t, 256, "%s\\utun_test_%08x", tmp_path, (unsigned)rand()); |
||||
_mkdir(t); |
||||
#else |
||||
(void)!mkdtemp(t); |
||||
#endif |
||||
} |
||||
static void t_rmdir(const char* p) { char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", p); system(cmd); } |
||||
|
||||
static int db_count(sqlite3* db, const char* tbl, const char* wh, uint64_t val) { |
||||
char sql[256]; |
||||
if (wh) snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s WHERE %s=%llu", tbl, wh, (unsigned long long)val); |
||||
else snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s", tbl); |
||||
sqlite3_stmt* s = NULL; |
||||
if (sqlite3_prepare_v2(db, sql, -1, &s, NULL) != SQLITE_OK) return -1; |
||||
int n = -1; |
||||
if (sqlite3_step(s) == SQLITE_ROW) n = sqlite3_column_int(s, 0); |
||||
sqlite3_finalize(s); return n; |
||||
} |
||||
|
||||
static void to_cb(void* arg) { (void)arg; fprintf(stderr, " TIMEOUT\n"); g_result = 2; } |
||||
|
||||
static void mon(void* arg) { |
||||
(void)arg; |
||||
if (g_result) return; |
||||
int ok = 0; |
||||
for (int i = 0; i < N_NODES; i++) { |
||||
struct ll_entry* e = g_inst[i]->connections->head; |
||||
while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; |
||||
if (ce->conn && ce->conn->links) { struct ETCP_LINK* l; for (l = ce->conn->links; l; l = l->next) |
||||
if (l->initialized && ce->conn->crypto_ctx.initialized) ok++; } e = e->next; } |
||||
} |
||||
if (ok >= N_NODES * 2 - 2) g_connected = 1; |
||||
if (!g_result) uasync_set_timeout(g_ua, 10, NULL, mon, "mdc_mon"); |
||||
} |
||||
|
||||
/* pubkey узла известен в in-memory registry (BGP сошёлся) */ |
||||
static int pubkey_known(struct UTUN_INSTANCE* inst, uint64_t node_id) { |
||||
struct TOPO_NODE* ni = inst && inst->topo_groups ? topo_node_registry_find(inst->topo_groups, node_id) : NULL; |
||||
if (!ni) return 0; |
||||
uint64_t chk; memcpy(&chk, ni->ed25519_public_key, 8); |
||||
return chk != 0; |
||||
} |
||||
|
||||
/* ждём, пока все узлы узнают ed25519-ключи всех остальных (нужно db_sync для верификации подписи) */ |
||||
static int wait_pubkeys(void) { |
||||
int a = 0; |
||||
while (a < 4000) { |
||||
int ok = 1; |
||||
for (int i = 0; i < N_NODES; i++) |
||||
for (int m = 0; m < N_NODES; m++) |
||||
if (m != i && !pubkey_known(g_inst[i], g_nid[m])) ok = 0; |
||||
if (ok) return 1; |
||||
uasync_poll(g_ua, POLL_MS); a++; |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
static void sha256_buf(const uint8_t* data, size_t len, uint8_t out[32]) { |
||||
EVP_MD_CTX* ctx = EVP_MD_CTX_new(); |
||||
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL); |
||||
EVP_DigestUpdate(ctx, data, len); |
||||
EVP_DigestFinal_ex(ctx, out, NULL); |
||||
EVP_MD_CTX_free(ctx); |
||||
} |
||||
|
||||
static int file_sha256(const char* path, uint8_t out[32]) { |
||||
FILE* f = fopen(path, "rb"); if (!f) return -1; |
||||
EVP_MD_CTX* ctx = EVP_MD_CTX_new(); |
||||
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL); |
||||
uint8_t buf[8192]; size_t rd; |
||||
while ((rd = fread(buf, 1, sizeof(buf), f)) > 0) EVP_DigestUpdate(ctx, buf, rd); |
||||
EVP_DigestFinal_ex(ctx, out, NULL); |
||||
EVP_MD_CTX_free(ctx); fclose(f); |
||||
return 0; |
||||
} |
||||
|
||||
static int file_exists(const char* p) { return access(p, F_OK) == 0; } |
||||
|
||||
/* nodeinfo_updated пересчитывает node_type по адресам и может сбросить тип суперузла,
|
||||
* если узел не имеет DIRECT-адреса. Перед каждой загрузкой принудительно выставляем |
||||
* node_type=4 для s1/s2 (md_dl_collect_supernodes читает именно это поле). */ |
||||
static void ensure_supernode_type(int idx) { |
||||
char sql[256]; |
||||
snprintf(sql, sizeof(sql), "UPDATE \"peers_%s\" SET node_type=4 WHERE node_id IN (%llu,%llu)", |
||||
CH_ID, (unsigned long long)g_nid[I_S1], (unsigned long long)g_nid[I_S2]); |
||||
sqlite3_exec(g_inst[idx]->topo_sqlite_db, sql, NULL, NULL, NULL); |
||||
} |
||||
|
||||
/* ── завершение загрузки ── */ |
||||
|
||||
static void dl_done_cb(void* arg, int err) { |
||||
intptr_t idx = (intptr_t)arg; |
||||
g_dl_done[idx] = 1; g_dl_err[idx] = err; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] download done node=%d err=%d", (int)idx, err); |
||||
} |
||||
|
||||
/* media_base узла = каталог db_path (как chat_core/md_start_download) */ |
||||
static void node_media_base(int idx, char* out, size_t sz) { snprintf(out, sz, "%s", g_db_dir[idx]); } |
||||
static void node_dest(int idx, char* out, size_t sz) { |
||||
snprintf(out, sz, "%s/media/%s/%s", g_db_dir[idx], CH_ID, MEDIA_FNAME); |
||||
} |
||||
|
||||
/* ── парсинг тела медиа-сообщения (как md_start_download в chat_msg.c) ── */ |
||||
|
||||
static int hex_byte(const char* p) { |
||||
int v = 0; |
||||
for (int i = 0; i < 2; i++) { char c = p[i]; v <<= 4; |
||||
v |= (c >= '0' && c <= '9') ? (c - '0') : (c >= 'a' && c <= 'f') ? (c - 'a' + 10) : (c - 'A' + 10); } |
||||
return v; |
||||
} |
||||
|
||||
static int parse_media_body(const char* body, struct media_index_result* out) { |
||||
const char* p = body; |
||||
while (*p && *p != '|') p++; |
||||
if (!*p) return -1; p++; |
||||
out->file_size = strtoll(p, (char**)&p, 10); |
||||
if (*p != '|') return -1; p++; |
||||
out->block_size = strtoll(p, (char**)&p, 10); |
||||
if (*p != '|') return -1; p++; |
||||
int nb = (int)strtol(p, (char**)&p, 10); |
||||
while (*p && *p != '|') p++; |
||||
if (!*p) return -1; p++; |
||||
if (nb <= 0 || nb > 100 || out->file_size <= 0) return -1; |
||||
|
||||
for (int i = 0; i < 16; i++) out->media_id[i] = (uint8_t)hex_byte(p + i * 2); |
||||
p += 32; if (*p != '|') return -1; p++; |
||||
for (int i = 0; i < 32; i++) out->content_hash[i] = (uint8_t)hex_byte(p + i * 2); |
||||
p += 64; if (*p != '|') return -1; p++; |
||||
|
||||
out->num_blocks = nb; |
||||
out->block_ids = u_malloc((size_t)nb * 16); |
||||
out->block_sigs = u_malloc((size_t)nb * 64); |
||||
if (!out->block_ids || !out->block_sigs) { media_index_result_free(out); return -1; } |
||||
for (int n = 0; n < nb; n++) { |
||||
for (int i = 0; i < 16; i++) out->block_ids[n * 16 + i] = (uint8_t)hex_byte(p + i * 2); |
||||
p += 32; if (*p != ',') { media_index_result_free(out); return -1; } p++; |
||||
for (int i = 0; i < 64; i++) out->block_sigs[n * 64 + i] = (uint8_t)hex_byte(p + i * 2); |
||||
p += 128; |
||||
if (n < nb - 1 && *p == ',') p++; |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
/* ── insert-callback: storage-узлы запускают auto-download ── */ |
||||
|
||||
static void storage_on_insert(struct DB_SYNC_INSTANCE* si, uint64_t ts, |
||||
const char* data, size_t len, uint64_t author, void* arg) { |
||||
(void)si; (void)ts; |
||||
int idx = (int)(intptr_t)arg; |
||||
struct UTUN_INSTANCE* inst = g_inst[idx]; |
||||
if (author == inst->node_id) return; |
||||
|
||||
char buf[8192]; size_t blen = len < sizeof(buf) - 1 ? len : sizeof(buf) - 1; |
||||
memcpy(buf, data, blen); buf[blen] = '\0'; |
||||
const char* ct = strstr(buf, "\"ct\":\""); |
||||
const char* dpos = strstr(buf, "\"d\":\""); |
||||
if (!ct || !dpos) return; |
||||
|
||||
const char* ct_val = ct + 6; |
||||
char content_type[32] = {0}; int cti = 0; |
||||
while (ct_val[cti] && ct_val[cti] != '\"' && cti < (int)sizeof(content_type) - 1) content_type[cti++] = ct_val[cti]; |
||||
|
||||
const char* d_start = dpos + 5; |
||||
char body[8192]; size_t bi = 0; |
||||
while (*d_start && *d_start != '\"' && bi < sizeof(body) - 1) body[bi++] = *d_start++; |
||||
body[bi] = '\0'; |
||||
|
||||
struct media_index_result result; |
||||
memset(&result, 0, sizeof(result)); |
||||
if (parse_media_body(body, &result) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[mdc] parse_media_body failed"); return; } |
||||
|
||||
char dest[1024], mbase[512]; |
||||
node_media_base(idx, mbase, sizeof(mbase)); |
||||
node_dest(idx, dest, sizeof(dest)); |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] auto-download on node=%d ch=%s ct=%s blocks=%d author=0x%016llx dest=%s", |
||||
idx, CH_ID, content_type, result.num_blocks, (unsigned long long)author, dest); |
||||
ensure_supernode_type(idx); |
||||
media_download_start(inst, g_group_id, &result, dest, mbase, author, dl_done_cb, (void*)(intptr_t)idx, NULL, NULL); |
||||
media_index_result_free(&result); |
||||
} |
||||
|
||||
/* ── публикация медиа автором ── */ |
||||
|
||||
static int author_index_file(void) { |
||||
char media_dir[512]; snprintf(media_dir, sizeof(media_dir), "%s/media", g_db_dir[I_N1]); |
||||
utun_mkdir(media_dir, 0755); |
||||
char dst_path[512]; snprintf(dst_path, sizeof(dst_path), "%s/%s", media_dir, MEDIA_FNAME); |
||||
char media_base[512]; snprintf(media_base, sizeof(media_base), "%s", g_db_dir[I_N1]); |
||||
|
||||
FILE* f = fopen(dst_path, "wb"); if (!f) return -1; |
||||
fwrite(g_file_data, 1, MEDIA_FILE_SIZE, f); fclose(f); |
||||
|
||||
sha256_buf(g_file_data, MEDIA_FILE_SIZE, g_content_hash); |
||||
|
||||
struct media_index_result result; memset(&result, 0, sizeof(result)); |
||||
media_index_init(g_inst[I_N1]->topo_sqlite_db); |
||||
media_index_generate_uuid(result.media_id); |
||||
memcpy(result.content_hash, g_content_hash, 32); |
||||
result.file_size = MEDIA_FILE_SIZE; result.block_size = MEDIA_BLOCK_SIZE; result.num_blocks = MEDIA_NUM_BLOCKS; |
||||
result.block_ids = u_malloc(MEDIA_NUM_BLOCKS * 16); |
||||
result.block_sigs = u_malloc(MEDIA_NUM_BLOCKS * 64); |
||||
for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { |
||||
media_index_generate_uuid(result.block_ids + n * 16); |
||||
int off = n * MEDIA_BLOCK_SIZE; |
||||
int sz = (n == MEDIA_NUM_BLOCKS - 1) ? MEDIA_FILE_SIZE - off : MEDIA_BLOCK_SIZE; |
||||
/* подпись блока = Ed25519(данные блока) — ровно как проверяет download */ |
||||
sc_ed25519_sign(g_inst[I_N1]->my_ed25519_privkey, g_file_data + off, (size_t)sz, result.block_sigs + n * 64); |
||||
} |
||||
int rc = media_index_commit(g_inst[I_N1]->topo_sqlite_db, &result, g_nid[I_N1], |
||||
g_inst[I_N1]->my_ed25519_privkey, CH_ID, dst_path, media_base); |
||||
|
||||
/* сохраняем глобально */ |
||||
memcpy(g_media_id, result.media_id, 16); |
||||
for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { |
||||
memcpy(g_block_ids[n], result.block_ids + n * 16, 16); |
||||
memcpy(g_block_sigs[n], result.block_sigs + n * 64, 64); |
||||
} |
||||
media_index_result_free(&result); |
||||
return rc; |
||||
} |
||||
|
||||
static int author_publish_message(void) { |
||||
/* body: <base>|<fsize>|<bsize>|<nb>|<mid_hex>|<hash_hex>|<id0>,<sig0>,... */ |
||||
char body[8192]; size_t off = 0; |
||||
off += snprintf(body + off, sizeof(body) - off, "%s|%d|%d|%d|", |
||||
MEDIA_FNAME, MEDIA_FILE_SIZE, MEDIA_BLOCK_SIZE, MEDIA_NUM_BLOCKS); |
||||
for (int i = 0; i < 16; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_media_id[i]); |
||||
off += snprintf(body + off, sizeof(body) - off, "|"); |
||||
for (int i = 0; i < 32; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_content_hash[i]); |
||||
off += snprintf(body + off, sizeof(body) - off, "|"); |
||||
for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { |
||||
if (n > 0) body[off++] = ','; |
||||
for (int i = 0; i < 16; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_block_ids[n][i]); |
||||
body[off++] = ','; |
||||
for (int i = 0; i < 64; i++) off += snprintf(body + off, sizeof(body) - off, "%02x", g_block_sigs[n][i]); |
||||
} |
||||
|
||||
char json[8700]; |
||||
snprintf(json, sizeof(json), "{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"application/octet-stream\",\"d\":\"%s\"}", |
||||
(unsigned long long)g_nid[I_N1], CH_ID, body); |
||||
|
||||
uint64_t ts = db_sync_next_timestamp(g_si[I_N1]); |
||||
size_t jl = strlen(json); |
||||
uint8_t sig_msg[8700]; size_t soff = 0; |
||||
memcpy(sig_msg + soff, &ts, 8); soff += 8; |
||||
memcpy(sig_msg + soff, json, jl); soff += jl; |
||||
uint8_t sig[64]; |
||||
if (sc_ed25519_sign(g_inst[I_N1]->my_ed25519_privkey, sig_msg, soff, sig) != SC_OK) return -1; |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[mdc] publish media message ts=%llu len=%zu", (unsigned long long)ts, jl); |
||||
return db_sync_insert_signed(g_si[I_N1], json, jl, sig, 64, ts, NULL); |
||||
} |
||||
|
||||
/* ── фазы ── */ |
||||
|
||||
static void setup_chat_group(void) { |
||||
uint8_t x25519_pub[32] = {0}, ed_pub[32] = {0}, ch_sig[64] = {0}; |
||||
int i; |
||||
|
||||
TEST("generate channel keys"); { |
||||
EVP_PKEY* xpkey = NULL, *epkey = NULL; |
||||
EVP_PKEY_CTX* ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_X25519, NULL); |
||||
if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_keygen(ctx, &xpkey); EVP_PKEY_CTX_free(ctx); } |
||||
if (xpkey) { size_t l = 32; EVP_PKEY_get_raw_public_key(xpkey, x25519_pub, &l); EVP_PKEY_free(xpkey); } |
||||
ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_ED25519, NULL); |
||||
if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_keygen(ctx, &epkey); EVP_PKEY_CTX_free(ctx); } |
||||
if (epkey) { size_t l = 32; EVP_PKEY_get_raw_public_key(epkey, ed_pub, &l); |
||||
EVP_MD_CTX* mctx = EVP_MD_CTX_new(); |
||||
if (mctx) { EVP_DigestSignInit(mctx, NULL, NULL, NULL, epkey); |
||||
size_t sl = 64; EVP_DigestSign(mctx, ch_sig, &sl, (const uint8_t*)CH_ID, strlen(CH_ID)); EVP_MD_CTX_free(mctx); } |
||||
EVP_PKEY_free(epkey); |
||||
} |
||||
if (x25519_pub[0] || ed_pub[0]) OK(); else FAIL(); |
||||
} |
||||
|
||||
TEST("create CHAT group + peers tables"); { |
||||
for (i = 0; i < N_NODES; i++) { |
||||
sqlite3* db = g_inst[i]->topo_sqlite_db; |
||||
struct TOPO_GROUPS* tg = g_inst[i]->topo_groups; |
||||
if (!db || !tg) continue; |
||||
topo_node_sqlite_channel_put(db, CH_ID, "test", g_nid[0], x25519_pub, NULL, ed_pub, NULL, ch_sig); |
||||
if (topo_groups_find(tg, g_group_id)) continue; |
||||
topo_groups_create_group(tg, g_group_id, TOPO_GROUP_TYPE_CHAT, CH_ID); |
||||
} |
||||
OK(); |
||||
} |
||||
|
||||
TEST("add 6 members (s1/s2 supernode, st1/st2 storage)"); { |
||||
const char* adm[N_NODES] = { NULL, NULL, "{\"supernode\":\"yes\"}", "{\"supernode\":\"yes\"}", "{\"storage\":\"yes\"}", "{\"storage\":\"yes\"}" }; |
||||
for (i = 0; i < N_NODES; i++) { |
||||
for (int m = 0; m < N_NODES; m++) { |
||||
member_sync_put(g_inst[i], CH_ID, g_nid[m], |
||||
g_inst[m]->my_keys.public_key, g_inst[m]->my_ed25519_pubkey, |
||||
NULL, 0, NULL, 0, "{\"name\":\"node\"}", adm[m], NULL, 0); |
||||
/* заполняем nodes-таблицу (нужно для верификации Ed25519 подписи блоков/автора) */ |
||||
topo_node_sqlite_node_update_verified(g_inst[i]->topo_sqlite_db, g_nid[m], |
||||
"node", g_inst[m]->my_keys.public_key, g_inst[m]->my_ed25519_pubkey, |
||||
(uint64_t)time(NULL), time(NULL)); |
||||
} |
||||
} |
||||
int ok = 1; |
||||
for (i = 0; i < N_NODES; i++) if (member_sync_count(g_inst[i], CH_ID) < N_NODES) ok = 0; |
||||
if (ok) OK(); else FAIL(); |
||||
} |
||||
|
||||
TEST("supernodes node_type=4 in peers table"); { |
||||
char sql[256]; |
||||
for (i = 0; i < N_NODES; i++) { |
||||
snprintf(sql, sizeof(sql), "UPDATE \"peers_%s\" SET node_type=4 WHERE node_id IN (%llu,%llu)", |
||||
CH_ID, (unsigned long long)g_nid[I_S1], (unsigned long long)g_nid[I_S2]); |
||||
sqlite3_exec(g_inst[i]->topo_sqlite_db, sql, NULL, NULL, NULL); |
||||
} |
||||
int ok = 1; |
||||
for (i = 0; i < N_NODES; i++) { |
||||
int n = db_count(g_inst[i]->topo_sqlite_db, "peers_" CH_ID, "node_type", 4); |
||||
if (n < 2) ok = 0; |
||||
} |
||||
if (ok) OK(); else FAIL(); |
||||
} |
||||
|
||||
TEST("media_delivery supernode s1,s2"); { |
||||
media_delivery_set_supernode(g_inst[I_S1], 1); |
||||
media_delivery_set_supernode(g_inst[I_S2], 1); |
||||
if (g_inst[I_S1]->md.is_supernode && g_inst[I_S2]->md.is_supernode) OK(); else FAIL(); |
||||
} |
||||
} |
||||
|
||||
static void setup_db_sync(void) { |
||||
TEST("db_sync instances on all nodes"); { |
||||
int ok = 1; |
||||
for (int i = 0; i < N_NODES; i++) { |
||||
g_si[i] = db_sync_instance_add(g_inst[i], "msg_" CH_ID, g_group_id, 1); |
||||
if (!g_si[i]) { ok = 0; DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[mdc] db_sync_instance_add failed node=%d", i); } |
||||
} |
||||
if (ok) OK(); else FAIL(); |
||||
} |
||||
|
||||
TEST("register auto-download insert-cb on storage nodes"); { |
||||
db_sync_set_insert_cb(g_si[I_ST1], storage_on_insert, (void*)(intptr_t)I_ST1); |
||||
db_sync_set_insert_cb(g_si[I_ST2], storage_on_insert, (void*)(intptr_t)I_ST2); |
||||
OK(); |
||||
} |
||||
} |
||||
|
||||
/* ── main ── */ |
||||
|
||||
int main(void) { |
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); |
||||
if (getenv("UTUN_TEST_DEBUG")) { |
||||
debug_set_category_level_by_name("media", "trace"); |
||||
debug_set_category_level_by_name("general", "info"); |
||||
debug_set_category_level_by_name("chatsync", "info"); |
||||
debug_set_category_level_by_name("etcproute", "info"); |
||||
} |
||||
utun_instance_set_tun_init_enabled(0); |
||||
srand((unsigned)time(NULL)); |
||||
printf("=== test_media_delivery_chat ===\n"); |
||||
t_mkdtemp(g_tdir); |
||||
|
||||
for (int i = 0; i < MEDIA_FILE_SIZE; i++) g_file_data[i] = (uint8_t)((i * 31 + 17) & 0xFF); |
||||
|
||||
for (int i = 0; i < N_NODES; i++) { |
||||
snprintf(g_cfg[i], sizeof(g_cfg[i]), "%s/n%d.conf", g_tdir, i); |
||||
snprintf(g_db_dir[i], sizeof(g_db_dir[i]), "%s/db%d", g_tdir, i); |
||||
utun_mkdir(g_db_dir[i], 0755); |
||||
g_port[i] = 52000 + (getpid() % 5000) + i * 1000; |
||||
} |
||||
|
||||
g_ua = uasync_create(); |
||||
for (int i = 0; i < N_NODES; i++) |
||||
wf(g_cfg[i], "[global]\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\ndb_path=%s\ndb_sync_enabled=1\n" |
||||
"[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", |
||||
i, i, g_db_dir[i], g_port[i]); |
||||
for (int i = 0; i < N_NODES; i++) { |
||||
config_ensure_keys_and_node_id(g_cfg[i]); |
||||
struct utun_config* c = parse_config(g_cfg[i]); g_nid[i] = c->global.my_node_id; free_config(c); |
||||
} |
||||
|
||||
for (int i = 0; i < N_NODES; i++) { |
||||
char *pr = gv(g_cfg[i], "priv"), *pu = gv(g_cfg[i], "pub"); |
||||
if (i == 0) { |
||||
wf(g_cfg[i], "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\n" |
||||
"db_path=%s\ndb_sync_enabled=1\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", |
||||
pr, pu, i, i, g_db_dir[i], g_port[i]); |
||||
} else { |
||||
int prev = i - 1; char *pv_pu = gv(g_cfg[prev], "pub"); |
||||
char link[256]; snprintf(link, sizeof(link), |
||||
"[client: to_n%d]\nkeepalive=1\npeer_public_key=%s\nlink=s1:127.0.0.1:%d\n", prev, pv_pu, g_port[prev]); |
||||
wf(g_cfg[i], "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.%d.1/24\ntun_ifname=tun%d0\n" |
||||
"db_path=%s\ndb_sync_enabled=1\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n%s[allowed_keys]\nallow_all=1\n", |
||||
pr, pu, i, i, g_db_dir[i], g_port[i], link); |
||||
u_free(pv_pu); |
||||
} |
||||
u_free(pr); u_free(pu); |
||||
} |
||||
|
||||
for (int i = 0; i < N_NODES; i++) { |
||||
g_inst[i] = utun_instance_create(g_ua, g_cfg[i]); |
||||
if (!g_inst[i]) { printf("FAIL: create n%d\n", i); goto done; } |
||||
utun_instance_init(g_inst[i]); |
||||
} |
||||
|
||||
mon(NULL); |
||||
uasync_set_timeout(g_ua, TIMEOUT_TB, NULL, (timeout_callback_t)to_cb, "mdc_to"); |
||||
{ int el = 0; while (!g_connected && !g_result && el < TIMEOUT_TB + 5000) { uasync_poll(g_ua, POLL_MS); el += POLL_MS; } |
||||
if (!g_connected) { printf("FAIL: connection timeout\n"); goto done; } } |
||||
printf(" connected\n"); |
||||
|
||||
g_group_id = strtoull(CH_ID, NULL, 10); |
||||
|
||||
setup_chat_group(); |
||||
setup_db_sync(); |
||||
|
||||
TEST("BGP pubkeys known (db_sync verification)"); { |
||||
if (wait_pubkeys()) OK(); else FAIL(); |
||||
} |
||||
|
||||
TEST("author: index file + media_index_commit"); { |
||||
if (author_index_file() == 0 && db_count(g_inst[I_N1]->topo_sqlite_db, "media_files", NULL, 0) == MEDIA_NUM_BLOCKS) OK(); |
||||
else FAIL(); |
||||
} |
||||
|
||||
TEST("author: publish media message via db_sync"); { |
||||
if (author_publish_message() == 0) OK(); else FAIL(); |
||||
} |
||||
|
||||
/* Фаза 1: сообщение распространилось на все узлы */ |
||||
TEST("message propagated to all 6 nodes"); { |
||||
int a = 0, all = 0; |
||||
while (a < 4000) { |
||||
all = 1; |
||||
for (int i = 0; i < N_NODES; i++) if (db_sync_count(g_si[i]) < 1) all = 0; |
||||
if (all) break; |
||||
uasync_poll(g_ua, POLL_MS); a++; |
||||
} |
||||
if (all) OK(); else FAIL("counts: %d %d %d %d %d %d", |
||||
db_sync_count(g_si[0]), db_sync_count(g_si[1]), db_sync_count(g_si[2]), |
||||
db_sync_count(g_si[3]), db_sync_count(g_si[4]), db_sync_count(g_si[5])); |
||||
} |
||||
|
||||
/* Фаза 2: storage скачали и собрали файл */ |
||||
TEST("storage st1/st2 downloaded + assembled media"); { |
||||
char d1[1024], d2[1024]; |
||||
node_dest(I_ST1, d1, sizeof(d1)); node_dest(I_ST2, d2, sizeof(d2)); |
||||
int a = 0; |
||||
while (a < 8000 && (!g_dl_done[I_ST1] || !g_dl_done[I_ST2])) { uasync_poll(g_ua, POLL_MS); a++; } |
||||
if (g_dl_done[I_ST1] && g_dl_done[I_ST2] && g_dl_err[I_ST1] == 0 && g_dl_err[I_ST2] == 0 |
||||
&& file_exists(d1) && file_exists(d2)) OK(); |
||||
else FAIL("st1 done=%d err=%d exists=%d; st2 done=%d err=%d exists=%d", |
||||
g_dl_done[I_ST1], g_dl_err[I_ST1], file_exists(d1), g_dl_done[I_ST2], g_dl_err[I_ST2], file_exists(d2)); |
||||
} |
||||
|
||||
TEST("storage file content hash matches"); { |
||||
char d1[1024], d2[1024]; |
||||
node_dest(I_ST1, d1, sizeof(d1)); node_dest(I_ST2, d2, sizeof(d2)); |
||||
uint8_t h1[32], h2[32]; |
||||
int ok = (file_sha256(d1, h1) == 0 && file_sha256(d2, h2) == 0 |
||||
&& memcmp(h1, g_content_hash, 32) == 0 && memcmp(h2, g_content_hash, 32) == 0); |
||||
if (ok) OK(); else FAIL(); |
||||
} |
||||
|
||||
/* Фаза 3: суперузлы узнали о блоках + репликация между собой */ |
||||
TEST("supernodes have block_availability (from storage HAVE_BLOCK)"); { |
||||
int a = 0, ok = 0; |
||||
while (a < 2000) { |
||||
int n1 = db_count(g_inst[I_S1]->topo_sqlite_db, "block_availability", NULL, 0); |
||||
int n2 = db_count(g_inst[I_S2]->topo_sqlite_db, "block_availability", NULL, 0); |
||||
if (n1 >= 2 && n2 >= 2) { ok = 1; break; } |
||||
uasync_poll(g_ua, POLL_MS); a++; |
||||
} |
||||
if (ok) OK(); else FAIL("s1=%d s2=%d", |
||||
db_count(g_inst[I_S1]->topo_sqlite_db, "block_availability", NULL, 0), |
||||
db_count(g_inst[I_S2]->topo_sqlite_db, "block_availability", NULL, 0)); |
||||
} |
||||
|
||||
TEST("supernode replication s1↔s2 (super_sync)"); { |
||||
int a = 0, ok = 0; |
||||
while (a < 2000) { |
||||
sqlite3_stmt* st = NULL; uint64_t lr = 0; |
||||
sqlite3_prepare_v2(g_inst[I_S2]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE peer_node_id=?", -1, &st, NULL); |
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)g_nid[I_S1]); |
||||
if (sqlite3_step(st) == SQLITE_ROW) lr = (uint64_t)sqlite3_column_int64(st, 0); |
||||
sqlite3_finalize(st); |
||||
if (lr > 0) { ok = 1; break; } |
||||
uasync_poll(g_ua, POLL_MS); a++; |
||||
} |
||||
if (ok) OK(); else FAIL(); |
||||
} |
||||
|
||||
/* Фаза 4: второй клиент n2 скачивает медиа */ |
||||
TEST("n2 downloads media (on-demand)"); { |
||||
struct media_index_result r; memset(&r, 0, sizeof(r)); |
||||
memcpy(r.media_id, g_media_id, 16); |
||||
memcpy(r.content_hash, g_content_hash, 32); |
||||
r.file_size = MEDIA_FILE_SIZE; r.block_size = MEDIA_BLOCK_SIZE; r.num_blocks = MEDIA_NUM_BLOCKS; |
||||
r.block_ids = u_malloc(MEDIA_NUM_BLOCKS * 16); |
||||
r.block_sigs = u_malloc(MEDIA_NUM_BLOCKS * 64); |
||||
for (int n = 0; n < MEDIA_NUM_BLOCKS; n++) { memcpy(r.block_ids + n * 16, g_block_ids[n], 16); memcpy(r.block_sigs + n * 64, g_block_sigs[n], 64); } |
||||
|
||||
char dest[1024], mbase[512]; |
||||
node_dest(I_N2, dest, sizeof(dest)); |
||||
node_media_base(I_N2, mbase, sizeof(mbase)); |
||||
ensure_supernode_type(I_N2); |
||||
media_download_start(g_inst[I_N2], g_group_id, &r, dest, mbase, g_nid[I_N1], dl_done_cb, (void*)(intptr_t)I_N2, NULL, NULL); |
||||
media_index_result_free(&r); |
||||
|
||||
int a = 0; |
||||
while (a < 8000 && !g_dl_done[I_N2]) { uasync_poll(g_ua, POLL_MS); a++; } |
||||
if (g_dl_done[I_N2] && g_dl_err[I_N2] == 0 && file_exists(dest)) OK(); |
||||
else FAIL("done=%d err=%d exists=%d", g_dl_done[I_N2], g_dl_err[I_N2], file_exists(dest)); |
||||
} |
||||
|
||||
TEST("n2 file content hash matches"); { |
||||
char dest[1024]; node_dest(I_N2, dest, sizeof(dest)); |
||||
uint8_t h[32]; |
||||
if (file_sha256(dest, h) == 0 && memcmp(h, g_content_hash, 32) == 0) OK(); else FAIL(); |
||||
} |
||||
|
||||
fflush(stdout); fflush(stderr); |
||||
printf("\n%d/%d passed, %d failed\n", G_PASSED, G_TOTAL, G_FAILED); fflush(stdout); |
||||
|
||||
done: |
||||
g_result = 1; |
||||
for (int i = 0; i < N_NODES; i++) { if (g_inst[i]) { g_inst[i]->running = 0; utun_instance_destroy(g_inst[i]); g_inst[i] = NULL; } } |
||||
if (g_ua) { uasync_destroy(g_ua, 0); g_ua = NULL; } |
||||
t_rmdir(g_tdir); |
||||
return G_FAILED > 0 ? 1 : 0; |
||||
} |
||||
Loading…
Reference in new issue