diff --git a/doc/stream.txt b/doc/stream.txt index 7497c760..ea51f4ab 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -411,7 +411,9 @@ adm_tags — блок владельца, для самоуправляемых со строгим разбором 64-битных IDs без потери точности через JSON numbers. Зафиксированный формат: stream_source = ":::" - stream_subscriptions = ":,..." + stream_subscriptions = ":[:r],..." +Суффикс :r означает внутренний приём только ради помощи, без локального зрителя. +Локальный viewer на общем экземпляре снимает :r; закрытие последнего viewer возвращает :r, если помощь ещё используется. Отсутствующая/пустая опция означает отсутствие активности. Подписки отсортированы по source_node_id, затем эпохе; нулевые IDs/эпохи, дубликаты, неверные разделители, ведущие нули decimal и переполнение отвергаются. Активная публикация требует записи собственной трансляции в stream_subscriptions. @@ -648,7 +650,7 @@ Stop/destroy выполняются по правилам ядра, вне callb 17. Зафиксированный сетевой формат и пределы реализации Все stream-числа на wire — big-endian, без packed-структур и зависимости от host byte order. -Версия stream wire — 3; обратная совместимость не поддерживается. +Версия stream wire — 4; обратная совместимость не поддерживается. Заголовок raw (52 байта): cmd:1, version:1, command:1, subblock:1, group_id:8, source_node_id:8, source_epoch:8, sender_epoch:8, receiver_epoch:8, handle:8. INFO_REQ/RESP согласуют sender/receiver epochs; receiver_epoch=0 используется INFO_REQ для обнаружения поколения и REVERSE_REQ. @@ -865,3 +867,62 @@ Desktop: Проверки: test_stream_view — настоящее ядро, старые эпохи/cancel, RX pressure, H.264 с поворотом и сменой revision без новой эпохи, Opus/громкость на настоящем null audio backend, закрытие канала и полный баланс tracked allocations. test_stream_service также проверяет общий каталог двух независимых источников одного канала. + +21. Суперузлы: помощь в раздаче по запросу + +Сам старт публикации не запускает приём на суперузлах. Selector зрителя запрашивает помощь после 5 секунд +недостатка действующих субпотоков для k=n-r. Для остановившихся выдач требуется также свидетельство более свежих данных; +пауза источника, отсутствие полной FEC-защиты и недобор всех n сами по себе не запускают помощь. +Выбирается один онлайн-суперузел данной CHAT-группы с допустимой NCD-попыткой; после неуспеха пробуется следующий. +Канальная роль берётся из принятого блока владельца adm_tags.supernode, не из client_type/storage/node_type. +Роль суперузла допускает кандидата и на мобильном устройстве; обычные мобильные ретрансляторы по-прежнему исключены. + +RELAY_REQUEST/STATUS/CANCEL идут только через подписанный и зашифрованный ETCP-router. +Routed ID стриминга — ETCP_RT_ID_STREAM=0x3C; 0x39 принадлежит MEMBER_FILES и не используется стримингом. +DATA остаётся raw ETCP. Управление помощью обрабатывается до local subscription check, но после проверки +канального членства отправителя. Отсутствие удалённой stream-опции из-за задержки Merkle не запрещает запрос. +Заголовок обычный, subblock=0, handle=случайный ненулевой nonce запроса. +sender_epoch REQUEST/CANCEL — отдельная случайная эпоха попытки; receiver_epoch — поколение сервиса помощи. +sender_epoch STATUS — случайное поколение сервиса помощи; receiver_epoch — эпоха исходного запроса. +Эти поколения независимы от raw stream peer epochs и не сбрасываются при транспортном REINIT. + +REQUEST и CANCEL не имеют payload. Первый REQUEST имеет receiver_epoch=0 и только получает подтверждение +поколения через STATUS. Приём/внутренний owner создаётся после повтора REQUEST с подтверждённым receiver_epoch. +Неизвестный запрос с ненулевым receiver_epoch отвергается: позднее подтверждение после удаления tombstone +не может снова создать приём. Начальный старый REQUEST без поколения также не создаёт owner без нового ответа клиента. +STATUS: state:1,reason:1,remaining_ms:4; для ENDED дополнительно неизменённый подписанный stream END (106 байт). +Состояния: STARTING=1, READY=2, BOUND=3, REJECTED=4, ENDED=5, CANCELLED=6. +ENDED в ответе помощи разрешён только для пустого источника (last_seq=UINT64_MAX) и проверяется ключом источника. + +Bootstrap ограничен 15 секундами с первого запроса, повторы срок не продлевают. READY означает наличие +реальных блоков хотя бы одного субпотока с допустимым путём к запросившему; окончательные индекс/путь/FEC +проверяются INFO/ACCEPT/DATA. Переход к READY даёт отдельные 5 секунд для SUBSCRIBE. +READY будит selector и снимает старый backoff данного пира. Подтверждённый поставщик временно допускается +и выборкой кандидатов, и receive_open, независимо от задержки его Merkle-опции. +После первой действующей выдачи запрос становится BOUND; дальнейшую потребность подтверждают обычные RENEW/аренда. +Если помощь не даёт прогресса потребителя 15 секунд и набор по-прежнему недостаточен, её выдачи закрываются, +selector пробует другого суперузла. Запросы/ответы не продлевают аренду raw-выдач. +Состояние закрытого запроса сохраняется 60 секунд; оно также ограничивает частоту повторных попыток к тому же суперузлу. + +Один внутренний owner/экземпляр на точную эпоху обслуживает всех получателей. У него обычные buffer/FEC/selector, +но нет player и пользовательских callbacks, пока нет публичного viewer. Добавление viewer делит тот же буфер. +Режим уже существующего ручного экземпляра сохраняется; запрос старой эпохи не открывает текущую вместо запрошенной. +Внутренний приём удерживает ACTIVE, освобождение восстанавливает запрошенную активность ядра. + +Внутренний owner удерживается, пока есть STARTING/READY запрос или любая downstream-выдача ACTIVE, +включая поздних подписчиков, не отправлявших REQUEST. INFO, upstream, ENDED и CLOSED не учитываются как получатели. +После исчезновения потребности выдерживаются 3 секунды, затем закрывается только внутренний owner. +Публичные viewers и чужое владение общим ETCP сохраняются. После END буфер живёт до завершения оставшихся ACTIVE-выдач. +Аренда downstream — 6 секунд, RENEW — 2 секунды; их существующий контракт для обычных handles не меняется. +Удаление группы/членства, новая эпоха и stop закрывают запросы; снятие supernode-роли закрывает внутреннее владение. + +Relay-only экземпляр не запрашивает новую помощь и не выбирает relay-only подписчика поставщиком, кроме источника. +Это предотвращает взаимное удержание двух суперузлов разными субпотоками после ухода всех зрителей. +Цепочки исключительно вспомогательных ретрансляторов не строятся; обычные подписчики, включая суперузлы с viewers, допускаются. +Максимум 8 внутренних экземпляров помощи и 128 записей запросов вместе с tombstones на ядро. +Память каждого кольца ограничена 32 МБ; пути/метрики имеют обычные дополнительные пределы stream. +Отказы по роли, эпохе, лимитам, таймауту и перегрузке диагностируются с причиной. + +test_stream_service проверяет отсутствие помощи при прямом успехе/паузе источника, автоматический fallback, +раздачу нескольким зрителям, общий owner с локальным viewer, остановку при уходе последнего, +поздний повтор запроса, исключение relay-only кандидатов и передачу пустого подписанного END. diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 893ad97d..9af76c09 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -32,14 +32,15 @@ struct stream_relay_request { struct stream_relay_request *next; struct stream_descriptor descriptor; uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq; - int outgoing, status, reason, dirty; + int outgoing, status, reason, dirty, granted; uint8_t end[STREAM_END_SIZE]; }; struct stream_relay { struct stream_relay *next; struct stream_descriptor descriptor; struct STREAM_HANDLE *owner; - uint64_t idle_ms; + uint64_t idle_ms, checked_ms; + int allowed; }; enum stream_peer_reset_flags { STREAM_RESET_ROTATE = 1, STREAM_RESET_KEEP_PENDING = 2 }; @@ -219,6 +220,10 @@ static int route_control(struct stream_network *net, uint64_t peer, struct strea if (length) memcpy(e->dgram + 1 + STREAM_WIRE_HEADER, payload, length); e->len = 1 + STREAM_WIRE_HEADER + length; int result = etcp_route_send(inst, m.descriptor.group_id, peer, e, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT); + if (m.command >= STREAM_CMD_RELAY_REQUEST) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, + "relay control queued self=%llx peer=%llx group=%llx command=%u id=%llx sender=%llx receiver=%llx result=%d", + (unsigned long long)inst->node_id, (unsigned long long)peer, (unsigned long long)m.descriptor.group_id, m.command, + (unsigned long long)m.handle, (unsigned long long)m.sender_epoch, (unsigned long long)m.receiver_epoch, result); if (result < 0) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control failed peer=%llx command=%u result=%d", (unsigned long long)peer, m.command, result); return result < 0 ? STREAM_ERR_STATE : STREAM_OK; @@ -743,6 +748,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t struct stream_message m; struct stream_local local; if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return; if (m.command >= STREAM_CMD_RELAY_REQUEST) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay dispatcher self=%llx sender=%llx command=%u routed=%d bytes=%zu", + (unsigned long long)net->inst->node_id, (unsigned long long)sender, m.command, routed, length); if (routed) relay_dispatch(net, sender, &m); else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u", (unsigned long long)sender, m.command); @@ -903,6 +910,11 @@ static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */ static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL; + if (entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER + && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] >= STREAM_CMD_RELAY_REQUEST) + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay routed entry self=%llx net=%p bytes=%u flags=%u cmd=%u", + (unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), (void *)net, entry->len, + entry->dgram[ROUTER_SVC_FLAGS_OFF], entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2]); if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER && (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) { uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8); @@ -1016,6 +1028,8 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) stream_network_destroy(net); return STREAM_ERR_MEMORY; } *out = net; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed handler installed self=%llx callback=%p net=%p", + (unsigned long long)inst->node_id, (void *)receive_routed, (void *)net); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network initialized raw=%u routed=%u peers=%u flows=%u controls=%u lease=6000ms renew=2000ms", ETCP_ID_STREAM, ETCP_RT_ID_STREAM, STREAM_PEER_LIMIT, STREAM_FLOW_LIMIT, STREAM_CONTROL_LIMIT); return STREAM_OK; @@ -1343,6 +1357,10 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request nonce conflicts peer=%llx id=%llx", (unsigned long long)sender, (unsigned long long)m->handle); return; } + if (!r && m->command == STREAM_CMD_RELAY_REQUEST && m->receiver_epoch) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay expired request confirmation ignored peer=%llx id=%llx", + (unsigned long long)sender, (unsigned long long)m->handle); return; + } if (!r) r = relay_request_new(net, &m->descriptor, sender, m->handle, m->sender_epoch, 0); if (!r) return; if (m->command == STREAM_CMD_RELAY_CANCEL) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); @@ -1359,8 +1377,11 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st && member.subscriptions[i].source_epoch == m->descriptor.source_epoch && member.subscriptions[i].relay_only) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_LOOP, now); return; } + /* Первый REQUEST только подтверждает поколение сервиса. Старый пакет после tombstone не запускает приём. */ + if (!m->receiver_epoch) return; if (!peer_open(net, sender)) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); return; } int result = relay_open(net, &m->descriptor); + if (result == STREAM_OK) r->granted = 1; if (result != STREAM_OK && result != STREAM_WAIT) relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now); } @@ -1396,7 +1417,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { (unsigned long long)r->id, r->status); relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); } - if (!r->outgoing && r->status == STREAM_RELAY_STARTING) { + if (!r->outgoing && r->granted && r->status == STREAM_RELAY_STARTING) { struct stream_local local; if (stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK) { if (local.finished && local.last == UINT64_MAX) { @@ -1437,8 +1458,11 @@ static void relay_tick(struct stream_network *net, uint64_t now) { for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) if (!r->outgoing && same_stream(&r->descriptor, &relay->descriptor) && r->status <= STREAM_RELAY_READY) pending++; unsigned active = relay_active(net, &relay->descriptor, 0, 0); - int valid = stream_handle_active(relay->owner) - && stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1; + if (now >= relay->checked_ms) { + relay->checked_ms = now + 1000; + relay->allowed = stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1; + } + int valid = stream_handle_active(relay->owner) && relay->allowed; if (valid && (pending || active)) { relay->idle_ms = 0; link = &relay->next; continue; } if (valid && !relay->idle_ms) relay->idle_ms = now; if (valid && now - relay->idle_ms < STREAM_RELAY_IDLE_MS) { link = &relay->next; continue; } diff --git a/src/transport_layer/etcp_api.h b/src/transport_layer/etcp_api.h index d28d54e0..4d3a0d0e 100644 --- a/src/transport_layer/etcp_api.h +++ b/src/transport_layer/etcp_api.h @@ -43,7 +43,8 @@ extern "C" { // router: 0x00 DATA, 0x02 NAT, 0x04 TCP_PROXY_SERVER, 0x05 UDP_PROXY, // 0x06 ICMP_PROXY, 0x07 MEDIA_DELIVERY, 0x08 TCP_PROXY_CLIENT, // 0x11 CONN_MGR, 0x32 ADMIN_KEY, 0x33 JOIN, 0x34 DM, -// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 STREAM, 0x3A GROUP_AVATAR, 0x3B NAT_DETECTION +// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 MEMBER_FILES, +// 0x3A GROUP_AVATAR, 0x3B NAT_DETECTION, 0x3C STREAM // // !!! Новый ID: впиши его в нужный реестр выше и убедись, что значение // не занято в том же пространстве. @@ -58,7 +59,7 @@ extern "C" { #define ETCP_ID_BROADCAST 0x05 // broadcast-сообщения между пирами #define ETCP_ID_RADIO 0x06 // рация (PTT мультикаст голоса) между пирами #define ETCP_ID_STREAM 0x07 // прямой stream INFO/выдачи/DATA -#define ETCP_RT_ID_STREAM 0x39 // stream REVERSE_REQ/CLOSED через группу +#define ETCP_RT_ID_STREAM 0x3C // stream REVERSE_REQ/CLOSED и помощь суперузлов через группу // === Router-сервисы (etcp_router_bind/etcp_route_send) === #define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 15eb121f..defd0527 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1018,6 +1018,10 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test routed callback self=%llx bytes=%u payload_command=%u flags=%u", + (unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), entry->len, + entry->len > ROUTER_SVC_PAYLOAD_OFF + 2 ? entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] : 0, + entry->len > ROUTER_SVC_FLAGS_OFF ? entry->dgram[ROUTER_SVC_FLAGS_OFF] : 0); if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) { memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++; @@ -1035,13 +1039,19 @@ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_eve /* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */ static void test_supernode_relay(void) { - struct TOPO_PEER_REQUEST *connection = NULL; - CHECK(topo_group_peer_open(topo_groups_find(nodes[3]->topo_groups, GROUP2), nodes[0]->node_id, &connection) == 0); + int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE]; + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_TRACE); + int diagnostic_level = g_debug_config.category_levels[DEBUG_CATEGORY_DEBUG]; + debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG); + struct TOPO_PEER_REQUEST *connections[3] = {0}; uint64_t deadline = get_time_tb() + 150000; - while (topo_group_peer_phase(connection) != TOPO_PEER_READY && get_time_tb() < deadline) test_clock_poll(ua, 100); - CHECK(topo_group_peer_phase(connection) == TOPO_PEER_READY); - CHECK(member_sync_start(nodes[3], nodes[0]->node_id, "710002", NULL, NULL) == 0); - CHECK(member_sync_start(nodes[0], nodes[3]->node_id, "710002", NULL, NULL) == 0); + for (unsigned i = 0; i < 3; i++) { + CHECK(topo_group_peer_open(topo_groups_find(nodes[3]->topo_groups, GROUP2), nodes[i]->node_id, &connections[i]) == 0); + while (topo_group_peer_phase(connections[i]) != TOPO_PEER_READY && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(topo_group_peer_phase(connections[i]) == TOPO_PEER_READY); + CHECK(member_sync_start(nodes[3], nodes[i]->node_id, "710002", NULL, NULL) == 0); + CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0); + } relay_role(1, 1, 100); for (unsigned i = 0; i < 3; i++) { relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; @@ -1049,6 +1059,8 @@ static void test_supernode_relay(void) { stream_set_event_callback(nodes[i], relay_test_event, NULL); } relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay test original routed handler=%p network=%p", (void *)relay_route_handler, + (void *)stream_network_get(nodes[1])); CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0); struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE}; struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0}; @@ -1127,7 +1139,9 @@ static void test_supernode_relay(void) { CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0); stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL); } - topo_group_peer_close(connection); + for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level); + debug_set_category_level(DEBUG_CATEGORY_DEBUG, diagnostic_level); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END"); } @@ -1157,7 +1171,7 @@ static void test_member_format(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] member format: full 64-bit IDs, canonical lists, duplicate/overflow rejection"); } -int main(void) { +int main(int argc, char **argv) { 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_MEMBER_SYNC, DEBUG_LEVEL_INFO); @@ -1219,6 +1233,15 @@ int main(void) { 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); } + if (argc == 2 && !strcmp(argv[1], "--relay-only")) { + test_supernode_relay(); + for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); + for (unsigned i = 0; i < 4; i++) { stream_set_event_callback(nodes[i], NULL, NULL); utun_instance_destroy(nodes[i]); } + uasync_destroy(ua, 1); test_clock_finish(); + CHECK(u_get_allocated_count() == allocations); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] isolated supernode relay checks complete, no tracked allocations remain"); + return 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;