jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
3264 Commits
15 Branches
2 Tags
156 MiB
 
 
 
 
 
 
Tree: 559d66ae20
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '559d66ae20'
${ noResults }
utun2/tests/test_stream_service.c

1839 lines
127 KiB
Raw Blame History

/* Живые 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/radio/radio.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:0"}, {"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);
}
}
/* Роль немедленно следует за подписанным блоком канала; deleted и чужой snapshot проверяются отдельно. */
static void test_supernode_lookup(void) {
uint64_t member = nodes[1]->node_id;
CHECK(stream_member_supernode(nodes[0], GROUP2, member) == 0);
relay_role(1, 1, 90);
CHECK(stream_member_supernode(nodes[0], GROUP1, member) == 0);
CHECK(stream_member_supernode(nodes[0], GROUP2, UINT64_MAX) == 0);
sqlite3 *db = nodes[0]->topo_sqlite_db; sqlite3_stmt *statement = NULL;
CHECK(sqlite3_exec(db, "SAVEPOINT role_lookup", NULL, NULL, NULL) == SQLITE_OK);
CHECK(sqlite3_prepare_v2(db, "UPDATE peers_710002 SET deleted=1 WHERE node_id=?", -1, &statement, NULL) == SQLITE_OK);
sqlite3_bind_int64(statement, 1, (sqlite3_int64)member);
CHECK(sqlite3_step(statement) == SQLITE_DONE && sqlite3_changes(db) == 1); sqlite3_finalize(statement);
CHECK(stream_member_supernode(nodes[0], GROUP2, member) == 0);
CHECK(sqlite3_exec(db, "ROLLBACK TO role_lookup", NULL, NULL, NULL) == SQLITE_OK);
CHECK(stream_member_supernode(nodes[0], GROUP2, member) == 1);
CHECK(sqlite3_prepare_v2(db, "UPDATE node_snapshots SET record=x'00' WHERE node_id=?", -1, &statement, NULL) == SQLITE_OK);
sqlite3_bind_int64(statement, 1, (sqlite3_int64)nodes[2]->node_id);
CHECK(sqlite3_step(statement) == SQLITE_DONE && sqlite3_changes(db) == 1); sqlite3_finalize(statement);
CHECK(stream_member_supernode(nodes[0], GROUP2, member) == 1);
CHECK(sqlite3_exec(db, "ROLLBACK TO role_lookup; RELEASE role_lookup", NULL, NULL, NULL) == SQLITE_OK);
relay_role(1, 0, 91);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode lookup: signed grant/revoke, channel isolation, absent/deleted member, unrelated snapshot");
}
/* Прямой 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;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test helper ACCEPT sub=%u start=%llu mask=%02x",
m.subblock, (unsigned long long)start, relay_race.accepted);
}
}
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;
/* Первый FEC готов после k ACCEPT; пустой ранний субпоток повторяется через 100ms плюс фазу network tick. */
deadline = get_time_tb() + 5000;
while (relay_race.accepted != 255 && get_time_tb() < deadline) test_clock_poll(ua, 100);
if (relay_race.accepted != 255)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "bootstrap test incomplete helper ACCEPT mask=%02x after first FEC", relay_race.accepted);
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, .audio_enabled = 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.audio_enabled && 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:1\"}",
"{\"stream_source\":\"0000000000000002:8:2:1189:2\",\"stream_subscriptions\":\"0000000000000001:0000000000000002\"}",
"{\"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");
}
static void wait_source_audio(uint64_t epoch, int enabled) {
uint64_t deadline = get_time_tb() + 100000;
for (;;) {
struct stream_member_state state;
if (stream_member_load(nodes[1], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK
&& state.published && state.source.source_epoch == epoch && state.audio_enabled == enabled) return;
CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100);
}
}
static void test_viewer_radio_mute(void) {
struct STREAM_HANDLE *publisher = NULL, *raw = NULL, *relay = NULL, *first = NULL, *second = NULL;
struct stream_info info;
uint64_t source = nodes[0]->node_id;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
CHECK(stream_publish_set_audio_enabled(publisher, 1) == STREAM_OK);
CHECK(stream_get_info(publisher, &info) == STREAM_OK);
uint64_t epoch = info.descriptor.source_epoch;
wait_source_audio(epoch, 1);
CHECK(!radio_member_silenced(nodes[1], GROUP1, source));
CHECK(stream_relay_open(nodes[1], &info.descriptor, &relay) == STREAM_OK);
CHECK(stream_subscribe_open(nodes[1], GROUP1, source, &raw) == STREAM_OK);
CHECK(!radio_member_silenced(nodes[1], GROUP1, source));
CHECK(stream_media_open(nodes[1], GROUP1, source, epoch, STREAM_MEDIA_TRACK_VIDEO, &first) == STREAM_OK);
CHECK(stream_media_open(nodes[1], GROUP1, source, epoch, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO, &second) == STREAM_OK);
CHECK(radio_member_silenced(nodes[1], GROUP1, source) && !radio_member_muted(nodes[1], GROUP1, source));
CHECK(!radio_member_silenced(nodes[1], GROUP2, source));
CHECK(!radio_member_silenced(nodes[1], GROUP1, nodes[2]->node_id));
stream_close(first); first = NULL;
CHECK(radio_member_silenced(nodes[1], GROUP1, source));
CHECK(radio_set_member_muted(nodes[1], GROUP1, source, 1) == 0);
CHECK(stream_publish_set_audio_enabled(publisher, 0) == STREAM_OK); wait_source_audio(epoch, 0);
CHECK(!stream_radio_muted(nodes[1], GROUP1, source) && radio_member_silenced(nodes[1], GROUP1, source));
CHECK(radio_set_member_muted(nodes[1], GROUP1, source, 0) == 0);
CHECK(!radio_member_silenced(nodes[1], GROUP1, source));
CHECK(stream_publish_set_audio_enabled(publisher, 1) == STREAM_OK); wait_source_audio(epoch, 1);
CHECK(radio_member_silenced(nodes[1], GROUP1, source));
stream_close(second); second = NULL;
CHECK(!radio_member_silenced(nodes[1], GROUP1, source));
CHECK(stream_media_open(nodes[1], GROUP1, source, epoch, STREAM_MEDIA_TRACK_VIDEO, &first) == STREAM_OK);
CHECK(radio_member_silenced(nodes[1], GROUP1, source));
stream_close(publisher); publisher = NULL;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
CHECK(stream_publish_set_audio_enabled(publisher, 1) == STREAM_OK);
CHECK(stream_get_info(publisher, &info) == STREAM_OK && info.descriptor.source_epoch != epoch);
wait_source_audio(info.descriptor.source_epoch, 1);
CHECK(!stream_handle_active(first) && !radio_member_silenced(nodes[1], GROUP1, source));
stream_close(first); stream_close(raw); stream_close(relay); first = NULL;
CHECK(stream_media_open(nodes[1], GROUP1, source, info.descriptor.source_epoch, STREAM_MEDIA_TRACK_VIDEO, &first) == STREAM_OK);
CHECK(radio_member_silenced(nodes[1], GROUP1, source));
stream_cancel_group(nodes[1], GROUP1);
CHECK(!radio_member_silenced(nodes[1], GROUP1, source));
stream_close(first); stream_close(publisher);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] viewer radio mute: source audio, manual mute, group isolation, owners, relay, epoch, close");
}
/* Пауза 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, 1, -1, &packet);
}
CHECK(result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.pts_us == 0);
CHECK(stream_media_read(preview, -1, 1, -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, 1, -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");
}
static struct {
etcp_recv_fn viewer_receive, source_receive;
struct ll_entry *held_info;
struct ETCP_CONN *held_conn;
unsigned data, ends, renews;
int hold_source;
} lifecycle;
/* Первый INFO источника задержан до DATA неполного ретранслятора; остальные пакеты не теряются. */
static void lifecycle_view_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (lifecycle.hold_source && conn->peer_node_id == nodes[0]->node_id
&& entry->len >= STREAM_WIRE_HEADER && entry->dgram[2] == STREAM_CMD_INFO_RESP) {
CHECK(!lifecycle.held_info);
lifecycle.held_info = entry; lifecycle.held_conn = conn; return;
}
lifecycle.viewer_receive(conn, entry);
}
static void lifecycle_source_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (conn->peer_node_id == nodes[2]->node_id && entry->len >= STREAM_WIRE_HEADER
&& entry->dgram[2] == STREAM_CMD_RENEW) lifecycle.renews++;
lifecycle.source_receive(conn, entry);
}
static void lifecycle_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
(void)inst; (void)arg;
if (event->type == STREAM_EVENT_DATA) {
CHECK(event->first_seq == 0 && event->length == sizeof(expected));
CHECK(!memcmp(event->data, expected, sizeof(expected))); lifecycle.data++;
} else if (event->type == STREAM_EVENT_END) lifecycle.ends++;
else CHECK(event->type == STREAM_EVENT_CLOSED);
}
/* Первый поставщик имеет лишь 3 из 6 нужных частей; пауза источника не оправдывает зависший bootstrap. */
static void test_partial_bootstrap_and_end(void) {
struct STREAM_HANDLE *publisher = NULL, *relay = NULL, *viewer = NULL;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
wait_state(1, 0, GROUP1, origin.descriptor.source_epoch, 0);
wait_state(2, 0, GROUP1, origin.descriptor.source_epoch, 0);
CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[0]->node_id, &relay) == STREAM_OK);
wait_state(2, 1, GROUP1, origin.descriptor.source_epoch, 1);
memset(expected, 0x37, sizeof(expected));
CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
struct stream_local source, received;
CHECK(stream_local_get(nodes[0], &origin.descriptor, &source) == STREAM_OK);
for (unsigned sub = 0; sub < 3; sub++) {
const uint8_t *part; size_t length; uint64_t actual;
CHECK(stream_buffer_read(source.buffer, 0, sub, &part, &length, &actual) == STREAM_OK);
CHECK(stream_ingest_block(nodes[1], part, length) == STREAM_OK);
}
lifecycle.viewer_receive = nodes[2]->api_bindings.callbacks[ETCP_ID_STREAM];
lifecycle.source_receive = nodes[0]->api_bindings.callbacks[ETCP_ID_STREAM];
CHECK(etcp_bind(nodes[2], ETCP_ID_STREAM, lifecycle_view_raw) == 0);
CHECK(etcp_bind(nodes[0], ETCP_ID_STREAM, lifecycle_source_raw) == 0);
stream_set_event_callback(nodes[2], lifecycle_event, NULL); lifecycle.hold_source = 1;
CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &viewer) == STREAM_OK);
uint64_t deadline = get_time_tb() + 40000;
do {
test_clock_poll(ua, 100);
CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK);
} while (!received.anchored && get_time_tb() < deadline);
CHECK(received.anchored && !received.decoded && lifecycle.held_info);
lifecycle.hold_source = 0;
struct ll_entry *held = lifecycle.held_info; lifecycle.held_info = NULL;
lifecycle.viewer_receive(lifecycle.held_conn, held);
run_until(&lifecycle.data, 1);
CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK);
CHECK(received.start == 0 && received.next == 1 && received.decoded);
struct stream_selection_info selected; CHECK(stream_selection_get_info(viewer, &selected) == STREAM_OK);
CHECK(!selected.restarts && selected.providers == 2);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] partial bootstrap: complementary source parts decoded original batch without restart");
/* END не зависит от существования relay; оставляем только прямую выдачу, чтобы измерять её RENEW. */
stream_close(viewer); stream_close(relay); viewer = NULL;
test_clock_run_for(ua, 10000);
CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &viewer) == STREAM_OK);
run_until(&lifecycle.data, 2); test_clock_run_for(ua, 20000);
CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&lifecycle.ends, 1);
test_clock_run_for(ua, 2000); lifecycle.renews = 0;
test_clock_run_for(ua, 100000);
CHECK(stream_selection_get_info(viewer, &selected) == STREAM_OK);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "finished reader retained: renews=%u flows=%u overlaps=%u probes=%u providers=%u",
lifecycle.renews, selected.subflows, selected.overlaps, selected.probes_active, selected.providers);
CHECK(!lifecycle.renews && !selected.subflows && !selected.overlaps && !selected.probes_active && !selected.providers);
CHECK(stream_handle_active(viewer));
CHECK(stream_local_get(nodes[2], &origin.descriptor, &received) == STREAM_OK && received.next == 1 && received.finished);
struct ETCP_CONN *shared = instance_find_conn(nodes[2], nodes[0]->node_id); CHECK(shared && shared->links_up);
stream_close(viewer); stream_close(publisher); test_clock_run_for(ua, 1000);
CHECK(etcp_bind(nodes[2], ETCP_ID_STREAM, lifecycle.viewer_receive) == 0);
CHECK(etcp_bind(nodes[0], ETCP_ID_STREAM, lifecycle.source_receive) == 0);
stream_set_event_callback(nodes[2], NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] completed receive releases upstream while retaining reader and shared transport");
}
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);
}
}
test_supernode_lookup();
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") || !strcmp(argv[1], "--lifecycle-only"))) {
if (!strcmp(argv[1], "--relay-only")) test_supernode_relay();
else test_partial_bootstrap_and_end();
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();
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 < 4; i++) stream_set_event_callback(nodes[i], NULL, NULL);
test_partial_bootstrap_and_end();
test_delayed_media_reader();
test_viewer_radio_mute();
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;
}