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

Exercise supernode relay delivery and preserve ended receive handles

master
evgeny 6 days ago
parent
29d91d4b19
commit
8dbfb4d297
2 changed files with 157 additions and 6 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. 16
      src/stream/stream_network.c
  2. 147
      tests/test_stream_service.c

16
src/stream/stream_network.c
Unescape View File

@ -199,9 +199,17 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin
/* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */
static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) {
struct UTUN_INSTANCE *inst = net->inst;
if (!etcp_router_has_route(inst, m.descriptor.group_id, peer)
|| !etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for route/room peer=%llx command=%u",
int route = etcp_router_has_route(inst, m.descriptor.group_id, peer);
if (!route) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for group route peer=%llx group=%llx command=%u",
(unsigned long long)peer, (unsigned long long)m.descriptor.group_id, m.command); return STREAM_WAIT;
}
if (!etcp_router_conn_get(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed channel allocation failed peer=%llx", (unsigned long long)peer);
return STREAM_ERR_MEMORY;
}
if (!etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for queue room peer=%llx command=%u",
(unsigned long long)peer, m.command); return STREAM_WAIT;
}
struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length));
@ -960,7 +968,7 @@ static void network_tick(void *arg) {
&& f->info.last_available != UINT64_MAX))
&& now >= f->deadline)
flow_close(f, f->outgoing ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
else if (f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
else if (f->outgoing && f->info.state >= STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
.command = STREAM_CMD_RENEW, .subblock = f->subblock};
if (control_put(p, m, NULL, 0) == STREAM_OK) f->renewed = now;

147
tests/test_stream_service.c
Unescape View File

@ -54,6 +54,10 @@ static int automatic_drop_relay;
static int automatic_drop_tail;
static etcp_recv_fn automatic_handlers[4];
static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
static etcp_recv_fn relay_raw_handlers[3], relay_route_handler;
static unsigned relay_data[3], relay_ends[3], relay_requests_seen;
static int relay_drop_source;
static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
static struct {
unsigned data, reinits, responses;
int hold_info;
@ -990,18 +994,156 @@ static void test_profile_atomicity(void) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] profile updates: retained avatar/name/options and atomic size/DB rollback");
}
/* Блок владельца действительно подписан канальным ключом и проходит member_sync, профиль A сохраняется. */
static void relay_role(unsigned member, int enabled, unsigned version) {
char tags[128]; uint8_t signature[64], body[256];
snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"supernode\":\"%s\"}", version, enabled ? "yes" : "no");
int length = member_sync_build_owner_msg(nodes[member]->node_id, tags, body, sizeof(body)); CHECK(length > 0);
CHECK(sc_ed25519_sign(group_private, body, (size_t)length, signature) == SC_OK);
for (unsigned i = 0; i < 4; i++) {
CHECK(member_sync_put(nodes[i], "710002", nodes[member]->node_id, nodes[member]->my_keys.public_key,
nodes[member]->my_ed25519_pubkey, NULL, 0, NULL, 0, NULL, tags, signature, 0, 0, NULL) >= 0);
CHECK(stream_member_supernode(nodes[i], GROUP2, nodes[member]->node_id) == enabled);
}
}
/* Прямой stream-вход источника блокируем только у зрителей; Merkle/BGP и получение суперузла остаются настоящими. */
static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
unsigned index = 0; while (index < 3 && conn->instance != nodes[index]) index++; CHECK(index < 3);
if (relay_drop_source && index != 1 && conn->peer_node_id == nodes[3]->node_id) {
queue_dgram_free(entry); queue_entry_free(entry); return;
}
relay_raw_handlers[index](conn, entry);
}
/* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */
static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) {
memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++;
}
relay_route_handler(conn, entry);
}
/* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */
static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
(void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3);
CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == sizeof(expected)); relay_data[index]++; }
if (event->type == STREAM_EVENT_END) relay_ends[index]++;
}
/* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */
static void test_supernode_relay(void) {
struct TOPO_PEER_REQUEST *connection = NULL;
CHECK(topo_group_peer_open(topo_groups_find(nodes[3]->topo_groups, GROUP2), nodes[0]->node_id, &connection) == 0);
uint64_t deadline = get_time_tb() + 150000;
while (topo_group_peer_phase(connection) != TOPO_PEER_READY && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(topo_group_peer_phase(connection) == TOPO_PEER_READY);
CHECK(member_sync_start(nodes[3], nodes[0]->node_id, "710002", NULL, NULL) == 0);
CHECK(member_sync_start(nodes[0], nodes[3]->node_id, "710002", NULL, NULL) == 0);
relay_role(1, 1, 100);
for (unsigned i = 0; i < 3; i++) {
relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM];
CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_test_raw) == 0);
stream_set_event_callback(nodes[i], relay_test_event, NULL);
}
relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0};
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
run_until(&relay_data[2], 1); test_clock_run_for(ua, 60000);
struct stream_local local;
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen);
stream_close(viewers[2]); viewers[2] = NULL;
test_clock_run_for(ua, 10000);
relay_drop_source = 1;
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
deadline = get_time_tb() + 300000;
while (relay_data[2] < 2 && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(relay_data[2] == 2 && relay_requests_seen && !relay_data[1]);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only);
/* Поздний обычный подписчик также удерживает раздачу, даже без собственного запроса помощи. */
CHECK(stream_subscribe_open(nodes[0], GROUP2, nodes[3]->node_id, &viewers[0]) == STREAM_OK);
run_until(&relay_data[0], 1);
stream_close(viewers[2]); viewers[2] = NULL;
test_clock_run_for(ua, 80000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK);
/* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only);
unsigned before = relay_data[1];
CHECK(stream_publish_block(publisher, expected, sizeof(expected)) == STREAM_OK);
run_until(&relay_data[0], 2); run_until(&relay_data[1], before + 1);
stream_close(viewers[0]); viewers[0] = NULL;
test_clock_run_for(ua, 100000);
CHECK(stream_handle_active(viewers[1]));
CHECK(stream_get_info(viewers[1], &origin) == STREAM_OK && origin.owners == 1);
stream_close(viewers[1]); viewers[1] = NULL;
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
/* Поздний повтор завершённого запроса не подписывает суперузел заново. */
struct ll_entry *entry = ll_alloc_lldgram(1 + STREAM_WIRE_HEADER); CHECK(entry);
entry->dgram[0] = ETCP_RT_ID_STREAM; memcpy(entry->dgram + 1, relay_request_wire, STREAM_WIRE_HEADER);
entry->len = 1 + STREAM_WIRE_HEADER;
CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
test_clock_run_for(ua, 50000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
/* Два relay-only экземпляра не выбирают друг друга, даже если имеют буферы и живые BGP-пути. */
struct STREAM_HANDLE *helpers[2] = {0};
relay_role(0, 1, 100);
CHECK(stream_relay_open(nodes[1], &origin.descriptor, &helpers[0]) == STREAM_OK);
CHECK(stream_relay_open(nodes[0], &origin.descriptor, &helpers[1]) == STREAM_OK);
test_clock_run_for(ua, 30000);
for (unsigned i = 0; i < 2; i++) {
struct stream_candidate *candidates = NULL; size_t count = 0;
CHECK(stream_get_candidates(helpers[i], &candidates, &count) == STREAM_OK);
for (size_t c = 0; c < count; c++) CHECK(candidates[c].node_id != nodes[i ? 1 : 0]->node_id);
u_free(candidates); stream_close(helpers[i]);
}
relay_role(0, 0, 101);
stream_close(publisher); publisher = NULL; test_clock_run_for(ua, 10000);
/* Пустой END проходит помощь до READY/DATA; буфер освобождается, хотя источник handle ещё жив. */
CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
for (unsigned i = 0; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0);
CHECK(stream_subscribe_manual_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
stream_relay_demand(nodes[2], &origin.descriptor, 1, get_time_tb() / 10);
deadline = get_time_tb() + 100000;
while (stream_local_get(nodes[1], &origin.descriptor, &local) != STREAM_OK && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK);
test_clock_run_for(ua, 20000); CHECK(stream_publish_finish(publisher) == STREAM_OK);
run_until(&relay_ends[2], 1); CHECK(!relay_ends[1]);
test_clock_run_for(ua, 50000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
stream_close(viewers[2]); stream_close(publisher);
relay_drop_source = 0; relay_role(1, 0, 101);
CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0);
for (unsigned i = 0; i < 3; i++) {
CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0);
stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);
}
topo_group_peer_close(connection);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supernode assistance: direct success, fallback, shared viewers, idle release, replay, relay-only exclusion, empty END");
}
/* Канонический список исключает неоднозначные IDs, дубли и обрезание значений. */
static void test_member_format(void) {
struct stream_member_state state = {.source = {.group_id = GROUP1, .source_node_id = 1, .source_epoch = 2,
.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}, .published = 1, .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];
.subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX, 1}}};
char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[2 * MS_USERINFO_MAX + 64];
CHECK(stream_member_format(&state, source, subscriptions) == STREAM_OK);
snprintf(json, sizeof(json), "{\"stream_source\":\"%s\",\"stream_subscriptions\":\"%s\"}", source, subscriptions);
struct stream_member_state parsed;
CHECK(stream_member_parse(GROUP1, 1, json, &parsed) == STREAM_OK);
CHECK(parsed.published && parsed.source.source_epoch == 2 && parsed.count == 2);
CHECK(parsed.subscriptions[1].source_node_id == UINT64_MAX && parsed.subscriptions[1].source_epoch == UINT64_MAX);
CHECK(!parsed.subscriptions[0].relay_only && parsed.subscriptions[1].relay_only);
const char *bad[] = {
"{\"stream_subscriptions\":\"1:2\"}",
"{\"stream_subscriptions\":\"0000000000000001:0000000000000002,\"}",
@ -1116,6 +1258,7 @@ int main(void) {
test_pressure();
test_global_pressure();
test_automatic();
test_supernode_relay();
test_pending_down();
/* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */

Write Preview
Loading…
Cancel
Save