Enable automatic upstream selection for subscriptions with lifecycle-safe cleanup

src/Makefile.am

@@ -7,6 +7,8 @@ stream_core_files = \
 	stream/stream.h \
 	stream/stream.c \
 	stream/stream_priv.h \
+	stream/stream_optimizer.h \
+	stream/stream_optimizer.c \
 	stream/stream_wire.h \
 	stream/stream_wire.c \
 	stream/stream_network.h \

src/stream/stream.c

@@ -10,6 +10,7 @@
 #include "stream_member.h"
 #include "stream_source.h"
 #include "stream_priv.h"
+#include "stream_optimizer.h"
 
 struct stream_session {
     struct stream_session *next;
@@ -20,13 +21,16 @@ struct stream_session {
     struct stream_buffer *buffer;
     uint8_t *output;
     struct stream_metrics *metrics;
+    struct stream_optimizer *optimizer;
     void *delivery;
     uint64_t next_seq;
     uint64_t start_seq, last_seq, passed_before;
+    uint64_t restart_from, restart_last;
     uint8_t end[STREAM_END_SIZE];
     int end_valid;
     unsigned owners, holds;
     int anchored, closing, invalidated, finished, end_delivered, reason;
+    int restarting, restart_gap;
 };
 
 struct STREAM_HANDLE {
@@ -119,6 +123,7 @@ 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);
+    stream_optimizer_destroy(s->optimizer);
     if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
     if (s->source) stream_source_destroy(s->source);
     else stream_buffer_destroy(s->buffer);
@@ -142,6 +147,7 @@ static void session_schedule(struct stream_session *s) {
 static void session_close(struct stream_session *s, int reason, int update_profile, int notify) {
     if (s->closing) return;
     s->closing = 1; s->reason = reason; s->owners = 0;
+    stream_optimizer_destroy(s->optimizer); s->optimizer = NULL;
     stream_network_cancel(s->service->network, &s->descriptor);
     for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next)
         if (h->session == s) { h->session = NULL; h->reason = reason; }
@@ -166,6 +172,9 @@ static void session_deliver(void *arg) {
     else if (s->invalidated) {
         DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx",
                     (unsigned long long)s->descriptor.source_node_id);
+    } else if (s->restart_gap) {
+        event.type = STREAM_EVENT_GAP; event.first_seq = s->restart_from; event.last_seq = s->restart_last;
+        s->restart_gap = 0; emit = again = 1;
     } else if (s->anchored) {
         struct stream_buffer_stats stats;
         stream_buffer_get_stats(s->buffer, &stats);
@@ -321,6 +330,9 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     service->stopping = 1;
     for (struct stream_session *s = service->sessions; s; s = s->next)
         if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
+    for (struct stream_session *s = service->sessions; s; s = s->next) {
+        stream_optimizer_destroy(s->optimizer); s->optimizer = NULL;
+    }
     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);
@@ -354,12 +366,12 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb
 }
 
 /* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */
-static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher,
+static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic,
                          struct STREAM_HANDLE **out) {
     struct stream_service *service = inst->stream;
     struct stream_session *s = session_find(service, d->group_id, d->source_node_id);
     int created = s == NULL;
-    if (s && (publisher || s->descriptor.source_epoch != d->source_epoch)) {
+    if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!s->source && !!s->optimizer != automatic))) {
         DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source already owned with another publication/epoch"); return STREAM_ERR_STATE;
     }
     struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h));
@@ -375,6 +387,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
             result = stream_source_create(&s->source, inst->ua, d, inst->my_ed25519_privkey, source_ready, s);
             if (result == STREAM_OK) s->buffer = stream_source_buffer(s->source);
         } else if (result == STREAM_OK) result = stream_buffer_create(&s->buffer, d, key, STREAM_BUFFER_BYTES);
+        if (result == STREAM_OK && automatic && !publisher) result = stream_optimizer_create(inst, d, &s->optimizer);
         s->next = service->sessions; service->sessions = s;
         if (result != STREAM_OK) {
             DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session initialization failed result=%d", result);
@@ -434,14 +447,14 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str
         }
     } while (!d.source_epoch);
     if (stream_descriptor_validate(&d) != STREAM_OK) return STREAM_ERR_ARGUMENT;
-    return session_open(inst, &d, inst->my_ed25519_pubkey, 1, out);
+    return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, out);
 }
 
-int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
+static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, int automatic, struct STREAM_HANDLE **out) {
     int result = stream_open_check(inst, group_id, out);
     if (result != STREAM_OK) return result;
     struct stream_session *s = session_find(inst->stream, group_id, source_id);
-    if (s) return session_open(inst, &s->descriptor, s->public_key, 0, out);
+    if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, out);
     struct stream_member_state state; uint8_t key[32];
     result = stream_member_load(inst, group_id, source_id, &state, key);
     if (result != STREAM_OK) return result;
@@ -449,7 +462,29 @@ int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_
         DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id);
         return STREAM_WAIT;
     }
-    return session_open(inst, &state.source, key, 0, out);
+    return session_open(inst, &state.source, key, 0, automatic, out);
+}
+
+int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
+    return subscribe_open(inst, group_id, source_id, 1, out);
+}
+
+int stream_subscribe_manual_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
+    return subscribe_open(inst, group_id, source_id, 0, out);
+}
+
+int stream_selection_configure(struct STREAM_HANDLE *h, const struct stream_selection_settings *settings) {
+    if (!stream_handle_active(h) || !h->session->optimizer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection configuration requires automatic receiver"); return STREAM_ERR_STATE;
+    }
+    return stream_optimizer_configure(h->session->optimizer, settings);
+}
+
+int stream_selection_get_info(const struct STREAM_HANDLE *h, struct stream_selection_info *info) {
+    if (!stream_handle_active(h) || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection info invalid handle/output"); return STREAM_ERR_STATE; }
+    memset(info, 0, sizeof(*info));
+    if (h->session->optimizer) stream_optimizer_info(h->session->optimizer, info);
+    return STREAM_OK;
 }
 
 /* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */
@@ -554,7 +589,11 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
     struct stream_path retained = *path; retained.nodes[retained.count++] = inst->node_id;
     int result = stream_buffer_put_path(s->buffer, wire, length, &retained);
     if (result == STREAM_OK || (result == STREAM_DUPLICATE && !s->anchored)) {
-        if (!s->anchored) { s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); }
+        if (!s->anchored) {
+            s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
+            if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; }
+            s->restarting = 0;
+        }
         session_schedule(s);
         stream_network_changed(s->service->network);
     }
@@ -568,7 +607,7 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
     if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT;
     struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
     if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT;
-    *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq,
+    *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq,
         .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, .end = s->end_valid ? s->end : NULL};
     return STREAM_OK;
 }
@@ -624,12 +663,24 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip
     }
     struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
     if (s->delivery) { uasync_call_soon_cancel(inst->ua, s->delivery); s->delivery = NULL; }
+    if (s->anchored) { s->restarting = 1; s->restart_from = s->next_seq; }
     s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u",
                (unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
                (unsigned long long)key->source_epoch, local.buffer != NULL); return STREAM_OK;
 }
 
+/* Заимствованный локальный handle нужен только на время tick; владельцем остаётся сама подписка. */
+void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) {
+    struct stream_service *service = inst ? inst->stream : NULL;
+    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};
+        stream_optimizer_tick(s->optimizer, &subscription, now_ms);
+    }
+}
+
 /* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */
 int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate **out, size_t *count) {
     if (!out || *out || !count || !stream_handle_active(h)) {

src/stream/stream_network.c

@@ -861,6 +861,7 @@ static void network_tick(void *arg) {
         p = next;
     }
     pressure_tick(net, now_ms);
+    stream_optimizers_tick(net->inst, now_ms);
     net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
     net->timer = uasync_set_timeout(net->inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network");
     if (!net->timer) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network timer scheduling failed");