diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index af00d8fb..0d28b0d7 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -7,6 +7,7 @@ #include "../transport_layer/etcp.h" #include "../transport_layer/node_conn_direct.h" #include "../routing_layer/route_crypto.h" +#include "../routing_layer/topo_node_sqlite.h" #include "stream_member.h" #include "stream_priv.h" @@ -19,6 +20,27 @@ #define STREAM_LEASE_TB 60000 #define STREAM_CONNECT_TB 50000 #define STREAM_TICK_TB 1000 +#define STREAM_RELAY_LIMIT 8 +#define STREAM_RELAY_REQUEST_LIMIT 128 +#define STREAM_RELAY_START_MS 15000 +#define STREAM_RELAY_HANDOFF_MS 5000 +#define STREAM_RELAY_IDLE_MS 3000 +#define STREAM_RELAY_RETAIN_MS 60000 + +/* Запросы не владеют буфером: один внутренний handle обслуживает всех получателей точной эпохи. */ +struct stream_relay_request { + struct stream_relay_request *next; + struct stream_descriptor descriptor; + uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq; + int outgoing, status, reason, dirty; + uint8_t end[STREAM_END_SIZE]; +}; +struct stream_relay { + struct stream_relay *next; + struct stream_descriptor descriptor; + struct STREAM_HANDLE *owner; + uint64_t idle_ms; +}; enum stream_peer_reset_flags { STREAM_RESET_ROTATE = 1, STREAM_RESET_KEEP_PENDING = 2 }; @@ -71,6 +93,10 @@ struct stream_network { int global_pressure; uint64_t local_busy_until_ms, global_since_ms; int global_candidate; + struct stream_relay_request *relay_requests; + struct stream_relay *relays; + uint64_t relay_epoch; + unsigned relay_count, relay_request_count; }; static void peer_pump(void *arg); @@ -79,6 +105,8 @@ 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); +static void relay_tick(struct stream_network *net, uint64_t now); +static void relay_dispatch(struct stream_network *net, uint64_t sender, const struct stream_message *message); /* Ключи очереди — cursor+кольцо, её возраст учитывает реальное время вставки и момент появления выдачи. */ static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) { @@ -169,22 +197,22 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin } /* 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)) { +static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) { + struct UTUN_INSTANCE *inst = net->inst; + if (!etcp_router_has_route(inst, m.descriptor.group_id, peer) + || !etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, 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; + (unsigned long long)peer, 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; + e->dgram[0] = ETCP_RT_ID_STREAM; 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); + int result = etcp_route_send(inst, m.descriptor.group_id, peer, 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); + (unsigned long long)peer, m.command, result); return result < 0 ? STREAM_ERR_STATE : STREAM_OK; } @@ -207,9 +235,10 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) { 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}; + .command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED, + .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; if (p->up) control_put(p, m, &payload, 1); - else if (!f->outgoing) route_control(p, m, &payload, 1); + else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1); } f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason; if (reason != STREAM_REASON_STOPPED) @@ -236,7 +265,10 @@ static void peer_sweep(struct stream_peer *p) { p->network->flow_count--; f->peer = NULL; f->next = NULL; if (!f->owned) u_free(f); } - if (p->flows) p->idle = 0; + int relay_pending = 0; + for (struct stream_relay_request *r = p->network->relay_requests; r; r = r->next) + if (r->peer == p->node_id && r->status <= STREAM_RELAY_READY) relay_pending = 1; + if (p->flows || relay_pending) p->idle = 0; else if (!p->idle) p->idle = get_time_tb(); } @@ -282,8 +314,9 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v 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); + .subblock = f->subblock, .command = STREAM_CMD_CLOSED, + .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; + route_control(p->network, p->node_id, m, &reason, 1); } peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_ROTATE, NULL); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id); @@ -701,6 +734,12 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st 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_RELAY_REQUEST) { + if (routed) relay_dispatch(net, sender, &m); + else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u", + (unsigned long long)sender, m.command); + 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) { @@ -921,7 +960,7 @@ static void network_tick(void *arg) { && 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) { + 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; @@ -931,10 +970,9 @@ static void network_tick(void *arg) { 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; + struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ, + .sender_epoch = p->own_epoch}; + if (route_control(net, p->node_id, m, NULL, 0) == STREAM_OK) p->reverse_at = now; } p->busy--; peer_sweep(p); if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p); @@ -942,6 +980,7 @@ static void network_tick(void *arg) { } pressure_tick(net, now_ms); stream_optimizers_tick(net->inst, now_ms); + relay_tick(net, 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"); @@ -954,7 +993,8 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) } 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; + net->inst = inst; net->relay_epoch = new_epoch(); + if (!net->relay_epoch) { u_free(net); return STREAM_ERR_CRYPTO; } 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; @@ -976,6 +1016,13 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) 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); + while (net->relays) { + struct stream_relay *relay = net->relays; net->relays = relay->next; + stream_close(relay->owner); u_free(relay); + } + while (net->relay_requests) { + struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next; u_free(r); + } 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); @@ -989,6 +1036,11 @@ void stream_network_changed(struct stream_network *net) { /* Кольцо освобождается только после отмены сетевых курсоров данной точной эпохи. */ void stream_network_cancel(struct stream_network *net, const struct stream_descriptor *d) { if (!net) return; + for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) + if (same_stream(&r->descriptor, d) && r->status < STREAM_RELAY_REJECTED) { + r->status = STREAM_RELAY_CANCELLED; r->reason = STREAM_REASON_STOPPED; r->dirty = 1; + r->retain_ms = get_time_tb() / 10 + STREAM_RELAY_RETAIN_MS; + } for (struct stream_peer *p = net->peers; p; p = p->next) { p->busy++; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) @@ -1026,7 +1078,8 @@ static int receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8 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)) { + || (!stream_member_has_subscription(&member, d.source_node_id, d.source_epoch) + && !stream_relay_confirmed(inst, &d, node))) { 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); @@ -1105,6 +1158,296 @@ int stream_receive_get_info(struct STREAM_RECEIVE *f, struct stream_receive_info return STREAM_OK; } +/* Терминальные записи остаются на минуту: повтор REQUEST/CANCEL не создаёт новое владение. */ +static struct stream_relay_request *relay_request_find(struct stream_network *net, const struct stream_descriptor *key, + uint64_t peer, uint64_t id, int outgoing) { + for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) + if (r->outgoing == outgoing && r->peer == peer && same_stream(&r->descriptor, key) && (!id || r->id == id)) return r; + return NULL; +} + +/* Лимит включает tombstones и проверяется до выделения; исчерпание не вытесняет защиту от повторов. */ +static struct stream_relay_request *relay_request_new(struct stream_network *net, const struct stream_descriptor *key, + uint64_t peer, uint64_t id, uint64_t epoch, int outgoing) { + if (net->relay_request_count >= STREAM_RELAY_REQUEST_LIMIT) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request limit peer=%llx count=%u", (unsigned long long)peer, + net->relay_request_count); return NULL; + } + struct stream_relay_request *r = u_calloc(1, sizeof(*r)); + if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay request allocation failed"); return NULL; } + r->descriptor = *key; r->peer = peer; r->id = id; r->epoch = epoch; r->outgoing = outgoing; + r->status = STREAM_RELAY_STARTING; r->dirty = 1; r->deadline_ms = get_time_tb() / 10 + STREAM_RELAY_START_MS; + r->next = net->relay_requests; net->relay_requests = r; net->relay_request_count++; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request opened group=%llx source=%llx epoch=%llx peer=%llx id=%llx out=%d deadline=%llu", + (unsigned long long)key->group_id, (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, + (unsigned long long)peer, (unsigned long long)id, outgoing, (unsigned long long)r->deadline_ms); + return r; +} + +/* Наличие транспортных handles/INFO/ENDED не является действующей раздачей. */ +static unsigned relay_active(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing) { + unsigned count = 0; + for (struct stream_peer *p = net->peers; p; p = p->next) { + if (peer && p->node_id != peer) continue; + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) + if (f->outgoing == outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE + && same_stream(&f->descriptor, key)) count++; + } + return count; +} + +/* Неуспешная помощь закрывает только выдачи своего пира и трансляции, не общий ETCP. */ +static void relay_finish(struct stream_network *net, struct stream_relay_request *r, int status, int reason, uint64_t now) { + if (r->status >= STREAM_RELAY_REJECTED) return; + r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1; + if (r->outgoing) { + struct stream_peer *p = peer_find(net, r->peer); + if (p) { + p->busy++; + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) + if (f->outgoing && same_stream(&f->descriptor, &r->descriptor)) flow_close(f, reason, 1); + p->busy--; peer_sweep(p); + } + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request finished source=%llx epoch=%llx peer=%llx id=%llx out=%d status=%d reason=%d", + (unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->descriptor.source_epoch, + (unsigned long long)r->peer, (unsigned long long)r->id, r->outgoing, status, reason); +} + +/* Одна подписка/кольцо на точную эпоху, независимо от числа ожидающих и действующих получателей. */ +static int relay_open(struct stream_network *net, const struct stream_descriptor *key) { + for (struct stream_relay *relay = net->relays; relay; relay = relay->next) + if (same_stream(&relay->descriptor, key)) return stream_handle_active(relay->owner) ? STREAM_OK : STREAM_ERR_STATE; + if (net->relay_count >= STREAM_RELAY_LIMIT) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay stream limit count=%u limit=%u", net->relay_count, STREAM_RELAY_LIMIT); + return STREAM_ERR_STATE; + } + struct stream_relay *relay = u_calloc(1, sizeof(*relay)); + if (!relay) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay context allocation failed"); return STREAM_ERR_MEMORY; } + int result = stream_relay_open(net->inst, key, &relay->owner); + if (result != STREAM_OK) { u_free(relay); return result; } + relay->descriptor = *key; relay->next = net->relays; net->relays = relay; net->relay_count++; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay receiving started group=%llx source=%llx epoch=%llx streams=%u limit=%u", + (unsigned long long)key->group_id, (unsigned long long)key->source_node_id, + (unsigned long long)key->source_epoch, net->relay_count, STREAM_RELAY_LIMIT); + return STREAM_OK; +} + +/* READY ускоряет поиск только в подтверждённом запросе; membership и DATA-подпись остаются обязательными. */ +int stream_relay_confirmed(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer) { + struct stream_network *net = stream_network_get(inst); + if (!net || net->stopping) return 0; + struct stream_relay_request *r = relay_request_find(net, key, peer, 0, 1); + return r && (r->status == STREAM_RELAY_BOUND + || (r->status == STREAM_RELAY_READY && get_time_tb() / 10 < r->deadline_ms)); +} + +/* Автоматическая помощь выбирает один доступный канальный суперузел; неудачные попытки имеют cooldown. */ +void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, int needed, uint64_t now) { + struct stream_network *net = stream_network_get(inst); struct stream_local local; + if (!net || net->stopping || stream_local_get(inst, key, &local) != STREAM_OK) return; + for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) { + if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue; + unsigned active = relay_active(net, key, r->peer, 1); + int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last)); + if (local.relay_only || drained || (!needed && !active) + || (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS)) + relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); + else return; + } + if (!needed || local.relay_only || local.finished) return; + struct TOPO_GROUP *group = topo_groups_find(inst->topo_groups, key->group_id); + struct topo_connect_peer *peers = NULL; size_t count = 0; + if (!group || topo_node_sqlite_get_connect_peers(inst->topo_sqlite_db, group->channel_id, &peers, &count) < 0) return; + uint64_t chosen = 0; + for (size_t i = 0; i < count; i++) { + uint64_t peer = peers[i].node_id; + if (!peers[i].supernode || peer == inst->node_id || peer == key->source_node_id + || !stream_member_online(inst, key->group_id, peer) || !node_conn_direct_compatible(inst, peer)) continue; + struct stream_relay_request *previous = relay_request_find(net, key, peer, 0, 1); + if (previous) continue; + struct stream_member_state member; + if (stream_member_load(inst, key->group_id, peer, &member, NULL) != STREAM_OK) continue; + chosen = peer; break; + } + u_free(peers); + if (!chosen) return; + uint64_t id = new_epoch(), epoch = new_epoch(); + if (!id || !epoch) return; + struct stream_relay_request *r = relay_request_new(net, key, chosen, id, epoch, 1); + if (r && !peer_open(net, chosen)) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_TRANSPORT, now); +} + +/* Только end-to-end signed router-вход; команды до local_get авторизуются отдельно по канальному членству. */ +static void relay_dispatch(struct stream_network *net, uint64_t sender, const struct stream_message *m) { + uint64_t now = get_time_tb() / 10; + if (!m->handle || m->subblock || (m->command != STREAM_CMD_RELAY_STATUS && m->length)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay malformed control peer=%llx command=%u bytes=%zu handle=%llx", + (unsigned long long)sender, m->command, m->length, (unsigned long long)m->handle); return; + } + struct stream_member_state member; + if (stream_member_load(net->inst, m->descriptor.group_id, sender, &member, NULL) != STREAM_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay control unauthorized peer=%llx group=%llx", (unsigned long long)sender, + (unsigned long long)m->descriptor.group_id); return; + } + if (m->command == STREAM_CMD_RELAY_STATUS) { + struct stream_relay_request *r = relay_request_find(net, &m->descriptor, sender, m->handle, 1); + if (!r || r->status >= STREAM_RELAY_REJECTED || m->receiver_epoch != r->epoch + || (r->server_epoch && r->server_epoch != m->sender_epoch)) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay stale status peer=%llx id=%llx", (unsigned long long)sender, + (unsigned long long)m->handle); return; + } + if (m->length < 6 || m->payload[0] < STREAM_RELAY_STARTING || m->payload[0] > STREAM_RELAY_CANCELLED + || m->length != (size_t)(6 + (m->payload[0] == STREAM_RELAY_ENDED ? STREAM_END_SIZE : 0))) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay invalid status peer=%llx bytes=%zu", (unsigned long long)sender, m->length); return; + } + if (stream_member_supernode(net->inst, m->descriptor.group_id, sender) != 1) { + relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_MEMBER, now); return; + } + r->server_epoch = m->sender_epoch; + unsigned status = m->payload[0]; uint64_t remaining = stream_read_be(m->payload + 2, 4); + if (status == STREAM_RELAY_READY || status == STREAM_RELAY_BOUND) { + if (!remaining && status == STREAM_RELAY_READY) return; + if (remaining > STREAM_RELAY_HANDOFF_MS) remaining = STREAM_RELAY_HANDOFF_MS; + if (r->status == STREAM_RELAY_STARTING) { + r->status = STREAM_RELAY_READY; r->deadline_ms = now + remaining; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay ready source=%llx peer=%llx id=%llx handoff_ms=%llu", + (unsigned long long)r->descriptor.source_node_id, (unsigned long long)sender, + (unsigned long long)r->id, (unsigned long long)remaining); + stream_relay_ready(net->inst, &r->descriptor, sender); + } + } else if (status == STREAM_RELAY_ENDED) { + if (stream_read_be(m->payload + 6 + 34, 8) != UINT64_MAX + || stream_ingest_end(net->inst, &r->descriptor, m->payload + 6, STREAM_END_SIZE) != STREAM_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay empty END verification failed peer=%llx", (unsigned long long)sender); + relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_PROTOCOL, now); + } else relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); + } else if (status >= STREAM_RELAY_REJECTED) { + relay_finish(net, r, STREAM_RELAY_CANCELLED, m->payload[1], now); + } + return; + } + if (m->receiver_epoch && m->receiver_epoch != net->relay_epoch) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay stale server generation peer=%llx", (unsigned long long)sender); return; + } + struct stream_relay_request *r = relay_request_find(net, &m->descriptor, sender, m->handle, 0); + if (r && r->epoch != m->sender_epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request nonce conflicts peer=%llx id=%llx", (unsigned long long)sender, + (unsigned long long)m->handle); return; + } + if (!r) r = relay_request_new(net, &m->descriptor, sender, m->handle, m->sender_epoch, 0); + if (!r) return; + if (m->command == STREAM_CMD_RELAY_CANCEL) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); + else { + r->dirty = 1; + if (r->status >= STREAM_RELAY_REJECTED) return; + if (!stream_member_online(net->inst, m->descriptor.group_id, sender) + || stream_member_supernode(net->inst, m->descriptor.group_id, net->inst->node_id) != 1) { + relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, now); return; + } + if (net->global_pressure) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_OVERLOAD, now); return; } + for (size_t i = 0; i < member.count; i++) + if (member.subscriptions[i].source_node_id == m->descriptor.source_node_id + && member.subscriptions[i].source_epoch == m->descriptor.source_epoch && member.subscriptions[i].relay_only) { + relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_LOOP, now); return; + } + if (!peer_open(net, sender)) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); return; } + int result = relay_open(net, &m->descriptor); + if (result != STREAM_OK && result != STREAM_WAIT) + relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now); + } +} + +/* Один network tick обслуживает bootstrap, передачу владения выдачам, отмену и остановку пустых ретрансляторов. */ +static void relay_tick(struct stream_network *net, uint64_t now) { + for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) { + if (r->status < STREAM_RELAY_REJECTED) { + if (now >= r->checked_ms) { + r->checked_ms = now + 1000; + struct stream_member_state member; + int valid = stream_member_online(net->inst, r->descriptor.group_id, r->peer) + && stream_member_load(net->inst, r->descriptor.group_id, r->peer, &member, NULL) == STREAM_OK + && stream_member_supernode(net->inst, r->descriptor.group_id, r->outgoing ? r->peer : net->inst->node_id) == 1; + if (!valid) relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, now); + } + unsigned active = relay_active(net, &r->descriptor, r->peer, r->outgoing); + if (r->status < STREAM_RELAY_REJECTED && active) { + if (r->status != STREAM_RELAY_BOUND) { + r->status = STREAM_RELAY_BOUND; r->dirty = 1; r->progress_ms = now; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request bound to issuance source=%llx peer=%llx out=%d flows=%u", + (unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->peer, r->outgoing, active); + } + struct stream_local local; + if (r->outgoing && stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK + && r->progress_seq != local.next) { r->progress_seq = local.next; r->progress_ms = now; } + } else if (r->status == STREAM_RELAY_BOUND) { + relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); + } else if (r->status < STREAM_RELAY_REJECTED && now >= r->deadline_ms) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay attempt expired source=%llx peer=%llx id=%llx status=%d", + (unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->peer, + (unsigned long long)r->id, r->status); + relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); + } + if (!r->outgoing && r->status == STREAM_RELAY_STARTING) { + struct stream_local local; + if (stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK) { + if (local.finished && local.last == UINT64_MAX) { + memcpy(r->end, local.end, STREAM_END_SIZE); + relay_finish(net, r, STREAM_RELAY_ENDED, STREAM_REASON_STOPPED, now); + } else { + uint8_t info[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0}; + int available = 0; + for (unsigned sub = 0; sub < local.descriptor.n && !available; sub++) + available = info_payload(&local, (uint8_t)sub, UINT64_MAX, r->peer, net->inst->node_id, info) == 0; + if (available) { + r->status = STREAM_RELAY_READY; r->deadline_ms = now + STREAM_RELAY_HANDOFF_MS; r->dirty = 1; + } + } + } + } + } + if (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000)) continue; + if (r->sent_ms && now - r->sent_ms < 1000) continue; + struct stream_message m = {.descriptor = r->descriptor, .handle = r->id, + .sender_epoch = r->outgoing ? r->epoch : net->relay_epoch, + .receiver_epoch = r->outgoing ? r->server_epoch : r->epoch}; + uint8_t payload[6 + STREAM_END_SIZE]; size_t length = 0; + if (r->outgoing) m.command = r->status >= STREAM_RELAY_REJECTED ? STREAM_CMD_RELAY_CANCEL : STREAM_CMD_RELAY_REQUEST; + else { + m.command = STREAM_CMD_RELAY_STATUS; payload[0] = (uint8_t)r->status; payload[1] = (uint8_t)r->reason; + uint64_t remaining = r->deadline_ms > now ? r->deadline_ms - now : 0; + stream_write_be(payload + 2, remaining, 4); length = 6; + if (r->status == STREAM_RELAY_ENDED) { memcpy(payload + 6, r->end, STREAM_END_SIZE); length += STREAM_END_SIZE; } + } + r->sent_ms = now; + if (route_control(net, r->peer, m, payload, length) == STREAM_OK) r->dirty = 0; + } + struct stream_relay **link = &net->relays; + while (*link) { + struct stream_relay *relay = *link; + unsigned pending = 0; + for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) + if (!r->outgoing && same_stream(&r->descriptor, &relay->descriptor) && r->status <= STREAM_RELAY_READY) pending++; + unsigned active = relay_active(net, &relay->descriptor, 0, 0); + int valid = stream_handle_active(relay->owner) + && stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1; + if (valid && (pending || active)) { relay->idle_ms = 0; link = &relay->next; continue; } + if (valid && !relay->idle_ms) relay->idle_ms = now; + if (valid && now - relay->idle_ms < STREAM_RELAY_IDLE_MS) { link = &relay->next; continue; } + *link = relay->next; net->relay_count--; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay receiving stopped source=%llx epoch=%llx pending=%u downstream=%u valid=%d idle_ms=%llu", + (unsigned long long)relay->descriptor.source_node_id, (unsigned long long)relay->descriptor.source_epoch, + pending, active, valid, (unsigned long long)(now - relay->idle_ms)); + stream_close(relay->owner); u_free(relay); + } + struct stream_relay_request **request_link = &net->relay_requests; + while (*request_link) { + struct stream_relay_request *r = *request_link; + if (r->status < STREAM_RELAY_REJECTED || now < r->retain_ms) { request_link = &r->next; continue; } + *request_link = r->next; net->relay_request_count--; u_free(r); + } +} + /* Добавление/вытеснение учитывает только ещё не переданные блоки действующей 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; diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 3a82e0de..fa4f196c 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -29,7 +29,7 @@ struct stream_supplier { struct STREAM_RECEIVE *probe; struct stream_offer *offers; unsigned probe_sub, failures; - int online, compatible, source; + int online, compatible, source, relay; }; struct stream_selection { struct STREAM_RECEIVE *current, *replacement; @@ -44,7 +44,7 @@ struct stream_optimizer { struct stream_selection_settings settings; struct stream_selection_info info; struct stream_probe_schedule schedule; - uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms; + uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms; int progress_valid, draining; }; @@ -78,6 +78,7 @@ static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, u s->node = candidates[i].node_id; s->next = o->suppliers; o->suppliers = s; } s->online = 1; s->source = candidates[i].source; + s->relay = stream_relay_confirmed(o->inst, &o->descriptor, s->node); s->compatible = node_conn_direct_compatible(o->inst, s->node); } u_free(candidates); @@ -165,7 +166,8 @@ static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, struct stream_supplier *chosen = NULL; for (struct stream_supplier *s = o->suppliers; s; s = s->next) { if (!s->compatible || s->cycle == o->schedule.cycle || s->probe || now < s->retry_ms) continue; - if (!chosen || s->source > chosen->source || (s->source == chosen->source && s->node < chosen->node)) chosen = s; + if (!chosen || s->relay > chosen->relay || (s->relay == chosen->relay && (s->source > chosen->source + || (s->source == chosen->source && s->node < chosen->node)))) chosen = s; } if (!chosen) break; probe_start(o, h, chosen, now); @@ -299,12 +301,25 @@ int stream_optimizer_configure(struct stream_optimizer *o, const struct stream_s void stream_optimizer_info(struct stream_optimizer *o, struct stream_selection_info *info) { *info = o->info; } +/* READY отменяет ожидание старого цикла/ошибки только для подтверждённого поставщика. */ +void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) { + o->roster_ms = o->schedule.next_ms = 0; + struct stream_supplier *s = supplier_find(o, peer); + if (s) { + s->cycle = s->retry_ms = 0; s->failures = 0; + if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) s->offers[sub].retry_ms = 0; + } + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer relay wake source=%llx peer=%llx", + (unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer); +} + void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) { struct stream_local local; if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return; if (local.finished) { /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */ if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) { + stream_relay_demand(o->inst, &o->descriptor, 0, now); for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; } if (o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer END drained source=%llx next=%llu last=%llu", (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last); @@ -441,6 +456,31 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, if (o->selected[sub].replacement) o->info.overlaps++; } if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0; + /* Помощь нужна для получения k субпотоков, а не для идеальной защиты n/r; пауза источника не является отказом пути. */ + unsigned active = 0, viable = 0; int newer = 0; + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct STREAM_RECEIVE *current = o->selected[sub].current; + if (!current) continue; + struct stream_receive_info info; stream_receive_get_info(current, &info); + if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested) continue; + active++; + if ((info.blocks && info.metrics_valid && !info.metrics.stopped) || (!info.blocks && info.last_available == UINT64_MAX)) viable++; + } + for (struct stream_supplier *s = o->suppliers; s; s = s->next) + if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct stream_receive_info *offer = &s->offers[sub].info; + if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS + && offer->last_available != UINT64_MAX && offer->last_available > local.next) newer = 1; + } + unsigned required = (unsigned)(o->descriptor.n - o->descriptor.r); + int deficient = active < required || (viable < required && newer); + if (!deficient) o->deficient_ms = 0; + else if (!o->deficient_ms) o->deficient_ms = now; + if (now >= o->relay_poll_ms) { + o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS; + int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS; + stream_relay_demand(o->inst, &o->descriptor, needed, now); + } /* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */ if (!local.anchored && now >= o->diagnostic_ms) { o->diagnostic_ms = now + STREAM_PROBE_PERIOD_MS;