Separate media reader cursors from live stream receive progress

src/stream/stream.c

@@ -26,6 +26,10 @@ struct stream_session {
     uint8_t *output;
     struct stream_metrics *metrics;
     struct stream_optimizer *optimizer;
+    struct stream_media_store *media;
+    void *media_work;
+    uint64_t media_next, media_seek;
+    int media_seeking, media_failed;
     void *delivery;
     uint64_t next_seq;
     uint64_t start_seq, last_seq, passed_before;
@@ -45,6 +49,7 @@ struct STREAM_HANDLE {
     int relay;
     int selection_owned;
     int reason;
+    struct stream_media_cursor *media_cursor;
 };
 
 struct stream_service {
@@ -64,6 +69,7 @@ struct stream_service {
 
 static void session_deliver(void *arg);
 static void service_reconcile(void *arg);
+static void session_media_schedule(struct stream_session *s);
 
 /* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */
 int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) {
@@ -230,6 +236,8 @@ static void session_free(struct stream_session *s) {
     while (*link != s) link = &(*link)->next;
     *link = s->next;
     if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery);
+    if (s->media_work) uasync_call_soon_cancel(s->service->inst->ua, s->media_work);
+    stream_media_store_destroy(s->media);
     stream_optimizer_destroy(s->optimizer);
     if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
     if (s->source) stream_source_destroy(s->source);
@@ -260,6 +268,7 @@ static void session_close(struct stream_session *s, int reason, int update_profi
     for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next)
         if (h->session == s) { h->session = NULL; h->reason = reason; }
     if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; }
+    if (s->media_work) { uasync_call_soon_cancel(s->service->inst->ua, s->media_work); s->media_work = NULL; }
     struct TOPO_GROUP *group = s->service->inst->topo_groups
                                ? topo_groups_find(s->service->inst->topo_groups, s->descriptor.group_id) : NULL;
     if (update_profile && group && service_publish_options(s->service, group) != STREAM_OK)
@@ -275,6 +284,62 @@ static void session_close(struct stream_session *s, int reason, int update_profi
     else if (!s->holds) session_free(s);
 }
 
+/* Поздний reader ищет ближайший сохранённый KEY с ограниченным бюджетом, затем индексирует вперёд. */
+static void session_media_work(void *arg) {
+    struct stream_session *s = arg; s->media_work = NULL; s->holds++;
+    int again = 0;
+    struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
+    if (!s->closing && !s->media_failed) {
+        for (unsigned work = 0; work < 8 && s->media_seeking; work++) {
+            size_t length = 0;
+            int result = stream_buffer_decode(s->buffer, s->media_seek, s->output, stream_input_size(&s->descriptor), &length);
+            if (result == STREAM_OK) {
+                struct stream_media_block b;
+                if (stream_media_parse(s->output, length, &b) != STREAM_OK) { s->media_failed = 1; break; }
+                for (unsigned i = 0; i < b.video_count; i++) {
+                    if (!b.video[i].offset && (b.video[i].flags & STREAM_MEDIA_KEY)) {
+                        s->media_next = s->media_seek; s->media_seeking = 0; break;
+                    }
+                }
+            }
+            if (!s->media_seeking) break;
+            if (s->media_seek <= stats.expired_before) { s->media_seeking = 0; s->media_next = s->next_seq; break; }
+            s->media_seek--;
+        }
+        if (s->media_seeking) again = 1;
+        for (unsigned work = 0; work < 8 && !s->media_seeking && !s->media_failed && s->media_next < s->next_seq; work++) {
+            size_t length = 0;
+            int result = stream_buffer_decode(s->buffer, s->media_next, s->output, stream_input_size(&s->descriptor), &length);
+            if (result == STREAM_OK) {
+                result = stream_media_store_feed(s->media, s->media_next++, s->output, length);
+                if (result < 0) s->media_failed = 1;
+            } else if (result == STREAM_EXPIRED || result == STREAM_REJECTED || result == STREAM_WAIT || result == STREAM_GAP) {
+                uint64_t next = s->next_seq, available;
+                if (stream_buffer_next_decodable(s->buffer, s->media_next + 1, &available) == STREAM_OK && available < next) next = available;
+                stream_media_store_gap(s->media, s->media_next, next - 1); s->media_next = next;
+            } else {
+                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index decode failed source=%llx seq=%llu result=%d",
+                            (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->media_next, result);
+                s->media_failed = 1;
+            }
+        }
+        if (!s->media_seeking && s->end_valid && (s->last_seq == UINT64_MAX || s->media_next > s->last_seq))
+            stream_media_store_finish(s->media);
+        if (!s->media_failed && s->media_next < s->next_seq) again = 1;
+    }
+    s->holds--;
+    if (s->closing) { if (!s->holds && !s->delivery) session_free(s); }
+    else if (again) session_media_schedule(s);
+}
+
+/* Индексация не зависит от позиции/паузы плеера и не создаёт отдельную очередь медиаданных. */
+static void session_media_schedule(struct stream_session *s) {
+    if (!s->media || s->media_work || s->closing || s->media_failed || s->service->stopping) return;
+    s->media_work = uasync_call_soon(s->service->inst->ua, s, session_media_work);
+    if (!s->media_work) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index scheduling failed source=%llx",
+                                   (unsigned long long)s->descriptor.source_node_id);
+}
+
 /* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */
 static void session_deliver(void *arg) {
     struct stream_session *s = arg;
@@ -334,6 +399,13 @@ static void session_deliver(void *arg) {
         s->end_delivered = 1; emit = 1;
     }
     if (!s->closing && s->owners == s->relay_owners) emit = 0;
+    if (!s->closing && s->media) {
+        if (event.type == STREAM_EVENT_DATA && emit && !s->media_seeking && s->media_next == event.first_seq) {
+            int result = stream_media_store_feed(s->media, s->media_next++, event.data, event.length);
+            if (result < 0) s->media_failed = 1;
+        }
+        session_media_schedule(s);
+    }
     if (emit && service->callback && !service->stopping) {
         service->dispatching++;
         service->callback(service->inst, &event, service->arg);
@@ -668,6 +740,7 @@ int stream_selection_get_info(const struct STREAM_HANDLE *h, struct stream_selec
 /* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */
 void stream_close(struct STREAM_HANDLE *h) {
     if (!h) return;
+    stream_media_cursor_destroy(h->media_cursor); h->media_cursor = NULL;
     struct stream_service *service = h->service;
     if (service) {
         struct STREAM_HANDLE **link = &service->handles;
@@ -704,6 +777,53 @@ int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) {
     return STREAM_OK;
 }
 
+/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */
+int stream_media_enable(struct STREAM_HANDLE *h, int reader) {
+    if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; }
+    struct stream_session *s = h->session;
+    if (!s->media) {
+        int result = stream_media_store_create(&s->media, s->buffer, &s->descriptor);
+        if (result != STREAM_OK) return result;
+        struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
+        s->media_next = stats.expired_before;
+        if (s->anchored && s->next_seq > stats.expired_before) {
+            s->media_seek = s->next_seq - 1; s->media_seeking = 1;
+        }
+    }
+    if (reader && !h->media_cursor) {
+        int result = stream_media_cursor_create(&h->media_cursor, s->media);
+        if (result != STREAM_OK) return result;
+    }
+    session_media_schedule(s);
+    return STREAM_OK;
+}
+
+/* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */
+int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int64_t lookahead, struct stream_media_packet *packet) {
+    if (!stream_handle_active(h) || !h->media_cursor) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media read: detached/unconfigured reader"); return STREAM_ERR_STATE;
+    }
+    if (h->session->media_failed) return STREAM_ERR_FORMAT;
+    if (h->session->media_seeking || h->session->media_next < h->session->next_seq) return STREAM_WAIT;
+    return stream_media_cursor_read(h->media_cursor, position, lookahead, packet);
+}
+
+int stream_media_get_status(struct STREAM_HANDLE *h, struct stream_media_status *status) {
+    if (!stream_handle_active(h) || !h->media_cursor || !status) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media status: detached/unconfigured reader"); return STREAM_ERR_STATE;
+    }
+    if (h->session->media_failed) return STREAM_ERR_FORMAT;
+    return stream_media_cursor_status(h->media_cursor, status);
+}
+
+int stream_media_resync(struct STREAM_HANDLE *h) {
+    if (!stream_handle_active(h) || !h->media_cursor) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media resync: detached/unconfigured reader"); return STREAM_ERR_STATE;
+    }
+    stream_media_cursor_reset(h->media_cursor);
+    return STREAM_OK;
+}
+
 size_t stream_publish_capacity(const struct STREAM_HANDLE *h) {
     if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream capacity not a publisher"); return 0; }
     return stream_source_input_capacity(h->session->source);

src/stream/stream.h

@@ -8,6 +8,7 @@
 #define STREAM_H
 
 #include "stream_buffer.h"
+#include "stream_media_buffer.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -86,6 +87,15 @@ int stream_handle_active(const struct STREAM_HANDLE *handle);
 int stream_handle_reason(const struct STREAM_HANDLE *handle);
 int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info);
 
+/* Медиаплеер использует тот же RS-buffer, но собственные позиции чтения.
+ * reader=0 включает только общий индекс (publisher), reader=1 создаёт cursor этого handle.
+ * Все вызовы — uasync. Заимствованный packet действует до следующего read/close.
+ * LIVE DATA/GAP/END callbacks описывают сетевой прогресс; конец отложенного просмотра получает media_read. */
+int stream_media_enable(struct STREAM_HANDLE *handle, int reader);
+int stream_media_read(struct STREAM_HANDLE *handle, int64_t position_us, int64_t lookahead_us, struct stream_media_packet *packet);
+int stream_media_get_status(struct STREAM_HANDLE *handle, struct stream_media_status *status);
+int stream_media_resync(struct STREAM_HANDLE *handle);
+
 size_t stream_publish_capacity(const struct STREAM_HANDLE *handle);
 int stream_publish_write(struct STREAM_HANDLE *handle, const uint8_t *data, size_t length, size_t *accepted);
 /* Публикует одну непустую FEC-пачку, без склейки/таймера. Не смешивать с publish_write в одном запуске. */