Connect local stream lifetime and signed END to network delivery

src/stream/stream.c

@@ -21,6 +21,9 @@ struct stream_session {
     uint8_t *output;
     void *delivery;
     uint64_t next_seq;
+    uint64_t start_seq, last_seq;
+    uint8_t end[STREAM_END_SIZE];
+    int end_valid;
     unsigned owners, holds;
     int anchored, closing, invalidated, finished, end_delivered, reason;
 };
@@ -37,6 +40,7 @@ struct stream_service {
     struct UTUN_INSTANCE *inst;
     struct stream_session *sessions;
     struct STREAM_HANDLE *handles;
+    struct stream_network *network;
     stream_event_fn callback;
     void *arg;
     void *reconcile;
@@ -122,6 +126,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_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; }
     if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; }
@@ -159,11 +164,11 @@ static void session_deliver(void *arg) {
             s->next_seq = stats.expired_before; emit = 1; again = 1;
         } else if (result < 0) {
             session_close(s, STREAM_CLOSE_FAILED, 1, 1);
-        } else if (s->finished && s->next_seq > stats.highest_seq && !s->end_delivered) {
+        } else if (s->finished && s->end_valid && s->next_seq > s->last_seq && !s->end_delivered) {
-            event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = stats.highest_seq;
+            event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = s->last_seq;
             s->end_delivered = 1; emit = 1;
         }
-    } else if (s->finished && !s->end_delivered) {
+    } else if (s->finished && s->end_valid && s->last_seq == UINT64_MAX && !s->end_delivered) {
         event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = UINT64_MAX;
         s->end_delivered = 1; emit = 1;
     }
@@ -182,8 +187,9 @@ static void source_ready(struct stream_source *source, uint64_t seq, void *arg)
     (void)source;
     struct stream_session *s = arg;
     if (s->closing) return;
-    if (!s->anchored) { s->anchored = 1; s->next_seq = seq; }
+    if (!s->anchored) { s->anchored = 1; s->start_seq = s->next_seq = seq; }
     session_schedule(s);
+    stream_network_changed(s->service->network);
 }
 
 /* Отложенная сверка не разрушает экземпляры из BGP/member callbacks. */
@@ -273,6 +279,7 @@ int stream_service_init(struct UTUN_INSTANCE *inst) {
         service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_MEMORY;
     }
     service->registered = 1; service->initializing = 0;
+    if (stream_network_init(inst, &service->network) != STREAM_OK) { stream_service_destroy(inst); return STREAM_ERR_STATE; }
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service initialized node=%llx", (unsigned long long)inst->node_id);
     return STREAM_OK;
 }
@@ -287,6 +294,7 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     for (struct stream_session *s = service->sessions; s; s = s->next)
         if (s->holds) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream stop during active operation rejected"); return; }
     service->stopping = 1;
+    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);
     while (service->sessions) {
@@ -474,8 +482,14 @@ int stream_publish_finish(struct STREAM_HANDLE *h) {
         return result != STREAM_OK ? result : STREAM_ERR_STATE;
     }
     result = stream_source_finish(s->source);
+    if (result == STREAM_OK) {
+        struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
+        s->last_seq = stats.blocks ? stats.highest_seq : UINT64_MAX;
+        result = stream_end_sign(&s->descriptor, s->last_seq, s->service->inst->my_ed25519_privkey, s->end);
+        s->end_valid = result == STREAM_OK;
+    }
     if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
-    else session_schedule(s);
+    else { session_schedule(s); stream_network_changed(s->service->network); }
     s->holds--;
     if (s->closing && !s->holds && !s->delivery) session_free(s);
     return result;
@@ -483,6 +497,13 @@ int stream_publish_finish(struct STREAM_HANDLE *h) {
 
 /* Сначала только ключ/наличие подписки, затем crypto/FEC. Парсер пути и peer authorization — выше этого входа. */
 int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length) {
+    struct stream_path path = {.count = 1};
+    if (wire && length >= STREAM_BLOCK_RS_OFF) path.nodes[0] = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8);
+    return stream_ingest_path(inst, wire, length, &path);
+}
+
+/* Кольцо хранит уже дополненный путь: источник ... отправитель ... этот узел. */
+int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path) {
     if (!inst || !wire || length < STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ingest invalid length=%zu", length); return STREAM_ERR_ARGUMENT;
     }
@@ -492,14 +513,58 @@ int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t
     if (!s) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "unsubscribed stream ignored group=%llx source=%llx",
                          (unsigned long long)group, (unsigned long long)source); return STREAM_WAIT; }
     if (s->source) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "incoming DATA to own publisher rejected"); return STREAM_ERR_STATE; }
-    int result = stream_buffer_put(s->buffer, wire, length);
+    if (!path || path->count >= STREAM_PATH_MAX || stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK)
+        return STREAM_ERR_FORMAT;
+    if (s->end_valid && (s->last_seq == UINT64_MAX || stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8) > s->last_seq)) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA beyond signed END source=%llx", (unsigned long long)source);
+        return STREAM_ERR_CONFLICT;
+    }
+    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) {
-        if (!s->anchored) { s->anchored = 1; 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); }
         session_schedule(s);
+        stream_network_changed(s->service->network);
     }
     return result;
 }
 
+struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst) { return inst && inst->stream ? inst->stream->network : NULL; }
+
+/* Доступ 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;
+    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,
+        .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, .end = s->end_valid ? s->end : NULL};
+    return STREAM_OK;
+}
+
+/* Выданный локальный 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->publisher) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active reader handle"); return STREAM_ERR_STATE;
+    }
+    *inst = h->service->inst; *d = h->session->descriptor; return STREAM_OK;
+}
+
+/* END ретранслируется без изменений и не может завершить поток по заявлению поставщика без подписи источника. */
+int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length) {
+    struct stream_local local;
+    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return STREAM_WAIT;
+    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
+    uint64_t last; int result = stream_end_verify(&s->descriptor, s->public_key, wire, length, &last);
+    if (result != STREAM_OK) return result;
+    struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
+    if ((s->end_valid && s->last_seq != last) || (stats.blocks && (last == UINT64_MAX || stats.highest_seq > last))) {
+        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "signed END conflicts with retained stream source=%llx", (unsigned long long)key->source_node_id);
+        return STREAM_ERR_CONFLICT;
+    }
+    memcpy(s->end, wire, STREAM_END_SIZE); s->end_valid = s->finished = 1; s->last_seq = last;
+    session_schedule(s); stream_network_changed(inst->stream->network); return STREAM_OK;
+}
+
 /* Выборка идёт из 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.h

@@ -0,0 +1,30 @@
+/* Сетевые выдачи raw ETCP с собственным NCD-владением пира; reverse/терминальные уведомления через router.
+ * Одна управляющая FIFO и waiter на пира для всех групп/трансляций. DATA читается из кольца в момент отправки.
+ * STREAM_RECEIVE принадлежит вызывающему даже после отмены/stop; close снимает только данную выдачу.
+ * API выполняется в uasync-потоке. Автоматический выбор/перенос поставщиков — отдельный планировщик. */
+#ifndef STREAM_NETWORK_H
+#define STREAM_NETWORK_H
+
+#include "stream.h"
+#include "stream_wire.h"
+
+struct stream_network;
+struct STREAM_RECEIVE;
+enum stream_receive_state { STREAM_RECEIVE_CONNECTING, STREAM_RECEIVE_INFO, STREAM_RECEIVE_REQUESTED,
+                            STREAM_RECEIVE_ACTIVE, STREAM_RECEIVE_ENDED, STREAM_RECEIVE_CLOSED };
+struct stream_receive_info {
+    uint64_t handle, peer, start, next, blocks, gaps;
+    enum stream_receive_state state;
+    int reason;
+};
+/* out заранее NULL; seq=UINT64_MAX: первая выдача с конца, остальные ждут её проверенный общий старт.
+ * peer должен быть онлайн-кандидатом группы. Повтор peer+stream+sub не создаёт вторую выдачу. */
+int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t peer, uint8_t subblock, uint64_t seq, struct STREAM_RECEIVE **out);
+void stream_receive_close(struct STREAM_RECEIVE *receive);
+int stream_receive_get_info(const struct STREAM_RECEIVE *receive, struct stream_receive_info *info);
+
+int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out);
+void stream_network_destroy(struct stream_network *network);
+void stream_network_changed(struct stream_network *network);
+void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor);
+#endif

src/stream/stream_priv.h

@@ -5,6 +5,21 @@
 #define STREAM_PRIV_H
 
 #include "stream.h"
+#include "stream_network.h"
+
+struct stream_local {
+    struct stream_descriptor descriptor;
+    struct stream_buffer *buffer;
+    uint64_t start, last;
+    int anchored, publisher, finished;
+    const uint8_t *end;
+};
+/* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */
+struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst);
+int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local);
+int stream_handle_context(struct STREAM_HANDLE *handle, struct UTUN_INSTANCE **inst, struct stream_descriptor *descriptor);
+int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path);
+int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length);
 
 int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length);