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

Validate stream lifecycle and Merkle BGP integration

master
evgeny 2 days ago
parent
1b406202b5
commit
78e0312fee
2 changed files with 30 additions and 8 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 17
      src/stream/stream.c
  2. 21
      tests/test_stream_service.c

17
src/stream/stream.c
Unescape View File

@ -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;

21
tests/test_stream_service.c
Unescape View File

@ -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;

Write Preview
Loading…
Cancel
Save