diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index b1240bc5..286f7b07 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -46,6 +46,11 @@ 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 etcp_recv_fn automatic_handlers[4]; +static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; 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); @@ -60,6 +65,36 @@ static void on_global_pressure(struct UTUN_INSTANCE *inst, const struct stream_e 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; + } + } + 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 == 1180 && !memcmp(event->data, pressure_payload, 1180)); + 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); +} + 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) { @@ -567,6 +602,95 @@ static void test_global_pressure(void) { 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, ¶meters, &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); + 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); + wait_state(2, 1, GROUP2, origin.descriptor.source_epoch, 1); + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &subscribers[2]) == STREAM_OK); + 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); + 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 empty source, shared owners, 30s cycle, FEC spread, overlap, stall and CLOSED failover"); +} + /* Ошибка размера/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]; @@ -713,6 +837,7 @@ int main(void) { test_network(); test_pressure(); test_global_pressure(); + test_automatic(); /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */ struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);