jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
3294f67dbffc2bbaed03dbea8badbd44532ed397
utun2/tests/test_stream_service.c
T

1047 lines
71 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/* Живые 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;
}
Reference in new issue View Git Blame Copy Permalink