#include <string.h>

#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/node_conn_direct.h"
#include "../routing_layer/route_crypto.h"
#include "stream_member.h"
#include "stream_priv.h"

#define STREAM_PEER_LIMIT 64
#define STREAM_FLOW_LIMIT 256
#define STREAM_PEER_FLOW_LIMIT 64
#define STREAM_CONTROL_LIMIT 128
#define STREAM_CONTROL_MAX (STREAM_WIRE_HEADER + STREAM_OVERLOAD_HEADER + STREAM_OVERLOAD_MAX * 16)
#define STREAM_RENEW_TB 20000
#define STREAM_LEASE_TB 60000
#define STREAM_CONNECT_TB 50000
#define STREAM_TICK_TB 1000

struct stream_peer;
struct stream_cut { uint64_t handle, rate; int closed, selected; };
struct stream_pressure_request {
    struct stream_cut cuts[STREAM_OVERLOAD_MAX];
    uint64_t id, required_rate, released_rate, deadline_ms;
    unsigned count, required_count, released_count;
    int active, sent;
};
struct STREAM_RECEIVE {
    struct STREAM_RECEIVE *next;
    struct stream_peer *peer;
    struct stream_descriptor descriptor;
    struct stream_receive_info info;
    uint64_t requested, deadline, renewed;
    uint8_t subblock;
    int outgoing, owned, bootstrap, info_sent, info_ready;
    struct stream_metrics metrics, offered_metrics;
    struct stream_load load;
    uint64_t admitted_ms, pending_highest, probe, probe_sent_ms;
    uint64_t gap_from;
    int gap_pending, account_failed;
};
struct stream_peer {
    struct stream_peer *next;
    struct stream_network *network;
    struct NODE_CONN_DIRECT *ncd;
    struct STREAM_RECEIVE *flows, *last_sent;
    struct ll_queue *control, *wait_q;
    struct queue_waiter_handle waiter;
    void *pump;
    uint64_t node_id, own_epoch, other_epoch, next_handle, highwater, born, idle, reverse_at;
    int up, busy;
    struct stream_load load;
    struct stream_pressure_request pressure, requested;
    uint64_t pressure_id, pressure_retry_ms, clear_since_ms;
    int cooling;
};
struct stream_network {
    struct UTUN_INSTANCE *inst;
    struct stream_peer *peers;
    void *timer;
    unsigned peer_count, flow_count;
    int stopping;
    struct stream_load load;
    struct uasync_load cpu;
    uint64_t next_tick_ms, callback_late_ms;
    int global_pressure;
    uint64_t local_busy_until_ms, global_since_ms;
    int global_candidate;
};

static void peer_pump(void *arg);
static void network_tick(void *arg);
static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry);
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry);
static void pressure_tick(struct stream_network *net, uint64_t now);
static void pressure_receive(struct stream_peer *peer, const struct stream_message *message);

/* Ключи очереди — cursor+кольцо, её возраст учитывает реальное время вставки и момент появления выдачи. */
static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) {
    if (f->outgoing || !f->load.queued.packets) return 0;
    struct stream_local local; uint64_t oldest = 0;
    if (stream_local_get(f->peer->network->inst, &f->descriptor, &local) != STREAM_OK) return 0;
    if (stream_buffer_pending(local.buffer, f->info.next, f->subblock, NULL, &oldest) != STREAM_OK) return 0;
    if (oldest < f->admitted_ms) oldest = f->admitted_ms;
    return now > oldest ? now - oldest : 0;
}

/* Один блок принадлежит логической очереди выдачи/пира/модуля; ни одна из них не хранит копию DATA. */
static void load_account(struct STREAM_RECEIVE *f, uint64_t packets, uint64_t bytes, int added, int sent) {
    if (!packets) return;
    struct stream_load *loads[] = {&f->load, &f->peer->load, &f->peer->network->load};
    for (unsigned i = 0; i < 3; i++) {
        int result = added ? stream_load_add(loads[i], packets, bytes) : stream_load_remove(loads[i], packets, bytes, sent);
        if (result != STREAM_OK) {
            f->account_failed = 1;
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ledger failed level=%u peer=%llx handle=%llu add=%d sent=%d packets=%llu bytes=%llu",
                        i, (unsigned long long)f->peer->node_id, (unsigned long long)f->info.handle, added, sent,
                        (unsigned long long)packets, (unsigned long long)bytes);
        }
    }
}

static int same_stream(const struct stream_descriptor *a, const struct stream_descriptor *b) {
    return a->group_id == b->group_id && a->source_node_id == b->source_node_id && a->source_epoch == b->source_epoch;
}

/* Водяной знак и управляющая очередь принадлежат сессии пира, а не отдельной группе. */
static struct stream_peer *peer_find(struct stream_network *net, uint64_t node) {
    for (struct stream_peer *p = net->peers; p; p = p->next) if (p->node_id == node) return p;
    return NULL;
}

static uint64_t new_epoch(void) {
    uint64_t value = 0;
    do {
        if (random_bytes((uint8_t *)&value, sizeof(value)) != 0) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer epoch generation failed"); return 0;
        }
    } while (!value);
    return value;
}

/* Waiter callback только планирует pump; рекурсивная отправка из ETCP-очереди исключена. */
static void peer_schedule(struct stream_peer *p) {
    if (p->network->stopping || p->pump) return;
    p->pump = uasync_call_soon(p->network->inst->ua, p, peer_pump);
    if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream pump scheduling failed peer=%llx", (unsigned long long)p->node_id);
}

static void send_ready(struct ll_queue *queue, void *arg) {
    (void)queue; struct stream_peer *p = arg; p->wait_q = NULL; peer_schedule(p);
}

/* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */
static void peer_cancel_wait(struct stream_peer *p) {
    if (p->waiter.internal || p->waiter.call_soon_id)
        queue_waiter_cancel(p->waiter.call_soon_id ? p->waiter.defer_q : p->wait_q, &p->waiter);
    memset(&p->waiter, 0, sizeof(p->waiter)); p->wait_q = NULL;
}

/* Управление мало и ограничено; FIFO фиксирует порядок выделенных handles. В ней никогда нет DATA. */
static int control_put(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
    if (length + STREAM_WIRE_HEADER > STREAM_CONTROL_MAX || p->control->count >= STREAM_CONTROL_LIMIT) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control limit peer=%llx count=%d command=%u bytes=%zu",
                   (unsigned long long)p->node_id, p->control->count, m.command, length);
        p->load.rejected++; p->network->load.rejected++; return STREAM_ERR_STATE;
    }
    struct ll_entry *e = queue_entry_new(sizeof(uint64_t));
    if (e) e->dgram = u_malloc((uint32_t)(STREAM_WIRE_HEADER + length));
    if (!e || !e->dgram) {
        if (e) queue_entry_free(e);
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY;
    }
    uint64_t queued_ms = get_time_tb() / 10; memcpy(e->data, &queued_ms, sizeof(queued_ms));
    m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
    stream_wire_header(e->dgram, &m);
    if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
    e->len = STREAM_WIRE_HEADER + length;
    if (queue_data_put(p->control, e) != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control enqueue failed peer=%llx", (unsigned long long)p->node_id);
        return STREAM_ERR_STATE;
    }
    peer_schedule(p); return STREAM_OK;
}

/* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */
static int route_control(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
    struct UTUN_INSTANCE *inst = p->network->inst;
    if (!etcp_router_has_route(inst, m.descriptor.group_id, p->node_id)
        || !etcp_router_send_q_has_room(inst, m.descriptor.group_id, p->node_id, ETCP_RT_ID_STREAM)) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for route/room peer=%llx command=%u",
                    (unsigned long long)p->node_id, m.command); return STREAM_WAIT;
    }
    struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length));
    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed control allocation failed"); return STREAM_ERR_MEMORY; }
    e->dgram[0] = ETCP_RT_ID_STREAM; m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
    stream_wire_header(e->dgram + 1, &m);
    if (length) memcpy(e->dgram + 1 + STREAM_WIRE_HEADER, payload, length);
    e->len = 1 + STREAM_WIRE_HEADER + length;
    int result = etcp_route_send(inst, m.descriptor.group_id, p->node_id, e, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
    if (result < 0) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control failed peer=%llx command=%u result=%d",
                              (unsigned long long)p->node_id, m.command, result);
    return result < 0 ? STREAM_ERR_STATE : STREAM_OK;
}

/* Закрытие не ждёт доставки уведомления. Память flow освобождается после выхода из pump/receiver. */
static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
    if (f->info.state == STREAM_RECEIVE_CLOSED) return;
    struct stream_peer *p = f->peer;
    if (!f->outgoing && p) {
        for (unsigned i = 0; i < p->pressure.count; i++) {
            struct stream_cut *cut = &p->pressure.cuts[i];
            if (p->pressure.active && !cut->closed && cut->handle == f->info.handle) {
                cut->closed = 1; p->pressure.released_rate += cut->rate; p->pressure.released_count++;
            }
        }
        load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0);
        stream_load_changed(&p->load); stream_load_changed(&p->network->load);
    }
    stream_metrics_end(&f->metrics); stream_metrics_end(&f->offered_metrics);
    if (notify && p && f->info.handle) {
        uint8_t payload = (uint8_t)reason;
        struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock,
            .command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED};
        if (p->up) control_put(p, m, &payload, 1);
        else if (!f->outgoing) route_control(p, m, &payload, 1);
    }
    f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
    if (reason != STREAM_REASON_STOPPED)
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance failure peer=%llx handle=%llu sub=%u reason=%d next=%llu",
                   (unsigned long long)f->info.peer, (unsigned long long)f->info.handle, f->subblock,
                   reason, (unsigned long long)f->info.next);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream issuance closed peer=%llx group=%llx source=%llx handle=%llu sub=%u out=%d reason=%d blocks=%llu",
               (unsigned long long)f->info.peer, (unsigned long long)f->descriptor.group_id,
               (unsigned long long)f->descriptor.source_node_id, (unsigned long long)f->info.handle, f->subblock,
               f->outgoing, reason, (unsigned long long)f->info.blocks);
}

/* Caller-owned receive после detach живёт до close; внутренние downstream принадлежат модулю. */
static void peer_sweep(struct stream_peer *p) {
    if (p->busy) return;
    struct STREAM_RECEIVE **link = &p->flows;
    while (*link) {
        struct STREAM_RECEIVE *f = *link;
        if (f->info.state != STREAM_RECEIVE_CLOSED) { link = &f->next; continue; }
        *link = f->next; if (p->last_sent == f) p->last_sent = NULL;
        p->network->flow_count--; f->peer = NULL; f->next = NULL;
        if (!f->owned) u_free(f);
    }
    if (p->flows) p->idle = 0;
    else if (!p->idle) p->idle = get_time_tb();
}

/* Invalidation сохраняет транспорт чужих владельцев; старые DATA больше не имеют живого handle. */
static void peer_reset(struct stream_peer *p, int reason) {
    peer_cancel_wait(p);
    struct ll_entry *e;
    while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, reason, 0);
    p->other_epoch = p->highwater = 0; p->next_handle = 1; p->last_sent = NULL;
    p->own_epoch = new_epoch(); peer_sweep(p);
    memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested));
    p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx reason=%d",
               (unsigned long long)p->node_id, (unsigned long long)p->own_epoch, reason);
}

/* NCD подтверждает UP собственного владельца, даже если ETCP уже использовался BGP/другим стримом. */
static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, void *arg) {
    (void)handle; struct stream_peer *p = arg;
    if (p->network->stopping) return;
    if (event == NCD_EVENT_UP) { p->up = 1; peer_schedule(p); }
    else if (event == NCD_EVENT_DOWN || event == NCD_EVENT_CLOSED) {
        p->up = 0;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
            if (!f->outgoing && f->info.handle) {
                uint8_t reason = STREAM_REASON_TRANSPORT;
                struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
                    .subblock = f->subblock, .command = STREAM_CMD_CLOSED};
                route_control(p, m, &reason, 1);
            }
        peer_reset(p, STREAM_REASON_TRANSPORT);
    } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id);
    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream NCD event peer=%llx event=%d up=%d", (unsigned long long)p->node_id, event, p->up);
}

/* На пира ровно один stream NCD owner, общий для всех его выдач и групп. */
static struct stream_peer *peer_open(struct stream_network *net, uint64_t node) {
    struct stream_peer *p = peer_find(net, node);
    if (p) {
        if (!node_conn_direct_get_conn(p->ncd)) {
            node_conn_direct_set_callback(p->ncd, NULL, NULL); node_conn_direct_close(p->ncd); p->ncd = NULL;
            p->born = get_time_tb(); p->reverse_at = 0;
            if (node_conn_direct_open(net->inst, node, ncd_changed, p, &p->ncd, NULL) < 0) {
                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer NCD reopen failed peer=%llx", (unsigned long long)node); return NULL;
            }
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer NCD reopened peer=%llx", (unsigned long long)node);
        }
        return p;
    }
    if (net->peer_count == STREAM_PEER_LIMIT) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream peer limit reached"); return NULL; }
    p = u_calloc(1, sizeof(*p));
    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer allocation failed"); return NULL; }
    p->network = net; p->node_id = node; p->own_epoch = new_epoch(); p->next_handle = 1; p->born = get_time_tb();
    stream_load_init(&p->load, p->born / 10);
    p->control = queue_new(net->inst->ua, 0, 0, 0, "stream_control");
    if (!p->control || !p->own_epoch || node_conn_direct_open(net->inst, node, ncd_changed, p, &p->ncd, NULL) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer NCD/init failed peer=%llx", (unsigned long long)node);
        if (p->ncd) node_conn_direct_close(p->ncd);
        if (p->control) queue_free(p->control); u_free(p); return NULL;
    }
    p->next = net->peers; net->peers = p; net->peer_count++;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer opened peer=%llx epoch=%llx", (unsigned long long)node, (unsigned long long)p->own_epoch);
    return p;
}

/* Все callbacks/waiters отменяются до NCD close и освобождения контекста. */
static void peer_free(struct stream_peer *p) {
    struct stream_network *net = p->network;
    if (p->pump) uasync_call_soon_cancel(net->inst->ua, p->pump);
    peer_cancel_wait(p);
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, STREAM_REASON_STOPPED, 0);
    peer_sweep(p);
    struct ll_entry *e;
    while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
    queue_free(p->control);
    node_conn_direct_set_callback(p->ncd, NULL, NULL); node_conn_direct_close(p->ncd);
    struct stream_peer **link = &net->peers;
    while (*link != p) link = &(*link)->next;
    *link = p->next; net->peer_count--;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer released peer=%llx", (unsigned long long)p->node_id); u_free(p);
}

static struct STREAM_RECEIVE *flow_find(struct stream_peer *p, const struct stream_message *m, int outgoing) {
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
        if (f->outgoing == outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &m->descriptor)
            && f->subblock == m->subblock && (m->handle ? f->info.handle == m->handle : !f->info.handle)) return f;
    return NULL;
}

/* Лимиты проверяются до выделения памяти, NCD не определяет право доступа. */
static struct STREAM_RECEIVE *flow_new(struct stream_peer *p, const struct stream_descriptor *d, uint8_t sub, int outgoing) {
    unsigned count = 0; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) count++;
    if (p->network->flow_count == STREAM_FLOW_LIMIT || count == STREAM_PEER_FLOW_LIMIT) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance limit peer=%llx local=%u global=%u",
                   (unsigned long long)p->node_id, count, p->network->flow_count); return NULL;
    }
    struct STREAM_RECEIVE *f = u_calloc(1, sizeof(*f));
    if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream issuance allocation failed"); return NULL; }
    f->peer = p; f->descriptor = *d; f->subblock = sub; f->outgoing = outgoing;
    f->admitted_ms = get_time_tb() / 10; stream_load_init(&f->load, f->admitted_ms);
    stream_metrics_init(&f->metrics); stream_metrics_init(&f->offered_metrics);
    if (!outgoing) { stream_load_changed(&p->load); stream_load_changed(&p->network->load); }
    f->info.peer = p->node_id; f->info.subblock = sub; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
    f->next = p->flows; p->flows = f; p->network->flow_count++; p->idle = 0; return f;
}

/* Точная локальная эпоха проверяется до чтения подписанного профиля пира (unsubscribed => нет crypto). */
static int authorized(struct stream_network *net, uint64_t peer, const struct stream_message *m, struct stream_local *local) {
    if (stream_local_get(net->inst, &m->descriptor, local) != STREAM_OK) return 0;
    struct stream_member_state state;
    int result = stream_member_load(net->inst, m->descriptor.group_id, peer, &state, NULL);
    if (result != STREAM_OK) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream peer authorization failed peer=%llx group=%llx result=%d",
                   (unsigned long long)peer, (unsigned long long)m->descriptor.group_id, result); return 0;
    }
    return 1;
}

/* INFO и ACCEPT описывают реальный последний/начальный блок с его историей, а не BGP-метрику. */
static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, uint64_t peer, uint64_t self, uint8_t *payload) {
    const uint8_t *wire; size_t length; uint64_t actual, first, last; struct stream_path path;
    int range = stream_buffer_range(local->buffer, sub, &first, &last);
    if (range != STREAM_OK && local->publisher && (seq == UINT64_MAX || seq == 0)) {
        path = (struct stream_path){.count = 1, .nodes = {self}};
        stream_write_be(payload + 8, seq, 8);
        stream_write_be(payload + 16, UINT64_MAX, 8); stream_write_be(payload + 24, UINT64_MAX, 8);
        payload[32] = local->descriptor.n; payload[33] = local->descriptor.r;
        stream_write_be(payload + 34, local->descriptor.block_size, 2); stream_path_write(payload + STREAM_INFO_PATH_OFF, &path);
        return 0;
    }
    int result = stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual);
    int tail = range == STREAM_OK && last < UINT64_MAX - 1 && seq == last + 1;
    if (tail) result = stream_buffer_read(local->buffer, UINT64_MAX, sub, &wire, &length, &actual);
    if (result == STREAM_EXPIRED) return STREAM_REASON_EXPIRED;
    if (result != STREAM_OK
        || range != STREAM_OK
        || stream_buffer_path(local->buffer, actual, sub, &path) != STREAM_OK) return STREAM_REASON_POSITION;
    if (stream_path_validate(&path, local->descriptor.source_node_id, self, peer) != STREAM_OK || path.count == STREAM_PATH_MAX)
        return STREAM_REASON_LOOP;
    stream_write_be(payload + 8, tail ? seq : actual, 8); stream_write_be(payload + 16, first, 8); stream_write_be(payload + 24, last, 8);
    payload[32] = local->descriptor.n; payload[33] = local->descriptor.r;
    stream_write_be(payload + 34, local->descriptor.block_size, 2);
    stream_path_write(payload + STREAM_INFO_PATH_OFF, &path); return 0;
}

static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self,
                      struct stream_receive_info *info, uint64_t *start) {
    struct stream_path path; size_t used;
    uint64_t actual = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 8, 8) : UINT64_MAX;
    uint64_t first = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 16, 8) : UINT64_MAX;
    uint64_t last = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 24, 8) : UINT64_MAX;
    int empty = first == UINT64_MAX && last == UINT64_MAX;
    if (m->length < STREAM_INFO_SIZE || m->payload[32] != d->n || m->payload[33] != d->r
        || stream_read_be(m->payload + 34, 2) != d->block_size
        || stream_path_parse(m->payload + STREAM_INFO_PATH_OFF, m->length - STREAM_INFO_PATH_OFF, &path, &used) != STREAM_OK
        || m->length != STREAM_INFO_PATH_OFF + used
        || path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK
        || (!empty && (actual == UINT64_MAX || first > last || last == UINT64_MAX
            || (actual > last && !(m->command == STREAM_CMD_ACCEPT && last < UINT64_MAX - 1 && actual == last + 1))))
        || (empty && (peer != d->source_node_id || path.count != 1
            || actual != (m->command == STREAM_CMD_ACCEPT ? 0 : UINT64_MAX)))) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length);
        return STREAM_ERR_FORMAT;
    }
    const uint8_t *offer = m->payload + STREAM_INFO_OFFER_OFF;
    if ((offer[0] & ~31u) || !!(offer[0] & 16) == empty || stream_read_be(offer + 3, 2) > 10000) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream invalid offer flags=%u cpu=%llu", offer[0],
                   (unsigned long long)stream_read_be(offer + 3, 2)); return STREAM_ERR_FORMAT;
    }
    *start = stream_read_be(m->payload + 8, 8);
    info->path = path; info->first_available = stream_read_be(m->payload + 16, 8);
    info->last_available = stream_read_be(m->payload + 24, 8); info->info_ms = get_time_tb() / 10;
    info->accepting = (offer[0] & 1) != 0; info->path_rtt_valid = (offer[0] & 2) != 0;
    info->progress_valid = (offer[0] & 4) != 0; info->cpu_valid = (offer[0] & 8) != 0;
    info->free_slots = (uint16_t)stream_read_be(offer + 1, 2); info->cpu_basis_points = (uint16_t)stream_read_be(offer + 3, 2);
    info->supplier_queue_bytes = stream_read_be(offer + 5, 8); info->progress_age_ms = stream_read_be(offer + 13, 8);
    info->path_rtt_ms = (uint32_t)stream_read_be(offer + 21, 4); return STREAM_OK;
}

/* Предложение привязано к фактическому пути последнего блока; другой upstream не подтверждает его RTT. */
static void offer_payload(struct stream_peer *p, struct stream_local *local, uint8_t sub, uint8_t *payload) {
    struct stream_network *net = p->network; uint64_t now = get_time_tb() / 10;
    uint8_t *offer = payload + STREAM_INFO_OFFER_OFF;
    unsigned count = 0; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) count++;
    unsigned slots = STREAM_PEER_FLOW_LIMIT - count;
    if (slots > STREAM_FLOW_LIMIT - net->flow_count) slots = STREAM_FLOW_LIMIT - net->flow_count;
    offer[0] = !p->pressure.active && !p->cooling ? 1 : 0;
    if (stream_read_be(payload + 16, 8) != UINT64_MAX) offer[0] |= 16;
    stream_write_be(offer + 1, slots, 2);
    double cpu = net->cpu.valid ? net->cpu.average_percent : 0;
    if (cpu < 0) cpu = 0; if (cpu > 100) cpu = 100;
    if (net->cpu.valid) offer[0] |= 8;
    stream_write_be(offer + 3, (uint16_t)(cpu * 100), 2); stream_write_be(offer + 5, p->load.queued.bytes, 8);
    struct stream_metric_snapshot snapshot = {0};
    int valid = stream_local_metrics(net->inst, &local->descriptor, sub, now, &snapshot) == STREAM_OK;
    stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
    if (valid) offer[0] |= 4;
    stream_write_be(offer + 13, snapshot.progress_age_ms, 8);
    if (local->publisher) { offer[0] |= 2; return; }
    struct stream_path path; size_t used;
    if (stream_path_parse(payload + STREAM_INFO_PATH_OFF, 1 + payload[STREAM_INFO_PATH_OFF] * 8, &path, &used) != STREAM_OK) return;
    for (struct stream_peer *up = net->peers; up; up = up->next) {
        for (struct STREAM_RECEIVE *f = up->flows; f; f = f->next) {
            if (!f->outgoing || f->info.probe_only || f->info.state != STREAM_RECEIVE_ACTIVE || f->subblock != sub
                || !same_stream(&f->descriptor, &local->descriptor) || !f->info.rtt_valid || !f->info.path_rtt_valid
                || now - f->info.info_ms > 30000 || path.count != f->info.path.count + 1
                || memcmp(path.nodes, f->info.path.nodes, f->info.path.count * sizeof(uint64_t))) continue;
            uint64_t total = (uint64_t)f->info.rtt_ms + f->info.path_rtt_ms;
            offer[0] |= 2; stream_write_be(offer + 21, total > UINT32_MAX ? UINT32_MAX : total, 4); return;
        }
    }
}

/* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение entry в etcp_send. */
static int raw_send(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
    struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
    struct ll_entry *e = ll_alloc_lldgram((uint16_t)(STREAM_WIRE_HEADER + length));
    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream raw allocation failed bytes=%zu", length); return STREAM_ERR_MEMORY; }
    m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
    stream_wire_header(e->dgram, &m); if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
    e->len = STREAM_WIRE_HEADER + length;
    if (etcp_send(conn, e) != 0) {
        queue_dgram_free(e); queue_entry_free(e);
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream raw send failed peer=%llx command=%u", (unsigned long long)p->node_id, m.command);
        return STREAM_ERR_STATE;
    }
    DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream raw sent peer=%llx command=%u handle=%llu bytes=%zu",
                (unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, length);
    return STREAM_OK;
}

/* Нижняя граница прогресса всех выбранных выдач. INFO/REQUESTED ещё могут принести старую пачку и запрещают скачок. */
static void receive_progress(struct stream_network *net, const struct stream_descriptor *d) {
    uint8_t seen[RS_MAX_BLOCKS] = {0}; unsigned distinct = 0; uint64_t before = UINT64_MAX;
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            if (!f->outgoing || f->info.probe_only || f->info.state == STREAM_RECEIVE_CLOSED || !same_stream(&f->descriptor, d)) continue;
            if (f->info.state < STREAM_RECEIVE_ACTIVE) return;
            if (!seen[f->subblock]) { seen[f->subblock] = 1; distinct++; }
            if (f->info.next < before) before = f->info.next;
        }
    }
    if (distinct >= (unsigned)(d->n - d->r)) stream_local_progress(net->inst, d, before);
}

/* Производитель не хранит готовые DATA: cursor проверяется непосредственно перед копированием в ETCP. */
static int downstream_send(struct STREAM_RECEIVE *f) {
    struct stream_peer *p = f->peer; struct stream_local local;
    if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; }
    struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock};
    if (f->gap_pending) {
        uint8_t payload[16]; stream_write_be(payload, f->gap_from, 8); stream_write_be(payload + 8, f->info.next, 8);
        m.command = STREAM_CMD_GAP;
        if (raw_send(p, m, payload, sizeof(payload)) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
        f->gap_pending = 0; f->info.gaps++; return 1;
    }
    const uint8_t *wire; size_t length; uint64_t actual;
    int result = f->info.next == UINT64_MAX ? STREAM_WAIT
                 : stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
    if (result == STREAM_OK) {
        struct stream_path path;
        if (stream_buffer_path(local.buffer, actual, f->subblock, &path) != STREAM_OK
            || stream_path_validate(&path, f->descriptor.source_node_id, p->network->inst->node_id, p->node_id) != STREAM_OK) {
            flow_close(f, STREAM_REASON_LOOP, 1); return 0;
        }
        uint8_t payload[1 + STREAM_PATH_MAX * 8 + STREAM_BLOCK_MAX_SIZE];
        size_t path_bytes = stream_path_write(payload, &path); memcpy(payload + path_bytes, wire, length);
        m.command = STREAM_CMD_DATA;
        if (raw_send(p, m, payload, path_bytes + length) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
        if (f->info.state == STREAM_RECEIVE_ACTIVE) {
            load_account(f, 1, length, 0, 1); f->info.next = actual + 1; f->info.blocks++;
        }
        return 1;
    }
    if (result == STREAM_EXPIRED || result == STREAM_GAP) {
        uint64_t next;
        int available = stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK;
        if (!available && local.finished && local.last != UINT64_MAX && f->info.next <= local.last) {
            next = local.last + 1; available = 1;
        }
        if (available && next > f->info.next) {
            uint8_t payload[16]; stream_write_be(payload, f->info.next, 8); stream_write_be(payload + 8, next, 8);
            m.command = STREAM_CMD_GAP;
            if (raw_send(p, m, payload, sizeof(payload)) == STREAM_OK) {
                f->info.next = next; f->info.gaps++; return 1;
            }
            flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0;
        }
    }
    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);
            stream_load_changed(&p->load); stream_load_changed(&p->network->load); return 1;
        }
        flow_close(f, STREAM_REASON_TRANSPORT, 0);
    }
    return 0;
}

/* INFO предшествует выделению handle; остальные latest-запросы ждут подтверждённый DATA старт первой выдачи. */
static void prepare_requests(struct stream_peer *p) {
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
        if (!f->outgoing || f->info.state >= STREAM_RECEIVE_REQUESTED) continue;
        struct stream_message m = {.descriptor = f->descriptor, .subblock = f->subblock};
        if (!f->info_sent) {
            m.command = STREAM_CMD_INFO_REQ;
            uint8_t probe[8];
            if (!f->probe) f->probe = new_epoch();
            if (!f->probe) { flow_close(f, STREAM_REASON_PROTOCOL, 0); continue; }
            stream_write_be(probe, f->probe, 8);
            if (control_put(p, m, probe, sizeof(probe)) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; }
        }
        if (!f->info_ready || f->info.probe_only || !p->other_epoch) continue;
        uint64_t requested = f->requested;
        struct stream_local local;
        if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 0); continue; }
        if (requested == UINT64_MAX) {
            if (local.anchored) requested = local.start;
            else if (!f->bootstrap) continue;
        }
        if (p->next_handle == UINT64_MAX) { flow_close(f, STREAM_REASON_LIMIT, 0); continue; }
        uint8_t payload[8]; stream_write_be(payload, requested, 8);
        m.command = STREAM_CMD_SUBSCRIBE; m.handle = p->next_handle;
        if (control_put(p, m, payload, sizeof(payload)) == STREAM_OK) {
            f->info.handle = p->next_handle++; f->requested = requested; f->info.state = STREAM_RECEIVE_REQUESTED;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE queued peer=%llx handle=%llu sub=%u index=%llu bootstrap=%d",
                       (unsigned long long)p->node_id, (unsigned long long)f->info.handle, f->subblock,
                       (unsigned long long)requested, f->bootstrap);
        }
    }
}

/* Управление FIFO прежде DATA, затем round-robin курсоров; один waiter на общий send_input_q. */
static void peer_pump(void *arg) {
    struct stream_peer *p = arg; p->pump = NULL;
    if (!p->up || !p->own_epoch || p->network->stopping) return;
    struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
    if (!conn || !conn->send_input_q || conn->state == 2 || conn->close_requested || conn->reinit_pending || !conn->initialized) return;
    if (conn->send_input_q->count) {
        p->wait_q = conn->send_input_q;
        if (queue_waiter_wait(p->wait_q, &p->waiter, send_ready, p) < 0)
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream transport waiter failed peer=%llx", (unsigned long long)p->node_id);
        return;
    }
    p->busy++; prepare_requests(p);
    int sent = 0;
    struct ll_entry *e = queue_data_get(p->control);
    if (e) {
        if (e->len >= STREAM_WIRE_HEADER && e->dgram[2] == STREAM_CMD_INFO_REQ) {
            struct stream_message m;
            if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) {
                struct STREAM_RECEIVE *f = flow_find(p, &m, 1); if (f) f->probe_sent_ms = get_time_tb() / 10;
            }
        }
        if (etcp_send(conn, e) != 0) {
            queue_dgram_free(e); queue_entry_free(e);
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control send failed peer=%llx", (unsigned long long)p->node_id);
            peer_reset(p, STREAM_REASON_TRANSPORT);
        } else sent = 1;
        queue_resume_callback(p->control);
    } else {
        struct STREAM_RECEIVE *first = p->last_sent && p->last_sent->next ? p->last_sent->next : p->flows;
        struct STREAM_RECEIVE *f = first;
        while (f) {
            if (f->account_failed) flow_close(f, STREAM_REASON_PROTOCOL, 1);
            if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && downstream_send(f)) { p->last_sent = f; sent = 1; break; }
            f = f->next ? f->next : p->flows;
            if (f == first) break;
        }
    }
    p->busy--; peer_sweep(p);
    if (sent && p->up) {
        p->wait_q = conn->send_input_q;
        if (queue_waiter_wait(p->wait_q, &p->waiter, send_ready, p) < 0)
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream post-send waiter failed peer=%llx", (unsigned long long)p->node_id);
    }
}

/* Известный повтор продлевает аренду без смены cursor; любой новый handle обязан превысить watermark. */
static void subscribe_request(struct stream_peer *p, struct stream_message m, struct stream_local *local) {
    uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
    int reason = 0;
    if (m.length != 8 || !m.handle || m.subblock >= local->descriptor.n) reason = STREAM_REASON_PROTOCOL;
    struct STREAM_RECEIVE *f = reason ? NULL : flow_find(p, &m, 0);
    uint64_t requested = m.length == 8 ? stream_read_be(m.payload, 8) : 0;
    if (f && f->requested != requested) reason = STREAM_REASON_CONFLICT;
    if (!f && !reason && m.handle <= p->highwater) reason = STREAM_REASON_STALE;
    if (!f && !reason && (p->pressure.active || p->cooling)) {
        p->highwater = m.handle; reason = STREAM_REASON_OVERLOAD;
        p->load.rejected++; p->network->load.rejected++;
    }
    if (!f && !reason) {
        p->highwater = m.handle;
        reason = info_payload(local, m.subblock, requested, p->node_id, p->network->inst->node_id, payload);
        if (!reason) {
            f = flow_new(p, &local->descriptor, m.subblock, 0);
            if (!f) reason = STREAM_REASON_LIMIT;
            else {
                f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload + 8, 8);
                if (f->info.start == UINT64_MAX) { f->info.start = f->info.next = 0; stream_write_be(payload + 8, 0, 8); }
                f->info.state = STREAM_RECEIVE_ACTIVE;
                uint64_t count = 0;
                stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL);
                load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0);
                f->pending_highest = stream_read_be(payload + 24, 8);
                if (f->pending_highest == UINT64_MAX) f->pending_highest = 0;
                const uint8_t *wire; size_t length; uint64_t seq;
                if (stream_buffer_read(local->buffer, UINT64_MAX, f->subblock, &wire, &length, &seq) == STREAM_OK)
                    stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
                                           wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
            }
        }
    } else if (f && !reason) {
        reason = info_payload(local, m.subblock, f->info.start, p->node_id, p->network->inst->node_id, payload);
        /* Идемпотентный ACCEPT возвращает первоначальный индекс даже если тот уже вытеснен. */
        if (reason == STREAM_REASON_POSITION || reason == STREAM_REASON_EXPIRED) {
            reason = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload);
            if (!reason) stream_write_be(payload + 8, f->info.start, 8);
        }
    }
    if (reason) {
        uint8_t rejection[17] = {(uint8_t)reason}; uint64_t first = UINT64_MAX, last = UINT64_MAX;
        if (m.subblock < local->descriptor.n) stream_buffer_range(local->buffer, m.subblock, &first, &last);
        stream_write_be(rejection + 1, first, 8); stream_write_be(rejection + 9, last, 8);
        m.command = STREAM_CMD_REJECT; control_put(p, m, rejection, sizeof(rejection));
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u index=%llu reason=%d range=%llu..%llu watermark=%llu",
                   (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)requested,
                   reason, (unsigned long long)first, (unsigned long long)last, (unsigned long long)p->highwater);
        return;
    }
    f->deadline = get_time_tb() + STREAM_LEASE_TB;
    offer_payload(p, local, m.subblock, payload);
    m.command = STREAM_CMD_ACCEPT;
    if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8) != STREAM_OK) {
        flow_close(f, STREAM_REASON_LIMIT, 0); return;
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms",
               (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)f->info.start);
}

/* Raw transport authenticates непосредственного отправителя; сначала local key, затем членство и peer epoch. */
static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t *wire, size_t length, int routed) {
    struct stream_message m; struct stream_local local;
    if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return;
    if (m.command == STREAM_CMD_OVERLOAD || m.command == STREAM_CMD_OVERLOAD_SELECT) {
        struct stream_peer *p = peer_find(net, sender);
        if (!routed && p && p->up && m.receiver_epoch == p->own_epoch && m.sender_epoch == p->other_epoch && !m.handle) {
            p->busy++; pressure_receive(p, &m); p->busy--; peer_sweep(p);
        } else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/unknown pressure request peer=%llx", (unsigned long long)sender);
        return;
    }
    if (stream_local_get(net->inst, &m.descriptor, &local) != STREAM_OK) {
        DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream packet ignored without local subscription command=%u", m.command); return;
    }
    struct stream_peer *p = peer_find(net, sender);
    if (routed && m.command == STREAM_CMD_REVERSE_REQ) {
        if (m.length || m.handle || m.receiver_epoch) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ malformed peer=%llx bytes=%zu handle=%llu receiver_epoch=%llx",
                       (unsigned long long)sender, m.length, (unsigned long long)m.handle, (unsigned long long)m.receiver_epoch); return;
        }
        if (!authorized(net, sender, &m, &local)) return;
        if (!p) p = peer_open(net, sender);
        if (p) { p->idle = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ peer=%llx attempt=%llx NCD=%p",
            (unsigned long long)sender, (unsigned long long)m.sender_epoch, (void *)p->ncd); }
        return;
    }
    if (routed && m.command != STREAM_CMD_CLOSED) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return;
    }
    if (m.command == STREAM_CMD_INFO_REQ && !routed) {
        if (m.handle || m.length != 8 || !stream_read_be(m.payload, 8) || m.subblock >= local.descriptor.n) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u",
                       (unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return;
        }
        if (!authorized(net, sender, &m, &local)) return;
        if (!p) p = peer_open(net, sender);
        if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return;
        if (p->other_epoch && p->other_epoch != m.sender_epoch) {
            peer_reset(p, STREAM_REASON_TRANSPORT);
            /* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */
        }
        p->other_epoch = m.sender_epoch;
        uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
        memcpy(payload, m.payload, 8);
        int reason = info_payload(&local, m.subblock, UINT64_MAX, sender, net->inst->node_id, payload);
        m.command = reason ? STREAM_CMD_REJECT : STREAM_CMD_INFO_RESP;
        if (reason) {
            uint8_t failure[9] = {(uint8_t)reason}; memcpy(failure + 1, m.payload, 8); control_put(p, m, failure, sizeof(failure));
        } else {
            offer_payload(p, &local, m.subblock, payload);
            control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8);
        }
        return;
    }
    if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
        || (p->other_epoch && m.sender_epoch != p->other_epoch)) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session peer=%llx command=%u handle=%llu sender=%llx/%llx receiver=%llx/%llx",
                    (unsigned long long)sender, m.command, (unsigned long long)m.handle,
                    (unsigned long long)m.sender_epoch, (unsigned long long)(p ? p->other_epoch : 0),
                    (unsigned long long)m.receiver_epoch, (unsigned long long)(p ? p->own_epoch : 0)); return;
    }
    p->busy++;
    struct STREAM_RECEIVE *f = flow_find(p, &m, m.command != STREAM_CMD_SUBSCRIBE && m.command != STREAM_CMD_RENEW
                                       && m.command != STREAM_CMD_UNSUBSCRIBE);
    if (m.command == STREAM_CMD_INFO_RESP && f && !m.handle && f->info_sent) {
        uint64_t start;
        if (m.length < 8 || stream_read_be(m.payload, 8) != f->probe) {
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale INFO probe peer=%llx", (unsigned long long)sender);
        } else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &start) == STREAM_OK
                   && stream_metric_read(m.payload + STREAM_INFO_METRIC_OFF, &f->info.supplier_metrics, &f->info.supplier_metrics_valid) == STREAM_OK) {
            p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; peer_schedule(p);
            uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms;
            f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt;
            for (struct STREAM_RECEIVE *active = p->flows; active; active = active->next) {
                if (active == f || !active->outgoing || active->info.probe_only || active->info.state != STREAM_RECEIVE_ACTIVE
                    || active->subblock != f->subblock || !same_stream(&active->descriptor, &f->descriptor)
                    || active->info.path.count != f->info.path.count
                    || memcmp(active->info.path.nodes, f->info.path.nodes, f->info.path.count * sizeof(uint64_t))) continue;
                active->info.rtt_ms = f->info.rtt_ms; active->info.rtt_valid = f->info.rtt_valid;
                active->info.path_rtt_ms = f->info.path_rtt_ms; active->info.path_rtt_valid = f->info.path_rtt_valid;
                active->info.info_ms = f->info.info_ms;
            }
        } else flow_close(f, STREAM_REASON_PROTOCOL, 0);
    } else if (m.command == STREAM_CMD_SUBSCRIBE) {
        if (!authorized(net, sender, &m, &local)) {
            uint8_t reason = STREAM_REASON_MEMBER; m.command = STREAM_CMD_REJECT; control_put(p, m, &reason, 1);
        } else subscribe_request(p, m, &local);
    } else if (m.command == STREAM_CMD_ACCEPT && f) {
        uint64_t start;
        if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &start) != STREAM_OK
            || (f->requested != UINT64_MAX && start != f->requested)) flow_close(f, STREAM_REASON_PROTOCOL, 1);
        else if (f->info.state == STREAM_RECEIVE_REQUESTED) {
            f->info.start = f->info.next = start; f->info.state = STREAM_RECEIVE_ACTIVE; f->renewed = get_time_tb();
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive ACCEPT peer=%llx handle=%llu sub=%u start=%llu",
                       (unsigned long long)sender, (unsigned long long)m.handle, m.subblock, (unsigned long long)start);
        }
    } else if (m.command == STREAM_CMD_DATA && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
        struct stream_path path; size_t used;
        if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || m.length - used != stream_block_size(&f->descriptor)
            || stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) != STREAM_OK) {
            flow_close(f, STREAM_REASON_LOOP, 1);
        } else {
            const uint8_t *block = m.payload + used; uint64_t seq = stream_read_be(block + STREAM_BLOCK_SEQ_OFF, 8);
            if (block[STREAM_BLOCK_SUB_OFF] != f->subblock || seq != f->info.next
                || stream_read_be(block + STREAM_BLOCK_GROUP_OFF, 8) != f->descriptor.group_id
                || stream_read_be(block + STREAM_BLOCK_SOURCE_OFF, 8) != f->descriptor.source_node_id
                || stream_read_be(block + STREAM_BLOCK_EPOCH_OFF, 8) != f->descriptor.source_epoch) flow_close(f, STREAM_REASON_PROTOCOL, 1);
            else {
                int result = stream_ingest_path(net->inst, block, m.length - used, &path);
                if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) {
                    stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2),
                                           block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10);
                    if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t)))
                        f->info.path_rtt_valid = 0;
                    f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB;
                } else flow_close(f, STREAM_REASON_PROTOCOL, 1);
            }
        }
    } else if (m.command == STREAM_CMD_GAP && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
        if (m.length != 16 || stream_read_be(m.payload, 8) != f->info.next
            || stream_read_be(m.payload + 8, 8) <= f->info.next) flow_close(f, STREAM_REASON_PROTOCOL, 1);
        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)
            || (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);
        }
        else flow_close(f, STREAM_REASON_PROTOCOL, 1);
    } else if (m.command == STREAM_CMD_RENEW && f && !m.length) {
        f->deadline = get_time_tb() + STREAM_LEASE_TB;
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream lease renewed peer=%llx handle=%llu deadline=%llu",
                    (unsigned long long)sender, (unsigned long long)m.handle, (unsigned long long)f->deadline);
    }
    else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1);
    else if (m.command == STREAM_CMD_REJECT && f && !m.handle && m.length == 9 && stream_read_be(m.payload + 1, 8) == f->probe)
        flow_close(f, m.payload[0], 0);
    else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.handle && (m.length == 1 || m.length == 17))
        flow_close(f, m.payload[0], 0);
    else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream unknown/inactive issuance peer=%llx command=%u handle=%llu sub=%u bytes=%zu",
                     (unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length);
    receive_progress(net, &local.descriptor);
    p->busy--; peer_sweep(p);
}

/* Entry/dgram принадлежат обработчику во всех ветках. */
static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
    struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
    if (net && entry && entry->dgram) dispatch(net, conn->peer_node_id, entry->dgram, entry->len, 0);
    if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
}

/* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
    struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
    if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
        && (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
        uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
        memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
        const uint8_t *wire = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
        if (stream_read_be(wire + 4, 8) == group) dispatch(net, sender, wire, entry->len - ROUTER_SVC_PAYLOAD_OFF, 1);
        else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed group mismatch");
    } else if (net && entry && entry->len > ROUTER_SVC_PAYLOAD_OFF)
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control unsigned/truncated bytes=%u", entry->len);
    if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
}

/* REINIT меняет stream generation при сохранении физического ETCP-соединения. */
static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) {
    struct stream_network *net = arg;
    if (status != ETCP_CONN_STATUS_REINIT || net->stopping) return;
    struct stream_peer *p = peer_find(net, conn->peer_node_id);
    if (p) peer_reset(p, STREAM_REASON_TRANSPORT);
}

/* FIFO ограничена и по времени. Отмена неподтверждённого SUBSCRIBE/ACCEPT не допускает DATA без живого согласования. */
static void expire_controls(struct stream_peer *p, uint64_t now) {
    while (p->control->head) {
        struct ll_entry *e = p->control->head; uint64_t queued_ms; memcpy(&queued_ms, e->data, sizeof(queued_ms));
        if (now - queued_ms <= STREAM_LOAD_MAX_AGE_MS) break;
        e = queue_data_get(p->control); struct stream_message m;
        if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) {
            int outgoing = m.command == STREAM_CMD_INFO_REQ || m.command == STREAM_CMD_SUBSCRIBE;
            struct STREAM_RECEIVE *f = flow_find(p, &m, outgoing);
            if (f && m.command == STREAM_CMD_INFO_REQ && (m.length != 8 || stream_read_be(m.payload, 8) != f->probe)) f = NULL;
            if (f && (outgoing || m.command == STREAM_CMD_ACCEPT)) flow_close(f, STREAM_REASON_OVERLOAD, 0);
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control expired peer=%llx command=%u handle=%llu age=%llu",
                       (unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, (unsigned long long)(now - queued_ms));
        }
        p->load.rejected++; p->network->load.rejected++;
        queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(p->control);
    }
}

/* Один таймер обслуживает сроки/продления и ограниченные reverse-попытки; DATA не опрашивается таймером. */
static void network_tick(void *arg) {
    struct stream_network *net = arg; net->timer = NULL;
    uint64_t now = get_time_tb();
    uint64_t now_ms = now / 10;
    net->callback_late_ms = now_ms > net->next_tick_ms ? now_ms - net->next_tick_ms : 0;
    uasync_get_load(net->inst->ua, &net->cpu);
    struct stream_peer *p = net->peers;
    while (p) {
        struct stream_peer *next = p->next;
        p->busy++;
        expire_controls(p, now_ms);
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
            f->info.metrics_valid = stream_metrics_poll(&f->metrics, now_ms, &f->info.metrics) == STREAM_OK;
            if ((!f->outgoing || (f->info.state < STREAM_RECEIVE_ACTIVE && !(f->info.probe_only && f->info.info_ready))
                || (f->outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE && !f->info.blocks
                    && f->info.last_available != UINT64_MAX))
                && now >= f->deadline)
                flow_close(f, f->outgoing ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
            else if (f->outgoing && f->info.state >= STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
                struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
                    .command = STREAM_CMD_RENEW, .subblock = f->subblock};
                if (control_put(p, m, NULL, 0) == STREAM_OK) f->renewed = now;
            }
        }
        struct STREAM_RECEIVE *pending = NULL;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
            if (f->outgoing && f->info.state < STREAM_RECEIVE_ACTIVE && now < f->deadline) { pending = f; break; }
        if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) {
            struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ};
            uint64_t saved = p->other_epoch; p->other_epoch = 0;
            if (route_control(p, m, NULL, 0) == STREAM_OK) p->reverse_at = now;
            p->other_epoch = saved;
        }
        p->busy--; peer_sweep(p);
        if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p);
        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");
}

/* Binding collision — ошибка запуска, другие обработчики не затираются. */
int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) {
    if (!inst || !out || *out || inst->api_bindings.callbacks[ETCP_ID_STREAM] || inst->router_bindings.callbacks[ETCP_RT_ID_STREAM]) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network invalid state/binding collision"); return STREAM_ERR_STATE;
    }
    struct stream_network *net = u_calloc(1, sizeof(*net));
    if (!net) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network allocation failed"); return STREAM_ERR_MEMORY; }
    net->inst = inst;
    stream_load_init(&net->load, get_time_tb() / 10); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
    if (etcp_bind(inst, ETCP_ID_STREAM, receive_raw) < 0 || etcp_router_bind(inst, ETCP_RT_ID_STREAM, receive_routed) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bindings registration failed"); stream_network_destroy(net); return STREAM_ERR_STATE;
    }
    etcp_add_conn_status_cbk(inst, transport_changed, net);
    int registered = 0;
    for (struct etcp_status_cbk_entry *e = inst->conn_status_cbks; e; e = e->next) if (e->fn == transport_changed && e->arg == net) registered = 1;
    net->timer = uasync_set_timeout(inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network");
    if (!registered || !net->timer) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network callback/timer registration failed");
        stream_network_destroy(net); return STREAM_ERR_MEMORY;
    }
    *out = net;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network initialized raw=%u routed=%u peers=%u flows=%u controls=%u lease=6000ms renew=2000ms",
               ETCP_ID_STREAM, ETCP_RT_ID_STREAM, STREAM_PEER_LIMIT, STREAM_FLOW_LIMIT, STREAM_CONTROL_LIMIT);
    return STREAM_OK;
}

void stream_network_destroy(struct stream_network *net) {
    if (!net) return;
    net->stopping = 1; if (net->timer) uasync_cancel_timeout(net->inst->ua, net->timer);
    etcp_remove_conn_status_cbk(net->inst, transport_changed, net);
    etcp_unbind(net->inst, ETCP_ID_STREAM); etcp_router_unbind(net->inst, ETCP_RT_ID_STREAM);
    while (net->peers) peer_free(net->peers);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network stopped flows=%u", net->flow_count); u_free(net);
}

void stream_network_changed(struct stream_network *net) {
    if (net && !net->stopping) for (struct stream_peer *p = net->peers; p; p = p->next) peer_schedule(p);
}

/* Кольцо освобождается только после отмены сетевых курсоров данной точной эпохи. */
void stream_network_cancel(struct stream_network *net, const struct stream_descriptor *d) {
    if (!net) return;
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        p->busy++;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
            if (same_stream(&f->descriptor, d)) flow_close(f, STREAM_REASON_STOPPED, 1);
        p->busy--; peer_sweep(p);
    }
}

/* Исключение поставщика/получателя или self закрывает выдачи, сохраняя локальное кольцо остальных трансляций. */
void stream_network_membership_changed(struct stream_network *net, uint64_t node, const char *channel) {
    if (!net || net->stopping || !channel) return;
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        if (node != p->node_id && node != net->inst->node_id) continue;
        p->busy++;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
            struct TOPO_GROUP *group = topo_groups_find(net->inst->topo_groups, f->descriptor.group_id);
            if (!group || strcmp(group->channel_id, channel)) continue;
            struct stream_member_state state;
            int result = stream_member_load(net->inst, group->group_id, node, &state, NULL);
            if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT)
                flow_close(f, STREAM_REASON_MEMBER, 1);
        }
        p->busy--; peer_sweep(p);
    }
}

static int receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, int probe_only,
                         struct STREAM_RECEIVE **out) {
    struct UTUN_INSTANCE *inst; struct stream_descriptor d;
    if (!out || *out || stream_handle_context(subscription, &inst, &d) != STREAM_OK || sub >= d.n || !node || node == inst->node_id) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive open invalid arguments peer=%llx sub=%u", (unsigned long long)node, sub);
        return STREAM_ERR_ARGUMENT;
    }
    struct stream_network *net = stream_network_get(inst);
    struct stream_member_state member;
    if (!net || !stream_member_online(inst, d.group_id, node) || stream_member_load(inst, d.group_id, node, &member, NULL) != STREAM_OK
        || !stream_member_has_subscription(&member, d.source_node_id, d.source_epoch)) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream receive supplier unavailable peer=%llx", (unsigned long long)node); return STREAM_WAIT;
    }
    struct stream_peer *p = peer_open(net, node);
    if (!p) return STREAM_ERR_STATE;
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
        if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d) && f->subblock == sub
            && (f->info.probe_only == probe_only || !f->info.handle)) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream receive already open peer=%llx sub=%u", (unsigned long long)node, sub);
            return STREAM_ERR_STATE;
        }
    int bootstrap = 1;
    for (struct stream_peer *other = net->peers; other; other = other->next)
        for (struct STREAM_RECEIVE *f = other->flows; f; f = f->next)
            if (f->outgoing && !f->info.probe_only && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d)) bootstrap = 0;
    struct STREAM_RECEIVE *f = flow_new(p, &d, sub, 1);
    if (!f) return STREAM_ERR_MEMORY;
    f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; f->info.probe_only = probe_only; *out = f; peer_schedule(p);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive opened peer=%llx group=%llx source=%llx epoch=%llx sub=%u index=%llu bootstrap=%d",
               (unsigned long long)node, (unsigned long long)d.group_id, (unsigned long long)d.source_node_id,
               (unsigned long long)d.source_epoch, sub, (unsigned long long)seq, bootstrap);
    return STREAM_OK;
}

int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, struct STREAM_RECEIVE **out) {
    return receive_open(subscription, node, sub, seq, 0, out);
}

int stream_probe_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, struct STREAM_RECEIVE **out) {
    return receive_open(subscription, node, sub, UINT64_MAX, 1, out);
}

int stream_probe_next(struct STREAM_RECEIVE *f, uint8_t sub) {
    if (!f || !f->peer || !f->info.probe_only || !f->info.info_ready || sub >= f->descriptor.n) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream probe next invalid state/sub=%u", sub); return STREAM_ERR_STATE;
    }
    for (struct STREAM_RECEIVE *other = f->peer->flows; other; other = other->next)
        if (other != f && other->outgoing && !other->info.handle && other->subblock == sub
            && same_stream(&other->descriptor, &f->descriptor)) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream probe next conflicts with pending INFO sub=%u", sub); return STREAM_WAIT;
        }
    f->subblock = f->info.subblock = sub; f->info_sent = f->info_ready = f->info.info_ready = 0;
    f->probe = f->probe_sent_ms = 0; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
    f->info.state = STREAM_RECEIVE_CONNECTING; peer_schedule(f->peer); return STREAM_OK;
}

void stream_receive_close(struct STREAM_RECEIVE *f) {
    if (!f) return;
    f->owned = 0;
    if (!f->peer) { u_free(f); return; }
    struct stream_peer *p = f->peer;
    flow_close(f, STREAM_REASON_STOPPED, 1); peer_sweep(p);
}

/* Смена стартовой попытки инвалидирует только upstream этой трансляции; номера новых SUBSCRIBE продолжают возрастать. */
int stream_receive_restart(struct STREAM_HANDLE *subscription) {
    struct UTUN_INSTANCE *inst; struct stream_descriptor d;
    if (stream_handle_context(subscription, &inst, &d) != STREAM_OK) return STREAM_ERR_STATE;
    struct stream_network *net = stream_network_get(inst);
    if (!net) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream restart without network"); return STREAM_ERR_STATE; }
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        p->busy++;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
            if (f->outgoing && same_stream(&f->descriptor, &d)) flow_close(f, STREAM_REASON_STOPPED, 1);
        p->busy--; peer_sweep(p);
    }
    return stream_local_restart(inst, &d);
}

int stream_receive_get_info(struct STREAM_RECEIVE *f, struct stream_receive_info *info) {
    if (!f || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive info invalid arguments"); return STREAM_ERR_ARGUMENT; }
    *info = f->info;
    info->metrics_valid = stream_metrics_poll(&f->metrics, get_time_tb() / 10, &info->metrics) == STREAM_OK;
    if (f->peer && f->peer->requested.active) for (unsigned i = 0; i < f->peer->requested.count; i++)
        if (f->peer->requested.cuts[i].selected && f->peer->requested.cuts[i].handle == f->info.handle) info->close_requested = 1;
    return STREAM_OK;
}

/* Добавление/вытеснение учитывает только ещё не переданные блоки действующей downstream-выдачи. */
void stream_network_cache_event(struct stream_network *net, const struct stream_descriptor *d, const uint8_t *wire, size_t length, int added) {
    if (!net || net->stopping) return;
    uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); uint8_t sub = wire[STREAM_BLOCK_SUB_OFF];
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            if (f->outgoing || f->info.state != STREAM_RECEIVE_ACTIVE || f->subblock != sub || seq < f->info.next
                || !same_stream(&f->descriptor, d)) continue;
            load_account(f, 1, length, added, 0);
            if (added) {
                if (seq > f->pending_highest) f->pending_highest = seq;
                stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
                                       wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
            }
        }
    }
}

/* Нужна скорость входящего вклада, а не урезанный D. При отсутствии source-дельты используем измеренные arrivals. */
static uint64_t flow_rate(struct STREAM_RECEIVE *f, uint64_t now) {
    struct stream_metric_snapshot metric = {0};
    stream_metrics_poll(&f->offered_metrics, now, &metric);
    uint64_t rate = metric.source_rate_valid && !metric.stopped ? metric.source_rate_bytes_sec : f->load.snapshot.input_rate;
    return rate ? rate : 1;
}

static struct STREAM_RECEIVE *flow_by_handle(struct stream_peer *p, uint64_t handle, int outgoing) {
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
        if (f->outgoing == outgoing && f->info.handle == handle && f->info.state == STREAM_RECEIVE_ACTIVE) return f;
    return NULL;
}

/* Rate фиксируется один раз. Высокие rates первыми определяют расчётное число закрытий, но получатель вправе выбрать другие handles. */
static int pressure_begin(struct stream_peer *p, uint64_t required, uint64_t now) {
    if (p->pressure.active || p->cooling) return 0;
    struct stream_pressure_request request = {0}; uint64_t total = 0;
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
        if (f->outgoing || f->info.state != STREAM_RECEIVE_ACTIVE) continue;
        struct stream_cut cut = {.handle = f->info.handle, .rate = flow_rate(f, now)};
        unsigned at = request.count++;
        while (at && request.cuts[at - 1].rate < cut.rate) { request.cuts[at] = request.cuts[at - 1]; at--; }
        request.cuts[at] = cut; total += cut.rate;
    }
    if (!request.count) return 0;
    if (p->pressure_id == UINT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream OVERLOAD id exhausted"); return 0; }
    request.required_rate = required && required < total ? required : total;
    uint64_t picked = 0;
    while (request.required_count < request.count && picked < request.required_rate) picked += request.cuts[request.required_count++].rate;
    request.id = ++p->pressure_id; request.deadline_ms = now + STREAM_LOAD_GRACE_MS; request.active = 1;
    p->pressure = request; p->pressure_retry_ms = 0; p->clear_since_ms = 0;
    DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream OVERLOAD begin peer=%llx id=%llu count=%u cut=%llu g=%llu A=%llu D=%llu X=%llu Q=%llu age=%llu global=%d",
               (unsigned long long)p->node_id, (unsigned long long)request.id, request.required_count,
               (unsigned long long)request.required_rate, (unsigned long long)p->load.snapshot.growth_rate,
               (unsigned long long)p->load.arrivals.bytes, (unsigned long long)p->load.sent.bytes,
               (unsigned long long)p->load.removed.bytes, (unsigned long long)p->load.queued.bytes,
               (unsigned long long)p->load.snapshot.oldest_age_ms, p->network->global_pressure);
    return 1;
}

/* Повтор запроса не меняет rate/срок. Header несёт один активный ключ, а список handles относится ко всем группам пира. */
static void pressure_send(struct stream_peer *p, uint64_t now) {
    if (!p->pressure.active || (p->pressure_retry_ms && now - p->pressure_retry_ms < 500)) return;
    struct STREAM_RECEIVE *first = NULL;
    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
        if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE) { first = f; break; }
    if (!first) return;
    uint8_t payload[STREAM_OVERLOAD_HEADER + STREAM_OVERLOAD_MAX * 16];
    stream_write_be(payload, p->pressure.id, 8); stream_write_be(payload + 8, p->pressure.required_rate, 8);
    payload[16] = (uint8_t)p->pressure.required_count; payload[17] = (uint8_t)p->pressure.count;
    for (unsigned i = 0; i < p->pressure.count; i++) {
        stream_write_be(payload + STREAM_OVERLOAD_HEADER + i * 16, p->pressure.cuts[i].handle, 8);
        stream_write_be(payload + STREAM_OVERLOAD_HEADER + i * 16 + 8, p->pressure.cuts[i].rate, 8);
    }
    struct stream_message m = {.descriptor = first->descriptor, .command = STREAM_CMD_OVERLOAD};
    if (control_put(p, m, payload, STREAM_OVERLOAD_HEADER + p->pressure.count * 16) == STREAM_OK) p->pressure_retry_ms = now;
}

/* Оценить возможность убрать один субпоток без потери k: учитываем дубликаты и остальных поставщиков. */
static int removal_cost(struct stream_peer *p, struct STREAM_RECEIVE *candidate) {
    uint8_t seen[RS_MAX_BLOCKS] = {0}; unsigned distinct = 0;
    for (struct stream_peer *other = p->network->peers; other; other = other->next) {
        for (struct STREAM_RECEIVE *f = other->flows; f; f = f->next) {
            if (!f->outgoing || f == candidate || f->info.state != STREAM_RECEIVE_ACTIVE
                || !same_stream(&f->descriptor, &candidate->descriptor)) continue;
            int selected = 0;
            if (other == p) for (unsigned i = 0; i < p->requested.count; i++)
                if (p->requested.cuts[i].selected && p->requested.cuts[i].handle == f->info.handle) selected = 1;
            if (!selected && !seen[f->subblock]) { seen[f->subblock] = 1; distinct++; }
        }
    }
    int parity = candidate->subblock >= candidate->descriptor.n - candidate->descriptor.r;
    return (distinct >= (unsigned)(candidate->descriptor.n - candidate->descriptor.r) ? 0 : 4) + (parity ? 0 : 1);
}

/* Выбор лишь предлагает source закрытия. Клиент не отсоединяет handles до CLOSED: поздний SELECT не режет лишние после fallback. */
static void pressure_select(struct stream_peer *p, uint64_t now) {
    struct stream_pressure_request *r = &p->requested;
    if (!r->active || r->sent || now >= r->deadline_ms) return;
    uint64_t reduction = 0; unsigned selected = 0;
    for (unsigned i = 0; i < r->count; i++) if (r->cuts[i].selected) { reduction += r->cuts[i].rate; selected++; }
    while (reduction < r->required_rate) {
        int best = -1, best_cost = 8;
        for (unsigned i = 0; i < r->count; i++) {
            struct STREAM_RECEIVE *f = flow_by_handle(p, r->cuts[i].handle, 1);
            if (r->cuts[i].selected || !f) continue;
            int cost = removal_cost(p, f);
            if (cost < best_cost || (cost == best_cost && (best < 0 || r->cuts[i].rate > r->cuts[best].rate))) {
                best = (int)i; best_cost = cost;
            }
        }
        if (best < 0) break;
        r->cuts[best].selected = 1; reduction += r->cuts[best].rate; selected++;
    }
    if (!selected) return;
    uint8_t payload[9 + STREAM_OVERLOAD_MAX * 8]; stream_write_be(payload, r->id, 8); payload[8] = (uint8_t)selected;
    struct STREAM_RECEIVE *first = NULL; unsigned at = 0;
    for (unsigned i = 0; i < r->count; i++) if (r->cuts[i].selected) {
        stream_write_be(payload + 9 + at++ * 8, r->cuts[i].handle, 8);
        if (!first) first = flow_by_handle(p, r->cuts[i].handle, 1);
    }
    if (!first) return;
    struct stream_message m = {.descriptor = first->descriptor, .command = STREAM_CMD_OVERLOAD_SELECT};
    if (control_put(p, m, payload, 9 + at * 8) == STREAM_OK) {
        r->sent = 1;
        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream OVERLOAD selected peer=%llx id=%llu handles=%u rate=%llu requested=%llu",
                   (unsigned long long)p->node_id, (unsigned long long)r->id, at,
                   (unsigned long long)reduction, (unsigned long long)r->required_rate);
    }
}

/* Проверяем весь список до действия: повторы handles и несуществующие элементы не создают частичных закрытий. */
static void pressure_receive(struct stream_peer *p, const struct stream_message *m) {
    uint64_t now = get_time_tb() / 10;
    if (m->command == STREAM_CMD_OVERLOAD) {
        if (m->length < STREAM_OVERLOAD_HEADER) goto malformed;
        unsigned count = m->payload[17]; uint64_t id = stream_read_be(m->payload, 8), required = stream_read_be(m->payload + 8, 8);
        if (!id || !required || !count || count > STREAM_OVERLOAD_MAX || !m->payload[16] || m->payload[16] > count
            || m->length != STREAM_OVERLOAD_HEADER + count * 16) goto malformed;
        if (id <= p->requested.id) return;
        struct stream_pressure_request request = {.id = id, .required_rate = required, .required_count = m->payload[16],
            .count = count, .deadline_ms = now + STREAM_LOAD_GRACE_MS, .active = 1};
        uint64_t total = 0;
        for (unsigned i = 0; i < count; i++) {
            request.cuts[i].handle = stream_read_be(m->payload + STREAM_OVERLOAD_HEADER + i * 16, 8);
            request.cuts[i].rate = stream_read_be(m->payload + STREAM_OVERLOAD_HEADER + i * 16 + 8, 8);
            if (!request.cuts[i].handle || !request.cuts[i].rate || request.cuts[i].rate > UINT64_MAX - total) goto malformed;
            for (unsigned j = 0; j < i; j++) if (request.cuts[j].handle == request.cuts[i].handle) goto malformed;
            total += request.cuts[i].rate;
        }
        if (required > total) goto malformed;
        p->requested = request; pressure_select(p, now); return;
    }
    if (m->length < 9) goto malformed;
    unsigned count = m->payload[8]; uint64_t id = stream_read_be(m->payload, 8);
    if (!id || !count || count > STREAM_OVERLOAD_MAX || m->length != 9 + count * 8) goto malformed;
    if (!p->pressure.active || id != p->pressure.id || now >= p->pressure.deadline_ms) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale OVERLOAD_SELECT peer=%llx id=%llu",
                    (unsigned long long)p->node_id, (unsigned long long)id); return;
    }
    for (unsigned i = 0; i < count; i++) {
        uint64_t handle = stream_read_be(m->payload + 9 + i * 8, 8); int found = 0;
        for (unsigned j = 0; j < p->pressure.count; j++) if (p->pressure.cuts[j].handle == handle) found = 1;
        for (unsigned j = 0; j < i; j++) if (stream_read_be(m->payload + 9 + j * 8, 8) == handle) goto malformed;
        if (!found) goto malformed;
    }
    for (unsigned i = 0; i < count && p->pressure.released_rate < p->pressure.required_rate; i++) {
        struct STREAM_RECEIVE *f = flow_by_handle(p, stream_read_be(m->payload + 9 + i * 8, 8), 0);
        if (f) flow_close(f, STREAM_REASON_OVERLOAD, 1);
    }
    return;
malformed:
    DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream malformed pressure command=%u peer=%llx bytes=%zu", m->command,
               (unsigned long long)p->node_id, m->length);
}

/* Просроченные ключи удаляются X; сохранённое кольцо не стирается и GAP обязательно предшествует дальнейшим DATA. */
static void expire_pending(struct STREAM_RECEIVE *f) {
    if (!f->gap_pending) f->gap_from = f->info.next;
    f->info.next = f->pending_highest + 1; f->gap_pending = 1;
    load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0);
    peer_schedule(f->peer);
    DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream pending expired peer=%llx handle=%llu first=%llu next=%llu",
               (unsigned long long)f->peer->node_id, (unsigned long long)f->info.handle,
               (unsigned long long)f->gap_from, (unsigned long long)f->info.next);
}

/* 100-мс обход не отправляет DATA и не заменяет waiter. Метрики фиксируются с фактическим периодом >=200мс. */
static void pressure_tick(struct stream_network *net, uint64_t now) {
    unsigned troubled = 0; uint64_t maximum_age = 0;
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        uint64_t oldest = 0;
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            uint64_t age = flow_age(f, now); if (age > oldest) oldest = age;
            struct stream_load_snapshot snapshot; stream_load_poll(&f->load, now, age, &snapshot);
        }
        struct stream_load_snapshot snapshot;
        int sampled = stream_load_poll(&p->load, now, oldest, &snapshot) == STREAM_OK;
        if (snapshot.pressure) troubled++;
        if (oldest > maximum_age) maximum_age = oldest;
        if (sampled && p->load.queued.bytes) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
            "stream load peer=%llx A=%llu D=%llu X=%llu Q=%llu age=%llu input=%llu output=%llu g=%llu interval=%llu confirm=%d",
            (unsigned long long)p->node_id, (unsigned long long)snapshot.arrivals.bytes, (unsigned long long)snapshot.sent.bytes,
            (unsigned long long)snapshot.removed.bytes, (unsigned long long)snapshot.queued.bytes, (unsigned long long)oldest,
            (unsigned long long)snapshot.input_rate, (unsigned long long)snapshot.output_rate,
            (unsigned long long)snapshot.growth_rate, (unsigned long long)snapshot.interval_ms, snapshot.confirmed);
    }
    struct stream_load_snapshot global; stream_load_poll(&net->load, now, maximum_age, &global);
    if (net->callback_late_ms >= 200 || (net->cpu.valid && net->cpu.average_percent >= 90)) net->local_busy_until_ms = now + 1000;
    int global_condition = troubled >= 2 && global.pressure && now < net->local_busy_until_ms;
    if (!global_condition) net->global_candidate = 0;
    else if (!net->global_candidate) { net->global_candidate = 1; net->global_since_ms = now; }
    net->global_pressure = net->global_candidate && now - net->global_since_ms >= STREAM_LOAD_CONFIRM_MS;
    if (net->global_pressure) {
        uint64_t required = global.growth_rate ? global.growth_rate + global.growth_rate / 100 + 1 : global.queued.bytes / 2 + 1;
        for (struct stream_peer *p = net->peers; p && required; p = p->next) if (p->pressure.active) {
            uint64_t pending = p->pressure.required_rate > p->pressure.released_rate
                               ? p->pressure.required_rate - p->pressure.released_rate : 0;
            required = pending >= required ? 0 : required - pending;
        }
        for (unsigned pass = 0; pass < net->peer_count && required; pass++) {
            struct stream_peer *best = NULL; uint64_t best_rate = 0;
            for (struct stream_peer *p = net->peers; p; p = p->next) {
                if (p->pressure.active || p->cooling) continue;
                uint64_t total = 0;
                for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
                    if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE) total += flow_rate(f, now);
                if (total > best_rate) { best = p; best_rate = total; }
            }
            if (!best) break;
            uint64_t cut = required < best_rate ? required : best_rate;
            if (!pressure_begin(best, cut, now)) break;
            required -= cut;
        }
    }
    for (struct stream_peer *p = net->peers; p; p = p->next) {
        p->busy++;
        if (!net->global_pressure && p->load.snapshot.confirmed && !p->pressure.active && !p->cooling) {
            uint64_t g = p->load.snapshot.growth_rate;
            pressure_begin(p, g ? g + g / 100 + 1 : p->load.queued.bytes / 2 + 1, now);
        }
        if (p->pressure.active && now >= p->pressure.deadline_ms) {
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream OVERLOAD deadline peer=%llx id=%llu released=%llu required=%llu",
                       (unsigned long long)p->node_id, (unsigned long long)p->pressure.id,
                       (unsigned long long)p->pressure.released_rate, (unsigned long long)p->pressure.required_rate);
            for (unsigned i = 0; i < p->pressure.count && p->pressure.released_rate < p->pressure.required_rate; i++) {
                struct STREAM_RECEIVE *f = flow_by_handle(p, p->pressure.cuts[i].handle, 0);
                if (f) flow_close(f, STREAM_REASON_OVERLOAD, 1);
            }
        }
        if (p->pressure.active && (p->pressure.released_rate >= p->pressure.required_rate || now >= p->pressure.deadline_ms)) {
            p->pressure.active = 0; p->cooling = 1;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream OVERLOAD complete peer=%llx id=%llu closed=%u rate=%llu requested=%llu",
                       (unsigned long long)p->node_id, (unsigned long long)p->pressure.id, p->pressure.released_count,
                       (unsigned long long)p->pressure.released_rate, (unsigned long long)p->pressure.required_rate);
        }
        pressure_send(p, now); pressure_select(p, now);
        if (p->requested.active && now >= p->requested.deadline_ms) p->requested.active = 0;
        if (p->cooling) {
            if (p->load.snapshot.pressure) p->clear_since_ms = 0;
            else if (!p->clear_since_ms) p->clear_since_ms = now;
            else if (now - p->clear_since_ms >= STREAM_LOAD_CONFIRM_MS) { p->cooling = 0; p->clear_since_ms = 0; }
        }
        for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
            if (f->account_failed) flow_close(f, STREAM_REASON_PROTOCOL, 1);
            if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && p->pressure.active
                && flow_age(f, now) > STREAM_LOAD_MAX_AGE_MS) expire_pending(f);
        }
        p->busy--; peer_sweep(p);
    }
}

int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct stream_peer_load_info *info) {
    struct stream_network *net = stream_network_get(inst); struct stream_peer *p = net ? peer_find(net, node) : NULL;
    if (!p || !info) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream load unavailable peer=%llx", (unsigned long long)node); return STREAM_ERR_STATE; }
    struct stream_peer_load_info result = {.load = p->load.snapshot, .request_id = p->pressure.id,
        .required_rate = p->pressure.required_rate, .released_rate = p->pressure.released_rate,
        .deadline_ms = p->pressure.deadline_ms, .required_count = p->pressure.required_count, .released_count = p->pressure.released_count,
        .request_active = p->pressure.active, .cpu_percent = net->cpu.valid ? net->cpu.average_percent : 0,
        .callback_late_ms = net->callback_late_ms, .global_pressure = net->global_pressure,
        .accepting = !p->pressure.active && !p->cooling};
    result.load.queued = p->load.queued; result.load.arrivals = p->load.arrivals; result.load.sent = p->load.sent;
    result.load.removed = p->load.removed; result.load.rejected = p->load.rejected;
    struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
    if (conn) {
        struct ll_queue *queues[] = {conn->send_input_q, conn->input_queue, conn->input_send_q};
        for (unsigned i = 0; i < 3; i++) if (queues[i]) {
            result.transport_queued_packets += queues[i]->count; result.transport_queued_bytes += queues[i]->total_bytes;
        }
        result.inflight_bytes = conn->unacked_bytes;
    }
    *info = result; return STREAM_OK;
}