Tie active-mode holds to stream ownership and deliver signed END for empty live sources

src/stream/stream.c

@@ -157,6 +157,12 @@ static void session_close(struct stream_session *s, int reason, int update_profi
     if (update_profile && group && service_publish_options(s->service, group) != STREAM_OK)
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream closed with stale own member options group=%llx",
                    (unsigned long long)s->descriptor.group_id);
+    if (!s->service->stopping) {
+        s->holds++;
+        if (utun_set_client_activity(s->service->inst, s->service->inst->client_activity_requested) < 0)
+            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream close failed to restore requested core activity");
+        s->holds--;
+    }
     if (notify && s->service->callback && !s->service->stopping) session_schedule(s);
     else if (!s->holds) session_free(s);
 }
@@ -322,6 +328,7 @@ int stream_service_init(struct UTUN_INSTANCE *inst) {
 void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     struct stream_service *service = inst ? inst->stream : NULL;
     if (!service) return;
+    if (service->stopping) return;
     if (service->dispatching || service->initializing) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service destroy inside callback/init rejected"); return;
     }
@@ -342,6 +349,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
         else session_free(s);
     }
     for (struct STREAM_HANDLE *h = service->handles; h; h = h->next) { h->service = NULL; h->session = NULL; h->reason = STREAM_CLOSE_SERVICE; }
+    if (utun_set_client_activity(inst, inst->client_activity_requested) < 0)
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream stop failed to restore requested core activity");
     inst->stream = NULL;
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service stopped node=%llx", (unsigned long long)inst->node_id);
     u_free(service);
@@ -401,6 +410,9 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
     h->next = service->handles; service->handles = h; s->owners++;
     s->holds++;
     int result = created ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK;
+    if (result == STREAM_OK && created && utun_set_client_activity(inst, inst->client_activity_requested) < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open failed to activate core"); result = STREAM_ERR_STATE;
+    }
     if (s->closing && result == STREAM_OK) {
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream open cancelled during member callbacks"); result = STREAM_ERR_STATE;
     }
@@ -603,6 +615,13 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
 
 struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst) { return inst && inst->stream ? inst->stream->network : NULL; }
 
+int stream_service_needs_activity(const struct UTUN_INSTANCE *inst) {
+    if (!inst || !inst->stream || inst->stream->stopping) return 0;
+    for (struct stream_session *s = inst->stream->sessions; s; s = s->next)
+        if (!s->closing && !s->invalidated && s->owners) return 1;
+    return 0;
+}
+
 /* Доступ network только к существующей локальной подписке точной эпохи. */
 int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local) {
     if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT;

src/stream/stream_network.c

@@ -511,7 +511,8 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
             flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0;
         }
     }
-    if (local.finished && local.last != UINT64_MAX && f->info.next > local.last) {
+    if (local.finished && ((local.last != UINT64_MAX && f->info.next > local.last)
+        || (local.last == UINT64_MAX && !f->info.blocks && !f->info.next))) {
         m.command = STREAM_CMD_END;
         if (raw_send(p, m, local.end, STREAM_END_SIZE) == STREAM_OK) {
             f->info.state = STREAM_RECEIVE_ENDED; stream_metrics_end(&f->offered_metrics);
@@ -787,7 +788,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
         else { f->info.next = stream_read_be(m.payload + 8, 8); f->info.gaps++; }
     } else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
         uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX;
-        if (m.length == STREAM_END_SIZE && last != UINT64_MAX && f->info.next == last + 1
+        if (m.length == STREAM_END_SIZE && ((last != UINT64_MAX && f->info.next == last + 1)
+            || (last == UINT64_MAX && !f->info.blocks && !f->info.next))
             && stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) {
             f->info.state = STREAM_RECEIVE_ENDED; stream_metrics_end(&f->metrics);
         }