From 1809416dceac4f672b4ca94f4848f15a69b21b11 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Wed, 14 Oct 2026 16:26:41 +0200
Subject: [PATCH] Add atomic media reader opening with selected epoch
 validation

---
 src/stream/stream.c | 31 +++++++++++++++++++++++++++++++
 src/stream/stream.h |  4 ++++
 2 files changed, 35 insertions(+)

diff --git a/src/stream/stream.c b/src/stream/stream.c
index 10777bb9..890b7edd 100644
--- a/src/stream/stream.c
+++ b/src/stream/stream.c
@@ -809,6 +809,37 @@ int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) {
     return STREAM_OK;
 }
 
+/* Выбранная эпоха проверяется до открытия и по реально используемой session; ошибка снимает только новый owner. */
+int stream_media_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, uint64_t epoch,
+                       unsigned tracks, struct STREAM_HANDLE **out) {
+    if (!out || *out || !epoch || !source_id || !(tracks & STREAM_MEDIA_TRACK_VIDEO) || (tracks & ~3u)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media open invalid source=%llx epoch=%llx tracks=%u",
+                    (unsigned long long)source_id, (unsigned long long)epoch, tracks); return STREAM_ERR_ARGUMENT;
+    }
+    struct stream_member_state member;
+    int result = stream_member_load(inst, group_id, source_id, &member, NULL);
+    if (result != STREAM_OK) return result;
+    struct stream_session *s = inst->stream ? session_find(inst->stream, group_id, source_id) : NULL;
+    if (!member.published || member.source.source_epoch != epoch || (s && s->descriptor.source_epoch != epoch)) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media selected source changed group=%llx source=%llx expected=%llx published=%llx local=%llx",
+            (unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch,
+            (unsigned long long)(member.published ? member.source.source_epoch : 0),
+            (unsigned long long)(s ? s->descriptor.source_epoch : 0)); return STREAM_ERR_CONFLICT;
+    }
+    struct STREAM_HANDLE *h = NULL;
+    result = stream_subscribe_open(inst, group_id, source_id, &h);
+    if (result == STREAM_OK) result = stream_media_enable(h, tracks);
+    if (result != STREAM_OK) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media reader open failed group=%llx source=%llx epoch=%llx result=%d",
+            (unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch, result);
+        stream_close(h); return result;
+    }
+    *out = h;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media reader opened group=%llx source=%llx epoch=%llx tracks=%u",
+        (unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch, tracks);
+    return STREAM_OK;
+}
+
 /* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */
 int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int want_video, int64_t audio_ahead,
                        struct stream_media_packet *packet) {
diff --git a/src/stream/stream.h b/src/stream/stream.h
index 9393909b..b1879d18 100644
--- a/src/stream/stream.h
+++ b/src/stream/stream.h
@@ -92,6 +92,10 @@ int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info
  * Все вызовы — uasync. Заимствованный packet действует до следующего read/close.
  * LIVE DATA/GAP/END callbacks описывают сетевой прогресс; конец отложенного просмотра получает media_read. */
 int stream_media_enable(struct STREAM_HANDLE *handle, unsigned tracks);
+/* Атомарно открыть viewer выбранной эпохи с собственным cursor. CONFLICT означает смену источника.
+ * tracks=VIDEO или VIDEO|AUDIO; out заранее NULL, при отказе остаётся NULL. Закрытие через stream_close. */
+int stream_media_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, uint64_t epoch,
+                       unsigned tracks, struct STREAM_HANDLE **out);
 /* Один следующий видеокадр при want_video=1; звук ограничен position_us+audio_ahead_us. */
 int stream_media_read(struct STREAM_HANDLE *handle, int64_t position_us, int want_video, int64_t audio_ahead_us,
                        struct stream_media_packet *packet);