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

Run media integration replication in its actual CHAT group

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 21:01:24 +02:00
1 parent 00502bf4d4
commit 63c0b9a655
2 files changed
+18 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_media_delivery_full.c
+15 -17
View File
@@ -4,7 +4,7 @@
// Топология: цепь 0→1 1→2 2→3
// Фазы:
// P1: CHAT-группа (topo_groups_create_group + member_sync_put)
// A1: SUPER_HELLO, HAVE_BLOCK, репликация
// A1: роли CHAT, запросы страниц, HAVE_BLOCK, репликация
// A2: admission control (OVERLOADED)
// B1: n1 создаёт файл → n2 инициирует загрузку → стриминг → сборка → проверка
@@ -104,13 +104,13 @@ static void mon(void* arg) {
if (!g_result) uasync_set_timeout(g_ua, 10, NULL, mon, "md_mon");
}
static int msend(struct UTUN_INSTANCE* inst, uint64_t dst, const uint8_t* data, size_t len) {
static int msend(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t dst, const uint8_t* data, size_t len) {
struct ll_entry* e = queue_entry_new(0); if (!e) return -1;
e->dgram = u_malloc(len + 1); if (!e->dgram) { queue_entry_free(e); return -1; }
e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY; memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "[msend] 0x%016llx → 0x%016llx svc=0x%02x subcmd=0x%02x len=%zu",
(unsigned long long)inst->node_id, (unsigned long long)dst, e->dgram[0], data[0], len);
int rc = etcp_route_send(inst, TOPO_GROUP_UTUN, dst, e, 1, 0);
int rc = etcp_route_send(inst, group, dst, e, 1, 0);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "[msend] FAILED rc=%d", rc);
return rc;
}
@@ -175,19 +175,16 @@ static void setup_chat_group(void) {
}
}
/* ── Phase A1: SUPER_HELLO ── */
/* ── Phase A1: автоматический запуск по подписанным ролям CHAT ── */
static void phase_a1_super_hello(void) {
TEST("s1,s2 are supernodes"); {
media_delivery_set_supernode(g_inst[2], 1);
media_delivery_set_supernode(g_inst[3], 1);
media_delivery_refresh_roles(g_inst[2]);
media_delivery_refresh_roles(g_inst[3]);
topo_node_sqlite_nodeinfo_updated(g_inst[2]->topo_sqlite_db, g_nid[2]);
topo_node_sqlite_nodeinfo_updated(g_inst[3]->topo_sqlite_db, g_nid[3]);
if (g_inst[2]->md.is_supernode && g_inst[3]->md.is_supernode) OK(); else FAIL();
}
TEST("SUPER_HELLO s1↔s2"); {
struct media_pkt_super_hello h; h.subcmd = MEDIA_SUBCMD_SUPER_HELLO; h.last_recv_id = 0; h.group_id = TOPO_GROUP_UTUN;
msend(g_inst[2], g_nid[3], (const uint8_t*)&h, sizeof(h));
msend(g_inst[3], g_nid[2], (const uint8_t*)&h, sizeof(h));
TEST("automatic CHAT replication s1↔s2"); {
int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; }
int ok1 = g_inst[2]->md.super_peers && g_inst[2]->md.super_peers->head != NULL;
int ok2 = g_inst[3]->md.super_peers && g_inst[3]->md.super_peers->head != NULL;
@@ -203,18 +200,19 @@ static void phase_a2_have_block_replication(void) {
hb.subcmd = MEDIA_SUBCMD_HAVE_BLOCK; hb.group_id = g_group_id;
memcpy(hb.block_id, uuid, 16); memcpy(hb.media_id, uuid, 16);
hb.chunk = 0; hb.timestamp = (int64_t)time(NULL);
msend(g_inst[1], g_nid[2], (const uint8_t*)&hb, sizeof(hb));
msend(g_inst[1], g_group_id, g_nid[2], (const uint8_t*)&hb, sizeof(hb));
int a = 0; while (a < 500) { if (db_count(g_inst[2]->topo_sqlite_db, "block_availability", "node_id", g_nid[1]) >= 1) break; uasync_poll(g_ua, POLL_MS); a++; }
if (db_count(g_inst[2]->topo_sqlite_db, "block_availability", "node_id", g_nid[1]) >= 1) OK(); else FAIL();
}
TEST("SUPER_REPL s1→s2"); {
TEST("SYNC pages s1→s2"); {
int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; }
if (db_count(g_inst[3]->topo_sqlite_db, "block_availability", NULL, 0) >= 1) OK(); else FAIL();
}
TEST("s2 super_sync updated"); {
sqlite3_stmt* st = NULL; uint64_t lr = 0;
sqlite3_prepare_v2(g_inst[3]->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[2]);
sqlite3_prepare_v2(g_inst[3]->topo_sqlite_db, "SELECT last_recv_id FROM super_sync WHERE group_id=? AND peer_node_id=?", -1, &st, NULL);
sqlite3_bind_int64(st, 1, (sqlite3_int64)g_group_id);
sqlite3_bind_int64(st, 2, (sqlite3_int64)g_nid[2]);
if (sqlite3_step(st) == SQLITE_ROW) lr = (uint64_t)sqlite3_column_int64(st, 0);
sqlite3_finalize(st);
if (lr > 0) OK(); else FAIL();
@@ -287,7 +285,7 @@ static void phase_a3_admission(void) {
struct media_pkt_block_req req; memset(&req, 0, sizeof(req));
req.subcmd = MEDIA_SUBCMD_BLOCK_REQ; assert(!media_index_generate_uuid(req.transfer_id));
memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid0, 16);
msend(g_inst[1], g_nid[0], (const uint8_t*)&req, sizeof(req));
msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req));
int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; }
if (g_inst[0]->md.streams_started >= 1) OK(); else FAIL("started=%d", g_inst[0]->md.streams_started);
}
@@ -297,7 +295,7 @@ static void phase_a3_admission(void) {
struct media_pkt_block_req req; memset(&req, 0, sizeof(req));
req.subcmd = MEDIA_SUBCMD_BLOCK_REQ; assert(!media_index_generate_uuid(req.transfer_id));
memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid0, 16);
msend(g_inst[1], g_nid[0], (const uint8_t*)&req, sizeof(req));
msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req));
}
int a = 0; while (a < 500) { uasync_poll(g_ua, POLL_MS); a++; }
int started = g_inst[0]->md.streams_started - before;
@@ -316,7 +314,7 @@ static void phase_b1_stream_test(void) {
memcpy(req.media_id, g_test_mid, 16); memcpy(req.block_id, g_test_bid1, 16);
req.chunk = 1;
g_inst[0]->md.stream_completed = 0;
msend(g_inst[1], g_nid[0], (const uint8_t*)&req, sizeof(req));
msend(g_inst[1], TOPO_GROUP_UTUN, g_nid[0], (const uint8_t*)&req, sizeof(req));
int a = 0;
while (a < 3000 && !g_inst[0]->md.stream_completed) { uasync_poll(g_ua, POLL_MS); a++; }
if (g_inst[0]->md.stream_completed) OK(); else FAIL("stream not completed after %dms", 3000 * POLL_MS);
tests/test_media_delivery_sql.c
+3 -3
View File
@@ -28,9 +28,9 @@ int main(void) {
/* Настоящая регистрация остаётся целой при пересоздании вспомогательного каталога. */
REQUIRE(!media_index_set_base(a->md.db, "/tmp"));
uint8_t media[16] = {1}, block[16] = {2}, sig[64] = {0};
REQUIRE(!media_index_register_downloaded(a->md.db, media, block, "111", "media-repl-preserved.bin",
SERVER, sig, 1, 1, 1, 0, 0));
uint8_t media[16] = {1}, block[16] = {2}, hash[32] = {0};
REQUIRE(!media_index_register_downloaded(a->md.db, media, block, hash, "111", "media-repl-preserved.bin",
SERVER, 1, 1, 0, 0));
REQUIRE(scalar(a->md.db, "SELECT COUNT(*) FROM media_files") == 1);
media_delivery_destroy(a);
sql(a->topo_sqlite_db, "DROP TABLE block_availability");