jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Exercise live stream relay, common start, leases and epochs

master
evgeny 6 days ago
parent
d83e85b8ca
commit
09e86a16f6
1 changed files with 167 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 167
      tests/test_stream_service.c

167
tests/test_stream_service.c
Unescape View File

@ -11,6 +11,8 @@
#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/utun_instance.h"
#include "../lib/debug_config.h"
#include "../lib/json_flat.h"
@ -30,6 +32,42 @@ 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 uint8_t first_request[STREAM_WIRE_HEADER + 8];
static etcp_recv_fn saved_raw[4], saved_route;
static int drop_renew;
/* Протокол проходит настоящие 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++;
if (node == 3 && command == STREAM_CMD_RENEW && drop_renew) {
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.group_id == GROUP2 && event->descriptor.source_node_id == nodes[3]->node_id);
if (event->type == STREAM_EVENT_DATA) {
CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected)));
CHECK(event->first_seq == 1 + network_data[node]); network_data[node]++;
} else if (event->type == STREAM_EVENT_END) network_end[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]);
@ -104,6 +142,134 @@ static void root_hash(unsigned node, uint64_t group, uint8_t hash[32]) {
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_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);
}
/* 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);
/* Одна и та же 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);
network_write(publisher, 2); run_until(&network_data[1], 2); run_until(&network_data[2], 2);
/* Снять два лишних проверочных субпотока: шесть оставшихся восстанавливают новые пачки. */
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], 3); run_until(&network_data[2], 3);
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_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");
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]); }
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);
stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher);
for (unsigned i = 0; i < 2; i++) topo_group_peer_close(connections[i]);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live raw ETCP chain: common index, FEC, retained paths/LOOP, duplicate/reverse, END, renew/lease, REINIT");
}
/* Ошибка размера/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];
@ -247,6 +413,7 @@ int main(void) {
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();
/* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */
struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);

Write Preview
Loading…
Cancel
Save