From 78e0312fee33a5060f1e9c6dd1390cddd77179b9 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:21:58 +0200 Subject: [PATCH] Validate stream lifecycle and Merkle BGP integration --- src/stream/stream.c | 17 ++++++++++++++--- tests/test_stream_service.c | 21 ++++++++++++++++----- 2 files changed, 30 insertions(+), 8 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index 2084a8bd..28eb89bd 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -401,7 +401,10 @@ int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_ struct stream_member_state state; uint8_t key[32]; result = stream_member_load(inst, group_id, source_id, &state, key); if (result != STREAM_OK) return result; - if (!state.published) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id); return STREAM_WAIT; } + if (!state.published) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id); + return STREAM_WAIT; + } return session_open(inst, &state.source, key, 0, out); } @@ -502,14 +505,22 @@ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates group missing"); return STREAM_ERR_STATE; } size_t capacity = (size_t)group->nodes->count; struct stream_candidate *candidates = capacity ? u_calloc((uint32_t)capacity, sizeof(*candidates)) : NULL; - if (capacity && !candidates) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates allocation failed count=%zu", capacity); return STREAM_ERR_MEMORY; } + if (capacity && !candidates) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates allocation failed count=%zu", capacity); + return STREAM_ERR_MEMORY; + } size_t found = 0; for (struct ll_entry *e = group->nodes->head; e; e = e->next) { struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e; if (node->node_id == inst->node_id || !stream_member_online(inst, group->group_id, node->node_id)) continue; struct TOPO_NODE *identity = topo_node_registry_find(inst->topo_groups, node->node_id); int source = node->node_id == s->descriptor.source_node_id; - if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) continue; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1", + (unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source); + if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay", + (unsigned long long)node->node_id); continue; + } struct stream_member_state state; if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK || !stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch)) continue; diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index e582f030..908813b8 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -101,7 +101,7 @@ static void profile(unsigned node, uint64_t group, char json[MS_USERINFO_MAX + 1 static void root_hash(unsigned node, uint64_t group, uint8_t hash[32]) { char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group); - CHECK(merkle_tree_root(nodes[node]->topo_sqlite_db, MT_DATA_MEMBERS, channel, hash) == 0); + CHECK(merkle_sync_read_hash(nodes[node], MT_DATA_MEMBERS, channel, 0, 0, hash) == 0); } /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */ @@ -161,7 +161,8 @@ static void test_member_format(void) { int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); - debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_DISABLED); + debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO); + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_WARN); test_member_format(); size_t allocations = u_get_allocated_count(); test_clock_init(100); utun_instance_set_tun_init_enabled(0); @@ -174,11 +175,11 @@ int main(void) { for (unsigned i = 0; i < 4; i++) { struct SC_MYKEYS keys; CHECK(sc_generate_keypair(&keys) == SC_OK); char priv[65], pub[65], config[1600]; hex(keys.private_key, priv); hex(keys.public_key, pub); - snprintf(config, sizeof(config), "[global]\nmy_private_key=%s\nmy_public_key=%s\ndb_path=%s/node%u\n" + unsigned type = i == 0 || i == 2 ? CLIENT_TYPE_MOBILE : CLIENT_TYPE_DESKTOP; + snprintf(config, sizeof(config), "[global]\nmy_private_key=%s\nmy_public_key=%s\ndb_path=%s/node%u\nclient_type=%u\n" "[server:udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n" - "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n", priv, pub, directory, i); + "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n", priv, pub, directory, i, type); nodes[i] = utun_instance_create_from_str(ua, config); CHECK(nodes[i] != NULL); - nodes[i]->client_type = i == 0 || i == 2 ? CLIENT_TYPE_MOBILE : CLIENT_TYPE_DESKTOP; snprintf(nodes[i]->name, sizeof(nodes[i]->name), "node%u", i); CHECK(utun_core_start(nodes[i]) == 0 && chat_service_start(nodes[i]) == 0 && nodes[i]->stream); for (unsigned g = 0; g < 2; g++) { @@ -207,6 +208,11 @@ int main(void) { test_clock_poll(ua, 100); } for (unsigned i = 0; i < 3; i++) CHECK(topo_group_peer_phase(connections[i]) == TOPO_PEER_READY); + /* Каналы/мемберы стенда добавлены после chat_sync_init: первый обмен запускается явно. */ + for (unsigned i = 1; i < 4; i++) { + CHECK(member_sync_start(nodes[0], nodes[i]->node_id, "710001", NULL, NULL) == 0); + CHECK(member_sync_start(nodes[i], nodes[0]->node_id, "710001", NULL, NULL) == 0); + } test_profile_atomicity(); stream_set_event_callback(nodes[0], on_stream, NULL); stream_set_event_callback(nodes[1], on_stream, NULL); struct STREAM_HANDLE *publisher = NULL, *publisher2 = NULL, *other_group = NULL, *mobile = NULL, *desktop = NULL; @@ -309,6 +315,11 @@ int main(void) { for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); for (unsigned i = 0; i < 4; i++) utun_instance_destroy(nodes[i]); uasync_destroy(ua, 1); test_clock_finish(); + if (u_get_allocated_count() != allocations) { + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR); + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "tracked allocations before=%zu after=%zu", allocations, u_get_allocated_count()); + u_report_unfreed_blocks(); + } CHECK(u_get_allocated_count() == allocations); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream service checks complete, no tracked allocations remain"); return 0;