diff --git a/tests/test_dm_group_routes.c b/tests/test_dm_group_routes.c index a877d1591..cfd997f1a 100644 --- a/tests/test_dm_group_routes.c +++ b/tests/test_dm_group_routes.c @@ -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()); diff --git a/tests/test_dm_schema.c b/tests/test_dm_schema.c index 8b2d734dc..9465bef24 100644 --- a/tests/test_dm_schema.c +++ b/tests/test_dm_schema.c @@ -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;