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.
2602 Commits
15 Branches
2 Tags
185 MiB
 
 
 
 
 
 
Tree: d44b4bc682
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'd44b4bc682'
${ noResults }
utun2/tests/test_stream_service.c

1046 lines
71 KiB
Raw Blame History

/* Живые CHAT-ядра и Merkle: атомарные опции, локальные владельцы, BGP-кандидаты и teardown. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <openssl/sha.h>
#include "../src/stream/stream.h"
#include "../src/stream/stream_member.h"
#include "../src/stream/stream_priv.h"
#include "../src/chat/chat_core.h"
#include "../src/chat/merkle_tree.h"
#include "../src/routing_layer/topo_node_sqlite.h"
#include "../src/routing_layer/route_crypto.h"
#include "../src/transport_layer/etcp.h"
#include "../src/transport_layer/node_conn_direct.h"
#include "../src/utun_instance.h"
#include "../lib/debug_config.h"
#include "../lib/json_flat.h"
#include "../lib/mem.h"
#include "test_clock.h"
#include "test_utils.h"
#define GROUP1 UINT64_C(710001)
#define GROUP2 UINT64_C(710002)
#define 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[7080], 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 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 == 1180); 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];
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) == STREAM_OK);
for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == 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) == STREAM_OK);
CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_OK);
CHECK(stream_receive_restart(subscriber) == STREAM_OK);
CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_DUPLICATE);
for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == 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], 1300) == 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);
stream_set_event_callback(nodes[3], NULL, NULL); stream_encoder_destroy(encoder);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated restart: deferred GAP retained; stale bootstrap never replays 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);
}
/* Источник -> 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);
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);
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[1180]; 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]);
memset(pressure_payload, 0xa9, 1180); 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, 1180, &accepted) == STREAM_OK && accepted == 1180);
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, 1180, &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, 1180, &accepted) == STREAM_OK);
test_clock_run_for(ua, 1200);
}
CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);
CHECK(state[2].failures > failures && state[2].providers == 2 && state[2].subflows == 2);
CHECK(automatic_data[2] == 400 && !automatic_gaps[2] && source_connection->links_up);
CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[1], 1); run_until(&automatic_end[2], 1);
for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
publisher = subscribers[1] = subscribers[2] = NULL;
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
wait_state(2, 3, GROUP2, origin.descriptor.source_epoch, 0);
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &subscribers[2]) == STREAM_OK);
test_clock_run_for(ua, 5000);
CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&automatic_end[2], 2);
CHECK(automatic_data[2] == 400); stream_close(subscribers[2]); stream_close(publisher);
subscribers[2] = publisher = NULL;
test_clock_run_for(ua, 1000);
/* END первого субпотока не выключает выбор недостающих: для 2+1 нужны ещё данные второй выдачи. */
parameters.n = 3;
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
for (unsigned i = 1; i < 3; i++) {
wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
CHECK(stream_subscribe_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK);
CHECK(stream_selection_configure(subscribers[i], &settings) == STREAM_OK);
automatic_next[i] = 0;
}
automatic_drop_tail = 1;
test_clock_run_for(ua, 10000);
size_t input_size = stream_input_size(&origin.descriptor);
memset(pressure_payload, 0xa9, input_size);
CHECK(stream_publish_write(publisher, pressure_payload, input_size, &accepted) == STREAM_OK && accepted == input_size);
CHECK(stream_publish_finish(publisher) == STREAM_OK);
struct stream_local terminal;
deadline = get_time_tb() + 100000;
do { test_clock_poll(ua, 100); CHECK(stream_local_get(nodes[2], &origin.descriptor, &terminal) == STREAM_OK); }
while (!terminal.finished && get_time_tb() < deadline);
CHECK(terminal.finished && terminal.anchored && !terminal.next && automatic_data[2] == 400 && automatic_end[2] == 2);
automatic_drop_tail = 0;
run_until(&automatic_data[2], 401); run_until(&automatic_end[2], 3);
CHECK(automatic_next[2] == 1 && !automatic_gaps[2]);
for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
test_clock_run_for(ua, 1000);
for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_handlers[i]) == 0);
stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] automatic streams: INFO-only, shared owners, 30s cycle, FEC spread, overlap, stall, CLOSED and early END drain");
}
/* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */
static void test_profile_atomicity(void) {
char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1];
uint8_t before_signature[64], after_signature[64], before_hash[32], after_hash[32];
uint64_t before_version, after_version;
profile(0, GROUP1, before, before_signature, &before_version); root_hash(0, GROUP1, before_hash);
memset(huge, 'p', MS_USERINFO_MAX); huge[MS_USERINFO_MAX] = 0;
struct chat_profile_option options[] = {{"stream_source", "0000000000000001:8:2:1189"}, {"stream_subscriptions", huge}};
CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) < 0);
profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash);
CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_signature, after_signature, 64));
CHECK(!memcmp(before_hash, after_hash, 32));
CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "CREATE TRIGGER reject_stream BEFORE UPDATE OF userinfo ON peers_710001 "
"BEGIN SELECT RAISE(ABORT,'injected stream profile failure'); END", NULL, NULL, NULL) == SQLITE_OK);
struct STREAM_HANDLE *failed = NULL;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &failed) < 0 && !failed);
CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "DROP TRIGGER reject_stream", NULL, NULL, NULL) == SQLITE_OK);
profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash);
CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_hash, after_hash, 32));
CHECK(!memcmp(before_signature, after_signature, 64));
options[0] = (struct chat_profile_option){"pad", "retained"}; options[1] = (struct chat_profile_option){"extra", "value"};
CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) == 0);
profile(0, GROUP1, after, after_signature, &after_version);
char value[64];
CHECK(json_flat_get(after, "avatar_file", value, sizeof(value)) == 0 && !strcmp(value, "retained-avatar"));
CHECK(json_flat_get(after, "avatar_server", value, sizeof(value)) == 0 && !strcmp(value, "0011223344556677"));
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] profile updates: retained avatar/name/options and atomic size/DB rollback");
}
/* Канонический список исключает неоднозначные IDs, дубли и обрезание значений. */
static void test_member_format(void) {
struct stream_member_state state = {.source = {.group_id = GROUP1, .source_node_id = 1, .source_epoch = 2,
.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}, .published = 1, .count = 2,
.subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX}}};
char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[MS_USERINFO_MAX + 1];
CHECK(stream_member_format(&state, source, subscriptions) == STREAM_OK);
snprintf(json, sizeof(json), "{\"stream_source\":\"%s\",\"stream_subscriptions\":\"%s\"}", source, subscriptions);
struct stream_member_state parsed;
CHECK(stream_member_parse(GROUP1, 1, json, &parsed) == STREAM_OK);
CHECK(parsed.published && parsed.source.source_epoch == 2 && parsed.count == 2);
CHECK(parsed.subscriptions[1].source_node_id == UINT64_MAX && parsed.subscriptions[1].source_epoch == UINT64_MAX);
const char *bad[] = {
"{\"stream_subscriptions\":\"1:2\"}",
"{\"stream_subscriptions\":\"0000000000000001:0000000000000002,\"}",
"{\"stream_subscriptions\":\"0000000000000001:0000000000000002,0000000000000001:0000000000000002\"}",
"{\"stream_source\":\"0000000000000002:256:2:1189\"}",
"{\"stream_source\":\"0000000000000002:8:2:1189\"}",
"{\"stream_source\":\"\",\"stream_source\":\"\"}"
};
for (unsigned i = 0; i < sizeof(bad) / sizeof(bad[0]); i++)
CHECK(stream_member_parse(GROUP1, 1, bad[i], &parsed) == STREAM_ERR_FORMAT);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] member format: full 64-bit IDs, canonical lists, duplicate/overflow rejection");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO);
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_WARN);
test_member_format();
size_t allocations = u_get_allocated_count();
test_clock_init(100); utun_instance_set_tun_init_enabled(0);
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:udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n"
"[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n", priv, pub, directory, i, type);
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);
for (unsigned g = 0; g < 2; g++) {
char id[32]; uint64_t group = g ? GROUP2 : GROUP1; snprintf(id, sizeof(id), "%llu", (unsigned long long)group);
uint8_t signature[64] = {0};
CHECK(topo_node_sqlite_channel_put(nodes[i]->topo_sqlite_db, id, "stream", nodes[0]->node_id,
group_x, NULL, group_public, i ? NULL : group_private, signature) == 0);
chat_core_ensure_channel_ready(nodes[i], id);
}
}
for (unsigned destination = 0; destination < 4; destination++) {
for (unsigned member = 0; member < 4; member++) {
put_member(destination, member, GROUP1); put_member(destination, member, GROUP2);
struct TOPO_NODE *identity = topo_node_registry_find(nodes[member]->topo_groups, nodes[member]->node_id);
CHECK(identity && topo_node_sqlite_snapshot_put(nodes[destination]->topo_sqlite_db, identity) >= 0);
}
}
struct TOPO_PEER_REQUEST *connections[3] = {0};
for (unsigned i = 1; i < 4; i++)
CHECK(topo_group_peer_open(topo_groups_find(nodes[0]->topo_groups, GROUP1), nodes[i]->node_id, &connections[i - 1]) == 0);
uint64_t deadline = get_time_tb() + 150000;
while (get_time_tb() < deadline) {
int ready = 1;
for (unsigned i = 0; i < 3; i++) if (topo_group_peer_phase(connections[i]) != TOPO_PEER_READY) ready = 0;
if (ready) break;
test_clock_poll(ua, 100);
}
for (unsigned i = 0; i < 3; i++) CHECK(topo_group_peer_phase(connections[i]) == TOPO_PEER_READY);
/* Каналы/мемберы стенда добавлены после chat_sync_init: первый обмен запускается явно. */
for (unsigned i = 1; i < 4; i++) {
CHECK(member_sync_start(nodes[0], nodes[i]->node_id, "710001", NULL, NULL) == 0);
CHECK(member_sync_start(nodes[i], nodes[0]->node_id, "710001", NULL, NULL) == 0);
}
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;
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();
/* 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];
CHECK(stream_encoder_encode(encoder, 100, 65530, 0, expected, sizeof(expected), blocks) == 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], 1300) == 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[1300]; memcpy(invalid, blocks[0], sizeof(invalid)); invalid[1299] ^= 1;
CHECK(stream_ingest_block(nodes[1], invalid, sizeof(invalid)) == STREAM_WAIT); /* Нет подписки — нет crypto. */
stream_encoder_destroy(encoder);
size_t accepted = 0;
CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
CHECK(stream_publish_finish(publisher) == STREAM_OK);
run_until(&data_events[0], 1); run_until(&end_events[0], 1);
CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && state.count == 1);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] opaque data/FEC delivery, close inside callback, independent group and DATA before END");
stream_close(publisher); publisher = NULL;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &info) == STREAM_OK && info.descriptor.source_epoch != epoch);
deadline = get_time_tb() + 150000;
while ((stream_handle_active(mobile) || stream_handle_active(desktop)) && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(!stream_handle_active(mobile) && !stream_handle_active(desktop));
CHECK(stream_handle_reason(mobile) == STREAM_CLOSE_SOURCE_CHANGED && stream_handle_reason(desktop) == STREAM_CLOSE_SOURCE_CHANGED);
CHECK(!stream_service_needs_activity(nodes[2]) && nodes[2]->client_activity == CLIENT_ACTIVITY_STANDBY);
stream_close(mobile); stream_close(desktop);
/* Повреждённое собственное членство удаляет обычная проверка Merkle; буфер не должен остаться активным. */
sqlite3_stmt *broken = NULL; uint8_t bad_signature[64]; memset(bad_signature, 1, sizeof(bad_signature));
CHECK(sqlite3_prepare_v2(nodes[1]->topo_sqlite_db, "UPDATE peers_710002 SET update_sig=? WHERE node_id=?",
-1, &broken, NULL) == SQLITE_OK);
sqlite3_bind_blob(broken, 1, bad_signature, sizeof(bad_signature), SQLITE_STATIC);
sqlite3_bind_int64(broken, 2, (sqlite3_int64)nodes[1]->node_id);
CHECK(sqlite3_step(broken) == SQLITE_DONE); sqlite3_finalize(broken);
CHECK(member_sync_verify_and_purge(nodes[1], "710002") == 1);
CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_MEMBER);
run_until(&closed_events[1], 2); stream_close(other_group); other_group = NULL;
put_member(1, 1, GROUP2);
CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK);
chat_core_delete_channel(nodes[1], "710002");
CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_GROUP);
stream_close(other_group);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source epoch, lost own membership and channel deletion detach local handles");
char stale[MS_USERINFO_MAX + 1]; uint8_t stale_signature[64]; uint64_t stale_version;
profile(0, GROUP1, stale, stale_signature, &stale_version);
CHECK(utun_service_start(nodes[0]) == 0);
sqlite3 *db = nodes[0]->topo_sqlite_db; struct ETCP_SOCKET *sockets = nodes[0]->etcp_sockets;
CHECK(stream_publish_write(publisher2, expected, 17, &accepted) == STREAM_OK);
chat_service_stop(nodes[0]);
CHECK(!nodes[0]->stream && nodes[0]->utun_started && nodes[0]->topo_sqlite_db == db && nodes[0]->etcp_sockets == sockets);
CHECK(!stream_handle_active(publisher) && !stream_handle_active(publisher2));
stream_close(publisher); stream_close(publisher2);
sqlite3_stmt *restore = NULL;
CHECK(sqlite3_prepare_v2(db, "UPDATE peers_710001 SET userinfo=?,update_sig=?,update_ts=? WHERE node_id=?", -1, &restore, NULL) == SQLITE_OK);
sqlite3_bind_text(restore, 1, stale, -1, SQLITE_STATIC); sqlite3_bind_blob(restore, 2, stale_signature, 64, SQLITE_STATIC);
sqlite3_bind_int64(restore, 3, (sqlite3_int64)stale_version); sqlite3_bind_int64(restore, 4, (sqlite3_int64)nodes[0]->node_id);
CHECK(sqlite3_step(restore) == SQLITE_DONE); sqlite3_finalize(restore);
CHECK(chat_service_start(nodes[0]) == 0 && nodes[0]->stream);
CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && !state.count);
CHECK(data_events[0] == 1); /* Pending 100ms batch не выполнилась после stop. */
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] CHAT stop keeps UTUN/core; source timer cancelled; stale signed options cleared on restart");
for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
for (unsigned i = 0; i < 4; i++) utun_instance_destroy(nodes[i]);
uasync_destroy(ua, 1); test_clock_finish();
if (u_get_allocated_count() != allocations) {
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "tracked allocations before=%zu after=%zu", allocations, u_get_allocated_count());
u_report_unfreed_blocks();
}
CHECK(u_get_allocated_count() == allocations);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream service checks complete, no tracked allocations remain");
return 0;
}