utun2/src/stream/stream.c

#include <stdio.h>
#include <string.h>
#include <limits.h>

#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include "../../lib/u_async.h"
#include "../../lib/strbuf.h"
#include "../../lib/sqlite3.h"
#include "../utun_instance.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "stream_member.h"
#include "stream_source.h"
#include "stream_priv.h"
#include "stream_optimizer.h"

struct stream_session {
    struct stream_session *next;
    struct stream_service *service;
    struct stream_descriptor descriptor;
    uint8_t public_key[SC_PUBKEY_SIZE];
    struct stream_source *source;
    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, relay_owners, holds;
    int anchored, decoded, closing, invalidated, finished, end_delivered, reason;
    int restarting, restart_gap, notify_close;
};

struct STREAM_HANDLE {
    struct STREAM_HANDLE *next;
    struct stream_service *service;
    struct stream_session *session;
    int publisher;
    int relay;
    int selection_owned;
    int reason;
};

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;
    void *sources_pending;
    uint64_t sources_group;
    char *sources_snapshot;
    int sources_valid;
    int stopping, dispatching, registered, initializing;
};

static void session_deliver(void *arg);
static void service_reconcile(void *arg);

/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */
int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) {
    if (!inst || !out || *out || !group_id) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT;
    }
    struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
    if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT;
    for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id);
        return STREAM_ERR_STATE;
    }
    sqlite3 *db = chat_core_get_db(inst);
    if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; }
    char query[320];
    snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
             "FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id);
    struct strbuf json = strbuf_new();
    sqlite3_stmt *statement = NULL;
    int result = STREAM_OK, count = 0;
    if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY;
    if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE;
    int step = SQLITE_DONE;
    while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) {
        uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
        if (!stream_member_online(inst, group_id, node)) continue;
        struct stream_member_state state;
        int loaded = stream_member_load(inst, group_id, node, &state, NULL);
        if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue;
        if (loaded < 0) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d",
                       (unsigned long long)group_id, (unsigned long long)node, loaded); continue;
        }
        if (!state.published) continue;
        if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
            (unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0
            || strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
            || strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY;
        count++;
    }
    if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE;
    sqlite3_finalize(statement);
    if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY;
    if (result == STREAM_OK) *out = strbuf_detach(&json);
    strbuf_free(&json);
    if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s",
                                       (unsigned long long)group_id, result, sqlite3_errmsg(db));
    else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count);
    return result;
}

/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */
static void sources_deliver(void *arg) {
    struct stream_service *service = arg;
    service->sources_pending = NULL;
    if (service->stopping || !service->sources_group) return;
    char *json = NULL;
    int result = stream_sources_json(service->inst, service->sources_group, &json);
    int valid = result == STREAM_OK || result == STREAM_WAIT;
    const char *list = json ? json : "[]";
    if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) {
        u_free(json); return;
    }
    size_t length = strlen(list);
    uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL;
    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;
    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;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
               (unsigned long long)service->sources_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) {
    struct stream_service *service = inst ? inst->stream : NULL;
    if (!service || service->stopping) {
        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) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY;
    }
    return STREAM_OK;
}

/* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */
static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) {
    struct stream_session *s = arg;
    if (s->closing || s->invalidated) return;
    uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
    if (added) {
        if (!s->metrics[sub].ended) stream_metrics_observe(&s->metrics[sub], seq,
            (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
        if (s->end_valid && seq == s->last_seq) stream_metrics_end(&s->metrics[sub]);
    }
    stream_network_cache_event(s->service->network, &s->descriptor, wire, length, added);
}

/* Активность CHAT проверяется до чтения профиля, выделения 32 МБ и создания handles. */
static struct TOPO_GROUP *stream_group(struct UTUN_INSTANCE *inst, uint64_t id) {
    struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, id) : NULL;
    return group && group->group_type == TOPO_GROUP_TYPE_CHAT && !group->stopping ? group : NULL;
}

/* Один локальный экземпляр на источник группы; закрывающийся старый запуск не мешает новому. */
static struct stream_session *session_find(struct stream_service *service, uint64_t group_id, uint64_t source_id) {
    for (struct stream_session *s = service->sessions; s; s = s->next)
        if (!s->closing && !s->invalidated && s->descriptor.group_id == group_id && s->descriptor.source_node_id == source_id) return s;
    return NULL;
}

/* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные handles. */
static int service_publish_options(struct stream_service *service, struct TOPO_GROUP *group) {
    if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream options group missing"); return STREAM_ERR_STATE; }
    struct stream_member_state state = {0};
    for (struct stream_session *s = service->sessions; s; s = s->next) {
        if (s->closing || s->invalidated || s->descriptor.group_id != group->group_id) continue;
        if (state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local subscription list exceeds member limit group=%llx",
                        (unsigned long long)group->group_id); return STREAM_ERR_ARGUMENT;
        }
        struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch,
                                                !s->source && s->owners == s->relay_owners};
        size_t at = state.count++;
        while (at && (state.subscriptions[at - 1].source_node_id > entry.source_node_id
               || (state.subscriptions[at - 1].source_node_id == entry.source_node_id
                   && state.subscriptions[at - 1].source_epoch > entry.source_epoch))) {
            state.subscriptions[at] = state.subscriptions[at - 1]; at--;
        }
        state.subscriptions[at] = entry;
        if (s->source && !s->finished) { state.source = s->descriptor; state.published = 1; }
    }
    char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1];
    int result = stream_member_format(&state, source, subscriptions);
    if (result != STREAM_OK) return result;
    struct chat_profile_option options[] = {
        {STREAM_MEMBER_SOURCE, source[0] ? source : NULL},
        {STREAM_MEMBER_SUBSCRIPTIONS, subscriptions[0] ? subscriptions : NULL}
    };
    if (chat_core_set_my_profile_options(service->inst, group->channel_id, options, 2) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local member stream options update failed group=%llx owners=%zu",
                    (unsigned long long)group->group_id, state.count); return STREAM_ERR_STATE;
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream options group=%llx published=%d subscriptions=%zu",
               (unsigned long long)group->group_id, state.published, state.count);
    return STREAM_OK;
}

/* Ресурсы живут до возврата пользовательского callback/write, даже если он закрыл последний handle. */
static void session_free(struct stream_session *s) {
    struct stream_session **link = &s->service->sessions;
    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);
    u_free(s->output);
    u_free(s->metrics);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream released group=%llx source=%llx epoch=%llx reason=%d",
               (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
               (unsigned long long)s->descriptor.source_epoch, s->reason);
    u_free(s);
}

/* Delivery выполняется отдельным callback, а не из source_ready при busy источнике. */
static void session_schedule(struct stream_session *s) {
    if (s->service->stopping || s->delivery) return;
    s->delivery = uasync_call_soon(s->service->inst->ua, s, session_deliver);
    if (!s->delivery) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream delivery scheduling failed source=%llx",
                                (unsigned long long)s->descriptor.source_node_id);
}

/* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */
static void session_close(struct stream_session *s, int reason, int update_profile, int notify) {
    if (s->closing) return;
    s->notify_close = notify && s->owners > s->relay_owners;
    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; }
    if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; }
    struct TOPO_GROUP *group = s->service->inst->topo_groups
                               ? topo_groups_find(s->service->inst->topo_groups, s->descriptor.group_id) : NULL;
    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);
}

/* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */
static void session_deliver(void *arg) {
    struct stream_session *s = arg;
    struct stream_service *service = s->service;
    s->delivery = NULL; s->holds++;
    struct stream_event event = { .descriptor = s->descriptor };
    int emit = 0, again = 0;
    if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = s->notify_close; }
    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);
        size_t length = 0;
        int result = s->next_seq == UINT64_MAX ? STREAM_WAIT
                     : stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length);
        if (result == STREAM_OK) {
            if (!s->decoded) {
                s->decoded = 1;
                DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u signature=verified",
                    (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
                    (unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq,
                    (unsigned)(s->descriptor.n - s->descriptor.r));
                stream_network_wake(service->network);
            }
            event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++;
            event.data = s->output; event.length = length; emit = 1; again = 1;
        } else if (result == STREAM_REJECTED) {
            event.type = STREAM_EVENT_GAP; event.first_seq = event.last_seq = s->next_seq++;
            emit = again = 1;
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues",
                       (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch,
                       (unsigned long long)event.first_seq);
        } else if (result == STREAM_EXPIRED) {
            event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = stats.expired_before - 1;
            s->next_seq = stats.expired_before; emit = 1; again = 1;
        } else if ((result == STREAM_WAIT || result == STREAM_GAP) && s->next_seq < s->passed_before) {
            uint64_t next = s->passed_before, decoded;
            if (stream_buffer_next_decodable(s->buffer, s->next_seq + 1, &decoded) == STREAM_OK && decoded < next) next = decoded;
            event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = next - 1;
            s->next_seq = next; emit = 1; again = 1;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream consumer GAP source=%llx first=%llu last=%llu next=%llu",
                       (unsigned long long)s->descriptor.source_node_id, (unsigned long long)event.first_seq,
                       (unsigned long long)event.last_seq, (unsigned long long)next);
        } else if (result < 0) {
            session_close(s, STREAM_CLOSE_FAILED, 1, 1);
        } 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 = s->last_seq;
            s->end_delivered = 1; emit = 1;
        }
    } 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;
    }
    if (!s->closing && s->owners == s->relay_owners) emit = 0;
    if (emit && service->callback && !service->stopping) {
        service->dispatching++;
        service->callback(service->inst, &event, service->arg);
        service->dispatching--;
    }
    s->holds--;
    if (s->closing) { if (!s->holds && !s->delivery) session_free(s); }
    else if (again) session_schedule(s);
}

/* Готовая source-пачка уже полностью в кольце. Пользовательский callback здесь не вызывается. */
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->start_seq = s->next_seq = seq; }
    session_schedule(s);
    stream_network_changed(s->service->network);
}

/* Отложенная сверка не разрушает экземпляры из BGP/member callbacks. */
static void service_reconcile(void *arg) {
    struct stream_service *service = arg;
    service->reconcile = NULL;
    struct stream_session *snapshot = service->sessions;
    for (struct stream_session *s = snapshot; s; s = s->next) s->holds++;
    struct stream_session *s = snapshot;
    while (s) {
        struct stream_session *next = s->next;
        if (!s->closing && s->invalidated) session_close(s, s->reason, 1, 1);
        else if (!s->closing && !s->source) {
            struct stream_member_state state; uint8_t key[32];
            int result = stream_member_load(service->inst, s->descriptor.group_id, s->descriptor.source_node_id, &state, key);
            if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT)
                session_close(s, STREAM_CLOSE_MEMBER, 1, 1);
            else if (result == STREAM_OK && (memcmp(key, s->public_key, 32)
                     || (state.published && (state.source.source_epoch != s->descriptor.source_epoch
                         || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r
                         || state.source.block_size != s->descriptor.block_size))))
                session_close(s, STREAM_CLOSE_SOURCE_CHANGED, 1, 1);
        }
        s = next;
    }
    s = snapshot;
    while (s) {
        struct stream_session *next = s->next;
        s->holds--;
        if (s->closing && !s->holds && !s->delivery) session_free(s);
        s = next;
    }
    stream_network_wake(service->network);
}

/* Общий props сигнал относится и к Merkle commit, и к BGP. Payload adm_tags здесь не используется. */
static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) {
    (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");
    }
    if (!service->sessions) return;
    stream_network_membership_changed(service->network, node_id, channel);
    int changed = 0;
    for (struct stream_session *s = service->sessions; s; s = s->next) {
        int own = node_id == service->inst->node_id;
        if (s->closing) continue;
        struct TOPO_GROUP *group = stream_group(service->inst, s->descriptor.group_id);
        if (!group || !channel || strcmp(group->channel_id, channel)) continue;
        changed = 1;
        if (s->optimizer) stream_optimizer_roster_changed(s->optimizer, node_id);
        if (!own && (s->source || s->descriptor.source_node_id != node_id)) continue;
        struct stream_member_state state; uint8_t key[32];
        int result = stream_member_load(service->inst, group->group_id, node_id, &state, key);
        if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER;
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream membership invalidated group=%llx member=%llx own=%d result=%d",
                       (unsigned long long)group->group_id, (unsigned long long)node_id, own, result);
        } else if (own && result == STREAM_OK && memcmp(key, service->inst->my_ed25519_pubkey, 32)) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER;
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream own member key changed group=%llx", (unsigned long long)group->group_id);
        } else if (!own && result == STREAM_OK && (memcmp(key, s->public_key, 32)
                   || (state.published && (state.source.source_epoch != s->descriptor.source_epoch
                       || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r
                       || state.source.block_size != s->descriptor.block_size)))) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
        }
    }
    if (!changed || service->reconcile) return;
    service->reconcile = uasync_call_soon(service->inst->ua, service, service_reconcile);
    if (!service->reconcile) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream member reconciliation scheduling failed");
}

/* Снять старые собственные опции после аварии; неполный/отсутствующий профиль не является активной подпиской. */
int stream_service_init(struct UTUN_INSTANCE *inst) {
    if (!inst || !inst->core_started || !chat_core_get_db(inst) || !inst->topo_groups) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream init requires core and chat DB"); return STREAM_ERR_STATE;
    }
    if (inst->stream) return STREAM_OK;
    struct stream_service *service = u_calloc(1, sizeof(*service));
    if (!service) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service allocation failed"); return STREAM_ERR_MEMORY; }
    service->inst = inst; service->initializing = 1; inst->stream = service;
    for (struct ll_entry *e = inst->topo_groups->group_list->head; e; e = e->next) {
        struct TOPO_GROUP *group = (struct TOPO_GROUP *)e;
        if (group->group_type != TOPO_GROUP_TYPE_CHAT) continue;
        struct stream_member_state state;
        int result = stream_member_load(inst, group->group_id, inst->node_id, &state, NULL);
        if (result == STREAM_OK && !state.published && !state.count) continue;
        if (result == STREAM_WAIT || result == STREAM_ERR_NOT_MEMBER) continue;
        if (service_publish_options(service, group) != STREAM_OK) {
            service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_STATE;
        }
    }
    if (member_sync_add_props_cbk(inst, member_changed, service) < 0) {
        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;
}

/* Detached handles принадлежат callers и остаются безопасными для close после остановки чата/ядра. */
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;
    }
    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;
    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);
    if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
    u_free(service->sources_snapshot);
    while (service->sessions) {
        struct stream_session *s = service->sessions;
        if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0);
        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);
}

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");
    }
    struct stream_session *s = service->sessions;
    while (s) {
        struct stream_session *next = s->next;
        if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 1);
        s = next;
    }
}

void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg) {
    if (!inst || !inst->stream || inst->stream->stopping) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream callback before init"); return; }
    inst->stream->callback = callback; inst->stream->arg = arg;
    if (callback) for (struct stream_session *s = inst->stream->sessions; s; s = s->next)
        if (!s->closing) session_schedule(s);
}

/* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */
static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic,
                         int relay,
                         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 || (!relay && !s->source && !!s->optimizer != automatic))) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream instance conflicts with publication/epoch/selection mode automatic=%d existing_auto=%d",
                   automatic, s->optimizer != NULL); return STREAM_ERR_STATE;
    }
    struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h));
    if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream handle allocation failed"); return STREAM_ERR_MEMORY; }
    if (created) {
        s = u_calloc(1, sizeof(*s));
        if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session allocation failed"); u_free(h); return STREAM_ERR_MEMORY; }
        s->service = service; s->descriptor = *d; memcpy(s->public_key, key, 32);
        s->output = u_malloc((uint32_t)stream_input_size(d));
        s->metrics = u_calloc(d->n, sizeof(*s->metrics));
        int result = s->output && s->metrics ? STREAM_OK : STREAM_ERR_MEMORY;
        if (result == STREAM_OK && publisher) {
            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);
            session_free(s); u_free(h); return result;
        }
        for (unsigned i = 0; i < d->n; i++) stream_metrics_init(&s->metrics[i]);
        stream_buffer_set_observer(s->buffer, cache_event, s);
    }
    int was_relay = s->owners && s->owners == s->relay_owners;
    h->service = service; h->session = s; h->publisher = publisher; h->relay = relay;
    h->next = service->handles; service->handles = h; s->owners++; s->relay_owners += relay;
    s->holds++;
    int result = created || was_relay != (s->owners == s->relay_owners)
                 ? 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;
    }
    if (result != STREAM_OK) {
        struct STREAM_HANDLE **link = &service->handles;
        while (*link != h) link = &(*link)->next;
        *link = h->next;
        if (!s->closing) {
            s->owners--; s->relay_owners -= relay;
            if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0);
        }
        u_free(h); s->holds--;
        if (s->closing && !s->holds && !s->delivery) session_free(s);
        return result;
    }
    s->holds--;
    *out = h;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream handle opened group=%llx source=%llx epoch=%llx publisher=%d owners=%u",
               (unsigned long long)d->group_id, (unsigned long long)d->source_node_id,
               (unsigned long long)d->source_epoch, publisher, s->owners);
    if (created && s->optimizer) stream_network_wake(service->network);
    return STREAM_OK;
}

/* Контекст не создаётся для неизвестной группы, неавторизованного self или неподтверждённого ключа. */
static int stream_open_check(struct UTUN_INSTANCE *inst, uint64_t group_id, struct STREAM_HANDLE **out) {
    if (!out || *out || !inst || !inst->chat_started || !inst->stream || inst->stream->stopping || !stream_group(inst, group_id)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open invalid state/group=%llx", (unsigned long long)group_id); return STREAM_ERR_STATE;
    }
    struct stream_member_state own; uint8_t key[32];
    int result = stream_member_load(inst, group_id, inst->node_id, &own, key);
    if (result != STREAM_OK) return result;
    if (memcmp(key, inst->my_ed25519_pubkey, 32)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream own member key differs from core identity"); return STREAM_ERR_SIGNATURE;
    }
    return STREAM_OK;
}

int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const struct stream_parameters *parameters,
                        struct STREAM_HANDLE **out) {
    int result = stream_open_check(inst, group_id, out);
    if (result != STREAM_OK) return result;
    struct stream_descriptor d = {.group_id = group_id, .source_node_id = inst->node_id,
        .n = parameters ? parameters->n : STREAM_DEFAULT_N, .r = parameters ? parameters->r : STREAM_DEFAULT_R,
        .block_size = parameters ? parameters->block_size : STREAM_RS_DEFAULT_SIZE};
    do {
        if (random_bytes((uint8_t *)&d.source_epoch, sizeof(d.source_epoch)) != 0) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source epoch generation failed"); return STREAM_ERR_CRYPTO;
        }
    } 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, 0, 0, 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, automatic, 0, 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;
    if (!state.published) {
        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, automatic, 0, out);
}

/* Запрос старой эпохи не может незаметно подписать суперузел на новый запуск. */
int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct STREAM_HANDLE **out) {
    int result = key ? stream_open_check(inst, key->group_id, out) : STREAM_ERR_ARGUMENT;
    if (result != STREAM_OK) return result;
    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 session_open(inst, &s->descriptor, s->public_key, 0, 1, 1, out);
    struct stream_member_state state; uint8_t public_key[32];
    result = stream_member_load(inst, key->group_id, key->source_node_id, &state, public_key);
    if (result != STREAM_OK) return result;
    if (!state.published || state.source.source_epoch != key->source_epoch) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay source epoch unavailable source=%llx requested=%llx published=%d actual=%llx",
                   (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, state.published,
                   (unsigned long long)state.source.source_epoch); return STREAM_ERR_STATE;
    }
    return session_open(inst, &state.source, public_key, 0, 1, 1, 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 снимает только собственное владение. */
void stream_close(struct STREAM_HANDLE *h) {
    if (!h) return;
    struct stream_service *service = h->service;
    if (service) {
        struct STREAM_HANDLE **link = &service->handles;
        while (*link != h) link = &(*link)->next;
        *link = h->next;
    }
    struct stream_session *s = h->session;
    if (s) {
        int was_relay = s->owners == s->relay_owners;
        if (h->publisher || s->owners == 1) session_close(s, STREAM_CLOSE_STOPPED, 1, !h->relay);
        else {
            s->owners--; s->relay_owners -= h->relay;
            if (was_relay != (s->owners == s->relay_owners)
                && service_publish_options(service, stream_group(service->inst, s->descriptor.group_id)) != STREAM_OK) {
                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream owner mode publication failed; closing inconsistent instance");
                session_close(s, STREAM_CLOSE_FAILED, 0, 1);
            }
        }
    }
    u_free(h);
}

int stream_handle_active(const struct STREAM_HANDLE *h) { return h && h->session && !h->session->closing && !h->session->invalidated; }
int stream_handle_reason(const struct STREAM_HANDLE *h) {
    return h && h->session && h->session->invalidated ? h->session->reason : h ? h->reason : STREAM_CLOSE_STOPPED;
}

int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) {
    if (!info || !stream_handle_active(h)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream info detached/invalid handle"); return STREAM_ERR_STATE; }
    struct stream_session *s = h->session;
    struct stream_info result = {.descriptor = s->descriptor, .owners = s->owners, .publisher = s->source != NULL, .finished = s->finished};
    stream_buffer_get_stats(s->buffer, &result.buffer);
    *info = result;
    return STREAM_OK;
}

size_t stream_publish_capacity(const struct STREAM_HANDLE *h) {
    if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream capacity not a publisher"); return 0; }
    return stream_source_input_capacity(h->session->source);
}

int stream_publish_write(struct STREAM_HANDLE *h, const uint8_t *data, size_t length, size_t *accepted) {
    if (accepted) *accepted = 0;
    if (!stream_handle_active(h) || !h->publisher || !accepted) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream write not an active publisher"); return STREAM_ERR_STATE;
    }
    struct stream_session *s = h->session; s->holds++;
    int result = stream_source_write(s->source, data, length, accepted);
    if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
    s->holds--;
    if (s->closing && !s->holds && !s->delivery) session_free(s);
    return result;
}

/* Блочная публикация использует те же holds и обработку отказов, что и источник непрерывных байтов. */
int stream_publish_block(struct STREAM_HANDLE *h, const uint8_t *data, size_t length) {
    if (!stream_handle_active(h) || !h->publisher) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream block write not an active publisher"); return STREAM_ERR_STATE;
    }
    struct stream_session *s = h->session; s->holds++;
    int result = stream_source_write_block(s->source, data, length);
    if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
    s->holds--;
    if (s->closing && !s->holds && !s->delivery) session_free(s);
    return result;
}

int stream_publish_finish(struct STREAM_HANDLE *h) {
    if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream finish not a publisher"); return STREAM_ERR_STATE; }
    struct stream_session *s = h->session;
    if (s->finished) return STREAM_OK;
    s->holds++;
    s->finished = 1;
    int result = service_publish_options(s->service, stream_group(s->service->inst, s->descriptor.group_id));
    if (result != STREAM_OK || s->closing) {
        if (!s->closing) s->finished = 0;
        s->holds--;
        if (s->closing && !s->holds && !s->delivery) session_free(s);
        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 (s->end_valid) for (unsigned i = 0; i < s->descriptor.n; i++) stream_metrics_end(&s->metrics[i]);
    }
    if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
    else { session_schedule(s); stream_network_changed(s->service->network); }
    s->holds--;
    if (s->closing && !s->holds && !s->delivery) session_free(s);
    return result;
}

/* Сначала только ключ/наличие подписки, затем 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;
    }
    if (!inst->stream || inst->stream->stopping) return STREAM_WAIT;
    uint64_t group = stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8), source = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8);
    struct stream_session *s = session_find(inst->stream, group, source);
    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; }
    if (!path || !path->count || 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 || (result == STREAM_DUPLICATE && !s->anchored)) {
        if (!s->anchored) {
            uint64_t received = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
            uint64_t start = s->restarting && received < s->restart_from ? s->restart_from : received;
            s->anchored = 1; s->start_seq = s->next_seq = start;
            if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; }
            if (s->restarting) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream restart anchored source=%llx received=%llu floor=%llu start=%llu",
                (unsigned long long)source, (unsigned long long)received, (unsigned long long)s->restart_from, (unsigned long long)start);
            s->restarting = 0;
        }
        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; }

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;
    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, .next = s->next_seq,
        .anchored = s->anchored, .decoded = s->decoded, .publisher = s->source != NULL, .finished = s->end_valid,
        .relay_only = !s->source && s->owners == s->relay_owners, .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->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;
}

/* 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;
    for (unsigned i = 0; i < s->descriptor.n; i++)
        if (last == UINT64_MAX || (s->metrics[i].initialized && s->metrics[i].last_seq == last)) stream_metrics_end(&s->metrics[i]);
    session_schedule(s); stream_network_changed(inst->stream->network); return STREAM_OK;
}

/* Общая метрика прихода субпотока в экземпляр, независимо от поставщика конкретного блока. */
int stream_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now,
                         struct stream_metric_snapshot *snapshot) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || sub >= local.descriptor.n || !snapshot) return STREAM_WAIT;
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    return stream_metrics_poll(&s->metrics[sub], now, snapshot);
}

/* Все выбранные upstream прошли эту границу; дальше старую неполную пачку они уже не выдадут. */
void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t before) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return;
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    if (before > s->passed_before) { s->passed_before = before; session_schedule(s); }
}

/* Явный новый старт не смешивается с DATA старых upstream handles; кольцо остаётся пригодным для relay. */
int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream restart requires local receiver"); return STREAM_ERR_STATE;
    }
    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) {
        if (!s->restart_gap) s->restart_from = s->next_seq;
        s->restarting = 1;
    }
    s->restart_gap = 0;
    s->anchored = s->decoded = 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 floor=%llu known=%d",
               (unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
               (unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); 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, .selection_owned = 1};
        stream_optimizer_tick(s->optimizer, &subscription, now_ms);
    }
}

/* Routed READY будит selector, не создавая пользовательского события или второго экземпляра. */
void stream_relay_ready(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer) {
    struct stream_session *s = inst && inst->stream ? session_find(inst->stream, key->group_id, key->source_node_id) : NULL;
    if (s && s->descriptor.source_epoch == key->source_epoch && s->optimizer) stream_optimizer_refresh(s->optimizer, peer);
}

/* Выборка идёт из 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)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates invalid handle/output"); return STREAM_ERR_ARGUMENT;
    }
    struct stream_session *s = h->session;
    struct UTUN_INSTANCE *inst = s->service->inst;
    struct TOPO_GROUP *group = stream_group(inst, s->descriptor.group_id);
    if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates group missing"); return STREAM_ERR_STATE; }
    size_t capacity = (size_t)group->nodes->count;
    struct stream_candidate *candidates = capacity ? u_calloc((uint32_t)capacity, sizeof(*candidates)) : NULL;
    if (capacity && !candidates) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates allocation failed count=%zu", capacity);
        return STREAM_ERR_MEMORY;
    }
    size_t found = 0;
    for (struct ll_entry *e = group->nodes->head; e; e = e->next) {
        struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
        if (node->node_id == inst->node_id || !stream_member_online(inst, group->group_id, node->node_id)) continue;
        struct TOPO_NODE *identity = topo_node_registry_find(inst->topo_groups, node->node_id);
        int source = node->node_id == s->descriptor.source_node_id;
        DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1",
                    (unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source);
        if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source
            && stream_member_supernode(inst, group->group_id, node->node_id) != 1) {
            DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay",
                        (unsigned long long)node->node_id); continue;
        }
        struct stream_member_state state;
        if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK) continue;
        int confirmed = stream_relay_confirmed(inst, &s->descriptor, node->node_id);
        if (!stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch) && !confirmed) continue;
        int relay_only = confirmed;
        for (size_t i = 0; i < state.count; i++)
            if (state.subscriptions[i].source_node_id == s->descriptor.source_node_id
                && state.subscriptions[i].source_epoch == s->descriptor.source_epoch) relay_only = state.subscriptions[i].relay_only;
        if (!source && s->owners == s->relay_owners && relay_only) continue;
        candidates[found++] = (struct stream_candidate){ .node_id = node->node_id, .source = (uint8_t)source,
            .client_type = identity ? identity->client_type : UINT8_MAX };
    }
    if (!found) { u_free(candidates); candidates = NULL; }
    *out = candidates; *count = found;
    DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu",
                (unsigned long long)s->descriptor.source_node_id, found, capacity);
    return STREAM_OK;
}