Test independent playback readers over one authenticated stream ring

tests/test_stream_playback.c

@@ -3,7 +3,9 @@
 #include <string.h>
 
 #include "../src/stream/stream_playback_clock.h"
+#include "../src/stream/stream_media_buffer.h"
 #include "../lib/debug_config.h"
+#include "../lib/mem.h"
 
 #define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback test line=%d: %s", __LINE__, #x); exit(1); } } while (0)
 
@@ -69,9 +71,96 @@ static void test_invalid_and_sampling_gap(void) {
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] clock: invalid/backward positions preserve result, missed samples reset healthy minute");
 }
 
+static const struct stream_descriptor descriptor = {.group_id = 1, .source_node_id = 2, .source_epoch = 3,
+    .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
+static uint8_t video_bytes[128], opus_bytes[10];
+
+/* Исходный медиаблок хранится только в RS; индекс получает проверенное восстановленное содержимое. */
+static void append_media(struct stream_encoder *encoder, struct stream_buffer *buffer, struct stream_media_store *store,
+                         uint64_t seq, uint64_t pts, int key, int parity_first) {
+    struct stream_media_block b = {.format = {64, 64, 0, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1},
+        .video_count = 1, .audio_count = 2};
+    b.video[0] = (struct stream_media_video){seq + 1, pts, sizeof(video_bytes), 0, sizeof(video_bytes), key, video_bytes};
+    b.audio[0] = (struct stream_media_audio){pts, sizeof(opus_bytes), 960, opus_bytes};
+    b.audio[1] = (struct stream_media_audio){pts + 20000, sizeof(opus_bytes), 960, opus_bytes};
+    uint8_t wire[1024], restored[1024]; size_t length = 0, part_size = 0, decoded = 0;
+    CHECK(stream_media_encode(&b, wire, sizeof(wire), &length) == STREAM_OK);
+    const uint8_t *parts[RS_MAX_BLOCKS];
+    CHECK(stream_encoder_encode(encoder, seq, (uint16_t)(pts / 1000), 0, wire, length, parts, &part_size) == STREAM_OK);
+    for (unsigned i = parity_first ? 2 : 0; i < 8; i++) CHECK(stream_buffer_put(buffer, parts[i], part_size) == STREAM_OK);
+    CHECK(stream_buffer_decode(buffer, seq, restored, sizeof(restored), &decoded) == STREAM_OK);
+    CHECK(decoded == length && !memcmp(restored, wire, length));
+    CHECK(stream_media_store_feed(store, seq, restored, decoded) == STREAM_OK);
+    if (parity_first) {
+        uint8_t bad[1300]; memcpy(bad, parts[0], part_size); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1;
+        CHECK(stream_buffer_put(buffer, bad, part_size) == STREAM_OK);
+        CHECK(stream_buffer_decode(buffer, seq, restored, sizeof(restored), &decoded) == STREAM_OK);
+        CHECK(decoded == length && !memcmp(restored, wire, length));
+    }
+}
+
+/* Два курсора делят кодированные данные; быстрый reader и late part не меняют результат медленного. */
+static void test_shared_media_ring(void) {
+    uint8_t key[32] = {1}; struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL;
+    struct stream_media_store *store = NULL; struct stream_media_cursor *fast = NULL, *slow = NULL;
+    CHECK(stream_encoder_create(&encoder, &descriptor, key) == STREAM_OK);
+    CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 32 * 1300) == STREAM_OK);
+    CHECK(stream_media_store_create(&store, buffer, &descriptor) == STREAM_OK);
+    CHECK(stream_media_cursor_create(&fast, store) == STREAM_OK && stream_media_cursor_create(&slow, store) == STREAM_OK);
+    append_media(encoder, buffer, store, 0, 0, STREAM_MEDIA_KEY, 1);
+    struct stream_media_status status;
+    CHECK(stream_media_cursor_status(fast, &status) == STREAM_OK && status.initialized && status.ready_us == 0);
+    CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.initialized && status.start_us == 0);
+    append_media(encoder, buffer, store, 1, 40000, 0, 0);
+    append_media(encoder, buffer, store, 2, 80000, 0, 0);
+    struct stream_media_packet packet;
+    CHECK(stream_media_cursor_read(fast, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
+    CHECK(packet.video.pts_us == 0 && !memcmp(packet.video.data, video_bytes, sizeof(video_bytes)));
+    unsigned videos = 1, audio = 0;
+    for (unsigned step = 0; step < 20; step++) {
+        int result = stream_media_cursor_read(fast, 80000, 0, &packet);
+        if (result == STREAM_WAIT) break;
+        CHECK(result == STREAM_OK);
+        if (packet.kind == STREAM_MEDIA_READ_VIDEO) videos++;
+        else { CHECK(packet.kind == STREAM_MEDIA_READ_AUDIO && packet.audio.pts_us <= 80000); audio++; }
+    }
+    CHECK(videos == 3 && audio == 5);
+    CHECK(stream_media_cursor_read(slow, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
+    CHECK(packet.video.pts_us == 0 && !memcmp(packet.video.data, video_bytes, sizeof(video_bytes)));
+    append_media(encoder, buffer, store, 3, 120000, STREAM_MEDIA_KEY, 0);
+    CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.start_us == 0 && status.ready_us == 120000);
+    stream_media_store_gap(store, 4, 4);
+    append_media(encoder, buffer, store, 5, 200000, STREAM_MEDIA_KEY, 0);
+    CHECK(stream_media_cursor_read(slow, 0, 0, &packet) == STREAM_EXPIRED);
+    int gap = 0;
+    for (unsigned step = 0; step < 20; step++) {
+        int result = stream_media_cursor_read(fast, 180000, 0, &packet);
+        if (result == STREAM_GAP) { gap = 1; break; }
+        CHECK(result == STREAM_OK || result == STREAM_WAIT);
+    }
+    CHECK(gap);
+    stream_media_cursor_reset(fast);
+    CHECK(stream_media_cursor_read(fast, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
+    CHECK(packet.video.pts_us == 200000);
+    stream_media_store_finish(store);
+    int ended = 0;
+    for (unsigned step = 0; step < 20; step++) {
+        int result = stream_media_cursor_read(fast, 1000000, 0, &packet);
+        CHECK(result == STREAM_OK || result == STREAM_WAIT);
+        if (result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_END) { ended = 1; break; }
+    }
+    CHECK(ended);
+    stream_media_cursor_destroy(fast); stream_media_cursor_destroy(slow); stream_media_store_destroy(store);
+    stream_buffer_destroy(buffer); stream_encoder_destroy(encoder);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] media ring: independent readers, authenticated frozen parts, lookahead, expiry, GAP and EOF");
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
     debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
-    test_history_correction(); test_catchup_device_queue(); test_invalid_and_sampling_gap();
+    size_t allocations = u_get_allocated_count();
+    for (unsigned i = 0; i < sizeof(video_bytes); i++) video_bytes[i] = (uint8_t)(i * 71 + 3);
+    test_history_correction(); test_catchup_device_queue(); test_invalid_and_sampling_gap(); test_shared_media_ring();
+    CHECK(u_get_allocated_count() == allocations);
     return 0;
 }