utun2/tests/test_stream_service.c

/* Живые CHAT-ядра и Merkle: атомарные опции, локальные владельцы, BGP-кандидаты и teardown. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

#include <openssl/sha.h>

#include "../src/stream/stream.h"
#include "../src/stream/stream_member.h"
#include "../src/stream/stream_priv.h"
#include "../src/chat/chat_core.h"
#include "../src/chat/merkle_tree.h"
#include "../src/routing_layer/topo_node_sqlite.h"
#include "../src/routing_layer/route_crypto.h"
#include "../src/transport_layer/etcp.h"
#include "../src/transport_layer/node_conn_direct.h"
#include "../src/utun_instance.h"
#include "../lib/debug_config.h"
#include "../lib/json_flat.h"
#include "../lib/mem.h"
#include "test_clock.h"
#include "test_utils.h"

#define GROUP1 UINT64_C(710001)
#define GROUP2 UINT64_C(710002)
#define SINGLE_INPUT_CAPACITY (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE - STREAM_BATCH_OVERHEAD)
#define CHECK(expr) do { if (!(expr)) { \
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \
} } while (0)

static struct UTUN_INSTANCE *nodes[4];
static struct UASYNC *ua;
static uint8_t group_private[32], group_public[32], group_x[32], expected[1000];
static struct STREAM_HANDLE *reader1, *reader2;
static unsigned data_events[4], end_events[4], closed_events[4];
static int close_readers_in_callback;
static unsigned network_data[4], network_end[4], accepts, reverse_requests, dropped_renew;
static unsigned network_gaps[4];
static unsigned parallel_data;
static uint64_t network_next[4] = {0, 1, 1, 0};
static uint8_t first_request[STREAM_WIRE_HEADER + 8];
static etcp_recv_fn saved_raw[4], saved_route;
static int drop_renew, gap_batch;
static unsigned pressure_data, overload_seen, selects_seen;
static int ignore_select;
static uint8_t pressure_payload[STREAM_DEFAULT_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER];
static size_t stale_select_length;
static etcp_recv_fn pressure_handlers[4];
static int simulated_cpu;
static unsigned global_data[4];
static unsigned automatic_data[4], automatic_gaps[4], automatic_end[4], automatic_probes[4], automatic_accepts;
static uint64_t automatic_next[4];
static int automatic_drop_relay;
static int automatic_drop_tail;
static etcp_recv_fn automatic_handlers[4];
static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
static etcp_recv_fn relay_raw_handlers[4], relay_route_handler, relay_status_handler;
static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen, relay_cancels_seen;
static int relay_drop_source;
static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
static struct {
    int hold_info, hold_request, hold_ready, check_starts, watch_provider;
    uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
    size_t info_length;
    unsigned source_subscribes, helper_subscribes, accepted, ready_held;
    uint64_t request_tb, start, epoch, provider_tb, provider_info_tb;
} relay_race;
static struct {
    unsigned data, reinits, responses;
    int hold_info;
    etcp_recv_fn receive;
    uint8_t first[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
    uint8_t latest[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
    size_t first_length, latest_length;
} bootstrap;
static void run_until(unsigned *value, unsigned expected_value);
err_t __real_uasync_get_load(struct UASYNC *async, struct uasync_load *load);
err_t __wrap_uasync_get_load(struct UASYNC *async, struct uasync_load *load) {
    err_t result = __real_uasync_get_load(async, load);
    if (simulated_cpu && result == ERR_OK) { load->average_percent = load->current_percent = 95; load->valid = 1; }
    return result;
}

static void on_global_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
    CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
    if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == SINGLE_INPUT_CAPACITY); global_data[node]++; }
    else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END);
}

static void automatic_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
    if (entry->len >= STREAM_WIRE_HEADER) {
        uint8_t command = entry->dgram[2];
        if (command == STREAM_CMD_INFO_REQ) automatic_probes[node]++;
        if (command == STREAM_CMD_ACCEPT) automatic_accepts++;
        if (node == 3 && conn->peer_node_id == nodes[2]->node_id && command == STREAM_CMD_SUBSCRIBE) {
            memcpy(automatic_request, entry->dgram, STREAM_WIRE_HEADER);
            automatic_request[2] = STREAM_CMD_UNSUBSCRIBE; automatic_request[STREAM_WIRE_HEADER] = STREAM_REASON_TRANSPORT;
        }
        if (automatic_drop_relay && node == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_DATA) {
            queue_dgram_free(entry); queue_entry_free(entry); return;
        }
        if (automatic_drop_tail && node == 2 && entry->dgram[3] && (command == STREAM_CMD_DATA || command == STREAM_CMD_END)) {
            queue_dgram_free(entry); queue_entry_free(entry); return;
        }
    }
    automatic_handlers[node](conn, entry);
}

static void on_automatic(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
    CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
    if (event->type == STREAM_EVENT_DATA) {
        CHECK(event->length == stream_input_size(&event->descriptor) && !memcmp(event->data, pressure_payload, event->length));
        CHECK(event->first_seq == automatic_next[node]++); automatic_data[node]++;
    } else if (event->type == STREAM_EVENT_GAP) {
        CHECK(event->first_seq == automatic_next[node] && event->last_seq >= event->first_seq);
        automatic_next[node] = event->last_seq + 1; automatic_gaps[node]++;
    } else if (event->type == STREAM_EVENT_END) automatic_end[node]++;
    else CHECK(event->type == STREAM_EVENT_CLOSED);
}

struct restart_events { uint64_t next; unsigned data, gaps; };
static void on_restart(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)inst; struct restart_events *events = arg;
    if (event->type == STREAM_EVENT_DATA) {
        CHECK(event->first_seq == events->next++ && event->length == sizeof(expected));
        CHECK(!memcmp(event->data, expected, sizeof(expected))); events->data++;
    } else if (event->type == STREAM_EVENT_GAP) {
        CHECK(event->first_seq == events->next && event->last_seq >= event->first_seq);
        events->next = event->last_seq + 1; events->gaps++;
    } else CHECK(event->type == STREAM_EVENT_CLOSED);
}

/* Два restart до отложенного GAP сохраняют ещё не сообщённую приложению границу. */
static void test_restart_bound(struct STREAM_HANDLE *subscriber, const struct stream_descriptor *descriptor) {
    struct restart_events events = {.next = 50};
    struct stream_encoder *encoder = NULL; const uint8_t *blocks[RS_MAX_BLOCKS];
    size_t size = 0;
    CHECK(stream_encoder_create(&encoder, descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK);
    stream_set_event_callback(nodes[3], on_restart, &events);
    CHECK(stream_encoder_encode(encoder, 50, 1, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK);
    for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK);
    run_until(&events.data, 1);
    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
    CHECK(stream_encoder_encode(encoder, 60, 2, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK);
    CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_OK);
    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
    CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_DUPLICATE);
    for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK);
    run_until(&events.data, 2);
    CHECK(events.gaps == 1 && events.next == 61);
    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
    CHECK(stream_ingest_block(nodes[3], blocks[0], size) == STREAM_DUPLICATE);
    test_clock_run_for(ua, 1000);
    struct stream_local local; CHECK(stream_local_get(nodes[3], descriptor, &local) == STREAM_OK);
    CHECK(local.anchored && local.start == 61 && local.next == 61 && events.data == 2 && events.gaps == 1);
    /* Неверная подпись пачки даёт ровно один GAP и сохраняет owner/приём следующего индекса. */
    CHECK(stream_encoder_encode(encoder, 61, 3, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK);
    uint8_t bad[STREAM_BLOCK_MAX_SIZE]; memcpy(bad, blocks[0], size); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1;
    CHECK(stream_ingest_block(nodes[3], bad, size) == STREAM_OK);
    for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK);
    run_until(&events.gaps, 2);
    CHECK(events.data == 2 && events.next == 62 && stream_handle_active(subscriber));
    CHECK(stream_encoder_encode(encoder, 62, 4, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK);
    for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], size) == STREAM_OK);
    run_until(&events.data, 3);
    CHECK(events.gaps == 2 && events.next == 63 && stream_handle_active(subscriber));
    struct stream_info info;
    CHECK(stream_get_info(subscriber, &info) == STREAM_OK && info.buffer.rejected_batches == 1);
    stream_set_event_callback(nodes[3], NULL, NULL); stream_encoder_destroy(encoder);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] restart and rejected batch: exact GAP, unchanged owner, following authenticated DATA");
}

static void pressure_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
    if (entry->len >= STREAM_WIRE_HEADER) {
        if (entry->dgram[2] == STREAM_CMD_OVERLOAD) overload_seen++;
        if (entry->dgram[2] == STREAM_CMD_OVERLOAD_SELECT) {
            selects_seen++; CHECK(entry->len <= sizeof(stale_select));
            memcpy(stale_select, entry->dgram, entry->len); stale_select_length = entry->len;
            if (ignore_select) { queue_dgram_free(entry); queue_entry_free(entry); return; }
        }
    }
    pressure_handlers[node](conn, entry);
}

static void on_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg; CHECK(inst == nodes[1] && event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
    if (event->type == STREAM_EVENT_DATA) {
        CHECK(event->length == sizeof(pressure_payload) && !memcmp(event->data, pressure_payload, sizeof(pressure_payload))); pressure_data++;
    } else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END);
}

static struct stream_peer_load_info source_load(void) {
    struct stream_peer_load_info info; CHECK(stream_peer_get_load(nodes[3], nodes[1]->node_id, &info) == STREAM_OK);
    CHECK(info.load.arrivals.bytes == info.load.sent.bytes + info.load.removed.bytes + info.load.queued.bytes);
    CHECK(info.load.arrivals.packets == info.load.sent.packets + info.load.removed.packets + info.load.queued.packets);
    return info;
}

/* Протокол проходит настоящие ETCP/router обработчики; перехват только считает команды и имитирует потерю RENEW. */
static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
    if (entry->len >= STREAM_WIRE_HEADER) {
        uint8_t command = entry->dgram[2];
        if (node == 3 && command == STREAM_CMD_SUBSCRIBE && entry->dgram[3] == 0 && !first_request[0]) {
            CHECK(entry->len == sizeof(first_request)); memcpy(first_request, entry->dgram, sizeof(first_request));
        }
        if (node == 1 && command == STREAM_CMD_ACCEPT) accepts++;
        /* Дыра поставщика внутри его окна: заменить DATA пачки 2 на честный GAP 2->3, сохранив сессию/handle. */
        if (node == 1 && command == STREAM_CMD_DATA && gap_batch) {
            struct stream_message message; struct stream_path path; size_t used;
            CHECK(stream_wire_parse(entry->dgram, entry->len, &message) == STREAM_OK);
            CHECK(stream_path_parse(message.payload, message.length, &path, &used) == STREAM_OK);
            if (stream_read_be(message.payload + used + STREAM_BLOCK_SEQ_OFF, 8) == 2) {
                entry->dgram[2] = STREAM_CMD_GAP; entry->len = STREAM_WIRE_HEADER + 16;
                stream_write_be(entry->dgram + STREAM_WIRE_HEADER, 2, 8);
                stream_write_be(entry->dgram + STREAM_WIRE_HEADER + 8, 3, 8);
            }
        }
        if (node == 3 && command == STREAM_CMD_RENEW && drop_renew && stream_read_be(entry->dgram + 4, 8) == GROUP2) {
            dropped_renew++; queue_dgram_free(entry); queue_entry_free(entry); return;
        }
    }
    saved_raw[node](conn, entry);
}

static void network_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
    if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
        && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_REVERSE_REQ) reverse_requests++;
    saved_route(conn, entry);
}

static void on_network(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg; unsigned node = inst == nodes[1] ? 1 : 2;
    CHECK(event->descriptor.source_node_id == nodes[3]->node_id);
    if (event->descriptor.group_id == GROUP1) {
        if (event->type == STREAM_EVENT_DATA) {
            CHECK(node == 1 && event->first_seq == parallel_data && event->length == sizeof(expected));
            CHECK(!memcmp(event->data, expected, sizeof(expected))); parallel_data++;
        } else CHECK(event->type == STREAM_EVENT_CLOSED);
        return;
    }
    CHECK(event->descriptor.group_id == GROUP2);
    if (event->type == STREAM_EVENT_DATA) {
        CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected)));
        CHECK(event->first_seq == network_next[node]++); network_data[node]++;
    } else if (event->type == STREAM_EVENT_END) network_end[node]++;
    else if (event->type == STREAM_EVENT_GAP) {
        CHECK(event->first_seq == network_next[node] && event->last_seq == 2);
        network_next[node] = event->last_seq + 1; network_gaps[node]++;
    }
    else CHECK(event->type == STREAM_EVENT_CLOSED);
}

static void hex(const uint8_t *input, char *out) {
    for (unsigned i = 0; i < 32; i++) snprintf(out + i * 2, 3, "%02x", input[i]);
}

/* Проверяем заимствованные байты и после закрытия последнего локального владельца внутри callback. */
static void on_stream(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg;
    unsigned node = 0;
    while (node < 4 && nodes[node] != inst) node++;
    CHECK(node < 4 && (event->descriptor.group_id == GROUP1 || event->type == STREAM_EVENT_CLOSED));
    if (event->type == STREAM_EVENT_DATA) {
        CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected)));
        data_events[node]++;
        if (node == 1 && close_readers_in_callback) {
            close_readers_in_callback = 0;
            stream_close(reader1); reader1 = NULL;
            stream_close(reader2); reader2 = NULL;
            CHECK(!memcmp(event->data, expected, sizeof(expected)));
        }
    } else if (event->type == STREAM_EVENT_END) end_events[node]++;
    else if (event->type == STREAM_EVENT_CLOSED) closed_events[node]++;
    else CHECK(0);
}

/* Реальные удостоверения участников: join/update/tree подписи, одинаковые на всех узлах. */
static void put_member(unsigned destination, unsigned member, uint64_t group) {
    char channel[32], userinfo[256], name[32];
    snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group);
    snprintf(name, sizeof(name), "node%u", member);
    snprintf(userinfo, sizeof(userinfo), "{\"name\":\"%s\",\"pad\":\"retained\",\"avatar_server\":\"0011223344556677\","
             "\"avatar_file\":\"retained-avatar\"}", name);
    uint8_t join[64], tree[64], update[64], body[MS_USERINFO_MAX + 112];
    int size = member_sync_build_join_msg(group_x, group_public, nodes[member]->node_id,
                                        nodes[member]->my_keys.public_key, 123, body, sizeof(body));
    CHECK(size > 0 && sc_ed25519_sign(nodes[member]->my_ed25519_privkey, body, (size_t)size, join) == SC_OK);
    CHECK(member_sync_sign_pubkey(group_private, nodes[member]->my_keys.public_key, tree) == 0);
    size = member_sync_build_update_msg(join, 124, userinfo, body, sizeof(body));
    CHECK(size > 0 && sc_ed25519_sign(nodes[member]->my_ed25519_privkey, body, (size_t)size, update) == SC_OK);
    CHECK(member_sync_put(nodes[destination], channel, nodes[member]->node_id, nodes[member]->my_keys.public_key,
          nodes[member]->my_ed25519_pubkey, join, 123, update, 124, userinfo, NULL, NULL, 0, 0, tree) >= 0);
}

static void run_until(unsigned *value, unsigned expected_value) {
    uint64_t deadline = get_time_tb() + 100000;
    while (*value < expected_value && get_time_tb() < deadline) test_clock_poll(ua, 100);
    if (*value != expected_value) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream test wait expected=%u actual=%u data=%u/%u end=%u/%u closed=%u/%u",
        expected_value, *value, data_events[1], data_events[2], end_events[1], end_events[2], closed_events[1], closed_events[2]);
    CHECK(*value == expected_value);
}

/* Дожидаемся именно Merkle-профиля, не копируем stream-опции вручную между ядрами. */
static void wait_state(unsigned receiver, unsigned member, uint64_t group, uint64_t epoch, int subscribed) {
    uint64_t deadline = get_time_tb() + 150000;
    while (get_time_tb() < deadline) {
        struct stream_member_state state;
        int result = stream_member_load(nodes[receiver], group, nodes[member]->node_id, &state, NULL);
        if (result == STREAM_OK && (subscribed ? stream_member_has_subscription(&state, nodes[0]->node_id, epoch)
                                             : state.published && state.source.source_epoch == epoch)) return;
        test_clock_poll(ua, 100);
    }
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Merkle stream state timeout receiver=%u member=%u group=%llu epoch=%llx",
                receiver, member, (unsigned long long)group, (unsigned long long)epoch);
    CHECK(0);
}

static void profile(unsigned node, uint64_t group, char json[MS_USERINFO_MAX + 1], uint8_t signature[64], uint64_t *version) {
    char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group);
    CHECK(chat_core_get_my_profile(nodes[node], channel, json, MS_USERINFO_MAX + 1, signature, version) == 0);
}

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_sync_read_hash(nodes[node], MT_DATA_MEMBERS, channel, 0, 0, hash) == 0);
}

static void network_write(struct STREAM_HANDLE *publisher, unsigned seq) {
    size_t accepted = 0;
    CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
    uint64_t deadline = get_time_tb() + 100000;
    struct stream_info info;
    do {
        CHECK(stream_get_info(publisher, &info) == STREAM_OK);
        if (info.buffer.blocks >= (seq + 1) * 8) return;
        test_clock_poll(ua, 100);
    } while (get_time_tb() < deadline);
    CHECK(0);
}

/* Начальный REINIT должен пройти настоящую ETCP-сессию, а не искусственное уведомление stream. */
static void bootstrap_status(struct ETCP_CONN *conn, int status, void *arg) {
    (void)arg;
    if (conn->peer_node_id == nodes[3]->node_id && status == ETCP_CONN_STATUS_REINIT) {
        bootstrap.reinits++;
        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "bootstrap test REINIT peer=%llx ordinal=%u",
                   (unsigned long long)conn->peer_node_id, bootstrap.reinits);
    }
}

/* Перехват удерживает только INFO_RESP проверяемого источника; остальной протокол проходит штатно. */
static void bootstrap_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    if (bootstrap.hold_info && conn->peer_node_id == nodes[3]->node_id && entry->len >= STREAM_WIRE_HEADER
        && entry->dgram[2] == STREAM_CMD_INFO_RESP) {
        CHECK(entry->len <= sizeof(bootstrap.first));
        if (!bootstrap.first_length) {
            memcpy(bootstrap.first, entry->dgram, entry->len); bootstrap.first_length = entry->len;
        }
        memcpy(bootstrap.latest, entry->dgram, entry->len); bootstrap.latest_length = entry->len; bootstrap.responses++;
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test holds INFO_RESP ordinal=%u bytes=%u", bootstrap.responses, entry->len);
        queue_dgram_free(entry); queue_entry_free(entry); return;
    }
    bootstrap.receive(conn, entry);
}

static void bootstrap_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    if (event->descriptor.source_node_id != nodes[3]->node_id) { on_stream(inst, event, arg); return; }
    if (event->type == STREAM_EVENT_CLOSED) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test local reader closed reason=%d", event->reason); return;
    }
    CHECK(event->descriptor.group_id == GROUP1 && event->type == STREAM_EVENT_DATA && event->length == sizeof(expected));
    CHECK(!memcmp(event->data, expected, sizeof(expected))); bootstrap.data++;
}

/* Холодное подключение, запоздалый ответ, ограниченный срок запроса и настоящий DOWN/DELETE. */
static void test_bootstrap(void) {
    debug_level_t transport_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCP];
    debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_INFO);
    struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL;
    struct stream_parameters parameters = {.n = 1, .r = 0, .block_size = STREAM_RS_DEFAULT_SIZE};
    CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    wait_state(1, 3, GROUP1, origin.descriptor.source_epoch, 0);
    CHECK(!instance_find_conn(nodes[1], nodes[3]->node_id));
    bootstrap.receive = nodes[1]->api_bindings.callbacks[ETCP_ID_STREAM];
    CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, bootstrap_raw) == 0);
    etcp_add_conn_status_cbk(nodes[1], bootstrap_status, NULL);
    stream_set_event_callback(nodes[1], bootstrap_event, NULL);
    for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3);
    size_t accepted = 0;
    CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
    uint64_t started = get_time_tb();
    CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[3]->node_id, &subscriber) == STREAM_OK);
    run_until(&bootstrap.data, 1);
    struct stream_selection_info selected; CHECK(stream_selection_get_info(subscriber, &selected) == STREAM_OK);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "bootstrap result REINIT=%u probes=%llu subflows=%u failures=%llu",
               bootstrap.reinits, (unsigned long long)selected.probes_started, selected.subflows,
               (unsigned long long)selected.failures);
    CHECK(bootstrap.reinits == 1 && selected.probes_started == 1 && selected.subflows == 1 && !selected.failures);
    CHECK(get_time_tb() - started < 100000);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] first automatic subscription survives initial REINIT: elapsed_ms=%llu probes=%llu",
               (unsigned long long)((get_time_tb() - started) / 10), (unsigned long long)selected.probes_started);
    stream_close(subscriber); subscriber = NULL;

    CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &subscriber) == STREAM_OK);
    bootstrap.hold_info = 1;
    struct STREAM_RECEIVE *probe = NULL;
    CHECK(stream_probe_open(subscriber, nodes[3]->node_id, 0, &probe) == STREAM_OK);
    run_until(&bootstrap.responses, 1);
    struct ETCP_CONN *conn = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(conn && conn->links_up);
    etcp_conn_reinit(conn, "stream pending INFO test");
    run_until(&bootstrap.responses, 2);
    struct stream_message old, current;
    CHECK(stream_wire_parse(bootstrap.first, bootstrap.first_length, &old) == STREAM_OK);
    CHECK(stream_wire_parse(bootstrap.latest, bootstrap.latest_length, &current) == STREAM_OK);
    CHECK(old.receiver_epoch != current.receiver_epoch && stream_read_be(old.payload, 8) != stream_read_be(current.payload, 8));
    struct stream_receive_info info;
    CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.state == STREAM_RECEIVE_INFO && !info.info_ready);
    for (unsigned replay = 0; replay < 2; replay++) {
        struct ll_entry *entry = ll_alloc_lldgram((uint16_t)bootstrap.first_length); CHECK(entry);
        memcpy(entry->dgram, bootstrap.first, bootstrap.first_length); entry->len = bootstrap.first_length;
        if (replay) {
            old.sender_epoch = current.sender_epoch; old.receiver_epoch = current.receiver_epoch;
            stream_wire_header(entry->dgram, &old); /* Отдельно проверяем nonce внутри уже актуальной сессии. */
        }
        bootstrap.receive(conn, entry);
        CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.state == STREAM_RECEIVE_INFO && !info.info_ready);
    }
    struct ll_entry *entry = ll_alloc_lldgram((uint16_t)bootstrap.latest_length); CHECK(entry);
    memcpy(entry->dgram, bootstrap.latest, bootstrap.latest_length); entry->len = bootstrap.latest_length;
    bootstrap.receive(conn, entry);
    CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.info_ready && !info.handle && !info.blocks);
    stream_receive_close(probe); probe = NULL;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] pending INFO survives REINIT; old epoch/nonce rejected; fresh response accepted");

    started = get_time_tb();
    CHECK(stream_probe_open(subscriber, nodes[3]->node_id, 0, &probe) == STREAM_OK);
    for (unsigned reset = 0; reset < 2; reset++) {
        test_clock_run_for(ua, 20000);
        CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.state != STREAM_RECEIVE_CLOSED && !info.info_ready);
        etcp_conn_reinit(conn, "stream pending deadline test");
    }
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(probe, &info) == STREAM_OK); }
    while (info.state != STREAM_RECEIVE_CLOSED && get_time_tb() - started < 60000);
    CHECK(info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_TRANSPORT && get_time_tb() - started < 60000);
    stream_receive_close(probe); probe = NULL;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated REINIT keeps original pending deadline: elapsed_ms=%llu",
               (unsigned long long)((get_time_tb() - started) / 10));

    stream_close(subscriber); stream_close(publisher);
    test_clock_run_for(ua, 1000); /* Доставить отложенные CLOSED до возврата общего обработчика стенда. */
    CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, bootstrap.receive) == 0);
    etcp_remove_conn_status_cbk(nodes[1], bootstrap_status, NULL);
    stream_set_event_callback(nodes[1], on_stream, NULL);
    debug_set_category_level(DEBUG_CATEGORY_ETCP, transport_level);
}

/* Разрушающий транспорт сценарий выполняется после проверок, которым нужен готовый общий стенд. */
static void test_pending_down(void) {
    struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL;
    CHECK(stream_publish_open(nodes[3], GROUP1, NULL, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    wait_state(1, 3, GROUP1, origin.descriptor.source_epoch, 0);
    stream_set_event_callback(nodes[1], bootstrap_event, NULL);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &subscriber) == STREAM_OK);
    struct STREAM_RECEIVE *probe = NULL;
    CHECK(stream_probe_open(subscriber, nodes[3]->node_id, 0, &probe) == STREAM_OK);
    struct ETCP_CONN *conn = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(conn && conn->links_up);
    etcp_connection_close(conn); /* Реальный DOWN и DELETE проходят через NCD. */
    struct stream_receive_info info;
    CHECK(stream_receive_get_info(probe, &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED
          && info.reason == STREAM_REASON_TRANSPORT);
    stream_receive_close(probe); stream_close(subscriber); stream_close(publisher);
    test_clock_run_for(ua, 1000);
    stream_set_event_callback(nodes[1], on_stream, NULL);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] transport DOWN/DELETE still cancels pending request");
}

/* Источник -> relay -> получатель; данные передаются через raw ETCP, никаких прямых stream_ingest в этой проверке. */
static void test_network(void) {
    for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3);
    struct TOPO_PEER_REQUEST *connections[2] = {0};
    struct TOPO_GROUP *root = topo_groups_find(nodes[3]->topo_groups, GROUP2);
    for (unsigned i = 1; i < 3; i++) CHECK(topo_group_peer_open(root, nodes[i]->node_id, &connections[i - 1]) == 0);
    uint64_t deadline = get_time_tb() + 150000;
    while (get_time_tb() < deadline
           && (topo_group_peer_phase(connections[0]) != TOPO_PEER_READY || topo_group_peer_phase(connections[1]) != TOPO_PEER_READY))
        test_clock_poll(ua, 100);
    for (unsigned i = 1; i < 3; i++) {
        CHECK(topo_group_peer_phase(connections[i - 1]) == 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);
    }
    struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0};
    CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    /* Подписанный END отдельно отвергает подмену подписи, индекса и группы. */
    uint8_t end[STREAM_END_SIZE]; uint64_t last;
    CHECK(stream_end_sign(&origin.descriptor, UINT64_MAX - 1, nodes[3]->my_ed25519_privkey, end) == STREAM_OK);
    CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_OK && last == UINT64_MAX - 1);
    end[105] ^= 1; CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_ERR_SIGNATURE);
    end[105] ^= 1; end[41] ^= 1;
    CHECK(stream_end_verify(&origin.descriptor, nodes[3]->my_ed25519_pubkey, end, sizeof(end), &last) == STREAM_ERR_SIGNATURE);
    for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
    for (unsigned i = 1; i < 3; i++) {
        CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
        stream_set_event_callback(nodes[i], on_network, NULL);
    }
    network_write(publisher, 0); network_write(publisher, 1);
    for (unsigned i = 1; i < 4; i++) {
        saved_raw[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(saved_raw[i] != NULL);
        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, network_raw) == 0);
    }
    saved_route = nodes[3]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; CHECK(saved_route != NULL);
    CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, network_routed) == 0);
    struct STREAM_RECEIVE *up[8] = {0}, *down[8] = {0};
    for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, i, UINT64_MAX, &up[i]) == STREAM_OK);
    run_until(&network_data[1], 1);
    for (unsigned i = 0; i < 8; i++) {
        struct stream_receive_info info; CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK);
        if (!info.blocks) { deadline = get_time_tb() + 100000;
            do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK); }
            while (!info.blocks && get_time_tb() < deadline);
        }
        CHECK(info.blocks == 1 && info.start == 1 && info.next == 2 && info.handle);
    }
    /* Меняется только локальная stream generation; физический ETCP и поколение источника сохраняются. */
    struct stream_receive_info previous_generation; CHECK(stream_receive_get_info(up[0], &previous_generation) == STREAM_OK);
    struct ETCP_CONN *stable = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(stable && stable->links_up);
    int notified = 0;
    for (struct etcp_status_cbk_entry *cb = nodes[1]->conn_status_cbks; cb; cb = cb->next)
        if (cb->arg == stream_network_get(nodes[1])) { cb->fn(stable, ETCP_CONN_STATUS_REINIT, cb->arg); notified = 1; break; }
    CHECK(notified && stable->links_up);
    /* Старый DUPLICATE устанавливает общий старт, но не повторяет уже выданную приложению пачку. */
    CHECK(stream_receive_restart(subscribers[1]) == STREAM_OK);
    first_request[0] = 0;
    for (unsigned i = 0; i < 8; i++) {
        struct stream_receive_info info; CHECK(stream_receive_get_info(up[i], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED);
        stream_receive_close(up[i]); up[i] = NULL;
        CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, i, UINT64_MAX, &up[i]) == STREAM_OK);
    }
    struct stream_receive_info current_generation; CHECK(stream_receive_get_info(up[0], &current_generation) == STREAM_OK);
    deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(up[0], &current_generation) == STREAM_OK); }
    while (!current_generation.blocks && get_time_tb() < deadline);
    test_clock_run_for(ua, 1000);
    struct stream_local restarted; CHECK(stream_local_get(nodes[1], &origin.descriptor, &restarted) == STREAM_OK);
    CHECK(current_generation.blocks == 1 && restarted.anchored && restarted.start == 2 && restarted.next == 2);
    CHECK(network_data[1] == 1 && network_next[1] == 2);
    CHECK(current_generation.local_epoch != previous_generation.local_epoch && current_generation.peer_epoch == previous_generation.peer_epoch);
    CHECK(stable == instance_find_conn(nodes[1], nodes[3]->node_id) && stable->links_up);
    /* Node2 видит подписку relay через обычный Merkle-обмен, а не копирование из его SQLite. */
    deadline = get_time_tb() + 150000;
    int ready = 0;
    while (!ready && get_time_tb() < deadline) {
        struct stream_member_state member;
        ready = stream_member_load(nodes[2], GROUP2, nodes[1]->node_id, &member, NULL) == STREAM_OK
                && stream_member_has_subscription(&member, nodes[3]->node_id, origin.descriptor.source_epoch);
        if (!ready) test_clock_poll(ua, 100);
    }
    CHECK(ready);
    for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscribers[2], nodes[1]->node_id, i, UINT64_MAX, &down[i]) == STREAM_OK);
    run_until(&network_data[2], 1);
    struct stream_local retained; CHECK(stream_local_get(nodes[2], &origin.descriptor, &retained) == STREAM_OK);
    struct stream_path path; CHECK(stream_buffer_path(retained.buffer, 1, 0, &path) == STREAM_OK);
    CHECK(path.count == 3 && path.nodes[0] == nodes[3]->node_id && path.nodes[1] == nodes[1]->node_id && path.nodes[2] == nodes[2]->node_id);
    /* Замкнуть сохранённую цепочку обратно в node1 нельзя, даже при актуальной Merkle-подписке. */
    struct STREAM_RECEIVE *loop = NULL;
    CHECK(stream_receive_open(subscribers[1], nodes[2]->node_id, 0, 1, &loop) == STREAM_OK);
    struct stream_receive_info loop_info;
    deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(loop, &loop_info) == STREAM_OK); }
    while (loop_info.state != STREAM_RECEIVE_CLOSED && get_time_tb() < deadline);
    CHECK(loop_info.state == STREAM_RECEIVE_CLOSED && loop_info.reason == STREAM_REASON_LOOP); stream_receive_close(loop);
    /* Повтор SUBSCRIBE сохраняет handle и текущий курсор, не отправляет старый DATA заново. */
    CHECK(first_request[0]); unsigned before = accepts;
    struct ETCP_CONN *conn = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(conn && conn->links_up);
    while (conn->send_input_q->count) test_clock_poll(ua, 100);
    struct ll_entry *duplicate = ll_alloc_lldgram(sizeof(first_request)); CHECK(duplicate != NULL);
    memcpy(duplicate->dgram, first_request, sizeof(first_request)); duplicate->len = sizeof(first_request); CHECK(etcp_send(conn, duplicate) == 0);
    run_until(&accepts, before + 1);
    struct stream_receive_info info; CHECK(stream_receive_get_info(up[0], &info) == STREAM_OK && info.blocks == 1 && info.next == 2);
    /* Вторая группа на том же peer/NCD/ETCP использует иной FEC и сохраняется после закрытия выдач GROUP2. */
    struct STREAM_HANDLE *parallel_publisher = NULL, *parallel_subscriber = NULL;
    struct STREAM_RECEIVE *parallel = NULL; struct stream_parameters parameters = {.n = 1, .r = 0, .block_size = STREAM_RS_DEFAULT_SIZE};
    CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &parallel_publisher) == STREAM_OK);
    struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK);
    wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0);
    /* Две независимые публикации одного канала видны через общий каталог; подписчики не становятся источниками. */
    char *catalog = NULL, origin_id[64], parallel_id[64], subscriber_id[64];
    CHECK(stream_sources_json(nodes[1], GROUP1, &catalog) == STREAM_OK);
    snprintf(origin_id, sizeof(origin_id), "\"id\":\"%llu\"", (unsigned long long)nodes[0]->node_id);
    snprintf(parallel_id, sizeof(parallel_id), "\"id\":\"%llu\"", (unsigned long long)nodes[3]->node_id);
    snprintf(subscriber_id, sizeof(subscriber_id), "\"id\":\"%llu\"", (unsigned long long)nodes[1]->node_id);
    CHECK(strstr(catalog, origin_id) && strstr(catalog, parallel_id) && !strstr(catalog, subscriber_id)); u_free(catalog);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] shared catalogue exposes two live publishers in the same channel");
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &parallel_subscriber) == STREAM_OK);
    size_t accepted = 0;
    CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
    deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK); }
    while (!parallel_info.buffer.blocks && get_time_tb() < deadline);
    CHECK(parallel_info.buffer.blocks == 1);
    CHECK(stream_receive_open(parallel_subscriber, nodes[3]->node_id, 0, UINT64_MAX, &parallel) == STREAM_OK);
    run_until(&parallel_data, 1);
    /* Одна и та же routed reverse-попытка не меняет сессию/действующие handles. */
    struct stream_message reverse; CHECK(stream_wire_parse(first_request, sizeof(first_request), &reverse) == STREAM_OK);
    reverse.command = STREAM_CMD_REVERSE_REQ; reverse.receiver_epoch = reverse.handle = 0;
    for (unsigned i = 0; i < 2; i++) {
        struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry != NULL);
        entry->dgram[0] = ETCP_RT_ID_STREAM; stream_wire_header(entry->dgram + 1, &reverse); entry->len = 1 + STREAM_WIRE_HEADER;
        CHECK(etcp_route_send(nodes[1], GROUP2, nodes[3]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
    }
    run_until(&reverse_requests, 2);
    gap_batch = 1; network_write(publisher, 2); run_until(&network_gaps[1], 1); gap_batch = 0;
    /* Снять два лишних проверочных субпотока: шесть оставшихся восстанавливают новые пачки. */
    for (unsigned i = 6; i < 8; i++) { stream_receive_close(up[i]); up[i] = NULL; stream_receive_close(down[i]); down[i] = NULL; }
    network_write(publisher, 3); run_until(&network_data[1], 2); run_until(&network_data[2], 2); run_until(&network_gaps[2], 1);
    /* Relay пересылает первую часть, пока у него и downstream ещё нет k частей для проверки общей подписи. */
    for (unsigned sub = 1; sub < 6; sub++) { stream_receive_close(up[sub]); up[sub] = NULL; }
    CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.blocks == 2);
    network_write(publisher, 4); deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK); }
    while (info.blocks < 3 && get_time_tb() < deadline);
    CHECK(info.blocks == 3 && network_data[1] == 2 && network_data[2] == 2);
    uint8_t pending_output[sizeof(expected)]; size_t pending_length = 0;
    for (unsigned node = 1; node < 3; node++) {
        struct stream_local pending; CHECK(stream_local_get(nodes[node], &origin.descriptor, &pending) == STREAM_OK);
        CHECK(pending.next == 4 && stream_buffer_decode(pending.buffer, 4, pending_output, sizeof(pending_output), &pending_length) == STREAM_WAIT);
    }
    for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, sub, 4, &up[sub]) == STREAM_OK);
    run_until(&network_data[1], 3); run_until(&network_data[2], 3);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live relay forwards before batch decode; consumers wait for k parts and source signature");
    CHECK(stream_publish_finish(publisher) == STREAM_OK);
    run_until(&network_end[1], 1); run_until(&network_end[2], 1);
    /* RENEW каждые 2с удерживает даже окончившуюся выдачу; затем имитируем отсутствие продлений. */
    test_clock_run_for(ua, 71000);
    CHECK(stream_receive_get_info(up[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_ENDED);
    drop_renew = 1; test_clock_run_for(ua, 71000); drop_renew = 0;
    CHECK(dropped_renew && stream_receive_get_info(up[0], &info) == STREAM_OK
          && info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_LEASE);
    CHECK(stream_receive_get_info(parallel, &info) == STREAM_OK && info.state == STREAM_RECEIVE_ACTIVE);
    CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
    run_until(&parallel_data, 2);
    CHECK(stream_handle_active(reader1) && stream_handle_active(reader2));
    /* REINIT не обязан разрушать ETCP_CONN, но старые stream handles становятся недействительны. */
    conn = instance_find_conn(nodes[2], nodes[1]->node_id); CHECK(conn != NULL);
    etcp_conn_reinit(conn, "stream network test");
    deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK); }
    while (info.state != STREAM_RECEIVE_CLOSED && get_time_tb() < deadline);
    CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED
          && info.reason == STREAM_REASON_TRANSPORT);
    for (unsigned i = 0; i < 8; i++) { stream_receive_close(up[i]); stream_receive_close(down[i]); }
    stream_receive_close(parallel); stream_close(parallel_subscriber); stream_close(parallel_publisher);
    for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, saved_raw[i]) == 0);
    CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, saved_route) == 0);
    for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
    test_clock_run_for(ua, 1000);
    stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
    for (unsigned i = 0; i < 2; i++) topo_group_peer_close(connections[i]);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live ETCP chain: common index/restart, FEC, paths/LOOP, GAP, duplicate/reverse, END, leases, REINIT");
}

/* Настоящий send_input_q намеренно держим непустым: короткий глитч не режет, устойчивый рост вызывает OVERLOAD. */
static void test_pressure(void) {
    memset(pressure_payload, 42, sizeof(pressure_payload));
    struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL;
    CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK);
    stream_set_event_callback(nodes[1], on_pressure, NULL);
    for (unsigned i = 1; i <= 3; i += 2) {
        pressure_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(pressure_handlers[i] != NULL);
        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_raw) == 0);
    }
    size_t accepted = 0;
    CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK && accepted == sizeof(pressure_payload));
    struct STREAM_RECEIVE *receives[8] = {0};
    for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK);
    run_until(&pressure_data, 1);
    uint64_t deadline = get_time_tb() + 100000;
    while (get_time_tb() < deadline) {
        int ready = 1;
        for (unsigned i = 0; i < 8; i++) {
            struct stream_receive_info info; CHECK(stream_receive_get_info(receives[i], &info) == STREAM_OK);
            if (!info.blocks) ready = 0;
            CHECK(info.rtt_valid && info.supplier_metrics_valid);
        }
        if (ready) break; test_clock_poll(ua, 100);
    }
    struct ETCP_CONN *conn = instance_find_conn(nodes[3], nodes[1]->node_id); CHECK(conn && conn->links_up);
    struct ll_queue *queue = conn->send_input_q;
    while (queue->count) test_clock_poll(ua, 100);
    queue_callback_fn callback = queue->callback; void *callback_arg = queue->callback_arg;
    CHECK(callback != NULL); queue_set_callback(queue, NULL, NULL);
    for (unsigned i = 0; i < 4; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
    test_clock_run_for(ua, 3000);
    struct stream_peer_load_info info = source_load(); CHECK(info.load.queued.bytes && !info.request_id && !overload_seen);
    queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
    run_until(&pressure_data, 5);
    test_clock_run_for(ua, 1000); info = source_load(); CHECK(!info.load.queued.bytes && !info.request_id);
    /* Вход 8 пакетов/интервал, выход 7: число parity-закрытий зависит от measured rate; должно сохраниться k=6. */
    while (queue->count) test_clock_poll(ua, 100);
    queue_set_callback(queue, NULL, NULL);
    for (unsigned round = 0; round < 60; round++) {
        CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
        uint64_t target = source_load().load.sent.packets + 7;
        deadline = get_time_tb() + 100000;
        while (source_load().load.sent.packets < target && !source_load().released_count && get_time_tb() < deadline) {
            if (queue->count) callback(queue, callback_arg);
            test_clock_poll(ua, 100);
        }
        info = source_load();
        if (info.released_count) break;
        CHECK(info.load.sent.packets >= target);
        test_clock_run_for(ua, 2000);
    }
    info = source_load(); CHECK(info.released_count >= 1 && info.released_count <= 2 && info.released_rate >= info.required_rate && selects_seen);
    unsigned parity_closed = info.released_count;
    queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
    deadline = get_time_tb() + 100000;
    for (;;) {
        unsigned received = 0;
        for (unsigned i = 6; i < 8; i++) {
            struct stream_receive_info state; CHECK(stream_receive_get_info(receives[i], &state) == STREAM_OK);
            if (state.state == STREAM_RECEIVE_CLOSED) received++;
        }
        if (received == parity_closed || get_time_tb() >= deadline) break;
        test_clock_poll(ua, 100);
    }
    for (unsigned i = 0; i < 8; i++) {
        struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
        if (receive.state == STREAM_RECEIVE_CLOSED) CHECK(i >= 6 && receive.reason == STREAM_REASON_OVERLOAD);
        else CHECK(receive.state == STREAM_RECEIVE_ACTIVE);
    }
    test_clock_run_for(ua, 10000); CHECK(!source_load().load.queued.bytes && conn->links_up);
    unsigned delivered = pressure_data;
    CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
    run_until(&pressure_data, delivered + 1);
    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
    for (unsigned i = 0; i < 8; i++) {
        stream_receive_close(receives[i]); receives[i] = NULL;
        CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK);
    }
    test_clock_run_for(ua, 1000); CHECK(pressure_data == delivered + 1);
    CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
    run_until(&pressure_data, delivered + 2); test_clock_run_for(ua, 1000);
    uint64_t previous_request = source_load().request_id;
    overload_seen = selects_seen = 0; stale_select_length = 0;
    /* Тот же транспорт остаётся UP. Теперь держим очередь дольше подтверждения и возвращаем управление получателю. */
    while (queue->count) test_clock_poll(ua, 100);
    queue_set_callback(queue, NULL, NULL);
    for (unsigned i = 0; i < 24; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
    deadline = get_time_tb() + 15000;
    do { test_clock_poll(ua, 100); info = source_load(); } while ((!info.request_active || info.request_id == previous_request) && get_time_tb() < deadline);
    CHECK(info.request_active && info.required_count > 1 && info.required_rate && !info.released_count);
    uint64_t request = info.request_id, expires = info.deadline_ms;
    test_clock_run_for(ua, 3000); info = source_load(); CHECK(info.request_id == request && info.deadline_ms == expires);
    ignore_select = 1; queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
    deadline = get_time_tb() + 100000;
    while (!selects_seen && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(overload_seen && selects_seen && stale_select_length);
    /* Повторы SELECT намеренно не доходят: source исполняет достаточный остаток на исходном deadline. */
    while (get_time_tb() / 10 < expires + 100) test_clock_poll(ua, 100);
    info = source_load(); CHECK(!info.request_active && info.released_count > 1 && info.released_rate >= info.required_rate);
    unsigned closed = 0;
    for (unsigned i = 0; i < 8; i++) {
        struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
        if (receive.state == STREAM_RECEIVE_CLOSED) { CHECK(receive.reason == STREAM_REASON_OVERLOAD); closed++; }
    }
    CHECK(closed == info.released_count);
    /* Поздний выбор старого id не закрывает дополнительную выдачу. */
    ignore_select = 0;
    struct ETCP_CONN *back = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(back != NULL);
    while (back->send_input_q->count) test_clock_poll(ua, 100);
    struct ll_entry *late = ll_alloc_lldgram((uint16_t)stale_select_length); CHECK(late != NULL);
    memcpy(late->dgram, stale_select, stale_select_length); late->len = (uint16_t)stale_select_length; CHECK(etcp_send(back, late) == 0);
    test_clock_run_for(ua, 1000); CHECK(source_load().released_count == closed && conn->links_up);
    for (unsigned i = 0; i < 8; i++) stream_receive_close(receives[i]);
    stream_close(subscriber); stream_close(publisher); test_clock_run_for(ua, 1000);
    for (unsigned i = 1; i <= 3; i += 2) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_handlers[i]) == 0);
    stream_set_event_callback(nodes[1], on_stream, NULL);
    CHECK(stream_handle_active(reader1) && stream_handle_active(reader2));
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: glitch, parity-only cooperation, frozen rates, fixed deadline, fallback and stale SELECT");
}

/* CPU сам по себе не режет выдачи. Подтверждённый backlog нескольких пиров с локальной загрузкой использует общий бюджет. */
static void test_global_pressure(void) {
    uint64_t ready_deadline = get_time_tb() + 10000;
    while (!source_load().accepting && get_time_tb() < ready_deadline) test_clock_poll(ua, 100);
    CHECK(source_load().accepting);
    struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0}; struct STREAM_RECEIVE *receives[3][2] = {{0}};
    struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 1; i < 3; i++) {
        wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
        CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
        stream_set_event_callback(nodes[i], on_global_pressure, NULL);
    }
    uint8_t payload[SINGLE_INPUT_CAPACITY]; memset(payload, 3, sizeof(payload)); size_t accepted;
    CHECK(stream_publish_write(publisher, payload, sizeof(payload), &accepted) == STREAM_OK && accepted == sizeof(payload));
    for (unsigned i = 1; i < 3; i++) for (unsigned sub = 0; sub < 2; sub++)
        CHECK(stream_receive_open(subscribers[i], nodes[3]->node_id, sub, UINT64_MAX, &receives[i][sub]) == STREAM_OK);
    for (unsigned i = 1; i < 3; i++) run_until(&global_data[i], 1);
    uint64_t previous_requests[3] = {0};
    for (unsigned i = 1; i < 3; i++) {
        struct stream_peer_load_info before; CHECK(stream_peer_get_load(nodes[3], nodes[i]->node_id, &before) == STREAM_OK);
        previous_requests[i] = before.request_id;
    }
    simulated_cpu = 1; test_clock_run_for(ua, 7000);
    struct stream_peer_load_info info[3];
    for (unsigned i = 1; i < 3; i++) {
        CHECK(stream_peer_get_load(nodes[3], nodes[i]->node_id, &info[i]) == STREAM_OK);
        CHECK(info[i].cpu_percent == 95 && info[i].request_id == previous_requests[i] && !info[i].load.queued.bytes);
    }
    struct ll_queue *queues[3] = {0}; queue_callback_fn callbacks[3] = {0}; void *args[3] = {0};
    for (unsigned i = 1; i < 3; i++) {
        struct ETCP_CONN *conn = instance_find_conn(nodes[3], nodes[i]->node_id); CHECK(conn != NULL);
        queues[i] = conn->send_input_q;
        while (queues[i]->count) test_clock_poll(ua, 100);
        callbacks[i] = queues[i]->callback; args[i] = queues[i]->callback_arg; CHECK(callbacks[i] != NULL);
        queue_set_callback(queues[i], NULL, NULL);
    }
    for (unsigned i = 0; i < 48; i++) CHECK(stream_publish_write(publisher, payload, sizeof(payload), &accepted) == STREAM_OK);
    uint64_t deadline = get_time_tb() + 20000;
    for (;;) {
        int ready = 1;
        for (unsigned i = 1; i < 3; i++) {
            CHECK(stream_peer_get_load(nodes[3], nodes[i]->node_id, &info[i]) == STREAM_OK);
            if (!info[i].global_pressure || !info[i].request_active) ready = 0;
        }
        if (ready || get_time_tb() >= deadline) break;
        test_clock_poll(ua, 100);
    }
    CHECK(info[1].global_pressure && info[2].global_pressure && info[1].request_active && info[2].request_active);
    for (unsigned i = 1; i < 3; i++) { queue_set_callback(queues[i], callbacks[i], args[i]); queue_resume_callback(queues[i]); }
    deadline = get_time_tb() + 100000;
    for (;;) {
        int done = 1;
        for (unsigned i = 1; i < 3; i++) {
            CHECK(stream_peer_get_load(nodes[3], nodes[i]->node_id, &info[i]) == STREAM_OK);
            if (info[i].request_active) done = 0;
            CHECK(info[i].load.arrivals.bytes == info[i].load.sent.bytes + info[i].load.removed.bytes + info[i].load.queued.bytes);
        }
        if (done || get_time_tb() >= deadline) break;
        test_clock_poll(ua, 100);
    }
    CHECK(info[1].released_rate >= info[1].required_rate && info[2].released_rate >= info[2].required_rate);
    simulated_cpu = 0;
    for (unsigned i = 1; i < 3; i++) { for (unsigned sub = 0; sub < 2; sub++) stream_receive_close(receives[i][sub]); stream_close(subscribers[i]); }
    stream_close(publisher); test_clock_run_for(ua, 1000);
    stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] global overload: CPU alone preserves flows; confirmed multi-peer backlog releases a shared rate budget");
}

/* Default API сам выбирает выдачи: пустой источник, общий старт, цикл 30 с, overlap, stall и реальный CLOSED. */
static void test_automatic(void) {
    test_clock_run_for(ua, 10000);
    struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0}, *second = NULL, *manual = NULL;
    struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
    stream_set_event_callback(nodes[1], on_automatic, NULL); stream_set_event_callback(nodes[2], on_automatic, NULL);
    for (unsigned i = 1; i < 4; i++) {
        automatic_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM];
        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_raw) == 0);
    }
    CHECK(node_conn_direct_compatible(nodes[1], nodes[3]->node_id));
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_OK);
    struct STREAM_RECEIVE *probe = NULL;
    CHECK(stream_probe_open(manual, nodes[3]->node_id, 0, &probe) == STREAM_OK);
    struct stream_receive_info probe_info; uint64_t deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(probe, &probe_info) == STREAM_OK); }
    while (!probe_info.info_ready && get_time_tb() < deadline);
    CHECK(probe_info.info_ready && probe_info.probe_only && !probe_info.handle && !probe_info.blocks && !automatic_accepts);
    CHECK(probe_info.first_available == UINT64_MAX && probe_info.last_available == UINT64_MAX);
    CHECK(probe_info.path.count == 1 && probe_info.path.nodes[0] == nodes[3]->node_id);
    CHECK(probe_info.accepting && probe_info.free_slots && probe_info.path_rtt_valid && probe_info.rtt_valid);
    CHECK(stream_probe_next(probe, 1) == STREAM_OK); deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(probe, &probe_info) == STREAM_OK); }
    while (!probe_info.info_ready && get_time_tb() < deadline);
    CHECK(probe_info.info_ready && probe_info.subblock == 1 && !automatic_accepts);
    stream_receive_close(probe); stream_close(manual); manual = NULL;
    CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscribers[1]) == STREAM_OK);
    CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &second) == STREAM_OK);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_ERR_STATE && !manual);
    struct STREAM_RECEIVE *foreign = NULL;
    CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, 0, UINT64_MAX, &foreign) == STREAM_ERR_ARGUMENT && !foreign);
    CHECK(stream_receive_restart(subscribers[1]) == STREAM_ERR_STATE);
    deadline = get_time_tb() + 150000;
    int relay_announced = 0;
    do {
        struct stream_member_state member;
        relay_announced = stream_member_load(nodes[2], GROUP2, nodes[1]->node_id, &member, NULL) == STREAM_OK
            && stream_member_has_subscription(&member, nodes[3]->node_id, origin.descriptor.source_epoch);
        if (!relay_announced) test_clock_poll(ua, 100);
    } while (!relay_announced && get_time_tb() < deadline);
    CHECK(relay_announced);
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &subscribers[2]) == STREAM_OK);
    CHECK(stream_service_needs_activity(nodes[2]) && nodes[2]->client_activity == CLIENT_ACTIVITY_ACTIVE);
    CHECK(utun_set_client_activity(nodes[2], 0) == 0 && nodes[2]->client_activity == CLIENT_ACTIVITY_ACTIVE);
    CHECK(nodes[2]->client_activity_requested == CLIENT_ACTIVITY_STANDBY);
    struct stream_selection_settings settings = {200, 1000, 100};
    for (unsigned i = 1; i < 3; i++) {
        CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK);
        stream_set_event_callback(nodes[i], on_automatic, NULL);
    }
    test_clock_run_for(ua, 3000);
    CHECK(!automatic_data[1] && !automatic_data[2]);
    size_t capacity = stream_publish_capacity(publisher);
    CHECK(capacity == SINGLE_INPUT_CAPACITY && capacity <= sizeof(pressure_payload));
    memset(pressure_payload, 0xa9, capacity); size_t accepted;
    uint64_t initial_probes[3] = {0}; struct stream_selection_info state[3];
    for (unsigned frame = 0; frame < 350; frame++) {
        CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK && accepted == capacity);
        test_clock_run_for(ua, 1200);
        if (frame == 80 || frame == 180) for (unsigned i = 1; i < 3; i++) {
            CHECK(stream_selection_get_info(subscribers[i], &state[i]) == STREAM_OK);
            CHECK(state[i].automatic && state[i].subflows == 2 && state[i].cycles == 1);
            if (frame == 80) initial_probes[i] = state[i].probes_started;
            else CHECK(state[i].probes_started == initial_probes[i]);
        }
    }
    for (unsigned i = 1; i < 3; i++) {
        CHECK(stream_selection_get_info(subscribers[i], &state[i]) == STREAM_OK);
        CHECK(state[i].cycles == 2 && state[i].probes_started > initial_probes[i] && state[i].subflows == 2);
        CHECK(automatic_data[i] == 350 && !automatic_gaps[i]);
    }
    CHECK(state[2].migrations && state[2].providers == 2 && state[2].protected_by_fec);
    CHECK(!automatic_probes[2]); /* Mobile relay не пробуется ни одним планировщиком. */
    struct stream_selection_info same; CHECK(stream_selection_get_info(second, &same) == STREAM_OK);
    CHECK(same.probes_started == state[1].probes_started); stream_close(second); second = NULL;
    uint64_t migrations = state[2].migrations; automatic_drop_relay = 1;
    for (unsigned frame = 0; frame < 30; frame++) {
        CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK);
        test_clock_run_for(ua, 1200);
    }
    CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
    CHECK(state[2].migrations > migrations && state[2].providers == 1 && state[2].subflows == 2);
    CHECK(automatic_data[2] == 380 && !automatic_gaps[2]); automatic_drop_relay = 0;
    CHECK(automatic_request[0]); uint64_t failures = state[2].failures;
    struct ETCP_CONN *source_connection = instance_find_conn(nodes[3], nodes[2]->node_id); CHECK(source_connection && source_connection->links_up);
    struct ll_entry *cancel = ll_alloc_lldgram(sizeof(automatic_request)); CHECK(cancel);
    memcpy(cancel->dgram, automatic_request, sizeof(automatic_request)); cancel->len = sizeof(automatic_request);
    automatic_handlers[3](source_connection, cancel);
    for (unsigned frame = 0; frame < 20; frame++) {
        CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK);
        test_clock_run_for(ua, 1200);
    }
    CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
    CHECK(state[2].failures > failures && state[2].providers == 2 && state[2].subflows == 2);
    CHECK(automatic_data[2] == 400 && !automatic_gaps[2] && source_connection->links_up);
    CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[1], 1); run_until(&automatic_end[2], 1);
    for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
    publisher = subscribers[1] = subscribers[2] = NULL;
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    wait_state(2, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &subscribers[2]) == STREAM_OK);
    test_clock_run_for(ua, 5000);
    CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[2], 2);
    CHECK(automatic_data[2] == 400); stream_close(subscribers[2]); stream_close(publisher);
    subscribers[2] = publisher = NULL;
    test_clock_run_for(ua, 1000);
    /* END первого субпотока не выключает выбор недостающих: для 2+1 нужны ещё данные второй выдачи. */
    parameters.n = 3;
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 1; i < 3; i++) {
        wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
        CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
        CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK);
        automatic_next[i] = 0;
    }
    automatic_drop_tail = 1;
    test_clock_run_for(ua, 10000);
    size_t input_size = stream_input_size(&origin.descriptor);
    memset(pressure_payload, 0xa9, input_size);
    CHECK(stream_publish_write(publisher, pressure_payload, input_size, &accepted) == STREAM_OK && accepted == input_size);
    CHECK(stream_publish_finish(publisher) == STREAM_OK);
    struct stream_local terminal;
    deadline = get_time_tb() + 100000;
    do { test_clock_poll(ua, 100); CHECK(stream_local_get(nodes[2], &origin.descriptor, &terminal) == STREAM_OK); }
    while (!terminal.finished && get_time_tb() < deadline);
    CHECK(terminal.finished && terminal.anchored && !terminal.next && automatic_data[2] == 400 && automatic_end[2] == 2);
    automatic_drop_tail = 0;
    run_until(&automatic_data[2], 401); run_until(&automatic_end[2], 3);
    CHECK(automatic_next[2] == 1 && !automatic_gaps[2]);
    for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
    test_clock_run_for(ua, 1000);
    for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_handlers[i]) == 0);
    stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only, shared owners, 30s cycle, FEC spread, overlap, stall, CLOSED and early END drain");
}

/* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */
static void test_profile_atomicity(void) {
    char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1];
    uint8_t before_signature[64], after_signature[64], before_hash[32], after_hash[32];
    uint64_t before_version, after_version;
    profile(0, GROUP1, before, before_signature, &before_version); root_hash(0, GROUP1, before_hash);
    memset(huge, 'p', MS_USERINFO_MAX); huge[MS_USERINFO_MAX] = 0;
    struct chat_profile_option options[] = {{"stream_source", "0000000000000001:8:2:1189"}, {"stream_subscriptions", huge}};
    CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) < 0);
    profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash);
    CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_signature, after_signature, 64));
    CHECK(!memcmp(before_hash, after_hash, 32));
    CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "CREATE TRIGGER reject_stream BEFORE UPDATE OF userinfo ON peers_710001 "
          "BEGIN SELECT RAISE(ABORT,'injected stream profile failure'); END", NULL, NULL, NULL) == SQLITE_OK);
    struct STREAM_HANDLE *failed = NULL;
    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &failed) < 0 && !failed);
    CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "DROP TRIGGER reject_stream", NULL, NULL, NULL) == SQLITE_OK);
    profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash);
    CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_hash, after_hash, 32));
    CHECK(!memcmp(before_signature, after_signature, 64));
    options[0] = (struct chat_profile_option){"pad", "retained"}; options[1] = (struct chat_profile_option){"extra", "value"};
    CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) == 0);
    profile(0, GROUP1, after, after_signature, &after_version);
    char value[64];
    CHECK(json_flat_get(after, "avatar_file", value, sizeof(value)) == 0 && !strcmp(value, "retained-avatar"));
    CHECK(json_flat_get(after, "avatar_server", value, sizeof(value)) == 0 && !strcmp(value, "0011223344556677"));
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] profile updates: retained avatar/name/options and atomic size/DB rollback");
}

/* Блок владельца действительно подписан канальным ключом и проходит member_sync, профиль A сохраняется. */
static void relay_role(unsigned member, int enabled, unsigned version) {
    char tags[128]; uint8_t signature[64], body[256];
    snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"supernode\":\"%s\"}", version, enabled ? "yes" : "no");
    int length = member_sync_build_owner_msg(nodes[member]->node_id, tags, body, sizeof(body)); CHECK(length > 0);
    CHECK(sc_ed25519_sign(group_private, body, (size_t)length, signature) == SC_OK);
    for (unsigned i = 0; i < 4; i++) {
        CHECK(member_sync_put(nodes[i], "710002", nodes[member]->node_id, nodes[member]->my_keys.public_key,
              nodes[member]->my_ed25519_pubkey, NULL, 0, NULL, 0, NULL, tags, signature, 0, 0, NULL) >= 0);
        CHECK(stream_member_supernode(nodes[i], GROUP2, nodes[member]->node_id) == enabled);
    }
}

/* Прямой stream-вход источника блокируем только у зрителей; Merkle/BGP и получение суперузла остаются настоящими. */
static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    unsigned index = 0; while (index < 4 && conn->instance != nodes[index]) index++; CHECK(index < 4);
    if (entry->len >= STREAM_WIRE_HEADER) {
        unsigned command = entry->dgram[2];
        if (relay_race.watch_provider && index == 2 && conn->peer_node_id == nodes[1]->node_id
            && command == STREAM_CMD_INFO_RESP && !relay_race.provider_info_tb) relay_race.provider_info_tb = get_time_tb();
        if (command == STREAM_CMD_SUBSCRIBE && conn->peer_node_id == nodes[2]->node_id) {
            if (index == 3) relay_race.source_subscribes++;
            if (index == 1) relay_race.helper_subscribes++;
        }
        if (index == 2 && conn->peer_node_id == nodes[3]->node_id && command == STREAM_CMD_INFO_RESP && relay_race.hold_info) {
            if (!relay_race.info_length) {
                CHECK(entry->len <= sizeof(relay_race.info)); relay_race.info_length = entry->len;
                memcpy(relay_race.info, entry->dgram, entry->len);
            }
            queue_dgram_free(entry); queue_entry_free(entry); return;
        }
        if (relay_race.check_starts && index == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_ACCEPT) {
            struct stream_message m; CHECK(stream_wire_parse(entry->dgram, entry->len, &m) == STREAM_OK && m.length >= 16);
            uint64_t start = stream_read_be(m.payload + 8, 8);
            if (!relay_race.accepted) relay_race.start = start;
            CHECK(start == relay_race.start && m.subblock < 8); relay_race.accepted |= 1u << m.subblock;
        }
    }
    if (relay_drop_source && index != 1 && conn->peer_node_id == nodes[3]->node_id) {
        queue_dgram_free(entry); queue_entry_free(entry); return;
    }
    relay_raw_handlers[index](conn, entry);
}

/* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */
static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
    if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
        && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_CANCEL) relay_cancels_seen++;
    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++;
        if (!relay_race.request_tb) relay_race.request_tb = get_time_tb();
        if (relay_race.hold_request) { queue_dgram_free(entry); queue_entry_free(entry); return; }
    }
    relay_route_handler(conn, entry);
}

/* Счётчик STATUS выявляет ответы на дубликаты, не меняя подписанное управление и его порядок. */
static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) {
    if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
        && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) {
        relay_statuses_seen++;
        size_t payload = ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER;
        if (relay_race.hold_ready && entry->len > payload
            && (entry->dgram[payload] == STREAM_RELAY_READY || entry->dgram[payload] == STREAM_RELAY_BOUND)) {
            relay_race.ready_held++;
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test holds availability status=%u ordinal=%u",
                        entry->dgram[payload], relay_race.ready_held);
            queue_dgram_free(entry); queue_entry_free(entry); return;
        }
    }
    relay_status_handler(conn, entry);
}

/* Фиксируем принятую Merkle/BGP-опцию перед stream callback; PROFILE без нужной подписки не является поставщиком. */
static void relay_test_provider(uint64_t node, const char *tags, const char *channel, void *arg) {
    (void)tags; (void)arg;
    if (!relay_race.watch_provider || relay_race.provider_tb || node != nodes[1]->node_id
        || !channel || strcmp(channel, "710002")) return;
    struct stream_member_state member;
    if (stream_member_load(nodes[2], GROUP2, node, &member, NULL) == STREAM_OK
        && stream_member_online(nodes[2], GROUP2, node)
        && stream_member_has_subscription(&member, nodes[3]->node_id, relay_race.epoch)) relay_race.provider_tb = get_time_tb();
}

/* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */
static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3);
    CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
    if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == sizeof(expected)); relay_data[index]++; }
    if (event->type == STREAM_EVENT_END) relay_ends[index]++;
}

/* Повторяем управление через штатный подписанный router, сохраняя nonce/эпохи исходной попытки. */
static void relay_test_control(const struct stream_message *message) {
    struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry);
    entry->dgram[0] = ETCP_RT_ID_STREAM; stream_wire_header(entry->dgram + 1, message);
    entry->len = 1 + STREAM_WIRE_HEADER;
    CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
}

/* Запоздалый настоящий INFO подаётся тому же raw dispatcher; старый probe nonce не должен менять победителя. */
static void relay_release_info(void) {
    CHECK(relay_race.info_length);
    struct ETCP_CONN *conn = instance_find_conn(nodes[2], nodes[3]->node_id); CHECK(conn && conn->links_up);
    struct ll_entry *entry = ll_alloc_lldgram((uint16_t)relay_race.info_length); CHECK(entry);
    memcpy(entry->dgram, relay_race.info, relay_race.info_length); entry->len = relay_race.info_length;
    relay_raw_handlers[2](conn, entry); relay_race.info_length = 0; relay_race.hold_info = 0;
}

/* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */
static void test_supernode_relay(void) {
    int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE];
    debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO);
    struct TOPO_PEER_REQUEST *connections[3] = {0};
    uint64_t deadline = get_time_tb() + 150000;
    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 < 4; i++) {
        relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM];
        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_test_raw) == 0);
        if (i < 3) stream_set_event_callback(nodes[i], relay_test_event, NULL);
    }
    relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
    CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
    relay_status_handler = nodes[2]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
    CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_test_status) == 0);
    CHECK(member_sync_add_props_cbk(nodes[2], relay_test_provider, NULL) == 0);
    struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
    struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0};
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    test_clock_run_for(ua, 10000);
    struct stream_local local;
    CHECK(!relay_requests_seen && !relay_data[2]);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000);
    CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    run_until(&relay_data[2], 1); test_clock_run_for(ua, 60000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen);
    stream_close(viewers[2]); viewers[2] = NULL;
    test_clock_run_for(ua, 10000);
    /* Помощь начинается через 100 мс, но прямой INFO выигрывает до ответа помощника и отменяет его запрос. */
    relay_race.hold_info = relay_race.hold_request = 1;
    uint64_t started = get_time_tb();
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    test_clock_run_for(ua, 900); CHECK(!relay_requests_seen);
    deadline = get_time_tb() + 10000;
    while (!relay_requests_seen && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(relay_requests_seen && relay_race.info_length);
    CHECK(relay_race.request_tb - started >= 990 && relay_race.request_tb - started < 1500);
    relay_release_info(); run_until(&relay_data[2], 2); test_clock_run_for(ua, 2000);
    CHECK(relay_race.source_subscribes && !relay_race.helper_subscribes);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    struct stream_message cancelled; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &cancelled) == STREAM_OK);
    relay_race.hold_request = 0; relay_test_control(&cancelled); test_clock_run_for(ua, 2000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap direct wins after helper request: request_delay_ms=%llu, late REQUEST cancelled",
               (unsigned long long)((relay_race.request_tb - started) / 10));
    stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000);
    /* Закрытие reader до дедлайна отменяет и таймер, и объединённую работу сети. */
    unsigned requests_before = relay_requests_seen;
    relay_race.hold_info = 1;
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    test_clock_run_for(ua, 500); stream_close(viewers[2]); viewers[2] = NULL;
    test_clock_run_for(ua, 2000); CHECK(relay_requests_seen == requests_before);
    /* Merkle объявляет буфер до STATUS READY/BOUND: выбранная выдача должна удержать STARTING-запрос. */
    memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_ready = relay_race.watch_provider = 1;
    relay_race.epoch = origin.descriptor.source_epoch;
    relay_drop_source = 1;
    unsigned cancellations_before = relay_cancels_seen;
    unsigned expected_data = relay_data[2] + 1;
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    deadline = get_time_tb() + 40000;
    while (relay_data[2] < expected_data && relay_cancels_seen == cancellations_before && get_time_tb() < deadline)
        test_clock_poll(ua, 100);
    CHECK(relay_race.ready_held && relay_cancels_seen == cancellations_before);
    CHECK(relay_data[2] == expected_data && relay_race.helper_subscribes);
    CHECK(relay_race.provider_tb && relay_race.provider_info_tb >= relay_race.provider_tb);
    CHECK(relay_race.provider_info_tb - relay_race.provider_tb < 2000);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] provider callback starts INFO without 1s roster wait: elapsed_ms=%llu",
               (unsigned long long)((relay_race.provider_info_tb - relay_race.provider_tb) / 10));
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] helper INFO before delayed READY/BOUND keeps selected pending issuance; no premature CANCEL");
    test_clock_run_for(ua, 2000);
    struct stream_selection_info stable_before, stable_after;
    CHECK(stream_selection_get_info(viewers[2], &stable_before) == STREAM_OK);
    for (unsigned version = 101; version <= 103; version++) relay_role(1, 1, version);
    test_clock_run_for(ua, 2000);
    CHECK(stream_selection_get_info(viewers[2], &stable_after) == STREAM_OK);
    CHECK(stable_after.probes_started == stable_before.probes_started);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated provider property callbacks preserve probes and their cycle");
    stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 40000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1;
    started = get_time_tb();
    expected_data = relay_data[2] + 1;
    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    deadline = get_time_tb() + 300000;
    while (relay_data[2] < expected_data && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(relay_data[2] == expected_data && relay_requests_seen && !relay_data[1]);
    CHECK(get_time_tb() - started < 10000 && relay_race.request_tb - started >= 990);
    uint64_t connected_ms = (get_time_tb() - started) / 10;
    test_clock_run_for(ua, 1000); CHECK(relay_race.accepted == 255);
    CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes);
    relay_release_info(); test_clock_run_for(ua, 1000);
    CHECK(!relay_race.source_subscribes && relay_data[2] == expected_data);
    struct stream_message duplicate; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &duplicate) == STREAM_OK);
    CHECK(duplicate.receiver_epoch);
    unsigned statuses_before = relay_statuses_seen;
    for (unsigned i = 0; i < 3; i++) { relay_test_control(&duplicate); test_clock_run_for(ua, 300); }
    CHECK(relay_statuses_seen == statuses_before);
    relay_race.check_starts = 0;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap helper wins: elapsed_ms=%llu, n=8 r=2, common_start=%llu, stale direct INFO ignored",
               (unsigned long long)connected_ms, (unsigned long long)relay_race.start);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only);
    /* Поздний обычный подписчик также удерживает раздачу, даже без собственного запроса помощи. */
    CHECK(stream_subscribe_open(nodes[0], GROUP2, nodes[3]->node_id, &viewers[0]) == STREAM_OK);
    run_until(&relay_data[0], 1);
    stream_close(viewers[2]); viewers[2] = NULL;
    test_clock_run_for(ua, 80000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK);
    /* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */
    CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "CREATE TRIGGER reject_relay_viewer BEFORE UPDATE OF userinfo ON peers_710002 "
          "BEGIN SELECT RAISE(ABORT,'injected relay viewer failure'); END", NULL, NULL, NULL) == SQLITE_OK);
    struct STREAM_HANDLE *failed = NULL;
    CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &failed) < 0 && !failed);
    CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "DROP TRIGGER reject_relay_viewer", NULL, NULL, NULL) == SQLITE_OK);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only);
    CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only);
    unsigned before = relay_data[1];
    CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
    run_until(&relay_data[0], 2); run_until(&relay_data[1], before + 1);
    stream_close(viewers[0]); viewers[0] = NULL;
    test_clock_run_for(ua, 100000);
    CHECK(stream_handle_active(viewers[1]));
    CHECK(stream_get_info(viewers[1], &origin) == STREAM_OK && origin.owners == 1);
    stream_close(viewers[1]); viewers[1] = NULL;
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    /* Поздний повтор завершённого запроса не подписывает суперузел заново. */
    struct stream_message replay; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &replay) == STREAM_OK);
    CHECK(replay.receiver_epoch); relay_test_control(&replay);
    test_clock_run_for(ua, 50000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    /* Даже после удаления tombstone старое подтверждение поколения не может снова выделить owner. */
    test_clock_run_for(ua, 700000); relay_test_control(&replay); test_clock_run_for(ua, 20000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    replay.handle ^= UINT64_C(0x123456789abcdef0); replay.sender_epoch ^= UINT64_C(0x9876543210fedcba);
    replay.receiver_epoch = 0; replay.command = STREAM_CMD_RELAY_CANCEL; relay_test_control(&replay);
    replay.command = STREAM_CMD_RELAY_REQUEST; relay_test_control(&replay);
    test_clock_run_for(ua, 20000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    /* Два relay-only экземпляра не выбирают друг друга, даже если имеют буферы и живые BGP-пути. */
    struct STREAM_HANDLE *helpers[2] = {0};
    relay_role(0, 1, 100);
    CHECK(stream_relay_open(nodes[1], &origin.descriptor, &helpers[0]) == STREAM_OK);
    CHECK(stream_relay_open(nodes[0], &origin.descriptor, &helpers[1]) == STREAM_OK);
    test_clock_run_for(ua, 30000);
    for (unsigned i = 0; i < 2; i++) {
        struct stream_candidate *candidates = NULL; size_t count = 0;
        CHECK(stream_get_candidates(helpers[i], &candidates, &count) == STREAM_OK);
        for (size_t c = 0; c < count; c++) CHECK(candidates[c].node_id != nodes[i ? 1 : 0]->node_id);
        u_free(candidates); stream_close(helpers[i]);
    }
    relay_role(0, 0, 101);
    stream_close(publisher); publisher = NULL; test_clock_run_for(ua, 10000);
    /* READY без SUBSCRIBE истекает; публичный reader сохраняется, внутренний приём останавливается. */
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
    CHECK(stream_subscribe_manual_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    stream_relay_demand(nodes[2], &origin.descriptor, 1, get_time_tb() / 10);
    deadline = get_time_tb() + 100000;
    while (!stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id) && get_time_tb() < deadline)
        test_clock_poll(ua, 100);
    CHECK(stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id));
    test_clock_run_for(ua, 100000);
    CHECK(!stream_relay_confirmed(nodes[2], &origin.descriptor, nodes[1]->node_id));
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && stream_handle_active(viewers[2]));
    stream_close(viewers[2]); viewers[2] = NULL; stream_close(publisher); publisher = NULL;
    test_clock_run_for(ua, 10000);
    /* Пустой END проходит помощь до READY/DATA; буфер освобождается, хотя источник handle ещё жив. */
    CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
    CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
    for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
    CHECK(stream_subscribe_manual_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
    stream_relay_demand(nodes[2], &origin.descriptor, 1, get_time_tb() / 10);
    deadline = get_time_tb() + 100000;
    while (stream_local_get(nodes[1], &origin.descriptor, &local) != STREAM_OK && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK);
    test_clock_run_for(ua, 20000); CHECK(stream_publish_finish(publisher) == STREAM_OK);
    run_until(&relay_ends[2], 1); CHECK(!relay_ends[1]);
    test_clock_run_for(ua, 50000);
    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
    stream_close(viewers[2]); stream_close(publisher);
    relay_drop_source = 0; relay_role(1, 0, 104);
    CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0);
    CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_status_handler) == 0);
    member_sync_remove_props_cbk(nodes[2], relay_test_provider, NULL);
    for (unsigned i = 0; i < 4; i++) {
        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0);
        if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);
    }
    for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
    debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END");
}

/* Канонический список исключает неоднозначные IDs, дубли и обрезание значений. */
static void test_member_format(void) {
    struct stream_member_state state = {.source = {.group_id = GROUP1, .source_node_id = 1, .source_epoch = 2,
        .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}, .published = 1, .count = 2,
        .subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX, 1}}};
    char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[2 * MS_USERINFO_MAX + 64];
    CHECK(stream_member_format(&state, source, subscriptions) == STREAM_OK);
    snprintf(json, sizeof(json), "{\"stream_source\":\"%s\",\"stream_subscriptions\":\"%s\"}", source, subscriptions);
    struct stream_member_state parsed;
    CHECK(stream_member_parse(GROUP1, 1, json, &parsed) == STREAM_OK);
    CHECK(parsed.published && parsed.source.source_epoch == 2 && parsed.count == 2);
    CHECK(parsed.subscriptions[1].source_node_id == UINT64_MAX && parsed.subscriptions[1].source_epoch == UINT64_MAX);
    CHECK(!parsed.subscriptions[0].relay_only && parsed.subscriptions[1].relay_only);
    const char *bad[] = {
        "{\"stream_subscriptions\":\"1:2\"}",
        "{\"stream_subscriptions\":\"0000000000000001:0000000000000002,\"}",
        "{\"stream_subscriptions\":\"0000000000000001:0000000000000002,0000000000000001:0000000000000002\"}",
        "{\"stream_source\":\"0000000000000002:256:2:1189\"}",
        "{\"stream_source\":\"0000000000000002:8:2:1189\"}",
        "{\"stream_source\":\"\",\"stream_source\":\"\"}"
    };
    for (unsigned i = 0; i < sizeof(bad) / sizeof(bad[0]); i++)
        CHECK(stream_member_parse(GROUP1, 1, bad[i], &parsed) == STREAM_ERR_FORMAT);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] member format: full 64-bit IDs, canonical lists, duplicate/overflow rejection");
}

/* Пауза player не останавливает live ingest/selector; FIN сохраняет источник до дочитывания локального owner. */
static void test_delayed_media_reader(void) {
    struct STREAM_HANDLE *publisher = NULL, *remote = NULL, *preview = NULL;
    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
    struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK);
    wait_state(1, 0, GROUP1, info.descriptor.source_epoch, 0);
    CHECK(stream_media_enable(publisher, 0) == STREAM_OK);
    CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &remote) == STREAM_OK);
    CHECK(stream_subscribe_open(nodes[0], GROUP1, nodes[0]->node_id, &preview) == STREAM_OK);
    CHECK(stream_media_enable(remote, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK);
    CHECK(stream_media_enable(preview, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK);
    uint8_t video[128] = {1}, wire[512]; size_t length = 0;
    struct stream_media_block media = {.format = {64, 64, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1};
    media.video[0] = (struct stream_media_video){1, 0, sizeof(video), 0, sizeof(video), STREAM_MEDIA_KEY, video};
    CHECK(stream_media_encode(&media, wire, sizeof(wire), &length) == STREAM_OK);
    CHECK(stream_publish_block(publisher, wire, length) == STREAM_OK);
    struct stream_media_packet packet; int result = STREAM_WAIT;
    uint64_t deadline = get_time_tb() + 100000;
    while (result == STREAM_WAIT || result == STREAM_MORE) {
        CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100);
        result = stream_media_read(remote, -1, 0, -1, &packet);
    }
    CHECK(result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.pts_us == 0);
    CHECK(stream_media_read(preview, -1, 0, -1, &packet) == STREAM_OK && packet.video.pts_us == 0);
    struct stream_selection_info before, after;
    CHECK(stream_selection_get_info(remote, &before) == STREAM_OK);
    for (unsigned frame = 1; frame <= 150; frame++) {
        media.video[0].frame_id = frame + 1; media.video[0].pts_us = (uint64_t)frame * 40000;
        media.video[0].flags = frame % 25 ? 0 : STREAM_MEDIA_KEY;
        CHECK(stream_media_encode(&media, wire, sizeof(wire), &length) == STREAM_OK);
        CHECK(stream_publish_block(publisher, wire, length) == STREAM_OK);
        test_clock_run_for(ua, 400);
    }
    struct stream_local live; CHECK(stream_local_get(nodes[1], &info.descriptor, &live) == STREAM_OK);
    CHECK(live.next == 151);
    CHECK(stream_selection_get_info(remote, &after) == STREAM_OK && after.restarts == before.restarts);
    struct stream_media_status status;
    CHECK(stream_media_get_status(remote, &status) == STREAM_OK && status.start_us == 0 && status.ready_us == 6000000);
    CHECK(stream_publish_finish(publisher) == STREAM_OK);
    stream_close(publisher); publisher = NULL;
    CHECK(stream_handle_active(preview));
    unsigned counts[2] = {0}; struct STREAM_HANDLE *readers[2] = {remote, preview};
    for (unsigned reader = 0; reader < 2; reader++) {
        deadline = get_time_tb() + 100000;
        for (;;) {
            result = stream_media_read(readers[reader], 10000000, 0, -1, &packet);
            if (result == STREAM_WAIT || result == STREAM_MORE) {
                CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100); continue;
            }
            CHECK(result == STREAM_OK);
            if (packet.kind == STREAM_MEDIA_READ_END) break;
            CHECK(packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.pts_us == (uint64_t)(counts[reader] + 1) * 40000);
            CHECK(packet.video.length == sizeof(video) && !memcmp(packet.video.data, video, sizeof(video))); counts[reader]++;
        }
        CHECK(counts[reader] == 150); stream_close(readers[reader]);
    }
    test_clock_run_for(ua, 1000);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] delayed media readers: 6s pause, live progress without restart, exact remote/local FIN drain");
}

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);
    debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_WARN);
#ifdef SIGPIPE
    CHECK(signal(SIGPIPE, SIG_IGN) != SIG_ERR);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "TCP stream test ignores SIGPIPE; closed socket errors are handled by transport");
#endif
    test_member_format();
    size_t allocations = u_get_allocated_count();
    test_clock_init(100); utun_instance_set_tun_init_enabled(0);
    ua = uasync_create(); CHECK(ua != NULL);
    struct SC_MYKEYS channel; CHECK(sc_generate_keypair(&channel) == SC_OK);
    memcpy(group_x, channel.public_key, 32);
    uint8_t seed[64]; SHA512(channel.private_key, 32, seed); memcpy(group_private, seed, 32);
    CHECK(sc_derive_ed25519_pubkey(channel.private_key, group_public) == SC_OK);
    char directory[512] = "/tmp/utun-stream-service-XXXXXX"; CHECK(test_mkdtemp(directory) == 0);
    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);
        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:direct]\naddr=127.0.0.1:0\ntype=public\ntransport=%s\n"
            "%s[allowed_keys]\nallow_all=1\n"
            "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n",
            priv, pub, directory, i, type, i == 3 ? "tcp" : "udp",
            i == 3 ? "" : "[server:tcp]\naddr=127.0.0.1:0\ntype=public\ntransport=tcp\n");
        nodes[i] = utun_instance_create_from_str(ua, config); CHECK(nodes[i] != NULL);
        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);
        CHECK(nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM]
              && nodes[i]->router_bindings.callbacks[ETCP_RT_ID_STREAM] != nodes[i]->router_bindings.callbacks[ETCP_RT_ID_MEMBER_FILES]);
        for (unsigned g = 0; g < 2; g++) {
            char id[32]; uint64_t group = g ? GROUP2 : GROUP1; snprintf(id, sizeof(id), "%llu", (unsigned long long)group);
            uint8_t signature[64] = {0};
            CHECK(topo_node_sqlite_channel_put(nodes[i]->topo_sqlite_db, id, "stream", nodes[0]->node_id,
                group_x, NULL, group_public, i ? NULL : group_private, signature) == 0);
            chat_core_ensure_channel_ready(nodes[i], id);
        }
    }
    for (unsigned destination = 0; destination < 4; destination++) {
        for (unsigned member = 0; member < 4; member++) {
            put_member(destination, member, GROUP1); put_member(destination, member, GROUP2);
            struct TOPO_NODE *identity = topo_node_registry_find(nodes[member]->topo_groups, nodes[member]->node_id);
            CHECK(identity && topo_node_sqlite_snapshot_put(nodes[destination]->topo_sqlite_db, identity) >= 0);
        }
    }
    struct TOPO_PEER_REQUEST *connections[3] = {0};
    for (unsigned i = 1; i < 4; i++)
        CHECK(topo_group_peer_open(topo_groups_find(nodes[0]->topo_groups, GROUP1), nodes[i]->node_id, &connections[i - 1]) == 0);
    uint64_t deadline = get_time_tb() + 150000;
    while (get_time_tb() < deadline) {
        int ready = 1;
        for (unsigned i = 0; i < 3; i++) if (topo_group_peer_phase(connections[i]) != TOPO_PEER_READY) ready = 0;
        if (ready) break;
        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);
    }
    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();
        if (u_get_allocated_count() != allocations) {
            debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay 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] isolated supernode relay checks complete, no tracked allocations remain");
        return 0;
    }
    test_profile_atomicity();
    test_delayed_media_reader();
    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;
    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
    CHECK(stream_publish_open(nodes[0], GROUP2, NULL, &publisher2) == STREAM_OK);
    struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK);
    uint64_t epoch = info.descriptor.source_epoch;
    for (unsigned i = 1; i < 4; i++) wait_state(i, 0, GROUP1, epoch, 0);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &reader1) == STREAM_OK);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &reader2) == STREAM_OK);
    CHECK(stream_get_info(reader1, &info) == STREAM_OK && info.owners == 2 && info.buffer.blocks == 0);
    CHECK(stream_subscribe_manual_open(nodes[2], GROUP1, nodes[0]->node_id, &mobile) == STREAM_OK);
    CHECK(stream_subscribe_manual_open(nodes[3], GROUP1, nodes[0]->node_id, &desktop) == STREAM_OK);
    struct stream_member_state state;
    CHECK(stream_member_load(nodes[1], GROUP1, nodes[1]->node_id, &state, NULL) == STREAM_OK && state.count == 1);
    for (unsigned i = 2; i < 4; i++) wait_state(1, i, GROUP1, epoch, 1);
    struct stream_candidate *candidates = NULL; size_t count = 0;
    CHECK(stream_get_candidates(reader1, &candidates, &count) == STREAM_OK);
    int source_found = 0, desktop_found = 0;
    for (size_t i = 0; i < count; i++) {
        CHECK(candidates[i].node_id != nodes[2]->node_id);
        if (candidates[i].node_id == nodes[0]->node_id) { CHECK(candidates[i].source); source_found = 1; }
        if (candidates[i].node_id == nodes[3]->node_id) desktop_found = 1;
    }
    CHECK(source_found && desktop_found); u_free(candidates); candidates = NULL;
    test_bootstrap();
    struct TOPO_GROUP *group = topo_groups_find(nodes[1]->topo_groups, GROUP1);
    struct TOPO_GROUP_NODE *peer = topo_node_find_by_id(group, nodes[3]->node_id); CHECK(peer != NULL);
    topo_nodeq_remove_node(group, peer); /* BGP withdraw не удаляет сохранённую опцию мембера. */
    CHECK(stream_member_load(nodes[1], GROUP1, nodes[3]->node_id, &state, NULL) == STREAM_OK);
    CHECK(stream_member_has_subscription(&state, nodes[0]->node_id, epoch));
    CHECK(stream_get_candidates(reader1, &candidates, &count) == STREAM_OK);
    for (size_t i = 0; i < count; i++) CHECK(candidates[i].node_id != nodes[3]->node_id);
    u_free(candidates);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] Merkle subscriptions and BGP gating; mobile source allowed, mobile relay excluded");
    test_network();
    test_pressure();
    test_global_pressure();
    test_automatic();
    test_supernode_relay();
    test_pending_down();

    /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */
    struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);
    char source_value[MS_USERINFO_MAX + 1], subs_value[MS_USERINFO_MAX + 1];
    CHECK(stream_member_load(nodes[0], GROUP2, nodes[0]->node_id, &state, NULL) == STREAM_OK);
    CHECK(stream_member_format(&state, source_value, subs_value) == STREAM_OK);
    char profile2[MS_USERINFO_MAX + 1]; uint8_t sig2[64], join2[64]; uint64_t version2, join_ts;
    profile(0, GROUP2, profile2, sig2, &version2);
    CHECK(topo_node_sqlite_member_get_join(nodes[0]->topo_sqlite_db, "710002", nodes[0]->node_id, join2, &join_ts) == 0);
    uint8_t tree2[64]; uint64_t parent;
    CHECK(topo_node_sqlite_member_get_sign(nodes[0]->topo_sqlite_db, "710002", nodes[0]->node_id, &parent, tree2) == 0);
    CHECK(member_sync_put(nodes[1], "710002", nodes[0]->node_id, nodes[0]->my_keys.public_key, nodes[0]->my_ed25519_pubkey,
        join2, join_ts, sig2, version2, profile2, NULL, NULL, 0, parent, tree2) >= 0);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK);

    for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3);
    test_restart_bound(desktop, &info.descriptor);
    struct stream_encoder *encoder = NULL;
    CHECK(stream_encoder_create(&encoder, &info.descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK);
    const uint8_t *blocks[RS_MAX_BLOCKS];
    size_t size = 0;
    CHECK(stream_encoder_encode(encoder, 100, 65530, 0, expected, sizeof(expected), blocks, &size) == STREAM_OK);
    close_readers_in_callback = 1;
    for (unsigned i = 0; i < 8; i++) if (i != 1 && i != 5) CHECK(stream_ingest_block(nodes[1], blocks[i], size) == STREAM_OK);
    run_until(&data_events[1], 1); run_until(&closed_events[1], 1);
    CHECK(!reader1 && !reader2 && stream_handle_active(other_group));
    uint8_t invalid[STREAM_BLOCK_MAX_SIZE]; memcpy(invalid, blocks[0], size); invalid[size - 1] ^= 1;
    CHECK(stream_ingest_block(nodes[1], invalid, size) == STREAM_WAIT); /* Нет подписки — нет decode/crypto. */
    stream_encoder_destroy(encoder);
    size_t accepted = 0;
    CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
    CHECK(stream_publish_finish(publisher) == STREAM_OK);
    run_until(&data_events[0], 1); run_until(&end_events[0], 1);
    CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && state.count == 1);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] opaque data/FEC delivery, close inside callback, independent group and DATA before END");

    stream_close(publisher); publisher = NULL;
    CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
    CHECK(stream_get_info(publisher, &info) == STREAM_OK && info.descriptor.source_epoch != epoch);
    deadline = get_time_tb() + 150000;
    while ((stream_handle_active(mobile) || stream_handle_active(desktop)) && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(!stream_handle_active(mobile) && !stream_handle_active(desktop));
    CHECK(stream_handle_reason(mobile) == STREAM_CLOSE_SOURCE_CHANGED && stream_handle_reason(desktop) == STREAM_CLOSE_SOURCE_CHANGED);
    CHECK(!stream_service_needs_activity(nodes[2]) && nodes[2]->client_activity == CLIENT_ACTIVITY_STANDBY);
    stream_close(mobile); stream_close(desktop);
    /* Повреждённое собственное членство удаляет обычная проверка Merkle; буфер не должен остаться активным. */
    sqlite3_stmt *broken = NULL; uint8_t bad_signature[64]; memset(bad_signature, 1, sizeof(bad_signature));
    CHECK(sqlite3_prepare_v2(nodes[1]->topo_sqlite_db, "UPDATE peers_710002 SET update_sig=? WHERE node_id=?",
                            -1, &broken, NULL) == SQLITE_OK);
    sqlite3_bind_blob(broken, 1, bad_signature, sizeof(bad_signature), SQLITE_STATIC);
    sqlite3_bind_int64(broken, 2, (sqlite3_int64)nodes[1]->node_id);
    CHECK(sqlite3_step(broken) == SQLITE_DONE); sqlite3_finalize(broken);
    CHECK(member_sync_verify_and_purge(nodes[1], "710002") == 1);
    CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_MEMBER);
    run_until(&closed_events[1], 2); stream_close(other_group); other_group = NULL;
    put_member(1, 1, GROUP2);
    CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK);
    chat_core_delete_channel(nodes[1], "710002");
    CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_GROUP);
    stream_close(other_group);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source epoch, lost own membership and channel deletion detach local handles");

    char stale[MS_USERINFO_MAX + 1]; uint8_t stale_signature[64]; uint64_t stale_version;
    profile(0, GROUP1, stale, stale_signature, &stale_version);
    CHECK(utun_service_start(nodes[0]) == 0);
    sqlite3 *db = nodes[0]->topo_sqlite_db; struct ETCP_SOCKET *sockets = nodes[0]->etcp_sockets;
    CHECK(stream_publish_write(publisher2, expected, 17, &accepted) == STREAM_OK);
    chat_service_stop(nodes[0]);
    CHECK(!nodes[0]->stream && nodes[0]->utun_started && nodes[0]->topo_sqlite_db == db && nodes[0]->etcp_sockets == sockets);
    CHECK(!stream_handle_active(publisher) && !stream_handle_active(publisher2));
    stream_close(publisher); stream_close(publisher2);
    sqlite3_stmt *restore = NULL;
    CHECK(sqlite3_prepare_v2(db, "UPDATE peers_710001 SET userinfo=?,update_sig=?,update_ts=? WHERE node_id=?", -1, &restore, NULL) == SQLITE_OK);
    sqlite3_bind_text(restore, 1, stale, -1, SQLITE_STATIC); sqlite3_bind_blob(restore, 2, stale_signature, 64, SQLITE_STATIC);
    sqlite3_bind_int64(restore, 3, (sqlite3_int64)stale_version); sqlite3_bind_int64(restore, 4, (sqlite3_int64)nodes[0]->node_id);
    CHECK(sqlite3_step(restore) == SQLITE_DONE); sqlite3_finalize(restore);
    CHECK(chat_service_start(nodes[0]) == 0 && nodes[0]->stream);
    CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && !state.count);
    CHECK(data_events[0] == 1); /* Pending 100ms batch не выполнилась после stop. */
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] CHAT stop keeps UTUN/core; source timer cancelled; stale signed options cleared on restart");
    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;
}