From 661cf032feb959575b4acbc4c10bc306c70efd34 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 01:14:49 +0200 Subject: [PATCH] Define routed stream relay requests and generation-scoped replies --- src/stream/stream.c | 6 ++++++ src/stream/stream_optimizer.h | 2 ++ src/stream/stream_priv.h | 1 + src/stream/stream_wire.c | 2 +- src/stream/stream_wire.h | 8 ++++++-- 5 files changed, 16 insertions(+), 3 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index 7f8016e3..d05a10ff 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -879,6 +879,12 @@ void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) { } } +/* Routed READY будит selector, не создавая пользовательского события или второго экземпляра. */ +void stream_relay_ready(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer) { + struct stream_session *s = inst && inst->stream ? session_find(inst->stream, key->group_id, key->source_node_id) : NULL; + if (s && s->descriptor.source_epoch == key->source_epoch && s->optimizer) stream_optimizer_refresh(s->optimizer, peer); +} + /* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate **out, size_t *count) { if (!out || *out || !count || !stream_handle_active(h)) { diff --git a/src/stream/stream_optimizer.h b/src/stream/stream_optimizer.h index 317cd501..526a9962 100644 --- a/src/stream/stream_optimizer.h +++ b/src/stream/stream_optimizer.h @@ -18,5 +18,7 @@ void stream_optimizer_destroy(struct stream_optimizer *optimizer); void stream_optimizer_tick(struct stream_optimizer *optimizer, struct STREAM_HANDLE *subscription, uint64_t now_ms); int stream_optimizer_configure(struct stream_optimizer *optimizer, const struct stream_selection_settings *settings); void stream_optimizer_info(struct stream_optimizer *optimizer, struct stream_selection_info *info); +/* Подтверждённая помощь немедленно обновляет кандидатов и снимает старый backoff конкретного пира. */ +void stream_optimizer_refresh(struct stream_optimizer *optimizer, uint64_t peer); #endif diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index 2bb8ce7d..f77b38a1 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -31,6 +31,7 @@ int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor int stream_relay_confirmed(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer); /* Вызывается selector один раз на экземпляр; relay-only экземпляры новых запросов помощи не создают. */ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, int needed, uint64_t now_ms); +void stream_relay_ready(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer); int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length); diff --git a/src/stream/stream_wire.c b/src/stream/stream_wire.c index 708d44eb..7bb9670a 100644 --- a/src/stream/stream_wire.c +++ b/src/stream/stream_wire.c @@ -7,7 +7,7 @@ /* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */ int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) { if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION - || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_OVERLOAD_SELECT) { + || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_RELAY_CANCEL) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT; } struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER, diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index c405a478..1c3a550b 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -10,7 +10,7 @@ extern "C" { #endif -#define STREAM_WIRE_VERSION 3 +#define STREAM_WIRE_VERSION 4 #define STREAM_WIRE_HEADER 52 #define STREAM_PATH_MAX 32 #define STREAM_END_SIZE 106 @@ -25,8 +25,12 @@ extern "C" { enum stream_command { STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT, STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED, - STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ, STREAM_CMD_OVERLOAD, STREAM_CMD_OVERLOAD_SELECT + STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ, STREAM_CMD_OVERLOAD, STREAM_CMD_OVERLOAD_SELECT, + STREAM_CMD_RELAY_REQUEST, STREAM_CMD_RELAY_STATUS, STREAM_CMD_RELAY_CANCEL }; +/* Routed помощь: handle — nonce запроса; эпохи запроса/сервиса независимы от raw ETCP-сессии. */ +enum stream_relay_status { STREAM_RELAY_STARTING = 1, STREAM_RELAY_READY, STREAM_RELAY_BOUND, + STREAM_RELAY_REJECTED, STREAM_RELAY_ENDED, STREAM_RELAY_CANCELLED }; enum stream_reject { STREAM_REASON_STOPPED = 1, STREAM_REASON_POSITION, STREAM_REASON_EXPIRED, STREAM_REASON_STALE, STREAM_REASON_CONFLICT, STREAM_REASON_LIMIT, STREAM_REASON_LOOP, STREAM_REASON_TRANSPORT,