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

Test pair-route revocation, shared custody and assignment migration

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 22:58:46 +01:00
1 parent d44dd362d3
commit d4ea78d619
2 files changed
+102 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_dm_group_routes.c
+70 -8
View File
@@ -7,8 +7,10 @@
#include "config_updater.h"
#include "chat/chat_core.h"
#include "chat/member_sync.h"
#include "chat/chat_setting.h"
#include "dm/dm_core.h"
#include "dm/dm_mailbox.h"
#include "dm/dm_media.h"
#include "routing_layer/topo_group.h"
#include "routing_layer/topo_node_sqlite.h"
#include "routing_layer/etcp_router.h"
@@ -22,7 +24,7 @@
#include "test_utils.h"
enum { A, S, M, B, NODES };
static const unsigned pairs[][2] = {{A,S}, {A,S}, {S,B}, {S,M}, {A,M}, {M,B}, {A,B}};
static const unsigned pairs[][2] = {{A,S}, {A,S}, {S,B}, {S,M}, {A,M}, {M,B}, {A,B}, {S,M}};
#define GROUPS (sizeof(pairs) / sizeof(pairs[0]))
#define REQUIRE(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "DM routes: %s line=%d", #x, __LINE__); return -1; } } while (0)
#define WAIT(x) do { uint64_t end = get_time_tb() + 70000; while (!(x) && get_time_tb() < end) uasync_poll(f.ua, 1); REQUIRE(x); } while (0)
@@ -32,7 +34,8 @@ static struct {
struct UTUN_INSTANCE* node[NODES];
struct TOPO_GROUP* group[NODES][GROUPS];
char directory[512], conv[64];
etcp_recv_fn receiver;
etcp_recv_fn receiver[NODES];
unsigned media_seen[NODES];
unsigned observed, passed, failed;
} f;
@@ -60,9 +63,14 @@ static int member(unsigned node, unsigned group, unsigned peer, const char* fiel
}
static void observed(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (entry->len >= sizeof(struct SVC_ROUTE_HDR) && entry->dgram[offsetof(struct SVC_ROUTE_HDR, svc_id)] == ETCP_RT_ID_DM_MAILBOX)
f.observed++;
f.receiver(conn, entry);
for (unsigned i = 0; i < NODES; i++) if (conn->instance == f.node[i]) {
if (entry->len >= sizeof(struct SVC_ROUTE_HDR)) {
uint8_t svc = entry->dgram[offsetof(struct SVC_ROUTE_HDR, svc_id)];
if (i == S && svc == ETCP_RT_ID_DM_MAILBOX) f.observed++;
if (svc == ETCP_RT_ID_DM_MEDIA) f.media_seen[i]++;
}
f.receiver[i](conn, entry); return;
}
}
/* Инъекция подписана настоящим A; обходим только исходящую политику отправителя. */
@@ -142,8 +150,10 @@ static int setup(void) {
}
for (unsigned g = 0; g < GROUPS; g++) for (unsigned side = 0; side < 2; side++)
WAIT(topo_group_peer_ready(f.group[pairs[g][side]][g], f.node[pairs[g][1-side]]->node_id));
f.receiver = f.node[S]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE]; REQUIRE(f.receiver);
f.node[S]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE] = observed;
for (unsigned i = 0; i < NODES; i++) {
f.receiver[i] = f.node[i]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE]; REQUIRE(f.receiver[i]);
f.node[i]->api_bindings.callbacks[ETCP_ID_SVC_ROUTE] = observed;
}
uint64_t conv;
REQUIRE(!dm_start(owner, f.node[B]->node_id, f.node[B]->my_keys.public_key, f.node[B]->my_ed25519_pubkey,
"recipient", "107", DM_MODE_DEVICE, &conv));
@@ -233,6 +243,58 @@ static int file_delivery(void) {
return 0;
}
static int storage_assignments(void) {
uint8_t request[24 + 8192]; sqlite3_stmt* st = NULL;
REQUIRE(sqlite3_prepare_v2(f.node[S]->topo_sqlite_db,
"SELECT h.assignment_id,j.body FROM dm_media_holders h JOIN dm_media_jobs j USING(id)", -1, &st, NULL) == SQLITE_OK);
REQUIRE(sqlite3_step(st) == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 16);
memcpy(request, sqlite3_column_blob(st, 0), 16); memcpy(request + 16, &f.node[B]->node_id, 8);
int len = sqlite3_column_bytes(st, 1); REQUIRE(len > 0 && len <= 8192);
memcpy(request + 24, sqlite3_column_blob(st, 1), len); REQUIRE(sqlite3_finalize(st) == SQLITE_OK);
int expires = value(M, "SELECT expires FROM dm_custody"), policy = value(M, "SELECT delete_on_open FROM dm_custody");
REQUIRE(expires > 0 && policy >= 0);
/* После отзыва первого пути прежнее назначение продолжается через 108 при том же ETCP. */
struct ETCP_CONN* transport = instance_find_conn(f.node[S], f.node[M]->node_id); REQUIRE(transport && transport->links_up);
f.group[S][3]->stopping = f.group[M][3]->stopping = 1;
topo_group_revalidate_members(f.group[S][3]); topo_group_revalidate_members(f.group[M][3]);
uint64_t route = 0; REQUIRE(dm_route_group(f.node[S], f.node[M]->node_id, 104, &route) && route == 108);
unsigned seen = f.media_seen[S];
REQUIRE(!dm_media_send(f.node[S], 104, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
WAIT(f.media_seen[S] > seen);
check("revoked S/M group keeps assignment and physical ETCP through second group",
instance_find_conn(f.node[S], f.node[M]->node_id) == transport && transport->links_up &&
value(M, "SELECT count(*) FROM dm_custody_requests") == 1 && value(M, "SELECT expires FROM dm_custody") == expires);
f.group[S][7]->stopping = f.group[M][7]->stopping = 1;
topo_group_revalidate_members(f.group[S][7]); topo_group_revalidate_members(f.group[M][7]);
check("no pair route pauses STORE without changing reservation", dm_media_send(f.node[S], 0, f.node[M]->node_id,
DM_MEDIA_STORE, request, len + 24) != 0 && value(M, "SELECT count(*) FROM dm_custody_requests") == 1 &&
value(M, "SELECT expires FROM dm_custody") == expires);
f.group[S][7]->stopping = f.group[M][7]->stopping = 0;
topo_group_revalidate_members(f.group[S][7]); topo_group_revalidate_members(f.group[M][7]);
/* A дополнительно становится вторым S только в своей группе с M. */
const char* tags = "{\"supernode\":\"yes\",\"ver\":\"2\"}";
uint8_t owner_body[256], signature[64];
int size = member_sync_build_owner_msg(f.node[A]->node_id, tags, owner_body, sizeof(owner_body)); REQUIRE(size > 0);
REQUIRE(!sc_ed25519_sign(f.node[A]->my_ed25519_privkey, owner_body, size, signature));
for (unsigned i = 0; i < 2; i++) REQUIRE(test_member_seed(f.node[i ? M : A], "105", f.node[A]->my_keys.public_key,
f.node[A]->my_ed25519_pubkey, f.node[A]->my_ed25519_privkey, f.node[A]->my_ed25519_privkey, "{}", tags, signature) >= 0);
REQUIRE(dm_peer_role(f.node[M], f.node[A]->node_id, "supernode"));
memset(request, 0x55, 16); seen = f.media_seen[A];
REQUIRE(!dm_media_send(f.node[A], 105, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
WAIT(value(M, "SELECT count(*) FROM dm_custody_requests") == 2 && f.media_seen[A] > seen);
check("two coordinators share one copy with first deadline and policy",
value(M, "SELECT count(*) FROM dm_custody WHERE ready=1") == 1 &&
value(M, "SELECT expires FROM dm_custody") == expires && value(M, "SELECT delete_on_open FROM dm_custody") == policy);
REQUIRE(!member(M, 4, A, "admitted", 0));
/* Без polling: исходящий путь A ещё допустим, входящий M проверяет свежую строку. */
seen = f.media_seen[M]; request[0]++;
REQUIRE(!dm_media_send(f.node[A], 105, f.node[M]->node_id, DM_MEDIA_STORE, request, len + 24));
WAIT(f.media_seen[M] > seen);
check("revoked requester cannot create another assignment", value(M, "SELECT count(*) FROM dm_custody_requests") == 2);
REQUIRE(!member(M, 4, A, "admitted", 1));
return 0;
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
if (getenv("UTUN_TEST_DEBUG")) {
@@ -241,7 +303,7 @@ int main(void) {
debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_INFO);
}
size_t before = u_get_allocated_count();
if (setup() || run() || file_delivery()) f.failed++;
if (setup() || run() || file_delivery() || storage_assignments()) f.failed++;
for (unsigned i = 0; i < NODES; i++) if (f.node[i]) utun_instance_destroy(f.node[i]);
if (f.ua) uasync_destroy(f.ua, 0);
check("fixture releases all allocations", before == u_get_allocated_count());
tests/test_dm_schema.c
+32
View File
@@ -30,8 +30,40 @@ static int fail_route_drop(void* arg, int action, const char* database, const ch
return action == SQLITE_ALTER_TABLE && name && !strcmp(name, "dm_mail_receipts") ? SQLITE_DENY : SQLITE_OK;
}
/* Ошибка удаления прежнего координатора откатывает и выдачу assignment_id. */
static int fail_custody_drop(void* arg, int action, const char* database, const char* name, const char* db, const char* trigger) {
(void)arg; (void)database; (void)db; (void)trigger;
return action == SQLITE_ALTER_TABLE && name && !strcmp(name, "dm_custody") ? SQLITE_DENY : SQLITE_OK;
}
static void media_migration(void) {
struct UTUN_INSTANCE inst = {0}; assert(sqlite3_open(":memory:", &inst.topo_sqlite_db) == SQLITE_OK);
sqlite3* db = inst.topo_sqlite_db;
sql(db, "CREATE TABLE dm_media_holders(id BLOB,node_id INTEGER,group_id INTEGER,released INTEGER DEFAULT 0,PRIMARY KEY(id,node_id));"
"INSERT INTO dm_media_holders VALUES(x'01',88,101,0);"
"CREATE TABLE dm_custody(id BLOB PRIMARY KEY,author INTEGER,recipient INTEGER,plain_size INTEGER,cipher_size INTEGER,"
"body BLOB,body_hash BLOB,group_id INTEGER,supernode INTEGER,expires INTEGER,ready INTEGER,receipt BLOB,retry_at INTEGER,"
"delete_on_open INTEGER); INSERT INTO dm_custody VALUES(x'01',11,22,100,116,x'ABCD',zeroblob(32),101,77,99999,1,NULL,123,0)");
assert(sqlite3_set_authorizer(db, fail_custody_drop, NULL) == SQLITE_OK);
assert(dm_media_schema_init(&inst) < 0);
assert(sqlite3_set_authorizer(db, NULL, NULL) == SQLITE_OK && sqlite3_get_autocommit(db));
assert(number(db, "SELECT count(*) FROM pragma_table_info('dm_media_holders') WHERE name='assignment_id'") == 0);
assert(number(db, "SELECT supernode FROM dm_custody") == 77);
assert(!dm_media_schema_init(&inst));
sql(db, "CREATE TEMP TABLE first_assignment AS SELECT assignment_id FROM dm_media_holders");
assert(!dm_media_schema_init(&inst));
assert(number(db, "SELECT count(*) FROM dm_media_holders h JOIN first_assignment f USING(assignment_id)"
" WHERE h.node_id=88 AND length(h.assignment_id)=16 AND h.released=0") == 1);
assert(number(db, "SELECT count(*) FROM dm_custody WHERE author=11 AND recipient=22 AND expires=99999"
" AND ready=1 AND delete_on_open=0 AND retry_at=123 AND hex(body)='ABCD'") == 1);
assert(number(db, "SELECT count(*) FROM pragma_table_info('dm_custody') WHERE name IN ('group_id','supernode')") == 0);
assert(number(db, "SELECT count(*) FROM dm_custody_requests") == 0);
assert(sqlite3_close(db) == SQLITE_OK);
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_DEBUG);
media_migration();
struct UTUN_INSTANCE inst = {.node_id = 55};
assert(sqlite3_open(":memory:", &inst.topo_sqlite_db) == SQLITE_OK);
sqlite3* db = inst.topo_sqlite_db;