From c2179b3959489dcdfc89bc4cf7ab03c9a2644f38 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:42:00 +0200 Subject: [PATCH] Implement NCD stream issuances, ring cursors and routed reverse --- src/Makefile.am | 2 + src/stream/stream_network.c | 705 ++++++++++++++++++++++++++++++++++++ 2 files changed, 707 insertions(+) create mode 100644 src/stream/stream_network.c diff --git a/src/Makefile.am b/src/Makefile.am index 0a8cab17..e78f10bc 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -9,6 +9,8 @@ stream_core_files = \ stream/stream_priv.h \ stream/stream_wire.h \ stream/stream_wire.c \ + stream/stream_network.h \ + stream/stream_network.c \ stream/stream_member.h \ stream/stream_member.c \ stream/stream_proto.h \ diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c new file mode 100644 index 00000000..10f9dd56 --- /dev/null +++ b/src/stream/stream_network.c @@ -0,0 +1,705 @@ +#include + +#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_INFO_SIZE + 1 + STREAM_PATH_MAX * 8) +#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_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_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_network { + struct UTUN_INSTANCE *inst; + struct stream_peer *peers; + void *timer; + unsigned peer_count, flow_count; + int stopping; +}; + +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 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); return STREAM_ERR_STATE; + } + struct ll_entry *e = ll_alloc_lldgram((uint16_t)(STREAM_WIRE_HEADER + length)); + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); 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 (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 (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; + 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); + 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) 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(); + 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->info.peer = p->node_id; 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; + if (stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual) != STREAM_OK + || stream_buffer_range(local->buffer, sub, &first, &last) != 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, actual, 8); stream_write_be(payload + 8, last, 8); + payload[16] = local->descriptor.n; payload[17] = local->descriptor.r; + stream_write_be(payload + 18, local->descriptor.block_size, 2); + /* byte 20 — число узлов, следует непосредственно за параметрами. */ + stream_path_write(payload + 20, &path); return 0; +} + +static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self, uint64_t *start) { + struct stream_path path; size_t used; + if (m->length < STREAM_INFO_SIZE || m->payload[16] != d->n || m->payload[17] != d->r + || stream_read_be(m->payload + 18, 2) != d->block_size + || stream_path_parse(m->payload + 20, m->length - 20, &path, &used) != STREAM_OK || m->length != 20 + used + || path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK + || stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 8, 8)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length); + return STREAM_ERR_FORMAT; + } + *start = stream_read_be(m->payload, 8); return STREAM_OK; +} + +/* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение 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; +} + +/* Производитель не хранит готовые 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}; + const uint8_t *wire; size_t length; uint64_t actual; + int result = 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) { f->info.next = actual + 1; f->info.blocks++; } + return 1; + } + if (result == STREAM_EXPIRED || result == STREAM_GAP) { + uint64_t next; + if (stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK && 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) { + m.command = STREAM_CMD_END; + if (raw_send(p, m, local.end, STREAM_END_SIZE) == STREAM_OK) { + f->info.state = STREAM_RECEIVE_ENDED; 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; + if (control_put(p, m, NULL, 0) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; } + } + if (!f->info_ready || !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) 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 (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->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]; + 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->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); + f->info.state = STREAM_RECEIVE_ACTIVE; + } + } + } 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 = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload); + if (!reason) stream_write_be(payload, f->info.start, 8); + } + } + if (reason) { + uint8_t value = (uint8_t)reason; m.command = STREAM_CMD_REJECT; control_put(p, m, &value, 1); + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u reason=%d watermark=%llu", + (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, reason, (unsigned long long)p->highwater); + return; + } + f->deadline = get_time_tb() + STREAM_LEASE_TB; + m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8); + 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 (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 || !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 || m.subblock >= local.descriptor.n || !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]; + 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 value = (uint8_t)reason; control_put(p, m, &value, 1); } + else control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8); + return; + } + if (!p || !p->up || m.receiver_epoch != p->own_epoch + || (p->other_epoch && m.sender_epoch != p->other_epoch)) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session packet peer=%llx command=%u handle=%llu", + (unsigned long long)sender, m.command, (unsigned long long)m.handle); 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 (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) == STREAM_OK) { + p->other_epoch = m.sender_epoch; f->info_ready = 1; peer_schedule(p); + } 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, &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) { + f->info.next = seq + 1; f->info.blocks++; 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) { + if (stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED; + 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; + else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1); + else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.length == 1) 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); + 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); +} + +/* Один таймер обслуживает сроки/продления и ограниченные reverse-попытки; DATA не опрашивается таймером. */ +static void network_tick(void *arg) { + struct stream_network *net = arg; net->timer = NULL; + uint64_t now = get_time_tb(); + struct stream_peer *p = net->peers; + while (p) { + struct stream_peer *next = p->next; + p->busy++; + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { + if (f->info.state == STREAM_RECEIVE_CLOSED) continue; + if ((!f->outgoing || f->info.state < STREAM_RECEIVE_ACTIVE) && now >= f->deadline) + flow_close(f, f->info.state < STREAM_RECEIVE_ACTIVE ? 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; + } + } + if (!p->up && p->flows && now - p->born < STREAM_CONNECT_TB && (!p->reverse_at || now - p->reverse_at >= 10000)) { + struct stream_message m = {.descriptor = p->flows->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; + } + 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; + 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); + } +} + +int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, 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) { + 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.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; *out = f; peer_schedule(p); 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); +} + +int stream_receive_get_info(const 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; return STREAM_OK; +}