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

338 lines
23 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/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 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);
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);
}
/* Ошибка размера/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_open(nodes[1], GROUP1, nodes[0]->node_id, &reader1) == STREAM_OK);
CHECK(stream_subscribe_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_open(nodes[2], GROUP1, nodes[0]->node_id, &mobile) == STREAM_OK);
CHECK(stream_subscribe_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");
/* 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_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);
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);
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_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;
}