From 6749d09025753024e178ce77c90b0f6c441cf777 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Sun, 11 Oct 2026 14:54:35 +0200
Subject: [PATCH] Enforce distinct manual and automatic upstream ownership

---
 src/stream/stream.c         | 7 ++++---
 src/stream/stream_network.h | 3 ++-
 tests/test_stream_service.c | 3 +++
 3 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/src/stream/stream.c b/src/stream/stream.c
index 1a1e0828..e7bddbbe 100644
--- a/src/stream/stream.c
+++ b/src/stream/stream.c
@@ -38,6 +38,7 @@ struct STREAM_HANDLE {
     struct stream_service *service;
     struct stream_session *session;
     int publisher;
+    int selection_owned;
     int reason;
 };
 
@@ -634,8 +635,8 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
 
 /* Выданный локальный handle проверяется до создания сетевых ресурсов. */
 int stream_handle_context(struct STREAM_HANDLE *h, struct UTUN_INSTANCE **inst, struct stream_descriptor *d) {
-    if (!inst || !d || !stream_handle_active(h) || h->session->source) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active reader handle"); return STREAM_ERR_STATE;
+    if (!inst || !d || !stream_handle_active(h) || h->session->source || (h->session->optimizer && !h->selection_owned)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active manual reader or internal selector handle"); return STREAM_ERR_STATE;
     }
     *inst = h->service->inst; *d = h->session->descriptor; return STREAM_OK;
 }
@@ -696,7 +697,7 @@ void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) {
     if (!service || service->stopping) return;
     for (struct stream_session *s = service->sessions; s; s = s->next) {
         if (!s->optimizer || s->closing || s->invalidated) continue;
-        struct STREAM_HANDLE subscription = {.service = service, .session = s};
+        struct STREAM_HANDLE subscription = {.service = service, .session = s, .selection_owned = 1};
         stream_optimizer_tick(s->optimizer, &subscription, now_ms);
     }
 }
diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h
index 54f9b3a2..3f16ef17 100644
--- a/src/stream/stream_network.h
+++ b/src/stream/stream_network.h
@@ -1,7 +1,8 @@
 /* Сетевые выдачи raw ETCP с собственным NCD-владением пира; reverse/терминальные уведомления через router.
  * Одна управляющая FIFO и waiter на пира для всех групп/трансляций. DATA читается из кольца в момент отправки.
  * STREAM_RECEIVE принадлежит вызывающему даже после отмены/stop; close снимает только данную выдачу.
- * API выполняется в uasync-потоке. Автоматический выбор/перенос поставщиков — отдельный планировщик. */
+ * API выполняется в uasync-потоке. Внешние ручные операции требуют stream_subscribe_manual_open;
+ * автоматический выбор/перенос поставщиков владеет своими handles через отдельный планировщик. */
 #ifndef STREAM_NETWORK_H
 #define STREAM_NETWORK_H
 
diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c
index ccb17b68..e882de00 100644
--- a/tests/test_stream_service.c
+++ b/tests/test_stream_service.c
@@ -637,6 +637,9 @@ static void test_automatic(void) {
     CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscribers[1]) == STREAM_OK);
     CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &second) == STREAM_OK);
     CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_ERR_STATE && !manual);
+    struct STREAM_RECEIVE *foreign = NULL;
+    CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, 0, UINT64_MAX, &foreign) == STREAM_ERR_ARGUMENT && !foreign);
+    CHECK(stream_receive_restart(subscribers[1]) == STREAM_ERR_STATE);
     deadline = get_time_tb() + 150000;
     int relay_announced = 0;
     do {