Track live stream catalogues independently for each owner

src/stream/stream.c

@@ -54,6 +54,15 @@ struct STREAM_HANDLE {
     struct stream_media_cursor *media_cursor;
 };
 
+struct source_watch {
+    struct source_watch *next;
+    struct stream_service *service;
+    uint64_t owner, group;
+    void *pending;
+    char *snapshot;
+    int valid;
+};
+
 struct stream_service {
     struct UTUN_INSTANCE *inst;
     struct stream_session *sessions;
@@ -62,10 +71,7 @@ struct stream_service {
     stream_event_fn callback;
     void *arg;
     void *reconcile;
-    void *sources_pending;
-    uint64_t sources_group;
-    char *sources_snapshot;
-    int sources_valid;
+    struct source_watch *watches;
     int stopping, dispatching, registered, initializing;
 };
 
@@ -125,14 +131,15 @@ int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **ou
 
 /* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */
 static void sources_deliver(void *arg) {
-    struct stream_service *service = arg;
-    service->sources_pending = NULL;
-    if (service->stopping || !service->sources_group) return;
+    struct source_watch *watch = arg;
+    struct stream_service *service = watch->service;
+    watch->pending = NULL;
+    if (service->stopping) return;
     char *json = NULL;
-    int result = stream_sources_json(service->inst, service->sources_group, &json);
-    int valid = result == STREAM_OK || result == STREAM_WAIT;
+    int result = stream_sources_json(service->inst, watch->group, &json);
+    int valid = result == STREAM_OK;
     const char *list = json ? json : "[]";
-    if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) {
+    if (watch->snapshot && valid == watch->valid && !strcmp(watch->snapshot, list)) {
         u_free(json); return;
     }
     size_t length = strlen(list);
@@ -140,30 +147,46 @@ static void sources_deliver(void *arg) {
     if (!event) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return;
     }
-    stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid;
+    stream_write_be(event, watch->group, 8); event[8] = (uint8_t)valid;
     memcpy(event + 9, list, length);
     char *snapshot = json ? json : u_strdup(list);
     if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed");
-    u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid;
+    u_free(watch->snapshot); watch->snapshot = snapshot; watch->valid = valid;
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
-               (unsigned long long)service->sources_group, valid, length);
+               (unsigned long long)watch->group, valid, length);
     chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9));
     u_free(event);
 }
 
 /* Наблюдение принадлежит сервису и снимается до освобождения ядра. */
-int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) {
+int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t owner) {
     struct stream_service *service = inst ? inst->stream : NULL;
-    if (!service || service->stopping) {
+    if (!service || (service->stopping && group_id)) {
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id);
         return STREAM_ERR_STATE;
     }
-    service->sources_group = group_id;
-    u_free(service->sources_snapshot); service->sources_snapshot = NULL;
-    if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
-    service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL;
-    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id);
-    if (group_id && !service->sources_pending) {
+    struct source_watch **link = &service->watches;
+    while (*link && (*link)->owner != owner) link = &(*link)->next;
+    struct source_watch *watch = *link;
+    if (!group_id) {
+        if (watch) {
+            if (watch->pending) uasync_call_soon_cancel(inst->ua, watch->pending);
+            *link = watch->next; u_free(watch->snapshot); u_free(watch);
+        }
+        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch removed owner=%llu", (unsigned long long)owner);
+        return STREAM_OK;
+    }
+    if (!watch) {
+        watch = u_calloc(1, sizeof(*watch));
+        if (!watch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch allocation failed"); return STREAM_ERR_MEMORY; }
+        watch->service = service; watch->owner = owner; *link = watch;
+    }
+    watch->group = group_id;
+    u_free(watch->snapshot); watch->snapshot = NULL;
+    if (!watch->pending) watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver);
+    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx owner=%llu",
+                (unsigned long long)group_id, (unsigned long long)owner);
+    if (!watch->pending) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY;
     }
     return STREAM_OK;
@@ -492,10 +515,12 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann
     (void)tags;
     struct stream_service *service = arg;
     if (service->stopping) return;
-    struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL;
-    if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) {
-        service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver);
-        if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
+    for (struct source_watch *watch = service->watches; watch; watch = watch->next) {
+        struct TOPO_GROUP *watched = stream_group(service->inst, watch->group);
+        if (watched && channel && !strcmp(watched->channel_id, channel) && !watch->pending) {
+            watch->pending = uasync_call_soon(service->inst->ua, watch, sources_deliver);
+            if (!watch->pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
+        }
     }
     if (!service->sessions) return;
     stream_network_membership_changed(service->network, node_id, channel);
@@ -579,8 +604,7 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     stream_network_destroy(service->network); service->network = NULL;
     if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
     if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
-    if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
-    u_free(service->sources_snapshot);
+    while (service->watches) stream_watch_sources(inst, 0, service->watches->owner);
     while (service->sessions) {
         struct stream_session *s = service->sessions;
         if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0);
@@ -597,9 +621,11 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
 void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) {
     struct stream_service *service = inst ? inst->stream : NULL;
     if (!service) return;
-    if (service->sources_group == group_id && !service->sources_pending) {
-        service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver);
-        if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed");
+    for (struct source_watch *watch = service->watches; watch; watch = watch->next) {
+        if (watch->group == group_id && !watch->pending) {
+            watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver);
+            if (!watch->pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed");
+        }
     }
     struct stream_session *s = service->sessions;
     while (s) {

src/stream/stream.h

@@ -64,9 +64,11 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb
 /* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self.
  * Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */
 int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out);
-/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES.
- * group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */
-int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id);
+/* Независимое наблюдение owner: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES.
+ * group_id=0 снимает только этого owner; owner=0 разрешён для выбранного канала frontend.
+ * Недоступный каталог имеет valid=0 и не означает завершение источников.
+ * Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */
+int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t owner);
 /* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */
 int stream_service_needs_activity(const struct UTUN_INSTANCE *inst);
 /* Только локальный viewer этой группы/эпохи подавляет рацию источника со включённым звуком. */