Add atomic media reader opening with selected epoch validation

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) {

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);