From 6163f34ce17d1e319920a722e984af527906d694 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 6 Sep 2026 22:20:08 +0300 Subject: [PATCH] =?UTF-8?q?etcp=5Frouter:=20=D1=8D=D0=BF=D0=BE=D1=85=D0=B0?= =?UTF-8?q?=20reset=5Fid=20=D0=BD=D0=B0=20wire=20+=20master/slave=20?= =?UTF-8?q?=E2=80=94=20=D1=84=D0=B8=D0=BA=D1=81=20=D0=BF=D0=BE=D1=82=D0=B5?= =?UTF-8?q?=D1=80=D0=B8=20=D0=BF=D0=B0=D0=BA=D0=B5=D1=82=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=BF=D1=80=D0=B8=20=D1=80=D0=B5=D1=81=D1=82=D0=B0=D1=80=D1=82?= =?UTF-8?q?=D0=B5=20=D0=BF=D0=B8=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- PLAN_etcp_router_reset_id.md | 147 ++++++++++++++++++++++++++++++++ src/routing_layer/etcp_router.c | 65 +++++++++----- src/routing_layer/etcp_router.h | 11 +-- tests/test_etcp_router_unit.c | 127 +++++++++++++++++++++------ 4 files changed, 297 insertions(+), 53 deletions(-) create mode 100644 PLAN_etcp_router_reset_id.md diff --git a/PLAN_etcp_router_reset_id.md b/PLAN_etcp_router_reset_id.md new file mode 100644 index 00000000..53dbc6a1 --- /dev/null +++ b/PLAN_etcp_router_reset_id.md @@ -0,0 +1,147 @@ +# План: эпоха (reset_id) для ETCP Router + +Аналог фикса `reset_id` из ETCP (см. `61c572c`), применённый к сервисному слою +маршрутизации `etcp_router` (src/routing_layer/etcp_router.c). + +## 1. Контекст + +`etcp_router` — «упрощённый TCP поверх ETCP»: своя seq-нумерация (`tx_seq`/`rx_seq`), +reorder (`recv_q`), inflight-контроль (`inflight_q`), ретрансмиты, START/RST handshake. + +Текущий механизм детекции рестарта пира: + +- `SVC_ROUTE_HDR.flags & ROUTER_FLAG_START` — флаг на первом data-пакете сеанса + (`seq==0 && !start_sent && pl_len>0`, `etcp_router.c:207`). +- `router_check_peer_restart` (`etcp_router.c:1055`) — если пришёл START на уже + синхронизированное соединение (`peer_sync_done==1`), то это рестарт пира; + **дедуп ретрансмитов** — побайтовое сравнение с `last_start_dgram` + (`etcp_router.c:1063`). Совпало → это ретрансмит (не рестарт). Не совпало → + `etcp_router_conn_restart` (полный сброс + `sess_id++`). +- `ROUTER_FLAG_RST` + `router_restart_send` (`etcp_router.c:285`) — сброс только + TX-стороны; повторные RST гасятся через `send_restart_pending`. +- `sess_id` (`etcp_router.h:121`) — **локальный счётчик переустановок, на wire + не пишется**. Используется только в `DEBUG_INFO`. + +## 2. Проблема (аналог ETCP seq-коллизии) + +Дедуп рестарта строится на **побайтовой идентичности** `last_start_dgram`. Это тот +же класс хрупкости, что убрали в ETCP: + +1. **Байт-идентичный START** может быть и ретрансмитом, и (теоретически) новым + сеансом — если у нового START совпали `seq==0`, `svc_id`, `timestamp` (16-битные + часы 0.1ms заворачиваются каждые ~6.5 сек) и payload. Тогда рестарт **не + детектится**. +2. Если рестарт не распознан: у пира `tx_seq` ушёл в 0, а у нас `rx_seq` остался + «высоким». В `router_handle_data_packet` (`etcp_router.c:1104`) ветка + `(rx_seq - seq) > 0` / `queue_find_data_by_index` считает пакеты новой сессии + **дубликатами** → молчаливый `c_dup_dropped` + пере-ACK (как `RX dup` в ETCP) → + потеря сервисных пакетов. +3. `sess_id` не на wire → пир вообще не может отличить «мы перезапустились» от + «продолжаем сеанс» иначе как по байтам START. + +То есть нужен надёжный маркер **эпохи сеанса** на wire — аналог `reset_id`. + +## 3. Решение (аналог reset_id) + +Семантика та же, что в ETCP: + +- эпоха генерится **один раз** при создании `ETCP_ROUTER_CONN`; +- меняется **только** при (фатальном) рестарте соединения; +- при приёме START/RST с **другой** эпохой → рестарт (adopt); +- **первая** увиденная эпоха принимается без рестарта (свежие при создании); +- дедуп ретрансмитов START — по эпохе, а не по байтам. + +Два варианта источника эпохи: + +- **(A) Перенести `sess_id` на wire** — он уже есть как `uint32` счётчик + (0 при создании, `++` при рестарте). Минимально инвазивно. +- **(B) Новый `reset_id` (uint64, случайный)** — симметрично ETCP; случайный + надёжнее счётчика (нет wrap-around и предсказуемости), но чуть больше поле. + +Рекомендация — **(A)** как минимальный шаг (поле уже есть), при желании апгрейд +до (B). + +### Разрыв цикла (master/slave) + +Как и в ETCP, симметричное «увидел другую эпоху → рестарт» зацикливается при +одновременном рестарте обеих сторон (A++ → B++ → A видит B → A++ → …). + +Ти-брейкер по `node_id` (как в ETCP): + +- на приёме START/RST с чужой эпохой: + - **master** (меньший `node_id`) — держит свою эпоху, шлёт RST пиру; + - **slave** (больший `node_id`) — принимает эпоху пира и рестартится. + +Существующие предохранители (`send_restart_pending`, `router_check_peer_restart` +на стороне master) сохраняются. + +## 4. Wire-формат + +`SVC_ROUTE_HDR` (`etcp_router.h:26`) сейчас 33 байта: + +``` +cmd(1) group_id(8) dst(8) src(8) seq(4) svc_id(1) flags(1) timestamp(2) +``` + +Добавить поле эпохи (только START/RST, в обычных data/ACK — 0): + +- вариант (A): `uint32_t sess_id` → заголовок 33→37 байт (упакован); +- вариант (B): `uint8_t reset_id[8]` → 33→41 байт. + +Заполняется в `router_build_packet` (START) и `router_send_rst`; читается в +`router_check_peer_restart` / обработчике RST. + +## 5. Шаги реализации + +Файлы: `src/routing_layer/etcp_router.h`, `src/routing_layer/etcp_router.c`, +(опц.) `tests/`. + +1. **Поле в заголовке**: добавить `sess_id`/`reset_id` в `struct SVC_ROUTE_HDR`; + обновить `SVC_ROUTE_HDR_SIZE`. +2. **Локальное состояние** (`struct ETCP_ROUTER_CONN`): + - `uint32_t sess_id` (уже есть) — оставить как «моя эпоха»; + - добавить `uint32_t peer_sess_id` — последняя увиденная эпоха пира. + - `router_conn_reset` (`etcp_router.c:380`) НЕ сбрасывает `sess_id` + (комментарий на :379 уже это гарантирует). +3. **Запись эпохи**: + - `router_build_packet` (`:193`) — при `ROUTER_FLAG_START` писать `sess_id`; + - `router_send_rst` (`:251`) — писать `sess_id`. +4. **Детект рестарта по эпохе** — переписать `router_check_peer_restart` (`:1055`): + - если `peer_sess_id==0` → первая эпоха, принять без рестарта (сохранить); + - если `hdr->sess_id == peer_sess_id` → ретрансмит (не рестарт); + - если отличается → master/slave: slave → `etcp_router_conn_restart` + adopt; + master → держать свою, при необходимости RST. + - `last_start_dgram`-байт-дедуп заменить на это сравнение (поле можно удалить). +5. **RST-обработчик** (`:1151`) — при RST также сверять `sess_id` (дедуп RST по + эпохе вместо/в дополнение к `send_restart_pending`). +6. **Инициализация** — `peer_sess_id = 0` в `router_conn_reset`/`etcp_router_conn_get`. + +## 6. Тест + +`tests/test_etcp_router_*` (существующие `test_etcp_router`, `test_etcp_router_reconnect`, +`test_etcp_router_unit`) — прогнать после правки заголовка (37/41 байт). + +Добавить (по возможности) регрессию, аналогичную `test_etcp_seq_collision`: + +- два инстанса, роутер-коннект, отправка N пакетов; +- принудительный рестарт одной стороны (смена эпохи); +- assert: пакеты новой эпохи доставлены (не `c_dup_dropped`), обе стороны сходятся + к одной эпохе. + +## 7. Решения (зафиксировано) + +1. **Вариант (B)** — `uint64_t reset_id`, случайный (`random_bytes`). Причина: вариант (A) + `sess_id` стартует с 0 и конфликтует с сентинелом «эпоха не задана». На wire — native + order (как остальные uint64-поля заголовка, SVC_ROUTE не использует htobe*). +2. **master/slave по node_id** — нужен: `etcp_router_conn_restart` делает полный + двунаправленный сброс, без ти-брейкера симметричный рестарт зациклится. master + (меньший node_id) держит эпоху и шлёт RST; slave принимает и ресетится. +3. **route_crypto** — совместим: заголовок не шифруется (только payload), подписывается + только в SIGN-режиме; `SVC_ROUTE_HDR_SIZE = sizeof(...)` авто-адаптируется. START + идёт через `route_crypto_encode`, RST/ACK/CLOSE — plain header-only. + +## 8. Статус + +Реализовано и проверено: `etcp_router.h/.c` + `tests/test_etcp_router_unit.c`. +Сборка `make clean && make` без ошибок; `./check.sh` → 74 passed / 0 failed / 1 skipped; +`test_etcp_router_unit` 34/34, `test_etcp_router` и `test_etcp_router_reconnect` — PASS. diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index ccaac870..b3383833 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -206,6 +206,7 @@ static int router_build_packet(struct ETCP_ROUTER_CONN* rconn, const uint8_t* pa uint8_t f = 0; if (!rconn->start_sent && hdr->seq == 0 && pl_len > 0) f |= ROUTER_FLAG_START; hdr->flags = f; + hdr->reset_id = (f & ROUTER_FLAG_START) ? rconn->reset_id : 0; } hdr->timestamp = get_current_timestamp(); if (pl_len > 0) memcpy(base + SVC_ROUTE_HDR_SIZE, payload, pl_len); @@ -235,6 +236,7 @@ static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flag_bits) hdr->svc_id = rconn->svc_id; hdr->flags = flag_bits; hdr->timestamp = get_current_timestamp(); + hdr->reset_id = 0; struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } entry->dgram = (uint8_t*)hdr; @@ -267,6 +269,7 @@ static void router_send_rst(struct ETCP_ROUTER_CONN* rconn) { hdr->flags = ROUTER_FLAG_RST; hdr->svc_id = rconn->svc_id; hdr->timestamp = get_current_timestamp(); + hdr->reset_id = rconn->reset_id; struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } entry->dgram = (uint8_t*)hdr; @@ -373,10 +376,9 @@ static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { queue_resume_callback(rconn->incoming_q); queue_free(rconn->incoming_q); rconn->incoming_q = NULL; } - if (rconn->last_start_dgram) { u_free(rconn->last_start_dgram); rconn->last_start_dgram = NULL; rconn->last_start_dgram_len = 0; } } -// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/last_dgram_ts). +// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/reset_id/last_dgram_ts). static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { router_conn_free_queues(rconn); @@ -403,6 +405,7 @@ static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { rconn->tx_seq = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0; rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; rconn->send_restart_pending = 0; + rconn->peer_reset_id = 0; rconn->no_route = 0; rconn->no_route_timer = NULL; rconn->no_ack_count = 0; rconn->closed = 0; rconn->c_pkts_sent = 0; rconn->c_pkts_send_err = 0; rconn->c_pkts_rcvd = 0; rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_retrans_done = 0; @@ -769,6 +772,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { hdr->seq = rconn->rx_seq; hdr->svc_id = rconn->svc_id; hdr->flags = 0; + hdr->reset_id = 0; if (rconn->last_recv_updated) { uint64_t now = get_time_tb(); hdr->timestamp = rconn->last_recv_pkt_ts + (uint32_t)(now - rconn->last_recv_pkt_local_tb); @@ -1054,20 +1058,37 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr, size_t wire_len) { + (void)wire_len; struct ETCP_ROUTER_CONN* rconn = *prconn; if (!(hdr->flags & ROUTER_FLAG_START)) return 0; // не START — не рестарт if (!rconn->peer_sync_done) return 0; // чистый: принимаем как первый пакет сеанса - // dirty + START: дедуп по payload. Ретрансмит уже принятого init-пакета побайтно идентичен — - // это НЕ новый сеанс, рестарт не делаем (пере-ACK сделает dup-ветка в router_handle_data_packet). - if (rconn->last_start_dgram && rconn->last_start_dgram_len == wire_len - && memcmp(rconn->last_start_dgram, hdr, wire_len) == 0) { + // dirty + START: детект рестарта по эпохе reset_id (аналог etcp_conn_apply_peer_reset_id). + uint64_t peer_rid = hdr->reset_id; + if (peer_rid == rconn->peer_reset_id) return 0; // ретрансмит известного START — не рестарт + + int i_am_master = inst->node_id < hdr->src_node_id; + if (peer_rid == rconn->reset_id) { + // пир принял НАШУ эпоху (slave-адопт с нашей стороны) — полный рестарт приёма, эпоху сохраняем. + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, + "router: peer adopted our epoch %016llx svc_id=%u from %016llx — resetting rx", + (unsigned long long)rconn->reset_id, hdr->svc_id, (unsigned long long)hdr->src_node_id); + } else if (i_am_master) { + // чужой неизвестный epoch, мы master — держим свою эпоху и шлём RST (slave пере-синхронизируется на нас). + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, + "router: peer reset_id=%016llx != mine=%016llx — master keeps epoch, RST to %016llx svc_id=%u", + (unsigned long long)peer_rid, (unsigned long long)rconn->reset_id, + (unsigned long long)hdr->src_node_id, hdr->svc_id); + router_send_rst(rconn); return 0; + } else { + // slave принимает эпоху мастера и ресетится. + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, + "router: peer reset_id=%016llx != mine=%016llx — slave adopts, svc_id=%u from %016llx", + (unsigned long long)peer_rid, (unsigned long long)rconn->reset_id, + hdr->svc_id, (unsigned long long)hdr->src_node_id); + rconn->reset_id = peer_rid; } - - DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, - "router: peer restart svc_id=%u from %016llx", - hdr->svc_id, (unsigned long long)hdr->src_node_id); etcp_router_conn_restart(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id); *prconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id); if (!*prconn) return -1; @@ -1083,13 +1104,8 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC if (!rconn->peer_sync_done) { rconn->peer_sync_done = 1; rconn->rx_seq = seq; - // сохранить init-пакет (START) для дедупа ретрансмитов - if (((const struct SVC_ROUTE_HDR*)wire)->flags & ROUTER_FLAG_START) { - if (rconn->last_start_dgram) u_free(rconn->last_start_dgram); - rconn->last_start_dgram = u_malloc(wire_len); - if (rconn->last_start_dgram) { memcpy(rconn->last_start_dgram, wire, wire_len); rconn->last_start_dgram_len = wire_len; } - else rconn->last_start_dgram_len = 0; - } + // запомнить эпоху сеанса пира (для детекта рестарта по reset_id) + rconn->peer_reset_id = ((const struct SVC_ROUTE_HDR*)wire)->reset_id; } {// SEQ out of bounds @@ -1155,6 +1171,13 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: RST (дубль) от %016llx svc_id=%u — игнор", (unsigned long long)hdr->src_node_id, hdr->svc_id); } else { + // slave (больший node_id) принимает эпоху мастера из RST перед рестартом отправки. + if (inst->node_id > hdr->src_node_id && hdr->reset_id != rconn->reset_id) { + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, + "router: RST от %016llx svc_id=%u — slave adopts epoch %016llx", + (unsigned long long)hdr->src_node_id, hdr->svc_id, (unsigned long long)hdr->reset_id); + rconn->reset_id = hdr->reset_id; + } DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: RST от %016llx svc_id=%u — рестарт отправки", (unsigned long long)hdr->src_node_id, hdr->svc_id); router_restart_send(rconn); @@ -1281,8 +1304,8 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uin struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, remote_node_id, svc_id); if (!rconn) return; - DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx sess_id=%u→%u — resetting local state", - svc_id, (unsigned long long)remote_node_id, rconn->sess_id, rconn->sess_id + 1); + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx reset_id=%016llx — resetting local state", + svc_id, (unsigned long long)remote_node_id, (unsigned long long)rconn->reset_id); etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id]; if (cb) { @@ -1301,7 +1324,6 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uin } router_conn_reset(rconn); - rconn->sess_id++; } void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, @@ -1351,6 +1373,9 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, memcpy(rconn->ll.data + 8, &remote_node_id, 8); rconn->ll.data[16] = svc_id; rconn->last_dgram_ts = get_current_timestamp(); + rconn->reset_id = 0; + if (random_bytes((uint8_t*)&rconn->reset_id, 8) != 0) // начальная случайная эпоха сеанса + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: random_bytes failed — reset_id=0 svc_id=%u", svc_id); router_conn_reset(rconn); if (!rconn->send_q || !rconn->recv_q || !rconn->inflight_q || !rconn->incoming_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new failed"); diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 8169d1ff..391f466c 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -4,8 +4,9 @@ // Подпись/шифрование вынесены в автономный модуль route_crypto (encode/decode). // // Формат SVC_ROUTE пакета: -// [cmd:1] [group_id:8] [dst_node_id:8] [src_node_id:8] [seq:4] [svc_id:1] [flags:1] [timestamp:2] [payload...] +// [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] [payload...] // flags: bit7=START, bit6=RST, bit3=SIGNED, bit2=ENCRYPTED, bit1=CLOSE +// reset_id — эпоха сеанса: пишется только в START/RST, в data/ACK = 0 // ACK-пакет: тот же заголовок, seq=rx_seq, payload_len=0 #ifndef ETCP_ROUTER_H #define ETCP_ROUTER_H @@ -32,9 +33,10 @@ struct SVC_ROUTE_HDR { uint8_t svc_id; // идентификатор сервиса uint8_t flags; // bit7=START, bit6=RST, bit3=SIGNED, bit2=ENCRYPTED, bit1=CLOSE uint16_t timestamp; // часы отправителя (0.1ms), для RTT; ACK: эхо последнего data-пакета + uint64_t reset_id; // эпоха сеанса (native order, как group_id/seq); только START/RST, в data/ACK = 0 }; #pragma pack(pop) -#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 33 +#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 41 #define SVC_ROUTE_MAX_BINDINGS 256 // Единый формат доставки сервисной кодограммы (router → сервис): @@ -118,11 +120,10 @@ struct ETCP_ROUTER_CONN { void* no_route_timer; // таймер 20ms проверки появления маршрута uint8_t no_ack_count; // счётчик последовательных ретрансмиссий без ACK uint8_t closed; // 1 = в процессе закрытия, таймеры игнорируют - uint32_t sess_id; // локальный счётчик переустановок (0 при создании, ++ при рестарте; на wire не пишется) + uint64_t reset_id; // метка эпохи сеанса (случайная при создании; master авторитет, slave принимает эпоху мастера) + uint64_t peer_reset_id; // последняя увиденная эпоха сеанса пира (0 = ещё не синхронизированы) uint8_t start_sent; // 0 = нужно отправить START в первом data uint8_t peer_sync_done; // 0 = rx_seq ещё не синхронизирован с удалённым seq (авто-создание/рестарт) - uint8_t* last_start_dgram; // копия wire-пакета последнего принятого START (дедуп ретрансмитов) - size_t last_start_dgram_len; uint8_t send_restart_pending; // 1 = RST принят и отправка переинициализирована (игнор повторных RST до первого ACK) // Ретрансмиты: inflight очередь diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index 633beb01..1028c8b1 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -88,6 +88,34 @@ static void inject(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst, recv_cb(&fake_conn, e); } +// Аналог inject, но с явным reset_id (для тестов эпохи/рестарта). +static void inject_epoch(etcp_recv_fn recv_cb, struct UTUN_INSTANCE* inst, + uint64_t src, uint8_t svc_id, uint32_t seq, + const uint8_t* pl, size_t pl_len, int is_ack, uint8_t flags, uint64_t reset_id) { + struct SVC_ROUTE_HDR hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.cmd = ETCP_RT_ID_SVC_ROUTE; + hdr.group_id = TOPO_GROUP_UTUN; + hdr.dst_node_id = inst->node_id; + hdr.src_node_id = src; + hdr.seq = seq; + hdr.svc_id = svc_id; + hdr.flags = flags; + hdr.reset_id = reset_id; + + size_t total = SVC_ROUTE_HDR_SIZE + (is_ack ? 0 : pl_len); + struct ll_entry* e = queue_entry_new(0); + if (!e) { printf(" inject_epoch: queue_entry_new failed\n"); return; } + e->dgram = u_malloc(total); + if (!e->dgram) { queue_entry_free(e); return; } + memcpy(e->dgram, &hdr, SVC_ROUTE_HDR_SIZE); + if (!is_ack && pl_len > 0) + memcpy(e->dgram + SVC_ROUTE_HDR_SIZE, pl, pl_len); + e->len = total; + + recv_cb(&fake_conn, e); +} + // ======================== Setup ======================== #define SETUP() do { \ ua = uasync_create(); \ @@ -699,22 +727,24 @@ static int test_rst_flag(void) { } static int test_start_retransmit_no_restart(void) { - TEST("START retransmit (same payload) — no restart"); + TEST("START retransmit (same epoch) — no restart"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; SETUP(); g_notify_count = 0; - int n = rand() % 10 + 1; + uint64_t epoch = 0x1111111111111111ULL; rx_reset(0xC1); - pump_data(recv_cb, &inst, 0xC1, n, 0); - if (rx.delivered != n) FAIL("initial pump failed"); + uint8_t data[] = { TEST_SVC_ID, 0xC1, 0 }; + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START, epoch); + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 1, data + 1, 2, 0, 0, 0); + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 2, data + 1, 2, 0, 0, 0); + if (rx.delivered != 3) FAIL("initial delivery failed"); g_notify_count = 0; - // ретрансмит первого пакета (seq=0, START, тот же payload) — не должен рестартовать - uint8_t data[] = { TEST_SVC_ID, 0xC1, 0 }; - inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START); - if (g_notify_count != 0) FAIL("START retransmit should NOT restart"); - if (rx.delivered != n) FAIL("duplicate START should not deliver"); + // ретрансмит первого пакета (seq=0, START, та же эпоха) — не должен рестартовать + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START, epoch); + if (g_notify_count != 0) FAIL("START retransmit (same epoch) should NOT restart"); + if (rx.delivered != 3) FAIL("duplicate START should not deliver"); TEARDOWN(); PASS(); @@ -722,44 +752,83 @@ static int test_start_retransmit_no_restart(void) { } static int test_client_restart(void) { - TEST("client restart — START flag detected after random data"); + TEST("client restart — new epoch after sync"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; SETUP(); g_notify_count = 0; - int n = rand() % 10 + 1; rx_reset(0xB1); - pump_data(recv_cb, &inst, 0xB1, n, 0); - if (rx.delivered != n) FAIL("initial pump failed"); + uint8_t data[] = { TEST_SVC_ID, 0xB1, 0 }; + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START, 0xAAAA000000000001ULL); + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 1, data + 1, 2, 0, 0, 0); + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 2, data + 1, 2, 0, 0, 0); + if (rx.delivered != 3) FAIL("initial delivery failed"); + + struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + if (!c) FAIL("rconn not found"); rx_reset(0xB2); - uint8_t data[] = { TEST_SVC_ID, 0xB2, 0 }; - inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START); + uint8_t data2[] = { TEST_SVC_ID, 0xB2, 0 }; + inject_epoch(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, data2 + 1, 2, 0, ROUTER_FLAG_START, 0xAAAA000000000002ULL); if (g_notify_count != 1) FAIL("restart not notified"); - if (rx.delivered < 1) FAIL("data after restart not delivered"); + if (rx.delivered != 1) FAIL("data after restart not delivered"); + // slave (мы 0xBBBB > пир 0xAAAA) должен принять эпоху пира + if (c->reset_id != 0xAAAA000000000002ULL) FAIL("slave did not adopt peer epoch"); + if (c->peer_reset_id != 0xAAAA000000000002ULL) FAIL("peer_reset_id not set to adopted epoch"); + etcp_router_conn_close(c); TEARDOWN(); PASS(); return 0; } -static int test_sess_id_increment(void) { - TEST("sess_id — 0 on create, increments on restart"); +static int test_master_keeps_epoch(void) { + TEST("master — keeps own epoch on peer epoch change"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; SETUP(); + g_notify_count = 0; - struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + // пир 0xCCCC > мы 0xBBBB → мы master. + uint64_t big_peer = 0xCCCC000000000001ULL; + rx_reset(0xD1); + uint8_t data[] = { TEST_SVC_ID, 0xD1, 0 }; + inject_epoch(recv_cb, &inst, big_peer, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START, 0x1111111111111111ULL); + if (rx.delivered != 1) FAIL("initial delivery failed"); + + struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, big_peer, TEST_SVC_ID); if (!c) FAIL("rconn not found"); - if (c->sess_id != 0) FAIL("sess_id != 0 on fresh conn"); + uint64_t my_rid = c->reset_id; - etcp_router_conn_restart(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); - c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); - if (!c) FAIL("rconn not found after first restart"); - if (c->sess_id != 1) FAIL("sess_id != 1 after first restart"); + g_notify_count = 0; + // пир перезапустился с новой эпохой — master держит свою эпоху (шлёт RST), локального рестарта нет. + inject_epoch(recv_cb, &inst, big_peer, TEST_SVC_ID, 0, data + 1, 2, 0, ROUTER_FLAG_START, 0x2222222222222222ULL); + if (g_notify_count != 0) FAIL("master should NOT restart"); + c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, big_peer, TEST_SVC_ID); + if (!c) FAIL("rconn not found after"); + if (c->reset_id != my_rid) FAIL("master reset_id changed"); + + etcp_router_conn_close(c); + TEARDOWN(); + PASS(); + return 0; +} + +static int test_reset_id(void) { + TEST("reset_id — random on create, preserved on restart"); + struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; + SETUP(); + + struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + if (!c) FAIL("rconn not found"); + uint64_t rid0 = c->reset_id; + if (rid0 == 0) FAIL("reset_id == 0 on fresh conn (should be random)"); + if (c->peer_reset_id != 0) FAIL("peer_reset_id != 0 on fresh conn"); etcp_router_conn_restart(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); - if (c->sess_id != 2) FAIL("sess_id != 2 after second restart"); + if (!c) FAIL("rconn not found after restart"); + if (c->reset_id != rid0) FAIL("reset_id changed after local restart"); + if (c->peer_reset_id != 0) FAIL("peer_reset_id not reset after restart"); etcp_router_conn_close(c); TEARDOWN(); @@ -768,7 +837,7 @@ static int test_sess_id_increment(void) { } static int test_server_reinit(void) { - TEST("server reinit — new rconn gets start_sent=0 sess_id=0"); + TEST("server reinit — new rconn gets start_sent=0 peer_reset_id=0"); struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; SETUP(); @@ -786,7 +855,8 @@ static int test_server_reinit(void) { struct ETCP_ROUTER_CONN* c = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); if (!c) FAIL("rconn after reinit"); if (c->start_sent != 0) FAIL("start_sent != 0 after reinit"); - if (c->sess_id != 0) FAIL("sess_id != 0 after reinit"); + if (c->peer_reset_id != 0) FAIL("peer_reset_id != 0 after reinit"); + if (c->reset_id == 0) FAIL("reset_id == 0 after reinit"); etcp_router_conn_close(c); TEARDOWN(); @@ -1184,7 +1254,8 @@ int main(void) { test_rst_flag(); test_start_retransmit_no_restart(); test_client_restart(); - test_sess_id_increment(); + test_master_keeps_epoch(); + test_reset_id(); test_server_reinit(); if (setup_sign_keys() != 0) { printf(" SKIP signed tests — crypto init failed\n"); }