From 9c01535c78b33309d86af2872dcea1f7d86af03e Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 00:02:59 +0200 Subject: [PATCH] Exercise paused ring readers and graceful source close across live nodes --- tests/test_stream_service.c | 61 +++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 56c9e8b1..f3546431 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1377,6 +1377,66 @@ static void test_member_format(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] member format: full 64-bit IDs, canonical lists, duplicate/overflow rejection"); } +/* Пауза player не останавливает live ingest/selector; FIN сохраняет источник до дочитывания локального owner. */ +static void test_delayed_media_reader(void) { + struct STREAM_HANDLE *publisher = NULL, *remote = NULL, *preview = NULL; + CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); + struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK); + wait_state(1, 0, GROUP1, info.descriptor.source_epoch, 0); + CHECK(stream_media_enable(publisher, 0) == STREAM_OK); + CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &remote) == STREAM_OK); + CHECK(stream_subscribe_open(nodes[0], GROUP1, nodes[0]->node_id, &preview) == STREAM_OK); + CHECK(stream_media_enable(remote, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK); + CHECK(stream_media_enable(preview, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK); + uint8_t video[128] = {1}, wire[512]; size_t length = 0; + struct stream_media_block media = {.format = {64, 64, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; + media.video[0] = (struct stream_media_video){1, 0, sizeof(video), 0, sizeof(video), STREAM_MEDIA_KEY, video}; + CHECK(stream_media_encode(&media, wire, sizeof(wire), &length) == STREAM_OK); + CHECK(stream_publish_block(publisher, wire, length) == STREAM_OK); + struct stream_media_packet packet; int result = STREAM_WAIT; + uint64_t deadline = get_time_tb() + 100000; + while (result == STREAM_WAIT || result == STREAM_MORE) { + CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100); + result = stream_media_read(remote, -1, 0, -1, &packet); + } + CHECK(result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.pts_us == 0); + CHECK(stream_media_read(preview, -1, 0, -1, &packet) == STREAM_OK && packet.video.pts_us == 0); + struct stream_selection_info before, after; + CHECK(stream_selection_get_info(remote, &before) == STREAM_OK); + for (unsigned frame = 1; frame <= 150; frame++) { + media.video[0].frame_id = frame + 1; media.video[0].pts_us = (uint64_t)frame * 40000; + media.video[0].flags = frame % 25 ? 0 : STREAM_MEDIA_KEY; + CHECK(stream_media_encode(&media, wire, sizeof(wire), &length) == STREAM_OK); + CHECK(stream_publish_block(publisher, wire, length) == STREAM_OK); + test_clock_run_for(ua, 400); + } + struct stream_local live; CHECK(stream_local_get(nodes[1], &info.descriptor, &live) == STREAM_OK); + CHECK(live.next == 151); + CHECK(stream_selection_get_info(remote, &after) == STREAM_OK && after.restarts == before.restarts); + struct stream_media_status status; + CHECK(stream_media_get_status(remote, &status) == STREAM_OK && status.start_us == 0 && status.ready_us == 6000000); + CHECK(stream_publish_finish(publisher) == STREAM_OK); + stream_close(publisher); publisher = NULL; + CHECK(stream_handle_active(preview)); + unsigned counts[2] = {0}; struct STREAM_HANDLE *readers[2] = {remote, preview}; + for (unsigned reader = 0; reader < 2; reader++) { + deadline = get_time_tb() + 100000; + for (;;) { + result = stream_media_read(readers[reader], 10000000, 0, -1, &packet); + if (result == STREAM_WAIT || result == STREAM_MORE) { + CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100); continue; + } + CHECK(result == STREAM_OK); + if (packet.kind == STREAM_MEDIA_READ_END) break; + CHECK(packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.pts_us == (uint64_t)(counts[reader] + 1) * 40000); + CHECK(packet.video.length == sizeof(video) && !memcmp(packet.video.data, video, sizeof(video))); counts[reader]++; + } + CHECK(counts[reader] == 150); stream_close(readers[reader]); + } + test_clock_run_for(ua, 1000); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] delayed media readers: 6s pause, live progress without restart, exact remote/local FIN drain"); +} + int main(int argc, char **argv) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); @@ -1456,6 +1516,7 @@ int main(int argc, char **argv) { return 0; } test_profile_atomicity(); + test_delayed_media_reader(); 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);