diff --git a/AGENTS.md b/AGENTS.md index 9d427756..3a5eaccf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -264,6 +264,11 @@ UASYNC/LL_QUEUE/MEMORY/TIMING объединены в SYS, NORMALIZER/BBR — в invite-ключ не даёт временного членства. Протокол описан в `src/chat/chat_join.h`, групповое согласование — в `src/routing_layer/topo_group.h`. - `TOPO_NODE` — общая подписанная запись узла. Более свежий timestamp заменяет запись целиком, с полями и подписью; адреса не версионируются отдельно. BGP обновляет запись и передаёт изменения NCD (см. `doc/node_snapshot.md`). +- BGP хранит один путь к назначению от каждого соседа. WITHDRAW удаляет только путь отправителя; + NODEINFO с прежним timestamp также обновляет путь. Выбор: живой путь с минимумом хопов, затем меньший ID соседа. + REINIT сохраняет пути до проверки нового снимка; TABLE_COMPLETE удаляет неподтверждённые, зависший обмен перезапускается. +- ETCP-router дописывает ID при транзите и отбрасывает повторное посещение для DATA/ACK/control. + Список переходов находится вне подписанного тела; формат описан в `src/routing_layer/etcp_router.h`. Все операции с ядром выполняются в его uasync-потоке; stop/destroy — вне callbacks останавливаемых сервисов. Порядок освобождения ресурсов и нюансы media workers: `doc/service_lifecycle.md`. diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index f8a0d8e0..0a2c7ab9 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -66,6 +66,57 @@ static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn); static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn); static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn); +/* Трейлер меняется независимо от подписанного тела. Не realloc: входной буфер + * может принадлежать пулу normalizer. При ошибке entry остаётся у вызывающего. */ +static int router_path_append(struct ll_entry* e, uint64_t node_id, uint8_t count) { + size_t kept = e->len - (count ? 1 : 0), size = kept + 9; + if (count >= ROUTER_MAX_VISITED || size > PKTNORM_MAX_DGRAM_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router path limit: count=%u len=%u", count, e->len); return -1; + } + uint8_t* wire = u_malloc(size); + if (!wire) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router path allocation failed size=%zu", size); return -1; } + memcpy(wire, e->dgram, kept); memcpy(wire + kept, &node_id, 8); wire[size - 1] = count + 1; + queue_dgram_free(e); e->dgram_pool = NULL; e->dgram_free_fn = NULL; + e->dgram = wire; e->len = size; e->memlen = size; + return 0; +} + +static int router_source_send(struct ETCP_CONN* conn, struct ll_entry* e) { + if (router_path_append(e, conn->instance->node_id, 0) < 0) return -1; + return etcp_send(conn, e); +} + +/* Возвращает размер неизменяемого тела или 0 при некорректном/зацикленном пакете. */ +static size_t router_path_check(struct ETCP_CONN* conn, const struct ll_entry* e) { + struct UTUN_INSTANCE* inst = conn->instance; + const struct SVC_ROUTE_HDR* h = (const struct SVC_ROUTE_HDR*)e->dgram; + uint8_t count = e->dgram[e->len - 1]; + size_t tail = 8 * (size_t)count + 1; + if (!count || count > ROUTER_MAX_VISITED || e->len < SVC_ROUTE_HDR_SIZE + tail) goto malformed; + size_t body = e->len - tail; + uint64_t ids[ROUTER_MAX_VISITED]; memcpy(ids, e->dgram + body, count * 8); + if (ids[0] != h->src_node_id || ids[count - 1] != conn->peer_node_id) goto malformed; + for (uint8_t i = 0; i < count; i++) { + if (!ids[i]) goto malformed; + if (ids[i] == inst->node_id) { + inst->router_loop_drops++; + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router loop: group=%016llx src=%016llx dst=%016llx via=%016llx self=%016llx visited=%u flags=%02x seq=%u dropped=%llu", + (unsigned long long)h->group_id, (unsigned long long)h->src_node_id, (unsigned long long)h->dst_node_id, + (unsigned long long)conn->peer_node_id, (unsigned long long)inst->node_id, count, h->flags, h->seq, + (unsigned long long)inst->router_loop_drops); + return 0; + } + for (uint8_t j = 0; j < i; j++) if (ids[i] == ids[j]) goto malformed; + } + return body; +malformed: + inst->router_path_errors++; + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router malformed path: peer=%016llx src=%016llx dst=%016llx len=%u count=%u dropped=%llu", + (unsigned long long)conn->peer_node_id, (unsigned long long)h->src_node_id, (unsigned long long)h->dst_node_id, + e->len, count, (unsigned long long)inst->router_path_errors); + return 0; +} + // Очередь-владелец waiter фиксируется при регистрации, не вычисляется по текущему маршруту. static void router_cancel_send_waiter(struct ETCP_ROUTER_CONN* rconn) { if (rconn->send_waiter_q) queue_waiter_cancel(rconn->send_waiter_q, &rconn->send_waiter); @@ -124,6 +175,7 @@ static struct TRANSIT_QUEUE* transit_queue_get_or_create(struct ETCP_CONN* conn, // Уничтожить транзитную очередь: снять waiter, дропнуть оставшиеся пакеты, удалить из реестра conn. static void transit_queue_destroy(struct ETCP_CONN* conn, struct TRANSIT_QUEUE* tq) { + if (tq->route_timer) uasync_cancel_timeout(conn->instance->ua, tq->route_timer); if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); int dropped = 0; struct ll_entry* e; @@ -138,19 +190,46 @@ static void transit_queue_destroy(struct ETCP_CONN* conn, struct TRANSIT_QUEUE* // Waiter-коллбэк транзитной очереди: send_input_q освободился → шлём один пакет, // при непустой очереди снова встаём в waiter, иначе уничтожаем очередь. +static void transit_queue_drain_cb(struct ll_queue* q, void* arg); +static void transit_queue_retry(void* arg) { + struct TRANSIT_QUEUE* tq = arg; + tq->route_timer = NULL; + transit_queue_drain_cb(tq->conn->send_input_q, tq); +} + +static void transit_queue_wait(struct TRANSIT_QUEUE* tq) { + if (!tq->route_timer) + tq->route_timer = uasync_set_timeout(tq->conn->instance->ua, ROUTER_NO_ROUTE_RETRY_TB, tq, transit_queue_retry, "transit_route"); + int waiting = tq->waiter.internal || tq->waiter.call_soon_id; + if (!tq->route_timer || (!waiting && queue_waiter_wait(tq->conn->send_input_q, &tq->waiter, transit_queue_drain_cb, tq) < 0)) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit wait failed dst=%016llx", (unsigned long long)tq->dst_node_id); + transit_queue_destroy(tq->conn, tq); + } +} + static void transit_queue_drain_cb(struct ll_queue* q, void* arg) { struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)arg; struct ETCP_CONN* conn = tq->conn; + struct TOPO_GROUP* group = topo_groups_find(conn->instance->topo_groups, tq->group_id); + struct ETCP_CONN* next = topo_group_find_conn_for_node(group, tq->dst_node_id); + if (next == conn && queue_entry_count(q)) { transit_queue_wait(tq); return; } + queue_waiter_cancel(conn->send_input_q, &tq->waiter); /* cancel only when sending or switching the route */ struct ll_entry* e = queue_data_get(tq->q); if (!e) { transit_queue_destroy(conn, tq); return; } - int ret = etcp_send(conn, e); + int ret = 0; + if (next != conn) { + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "transit reroute: dst=%016llx old=%016llx new=%016llx", + (unsigned long long)tq->dst_node_id, (unsigned long long)conn->peer_node_id, + (unsigned long long)(next ? next->peer_node_id : 0)); + router_forward_transit(conn->instance, e, (struct SVC_ROUTE_HDR*)e->dgram); + } else ret = etcp_send(conn, e); if (ret != 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_drain: etcp_send failed ret=%d src=%016llx dst=%016llx", ret, (unsigned long long)tq->src_node_id, (unsigned long long)tq->dst_node_id); free_entry(e); } if (queue_entry_count(tq->q) > 0) - queue_waiter_wait(conn->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); + transit_queue_wait(tq); else transit_queue_destroy(conn, tq); } @@ -161,6 +240,7 @@ void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn) { struct ll_entry* entry; while ((entry = queue_data_get(conn->transit_queues)) != NULL) { struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)entry; + if (tq->route_timer) uasync_cancel_timeout(conn->instance->ua, tq->route_timer); if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); struct ll_entry* e; while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); @@ -283,7 +363,7 @@ static void router_control(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* con "router control: tx group=%016llx peer=%016llx svc=%u flags=%02x local=%016llx remote=%016llx challenge=%016llx", (unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, flags, (unsigned long long)h->reset_id, (unsigned long long)peer_id, (unsigned long long)challenge); - if (etcp_send(conn, e) != 0) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router control: send failed"); free_entry(e); } + if (router_source_send(conn, e) != 0) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router control: send failed"); free_entry(e); } } static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flags) { @@ -555,7 +635,7 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) { uint32_t seq = inf->seq; queue_data_put_with_index(rconn->inflight_q, &inf->ll); rconn->tx_sent = seq + 1; - if (etcp_send(conn, e) != 0) { + if (router_source_send(conn, e) != 0) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router send failed svc=%u seq=%u; packet retained", rconn->svc_id, seq); rconn->tx_sent = seq; queue_remove_data(rconn->inflight_q, &inf->ll); @@ -640,7 +720,7 @@ static int router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_I rconn->last_dgram_ts = get_current_timestamp(); DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "RETRANS: svc_id=%u seq=%u attempt=%d → %016llx", rconn->svc_id, inf->seq, inf->send_count + 1, (unsigned long long)rconn->remote_node_id); - int ret = etcp_send(conn, entry); + int ret = router_source_send(conn, entry); if (ret != 0) { queue_dgram_free(entry); queue_entry_free(entry); rconn->c_pkts_send_err++; @@ -703,7 +783,7 @@ static void router_retrans_timer_cb(void* arg) { // Собрать+закодировать пакет в начале отправки и положить в send_q. static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t len, int force, int mode) { - size_t overhead = SVC_ROUTE_HDR_SIZE + ((mode & ROUTE_CRYPTO_SIGN) ? SC_SIGN_SIZE : 0) + size_t overhead = SVC_ROUTE_HDR_SIZE + ROUTER_PATH_MAX_SIZE + ((mode & ROUTE_CRYPTO_SIGN) ? SC_SIGN_SIZE : 0) + ((mode & ROUTE_CRYPTO_ENCRYPT) ? SC_NONCE_SIZE + SC_CRC32_SIZE + SC_TAG_SIZE : 0); if (rconn->closed || !payload || !len || len > PKTNORM_MAX_DGRAM_SIZE - overhead || (mode & ~(ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT))) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: invalid payload len=%zu mode=%x closed=%u", len, mode, rconn->closed); @@ -864,7 +944,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ACK_SEND: svc_id=%u rx_seq=%u → %016llx", rconn->svc_id, rconn->rx_seq, (unsigned long long)rconn->remote_node_id); - int ret = etcp_send(conn, entry); + int ret = router_source_send(conn, entry); if (ret != 0) { queue_dgram_free(entry); queue_entry_free(entry); DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: etcp_send failed ret=%d svc_id=%u rx_seq=%u", ret, rconn->svc_id, rconn->rx_seq); @@ -1127,7 +1207,7 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* hdr->svc_id, (unsigned long long)hdr->src_node_id, (unsigned long long)hdr->dst_node_id, queue_entry_count(tq->q)); if (was_empty && next->send_input_q) - queue_waiter_wait(next->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); + transit_queue_wait(tq); } // Принять data-пакет: зафиксировать recv_conn, синхронизировать rx_seq при первом пакете сеанса, @@ -1185,7 +1265,15 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) free_entry(entry); return; } struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)entry->dgram; - if (h->dst_node_id != inst->node_id) { router_forward_transit(inst, entry, h); return; } + size_t body = router_path_check(conn, entry); + if (!body) { free_entry(entry); return; } + if (h->dst_node_id != inst->node_id) { + if (router_path_append(entry, inst->node_id, entry->dgram[entry->len - 1]) < 0) { + inst->router_path_errors++; free_entry(entry); return; + } + router_forward_transit(inst, entry, (struct SVC_ROUTE_HDR*)entry->dgram); return; + } + entry->len = body; struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, h->group_id, h->src_node_id, h->svc_id); if (h->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) { uint8_t* decoded = NULL; size_t len = 0; uint8_t flags = 0; diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 8ea480eb..8ee5081a 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -6,9 +6,20 @@ // Сервисы регистрируют приём через bind, отправляют через route_send и соблюдают backpressure. // Все операции и callbacks — в потоке uasync. Закрытие группы удаляет её каналы и ожидающий транзит. // +// Построение маршрутов описано в topo_group.h. Роутер берёт текущий живой next hop +// из таблицы группы; при ожидании свободной очереди проверяет выбор повторно. +// Во время сходимости BGP возможны временные петли. Источник записывает свой ID, +// каждый транзитный узел проверяет список пройденных узлов и дописывает свой ID. +// Повторное посещение запрещено для всех пакетов, включая ACK и handshake. +// Ожидание в очереди не добавляет ID повторно; ретрансмит источника начинает новый список. +// // Формат SVC_ROUTE пакета: // [cmd:1] [group_id:8] [dst_node_id:8] [src_node_id:8] [seq:4] [svc_id:1] [flags:1] [timestamp:2] [reset_id:8] -// [peer_reset_id:8] [challenge:8] [payload...] +// [peer_reset_id:8] [challenge:8] [payload...] [signature?:64] [visited_ids:8*count] [count:1] +// count=1..ROUTER_MAX_VISITED; IDs идут от источника до отправителя текущего хопа. +// Изменяемый список находится вне end-to-end подписи/шифрования. Получатель +// проверяет его границы, уникальность, первый ID=src и последний ID=ETCP peer. +// Это защита от ошибок маршрутизации, а не доказательство пути от недоверенного пира. // flags: bit7=START, bit6=RST, bit3=SIGNED, bit2=ENCRYPTED, bit1=CLOSE // reset_id/peer_reset_id — идентификаторы отправителя/получателя во всех пакетах. // START=HELLO, RST=CHALLENGE, START|RST=CONFIRM; challenge используется только в handshake. @@ -45,6 +56,8 @@ struct SVC_ROUTE_HDR { #pragma pack(pop) #define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 57 #define SVC_ROUTE_MAX_BINDINGS 256 +#define ROUTER_MAX_VISITED 64 // предел размера списка, не замена проверки повторных ID +#define ROUTER_PATH_MAX_SIZE (8 * ROUTER_MAX_VISITED + 1) // Единый формат доставки сервисной кодограммы (router → сервис): // [svc_id:1][src_node_id:8][dst_node_id:8][rx_flags:1][group_id:8][payload...] @@ -82,7 +95,8 @@ struct TRANSIT_QUEUE { uint64_t dst_node_id; // = data[16..23] struct ll_queue* q; // FIFO ещё не переданных транспорту пакетов struct queue_waiter_handle waiter; // ожидание свободной conn->send_input_q - struct ETCP_CONN* conn; // next_hop (для etcp_send в drain_cb) + struct ETCP_CONN* conn; // владелец очереди и waiter; текущий маршрут проверяется при выгрузке + void* route_timer; // повторный выбор пути, даже если прежняя очередь остаётся занятой }; // Состояние одного логического подключения (group_id + remote_node_id + svc_id) @@ -184,13 +198,13 @@ struct ETCP_ROUTER_CONN { #define MINRTT_WINDOW_SIZE 10 #define MINRTT_DEFAULT_TB 100 // default minRTT = 10ms (заменяется первым RTT) -// Inflight запись — копия финального wire-пакета (уже закодирован encode'ом) для ретрансмита +// Inflight хранит подписанное/зашифрованное тело. Каждый ретрансмит добавляет новый список переходов. struct ROUTER_INFLIGHT { struct ll_entry ll; // индекс по seq (4 байта, offset 0) uint32_t seq; uint64_t last_sent_tb; // время последней отправки (0.1ms) uint8_t send_count; // число переотправок - uint8_t* dgram; // копия финального пакета [SVC_ROUTE_HDR][payload][sig?] + uint8_t* dgram; // неизменяемое тело [SVC_ROUTE_HDR][payload][sig?], без visited_ids/count size_t dgram_len; }; diff --git a/src/routing_layer/etcp_router_doc.md b/src/routing_layer/etcp_router_doc.md index 7c5b063a..1a956a54 100644 --- a/src/routing_layer/etcp_router_doc.md +++ b/src/routing_layer/etcp_router_doc.md @@ -10,6 +10,19 @@ Packed `SVC_ROUTE_HDR`, 57 байт: Числовое представление полей сохраняет существующий формат проекта (native endian). Отдельной версии/согласования формата нет; узлы должны обновляться совместно. +Полный пакет: `[SVC_ROUTE_HDR][payload][signature?][visited_ids:8*count][count:1]`. +Источник начинает список своим ID; транзит проверяет отсутствие собственного ID и дописывает его. +Правило действует для DATA, ACK и всех управляющих пакетов. Повторное посещение отбрасывается; +также проверяются размер, уникальность ID, первый ID=src и последний ID=ETCP peer. +Предел — 64 ID; на конечном узле полный список допустим, для следующего перехода места уже нет. +Список находится вне end-to-end подписи/шифрования и снимается перед crypto decode. +Это защита от временных петель при сходимости BGP, а не криптографическое подтверждение пути. + +Маршруты строит BGP по правилам `topo_group.h`. Транзит использует текущий next hop. +При backpressure пакет ждёт в очереди; waiter и таймер 20 ms повторно проверяют маршрут. +Смена пути не требует освобождения старой физической очереди. ID добавляется один раз на приёме, +а не при каждой попытке выгрузки. При исчезновении маршрута транзит отбрасывается; доставку повторяет источник. + `reset_id` — случайный ненулевой ID локального экземпляра соединения; `peer_reset_id` — ID получателя. DATA, ACK и CLOSE принимаются только при совпадении обоих ID с текущей парой. Смена физического транспорта пару не меняет. | Пакет | Флаги | Payload | Поля | @@ -31,7 +44,7 @@ Challenge защищает от случайных старых/перестав - `send_q`: plaintext + seq + crypto mode, до 64 пакетов при обычной отправке. Кодирование происходит при отправке после handshake, с актуальной парой ID. - `tx_seq`: следующий выделяемый seq; `tx_sent`: граница фактически отправленного; `tx_acked`: граница подтверждённого. -- `inflight_q`: финальные wire-копии, обычно до 256 пакетов. Копия создаётся до передачи в ETCP; ошибка не теряет pending-пакет. +- `inflight_q`: неизменяемое тело с подписью, без списка посещённых узлов, обычно до 256 пакетов. Каждый ретрансмит начинает новый список; ошибка не теряет pending-пакет. - ACK принимается только в модульном интервале `(tx_acked, tx_sent]`. Дубликаты, старые и выходящие за границу ACK не продлевают ожидание прогресса. Удаляются существующие inflight-записи, а не перебираются все номера до произвольного ACK. - Новые DATA и ретрансмиты идут по одному через waiter физической `send_input_q`. При смене пути отменяется waiter на очереди, где он действительно был зарегистрирован, включая уже запланированный callback. - Каждые 300 ms без прогресса планируется проход повторной отправки. После 17 попыток без ACK соединение закрывается. Отсутствующий маршрут проверяется каждые 20 ms; очередь сохраняется. Пока первичный handshake не завершён, HELLO повторяется без отдельного таймаута. @@ -39,7 +52,7 @@ Challenge защищает от случайных старых/перестав - Сброс физического ETCP сохраняет router inflight и seq. Восстановление самого ETCP-handshake остаётся обязанностью транспортного слоя. - Crypto decode/проверка подписи выполняются до помещения в reorder и изменения rx_seq. Повреждённый пакет не подтверждается как доставленный. - DATA выдаются сервису строго по seq; входящие дубликаты повторно проверяются после извлечения из очереди. -- Wire-пакет ограничен `PKTNORM_MAX_DGRAM_SIZE` (16384), включая заголовок и crypto overhead. Пустой payload запрещён: он используется для управления. +- Wire-пакет ограничен `PKTNORM_MAX_DGRAM_SIZE` (16384), включая заголовок, crypto overhead и резерв полного списка переходов. Пустой payload запрещён: он используется для управления. ## Сервисный API @@ -55,6 +68,12 @@ Callback может закрыть соединение. Освобождени ## Проверки и диагностика +`test_bgp_paths` использует настоящий BGP и роутер в управляемой сети четырёх узлов: +все 64 графа, разбиение/слияние, изменения при непустых очередях, REINIT, выбор кратчайшего пути +с устойчивым разрешением равенства. Дополнительно проверяет подписанный зашифрованный транзит, +петли DATA/ACK/control, повреждённые списки переходов, смену заблокированного next hop и очистку таймеров. +Счётчики `UTUN_INSTANCE.router_loop_drops` и `router_path_errors` сопровождаются WARN с идентичностью пакета. + `tests/test_etcp_router_faults` перехватывает SVC_ROUTE между настоящими роутерами, очередями и UASYNC-таймерами. Физическая очередь тестовая: нижележащий ETCP не скрывает потери от роутера. Проверяются: 1. Потери DATA/ACK, перестановка, дубликаты, четыре потока (два направления × два сервиса), backpressure и ограниченность физической очереди. diff --git a/src/routing_layer/route_crypto.h b/src/routing_layer/route_crypto.h index 5a44f2f8..e2b34123 100644 --- a/src/routing_layer/route_crypto.h +++ b/src/routing_layer/route_crypto.h @@ -6,7 +6,8 @@ // decode(): вызывается перед вызовом сервисного коллбэка. // [SVC_ROUTE_HDR][payload][sig?] → [SVC_ROUTE_HDR][plaintext] // -// Порядок кодирования: encrypt-then-sign (подпись поверх всего протокола, включая шифртекст). +// Порядок кодирования: encrypt-then-sign (подпись заголовка и payload, включая шифртекст). +// Список пройденных узлов добавляет/снимает etcp_router снаружи этого формата. // При ошибке декодирования (плохая подпись / не удалось расшифровать) пакет дропается. #ifndef ROUTE_CRYPTO_H #define ROUTE_CRYPTO_H diff --git a/src/routing_layer/topo_group.c b/src/routing_layer/topo_group.c index c13944da..dc0180b7 100644 --- a/src/routing_layer/topo_group.c +++ b/src/routing_layer/topo_group.c @@ -1435,7 +1435,9 @@ static int topo_emit_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* } DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "send_nodeinfo: node %016llx ver=%d grp=%016llx to conn=%p name='%s'", (unsigned long long)node->node_id, sni->ver, (unsigned long long)group->group_id, (void*)conn, conn->log_name); uint8_t sflags = (group->group_type == TOPO_GROUP_TYPE_CHAT) ? 0 : TOPO_FLAG_SEND_SUBNETS; - int ser_len = topo_node_serialize(sni, node, group->group_id, sflags, p + sizeof(h), max_sz - sizeof(h), cumulative_rtt); + /* RTT не определяет маршрут. Его непрерывные колебания не должны запускать + * волну NODEINFO при каждом принятом альтернативном пути. Хешируем с RTT=0. */ + int ser_len = topo_node_serialize(sni, node, group->group_id, sflags, p + sizeof(h), max_sz - sizeof(h), 0); if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: serialize failed for node %016llx", (unsigned long long)node->node_id); u_free(p); return -1; @@ -1447,6 +1449,7 @@ static int topo_emit_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* SHA256(p + sizeof(h), ser_len, digest); struct topo_advert* advert = (struct topo_advert*)queue_find_data_by_index(peer->advertised, &node->node_id); if (advert && memcmp(advert->digest, digest, sizeof(digest)) == 0) { u_free(p); return 0; } + ((struct TOPOMSG_NODEINFO_PKT*)p)->node.cumulative_rtt = cumulative_rtt; int fresh = advert == NULL; if (fresh) advert = (struct topo_advert*)queue_entry_new(sizeof(*advert) - sizeof(struct ll_entry)); if (!advert) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "advert allocation failed"); u_free(p); return -1; } @@ -1467,8 +1470,8 @@ static int topo_emit_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* } advert->node_id = node->node_id; memcpy(advert->digest, digest, sizeof(digest)); if (fresh) queue_data_put_with_index(peer->advertised, &advert->ll); - DEBUG_INFO(DEBUG_CATEGORY_BGP, "NODEINFO sent: group=%016llx node=%016llx to=%016llx", - (unsigned long long)group->group_id, (unsigned long long)node->node_id, (unsigned long long)peer->node_id); + DEBUG_INFO(DEBUG_CATEGORY_BGP, "NODEINFO sent: group=%016llx node=%016llx to=%016llx rtt=%u", + (unsigned long long)group->group_id, (unsigned long long)node->node_id, (unsigned long long)peer->node_id, cumulative_rtt); return 1; } diff --git a/src/routing_layer/topo_group.h b/src/routing_layer/topo_group.h index 87de7773..08b5ee21 100644 --- a/src/routing_layer/topo_group.h +++ b/src/routing_layer/topo_group.h @@ -25,6 +25,8 @@ * - Каждая сессия хранит хеш последнего отправленного анонса назначения. * Очередь хранит node_id; пакет строится из актуального состояния при отправке. * Поэтому изменение пути распространяется без изменения timestamp и без эха дублей. + * RTT передаётся как оценка на момент анонса, но исключён из хеша: колебания + * задержки сами по себе не меняют маршрут и не вызывают повторных объявлений. * - Альтернатива сохраняет доступность узла. После потери последнего пути * удаляется только маршрут группы: членство и сохранённая запись узла остаются. * @@ -193,7 +195,7 @@ struct TOPO_GROUP_CONN_ITEM { struct topo_tx_item *tx_head, *tx_tail; // FIFO команд, ещё не переданных транспорту struct queue_waiter_handle tx_waiter; // ожидание свободной send_input_q void* tx_wake; // отложенный запуск отправки - struct ll_queue* advertised; // node_id -> хеш последнего успешно отправленного NODEINFO; владеет сессия + struct ll_queue* advertised; // node_id -> хеш отправленного NODEINFO без RTT; владеет сессия uint8_t table_received; // TABLE_COMPLETE текущей сессии принят uint8_t table_sent; // ответ на запрос пира целиком передан транспорту }; diff --git a/src/utun_instance.h b/src/utun_instance.h index ffaf2c85..6efc6520 100644 --- a/src/utun_instance.h +++ b/src/utun_instance.h @@ -210,6 +210,8 @@ struct UTUN_INSTANCE { // etcp_router bindings и seq-connections (per-instance service routing) struct ETCP_ROUTER_BINDINGS router_bindings; struct ll_queue* router_conns; + uint64_t router_loop_drops; // пакеты, вернувшиеся в уже посещённый узел + uint64_t router_path_errors; // некорректный или переполненный список пройденных узлов struct CONN_MGR* conn_mgr; // заимствован у UTUN-группы; NULL при остановленном UTUN struct DB_SYNC* db_sync; // Distributed DB sync (может быть NULL) diff --git a/tests/test_bgp_paths.c b/tests/test_bgp_paths.c index 2b1a67a8..b30fff86 100644 --- a/tests/test_bgp_paths.c +++ b/tests/test_bgp_paths.c @@ -8,6 +8,8 @@ #include "topo_group.h" #include "etcp.h" #include "etcp_api.h" +#include "etcp_router.h" +#include "route_crypto.h" #include "node_conn_direct.h" #include "../lib/debug_config.h" #include "test_utils.h" @@ -21,6 +23,8 @@ static struct ETCP_CONN* conns[N][N]; static struct NODE_CONN_DIRECT* owners[N][N]; static unsigned graph, delivered; static uint32_t random_state = 1; +static int router_test, blocked_from = -1, blocked_to = -1; +static unsigned received, transit_packets; static unsigned next_random(void) { random_state = random_state * 1664525U + 1013904223U; @@ -44,14 +48,21 @@ static int step(void) { for (int k = 0; k < N * N; k++) { int a = ((k + offset) % (N * N)) / N, b = (k + offset) % N; if (a == b) continue; + if (a == blocked_from && b == blocked_to) continue; struct ll_queue* q = conns[a][b]->send_input_q; - assert(queue_entry_count(q) <= 1); + assert(queue_entry_count(q) <= (router_test ? 8 : 1)); struct ll_entry* e = queue_data_get(q); if (!e) continue; work++; if ((graph & edge(a, b)) && e->len && e->dgram[0] == ETCP_ID_TOPO_ENTRY) { delivered++; nodes[b]->api_bindings.callbacks[ETCP_ID_TOPO_ENTRY](conns[b][a], e); + } else if ((graph & edge(a, b)) && e->len && e->dgram[0] == ETCP_RT_ID_SVC_ROUTE) { + unsigned count = e->dgram[e->len - 1]; + assert(count && count <= ROUTER_MAX_VISITED); + uint64_t last; memcpy(&last, e->dgram + e->len - 9, 8); assert(last == nodes[a]->node_id); + if (count > 1) transit_packets++; + nodes[b]->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conns[b][a], e); } else release(e); queue_resume_callback(q); } @@ -102,6 +113,14 @@ static void verify(void) { struct TOPO_GROUP_NODE* node = topo_node_find_by_id(groups[a], nodes[b]->node_id); uint8_t count; uint64_t* hops = topo_node_best_hop_list(node, &count, NULL); assert(count == distance[a][b]); + uint64_t best_neighbor = UINT64_MAX; + for (int n = 0; n < N; n++) + if ((graph & edge(a, n)) && distance[n][b] + 1 == distance[a][b] && nodes[n]->node_id < best_neighbor) + best_neighbor = nodes[n]->node_id; + assert(conn->peer_node_id == best_neighbor); + for (struct ll_entry* p = node->paths->head; p; p = p->next) + for (struct ll_entry* other = p->next; other; other = other->next) + assert(((struct TOPO_NODEPATH*)p)->conn != ((struct TOPO_NODEPATH*)other)->conn); int previous = a; for (int h = count - 1; h >= 0; h--) { int current = 0; @@ -138,12 +157,157 @@ static void create(void) { } } +static void metric_jitter(void) { + unsigned before = delivered; + for (int a = 0; a < N; a++) for (int b = 0; b < N; b++) { + if (!(graph & edge(a, b))) continue; + conns[a][b]->rtt_last = 17 + 13 * a + b; + } + for (int a = 0; a < N; a++) for (int b = 0; b < N; b++) { + if (!(graph & edge(a, b))) continue; + for (struct ll_entry* e = groups[a]->nodes->head; e; e = e->next) + assert(topo_group_send_nodeinfo(groups[a], (struct TOPO_GROUP_NODE*)e, conns[a][b]) == 0); + } + settle(); verify(); + DEBUG_INFO(DEBUG_CATEGORY_BGP, "metric-only change: unexpected announcements=%u", delivered - before); + assert(delivered == before); /* RTT is telemetry, not routing state. */ +} + +static void service_receive(struct ETCP_CONN* conn, struct ll_entry* e) { + assert(conn->instance == nodes[3]); + if (e->len > ROUTER_SVC_HDR_SIZE) { + uint64_t source, group; + memcpy(&source, e->dgram + ROUTER_SVC_SRC_OFF, 8); memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); + assert(source == nodes[0]->node_id && group == GROUP); + assert(e->dgram[ROUTER_SVC_FLAGS_OFF] == (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)); + assert(e->len == ROUTER_SVC_HDR_SIZE + 4 && !memcmp(e->dgram + ROUTER_SVC_HDR_SIZE, "test", 4)); + received++; + } + release(e); +} + +static struct ll_entry* route_packet(uint8_t flags, const uint64_t* ids, unsigned count) { + size_t body = SVC_ROUTE_HDR_SIZE + (flags ? 0 : 4); + struct ll_entry* e = ll_alloc_lldgram(body + count * 8 + 1); assert(e); + struct SVC_ROUTE_HDR h = { .cmd = ETCP_RT_ID_SVC_ROUTE, .group_id = GROUP, + .src_node_id = ids[0], .dst_node_id = nodes[3]->node_id, .svc_id = 0x41, .flags = flags, .reset_id = 1 }; + memcpy(e->dgram, &h, sizeof(h)); + if (!flags) memcpy(e->dgram + sizeof(h), "loop", 4); + memcpy(e->dgram + body, ids, count * 8); e->dgram[body + count * 8] = count; + e->len = body + count * 8 + 1; + return e; +} + +static void inject_router(struct ll_entry* e) { + nodes[1]->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conns[1][0], e); +} + +static void blocked_ready(struct ll_queue* q, void* arg) { + (void)q; (void)arg; + assert(0 && "busy transport must not release a waiter"); +} + +static void router_paths(void) { + router_test = 1; + unsigned chain = edge(0, 1) | edge(1, 2) | edge(2, 3); + change(chain); settle(); verify(); + assert(etcp_router_bind(nodes[3], 0x42, service_receive) == 0); + struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(nodes[0], GROUP, nodes[3]->node_id, 0x42); assert(c); + assert(etcp_router_conn_send(c, (const uint8_t*)"test", 4, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0); + uint64_t deadline = get_time_tb() + 20000; + while ((!received || c->tx_acked != c->tx_seq) && get_time_tb() < deadline) { step(); uasync_poll(ua, 1); } + assert(received == 1 && c->tx_acked == c->tx_seq && transit_packets > 0); + settle(); + + /* Actual transient two-hop loop B -> C -> B. Each hop must append once, + * for DATA, ACK and every control flag, before B rejects its second visit. */ + struct TOPO_GROUP_NODE* target = topo_node_find_by_id(groups[2], nodes[3]->node_id); + assert(topo_group_remove_path(target, conns[2][3]) == 1); + uint64_t cycle[] = { nodes[3]->node_id, nodes[1]->node_id }; + assert(topo_group_add_path(target, conns[2][1], cycle, 2, 0) == 0); + uint8_t flags[] = { 0, 0, ROUTER_FLAG_START, ROUTER_FLAG_RST, ROUTER_FLAG_START | ROUTER_FLAG_RST, ROUTER_FLAG_CLOSE }; + for (unsigned i = 0; i < sizeof(flags); i++) { + uint64_t id = nodes[0]->node_id, before = nodes[1]->router_loop_drops; + struct ll_entry* e = route_packet(flags[i], &id, 1); + if (i == 1) { /* Header-only ACK. */ + memcpy(e->dgram + SVC_ROUTE_HDR_SIZE, &id, 8); e->dgram[SVC_ROUTE_HDR_SIZE + 8] = 1; e->len -= 4; + } + inject_router(e); settle(); + assert(nodes[1]->router_loop_drops == before + 1); + } + assert(topo_group_remove_path(target, conns[2][1]) == 1); + uint64_t direct = nodes[3]->node_id; + assert(topo_group_add_path(target, conns[2][3], &direct, 1, 0) == 0); + + /* Reject malformed metadata before touching a router session. */ + uint64_t ids[ROUTER_MAX_VISITED + 1]; + for (unsigned i = 0; i < ROUTER_MAX_VISITED + 1; i++) ids[i] = 100 + i; + ids[0] = nodes[0]->node_id; + uint64_t errors = nodes[1]->router_path_errors; + struct ll_entry* e = route_packet(ROUTER_FLAG_START, ids, 1); e->dgram[e->len - 1] = 0; inject_router(e); + e = route_packet(ROUTER_FLAG_START, ids, 1); e->dgram[e->len - 1] = 2; inject_router(e); + e = route_packet(ROUTER_FLAG_START, ids, 1); ((struct SVC_ROUTE_HDR*)e->dgram)->src_node_id++; inject_router(e); + ids[1] = ids[0]; inject_router(route_packet(ROUTER_FLAG_START, ids, 2)); /* repeated source */ + ids[1] = 101; inject_router(route_packet(ROUTER_FLAG_START, ids, 2)); /* last hop != ETCP peer */ + inject_router(route_packet(ROUTER_FLAG_START, ids, ROUTER_MAX_VISITED + 1)); + ids[0] = 99; ids[ROUTER_MAX_VISITED - 1] = nodes[0]->node_id; + inject_router(route_packet(ROUTER_FLAG_START, ids, ROUTER_MAX_VISITED)); /* valid but cannot append */ + assert(nodes[1]->router_path_errors == errors + 7); + e = route_packet(ROUTER_FLAG_START, ids, ROUTER_MAX_VISITED); + ((struct SVC_ROUTE_HDR*)e->dgram)->dst_node_id = nodes[1]->node_id; + ((struct SVC_ROUTE_HDR*)e->dgram)->flags = ROUTER_FLAG_CLOSE; /* valid max-sized list at destination */ + inject_router(e); assert(nodes[1]->router_path_errors == errors + 7); + + /* Queued transit must leave the old busy next hop even if it never drains. */ + change(chain | edge(1, 3)); settle(); verify(); + blocked_from = 1; blocked_to = 3; + struct ll_entry* blocker = queue_entry_new(0); assert(blocker); queue_data_put(conns[1][3]->send_input_q, blocker); + assert(etcp_router_conn_send(c, (const uint8_t*)"test", 4, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0); + settle(); + assert(received == 1 && conns[1][3]->transit_queues && queue_entry_count(conns[1][3]->transit_queues) == 1); + struct ll_queue* busy = conns[1][3]->send_input_q; + struct TRANSIT_QUEUE* waiting = (struct TRANSIT_QUEUE*)conns[1][3]->transit_queues->head; + struct queue_waiter_handle other = {0}; + assert(queue_waiter_wait(busy, &other, blocked_ready, NULL) == 0); + deadline = get_time_tb() + 600; + while (get_time_tb() < deadline) uasync_poll(ua, 1); + assert(busy->waiter_head == waiting->waiter.internal); /* polling must preserve FIFO priority */ + queue_waiter_cancel(busy, &other); + target = topo_node_find_by_id(groups[1], nodes[3]->node_id); + assert(topo_group_remove_path(target, conns[1][3]) == 1); + deadline = get_time_tb() + 20000; + while ((received != 2 || c->tx_acked != c->tx_seq) && get_time_tb() < deadline) { step(); uasync_poll(ua, 1); } + assert(received == 2 && c->tx_acked == c->tx_seq); + assert(queue_entry_count(conns[1][3]->transit_queues) == 0 && !conns[1][3]->send_input_q->waiter_head); + assert(queue_data_get(conns[1][3]->send_input_q) == blocker); release(blocker); + blocked_from = blocked_to = -1; + queue_resume_callback(conns[1][3]->send_input_q); + /* Pending waiter + route timer must both disappear with their owning group. */ + settle(); blocked_from = 1; blocked_to = 2; + blocker = queue_entry_new(0); assert(blocker); queue_data_put(conns[1][2]->send_input_q, blocker); + uint64_t source = nodes[0]->node_id; inject_router(route_packet(ROUTER_FLAG_START, &source, 1)); + assert(conns[1][2]->transit_queues && queue_entry_count(conns[1][2]->transit_queues) == 1); + etcp_router_close_group(nodes[1], GROUP); + assert(queue_entry_count(conns[1][2]->transit_queues) == 0 && !conns[1][2]->send_input_q->waiter_head); + assert(queue_data_get(conns[1][2]->send_input_q) == blocker); release(blocker); + blocked_from = blocked_to = -1; + queue_resume_callback(conns[1][2]->send_input_q); + settle(); + /* A direct physical connection does not authorize transit in an unknown group. */ + e = route_packet(ROUTER_FLAG_START, &source, 1); + ((struct SVC_ROUTE_HDR*)e->dgram)->group_id = GROUP + 1; + inject_router(e); + for (int b = 0; b < N; b++) if (b != 1) assert(queue_entry_count(conns[1][b]->send_input_q) == 0); + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router paths: encrypted transit, loops, malformed paths and blocked-route switch passed"); +} + int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_BGP, DEBUG_LEVEL_DEBUG); utun_instance_set_tun_init_enabled(0); create(); unsigned triangle = edge(0, 1) | edge(1, 2) | edge(0, 2) | edge(2, 3); change(triangle); settle(); verify(); + metric_jitter(); /* One-sided REINIT with routes retained, then simultaneous REINIT at all peers. */ etcp_fire_conn_status(conns[0][1], ETCP_CONN_STATUS_REINIT); settle(); verify(); for (int a = 0; a < N; a++) for (int b = 0; b < N; b++) @@ -160,10 +324,13 @@ int main(void) { for (unsigned n = next_random() % 5; n; n--) step(); } settle(); verify(); + router_paths(); change(0); settle(); verify(); for (int a = 0; a < N; a++) for (int b = 0; b < N; b++) if (a != b) node_conn_direct_close(owners[a][b]); for (int a = 0; a < N; a++) utun_instance_destroy(nodes[a]); - uasync_poll(ua, 0); uasync_destroy(ua, 0); + uasync_poll(ua, 0); + assert(ua->timer_alloc_count == ua->timer_free_count); + uasync_destroy(ua, 0); DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP paths: triangle, partitions, all four-node graphs and churn passed packets=%u", delivered); return 0; } diff --git a/tests/test_etcp_router_faults.c b/tests/test_etcp_router_faults.c index fe3eeab3..04ce20ac 100644 --- a/tests/test_etcp_router_faults.c +++ b/tests/test_etcp_router_faults.c @@ -70,7 +70,8 @@ static void wire(int from, struct ll_queue* q) { if (!e) return; if (e->len < SVC_ROUTE_HDR_SIZE) { release(e); return; } struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)e->dgram; - int data = e->len > SVC_ROUTE_HDR_SIZE; + if (e->dgram[e->len - 1] != 1) { failed = 1; release(e); return; } /* every source retry starts afresh */ + int data = e->len > SVC_ROUTE_HDR_SIZE + 9; if (!data && h->flags == ROUTER_FLAG_START && !saved_hello && from == 0) saved_hello = copy_packet(e); if (!data && h->flags == (ROUTER_FLAG_START | ROUTER_FLAG_RST) && !saved_confirm && from == 0) saved_confirm = copy_packet(e); int* drop = NULL; @@ -336,7 +337,8 @@ static int size_bounds(void) { REQUIRE(etcp_router_conn_send(c, data, PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE, ROUTER_FLAG_SIGNED) == -1); REQUIRE(etcp_router_conn_send(c, data, SIZE_MAX, 0) == -1); REQUIRE(c->tx_seq == 0 && c->send_q->count == 0); - REQUIRE(etcp_router_conn_send(c, data, PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE, 0) == 0); + REQUIRE(etcp_router_conn_send(c, data, PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE - ROUTER_PATH_MAX_SIZE + 1, 0) == -1); + REQUIRE(etcp_router_conn_send(c, data, PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE - ROUTER_PATH_MAX_SIZE, 0) == 0); u_free(data); cleanup(); return 0; } diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index f1f74994..36224179 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -62,6 +62,17 @@ static void rx_reset(uint8_t marker) { // ======================== Inject helper ======================== static struct ETCP_CONN fake_conn; +/* Older unit cases construct the immutable body; add the mandatory direct-hop trailer. */ +static void inject_wire(struct ETCP_CONN* conn, struct ll_entry* e) { + size_t len = e->len; + uint64_t src = ((struct SVC_ROUTE_HDR*)e->dgram)->src_node_id; + uint8_t* wire = u_malloc(len + 9); if (!wire) abort(); + memcpy(wire, e->dgram, len); memcpy(wire + len, &src, 8); wire[len + 8] = 1; + queue_dgram_free(e); e->dgram = wire; e->len = len + 9; + conn->peer_node_id = src; + conn->instance->api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE](conn, e); +} + // Настоящий HELLO/challenge/CONFIRM через публичный wire callback; состояние не подменяем. static void establish(etcp_recv_fn cb, struct UTUN_INSTANCE* inst, uint64_t src, uint8_t svc, uint64_t epoch) { struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(inst, TOPO_GROUP_UTUN, src, svc); @@ -152,7 +163,7 @@ static void inject_epoch(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst, fake_conn.send_input_q = queue_new(ua, 0, 0, 0, "test_wire"); \ TESTASSERT(etcp_router_init(&inst) == 0); \ TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, test_handler) == 0); \ - recv_cb = inst.api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE]; \ + recv_cb = inject_wire; \ TESTASSERT(recv_cb != NULL); \ } while(0) @@ -880,7 +891,7 @@ static int test_server_reinit(void) { memset(&inst.router_bindings, 0, sizeof(inst.router_bindings)); if (etcp_router_init(&inst) != 0) FAIL("reinit failed"); etcp_router_bind(&inst, TEST_SVC_ID, test_handler); - recv_cb = inst.api_bindings.callbacks[ETCP_RT_ID_SVC_ROUTE]; + recv_cb = inject_wire; if (!recv_cb) FAIL("recv_cb not bound"); struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID);