diff --git a/src/stream/stream_media_buffer.c b/src/stream/stream_media_buffer.c index 9ae76a63..17bf364d 100644 --- a/src/stream/stream_media_buffer.c +++ b/src/stream/stream_media_buffer.c @@ -30,8 +30,8 @@ struct cursor_track { uint64_t seq; unsigned entry; int ended; }; struct stream_media_cursor { struct stream_media_store *store; struct cursor_track video, audio; - uint32_t generation; - int initialized, key_delivered; + uint32_t generation, video_revision; + int initialized; unsigned tracks; int64_t start_us; uint8_t *scratch[2], *frame; @@ -257,7 +257,8 @@ void stream_media_cursor_destroy(struct stream_media_cursor *c) { void stream_media_cursor_reset(struct stream_media_cursor *c) { if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor reset: null cursor"); return; } - c->initialized = c->key_delivered = 0; c->pending_used = 0; + c->initialized = 0; c->video_revision = 0; c->pending_used = 0; + c->pending = (struct stream_media_video){0}; c->pending_format = (struct stream_media_format){0}; c->scratch_valid[0] = c->scratch_valid[1] = 0; c->video = (struct cursor_track){0}; c->audio = (struct cursor_track){0}; c->audio.ended = !(c->tracks & STREAM_MEDIA_TRACK_AUDIO); @@ -358,10 +359,19 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe int result = cursor_peek(c, &c->video, 1, &b, &pts); if (result != STREAM_OK) return result; const struct stream_media_video *v = &b.video[c->video.entry++]; - if (!c->key_delivered && !c->pending.frame_size + if (b.format.revision != c->pending_format.revision) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "media cursor revision source=%llx seq=%llu revision=%u->%u pending=%u; require KEY", + (unsigned long long)c->store->descriptor.source_node_id, (unsigned long long)c->video.seq, + c->pending_format.revision, b.format.revision, c->pending_used); + c->pending = (struct stream_media_video){0}; c->pending_used = 0; c->pending_format = b.format; + } + /* Revision относится к уже выданному видео этого reader, а не к текущему live store. */ + if (b.format.revision != c->video_revision && !(v->flags & STREAM_MEDIA_KEY)) continue; + if (!c->pending.frame_size && v->offset) continue; + if (!c->video_revision && !c->pending.frame_size && (v->offset || !(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; if (!v->offset) { - if (!c->key_delivered && (!(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; + if (!c->video_revision && (!(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; if (v->frame_size > c->frame_capacity) { uint8_t *frame = u_realloc(c->frame, v->frame_size); if (!frame) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor frame allocation failed bytes=%u", v->frame_size); return STREAM_ERR_MEMORY; } @@ -370,7 +380,7 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe c->pending = *v; c->pending.data = NULL; c->pending_used = 0; c->pending_format = b.format; } if (!c->pending.frame_size || v->frame_id != c->pending.frame_id || v->offset != c->pending_used - || v->pts_us != c->pending.pts_us || v->frame_size != c->pending.frame_size) { + || v->pts_us != c->pending.pts_us || v->frame_size != c->pending.frame_size || v->flags != c->pending.flags) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media cursor fragment conflict seq=%llu frame=%llu offset=%u used=%u", (unsigned long long)c->video.seq, (unsigned long long)v->frame_id, v->offset, c->pending_used); return STREAM_GAP; @@ -379,7 +389,11 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe if (c->pending_used != c->pending.frame_size) continue; struct stream_media_packet out = {.kind = STREAM_MEDIA_READ_VIDEO, .format = c->pending_format, .video = c->pending}; out.video.offset = 0; out.video.length = out.video.frame_size; out.video.data = c->frame; - c->pending_used = 0; c->pending.frame_size = 0; c->key_delivered = 1; *packet = out; + if (c->video_revision != b.format.revision) DEBUG_INFO(DEBUG_CATEGORY_STREAM, + "media cursor KEY ready source=%llx revision=%u frame=%llu pts=%llu bytes=%u", + (unsigned long long)c->store->descriptor.source_node_id, b.format.revision, + (unsigned long long)v->frame_id, (unsigned long long)v->pts_us, v->frame_size); + c->pending_used = 0; c->pending.frame_size = 0; c->video_revision = b.format.revision; *packet = out; return STREAM_OK; } return STREAM_MORE; diff --git a/tests/test_stream_media.c b/tests/test_stream_media.c index 2d63f9b3..af35f71c 100644 --- a/tests/test_stream_media.c +++ b/tests/test_stream_media.c @@ -260,26 +260,94 @@ static int media_fixture_feed(struct media_fixture *f, const struct stream_media return stream_media_store_feed(f->store, f->seq++, decoded, decoded_size); } -/* Диагностика расхождения store/cursor; отдельный запуск --revision-transition до согласования контракта. */ +/* Новая revision начинается с целого KEY; аудио и отстающий reader сохраняют собственные позиции. */ static void test_revision_transition(void) { struct media_fixture f; media_fixture_open(&f); - struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; + struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1}, + .video_count = 1, .audio_count = 1}; b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; + b.audio[0] = (struct stream_media_audio){1000000, 3, 960, (const uint8_t *)"ops"}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); - struct stream_media_packet packet; + struct stream_media_packet packet; struct stream_media_status status; struct stream_media_cursor *slow = NULL; + CHECK(stream_media_cursor_create(&slow, f.store, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK); + CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.initialized); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1); + b.audio_count = 0; b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); - b.format = (struct stream_media_format){1920, 1080, 270, STREAM_MEDIA_H264, 0, 2}; + CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); + b.format = (struct stream_media_format){1920, 1080, 270, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 2}; b.video[0] = (struct stream_media_video){3, 1080000, 12, 0, 12, 0, expected_video}; CHECK(media_fixture_feed(&f, &b) == STREAM_OK); - struct stream_media_status status; CHECK(stream_media_cursor_status(f.cursor, &status) == STREAM_OK && status.video_us == 1000000); - int result = stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "revision transition before KEY: indexed_video_us=%lld result=%d kind=%d revision=%u frame=%llu flags=%u", - (long long)status.video_us, result, packet.kind, packet.format.revision, (unsigned long long)packet.video.frame_id, packet.video.flags); + CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); + CHECK(stream_media_cursor_read(f.cursor, 1000000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO); + CHECK(packet.audio.pts_us == 1000000 && packet.format.audio_codec == STREAM_MEDIA_OPUS_STEREO); + b.video_count = 0; b.audio_count = 1; b.audio[0].pts_us = 1020000; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, 1020000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO); + CHECK(packet.audio.pts_us == 1020000); + b.video_count = 1; b.audio_count = 0; + b.video[0] = (struct stream_media_video){4, 1120000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); + b.video[0].offset = 6; b.video[0].data = expected_video + 6; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); + CHECK(packet.format.revision == 2 && packet.video.frame_id == 4 && packet.video.flags == STREAM_MEDIA_KEY); + CHECK(packet.video.length == 12 && !memcmp(packet.video.data, expected_video, 12)); + CHECK(stream_media_cursor_read(slow, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1); + CHECK(stream_media_cursor_read(slow, 1000000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 2); + CHECK(packet.video.frame_id == 4); + /* Полный KEY новой конфигурации выдаётся сразу, без таймера и искусственной задержки. */ + b.format.revision = 3; b.video[0] = (struct stream_media_video){5, 1160000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, 1120000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 3); + CHECK(packet.video.frame_id == 5 && packet.video.flags == STREAM_MEDIA_KEY); + stream_media_cursor_destroy(slow); media_fixture_close(&f); - CHECK(result == STREAM_WAIT); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor revisions: missing/fragmented KEY, immediate complete KEY, independent audio and history"); +} + +/* Пропущенный блок, нарушенные фрагменты и вытеснение запрещают P-кадры до нового целого KEY. */ +static void test_cursor_key_recovery(void) { + for (unsigned failure = 0; failure < 3; failure++) { + struct media_fixture f; media_fixture_open(&f); + struct stream_media_block b = {.format = {1280, 720, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; + b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + struct stream_media_packet packet; + CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK); + b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT); + if (failure == 0) { stream_media_store_gap(f.store, f.seq, f.seq); f.seq++; } + else if (failure == 1) { + b.video[0].offset = 5; b.video[0].data = expected_video + 5; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + } else { + for (unsigned i = 0; i < 48; i++) { + b.video[0] = (struct stream_media_video){3 + i, 1080000 + i * 40000, 12, 0, 12, 0, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + } + } + int result = stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet); + CHECK(result == (failure == 2 ? STREAM_EXPIRED : STREAM_GAP)); + stream_media_cursor_reset(f.cursor); + b.video[0] = (struct stream_media_video){100, 4000000, 12, 0, 12, 0, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT); + b.video[0] = (struct stream_media_video){101, 4040000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video}; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT); + b.video[0].offset = 6; b.video[0].data = expected_video + 6; + CHECK(media_fixture_feed(&f, &b) == STREAM_OK); + CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); + CHECK(packet.video.frame_id == 101 && packet.video.flags == STREAM_MEDIA_KEY && packet.video.length == 12); + CHECK(!memcmp(packet.video.data, expected_video, 12)); + media_fixture_close(&f); + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor recovery: GAP, fragment discontinuity and expiry resume only from a complete KEY"); } int main(int argc, char **argv) { @@ -290,10 +358,12 @@ int main(int argc, char **argv) { for (unsigned i = 0; i < sizeof(expected_video); i++) expected_video[i] = (uint8_t)(i * 31 + 7); if (argc == 2 && !strcmp(argv[1], "--revision-transition")) { debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); - test_revision_transition(); + test_revision_transition(); test_cursor_key_recovery(); } else { CHECK(argc == 1); test_media(); test_batches(); test_revisions(); test_stereo(); test_codecs(); + debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); + test_revision_transition(); test_cursor_key_recovery(); } CHECK(u_get_allocated_count() == before); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);