Fix ETCP session recovery and router lifecycle; confirm chat invites before publishing
This commit is contained in:
evgeny committed
1 parent
018c8e795d
commit
3131dcbb7c
67 files changed
+3005
-1249
No files matched your search
@@ -252,6 +252,7 @@ void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void
|
||||
if (!entry || !entry->mgr || entry->state == CONN_MGR_STATE_DISCONNECTED) return;
|
||||
switch (ncd_ev) {
|
||||
case NCD_EVENT_UP:
|
||||
entry->ncd_down_notified = 0;
|
||||
if (entry->state == CONN_MGR_STATE_CONNECTED) return;
|
||||
{ int won = entry->reverse_active ? CONN_TYPE_REVERSE : CONN_TYPE_DIRECT;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "link: ETCP handshake OK 0x%016llx — %s (reverse_active=%d indirect_active=%d)",
|
||||
@@ -303,7 +304,18 @@ void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void
|
||||
if (entry->reverse_active || entry->indirect_active) return;
|
||||
cm_start_parallel(entry);
|
||||
break;
|
||||
case NCD_EVENT_CLOSED:
|
||||
if (entry->ncd_handle == ncd_h) entry->ncd_handle = NULL;
|
||||
node_conn_direct_close(ncd_h);
|
||||
if (entry->state != CONN_MGR_STATE_CONNECTED) {
|
||||
cm_ncd_callback(NULL, NCD_EVENT_TIMEOUT, entry);
|
||||
return;
|
||||
}
|
||||
entry->state = CONN_MGR_STATE_FAILED; entry->conn_type = CONN_TYPE_NONE;
|
||||
if (!entry->ncd_down_notified) cm_deliver_event(entry, CONN_EVENT_DOWN);
|
||||
return;
|
||||
case NCD_EVENT_DOWN:
|
||||
entry->ncd_down_notified = 1;
|
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "link: ETCP link DROPPED to 0x%016llx",
|
||||
(unsigned long long)entry->node_id);
|
||||
cm_deliver_event(entry, CONN_EVENT_DOWN);
|
||||
|
||||
@@ -55,6 +55,15 @@ void conn_mgr_update_best_candidates(struct CONN_MGR* mgr, uint64_t node_id, uin
|
||||
{ struct CONN_MGR_CANDIDATE t = mgr->best_candidates[i]; mgr->best_candidates[i]=mgr->best_candidates[i-1]; mgr->best_candidates[i-1]=t; }
|
||||
}
|
||||
|
||||
/* Маршрут может вести через посредника. Для прямой пробы нужны transport peer и его живой линк. */
|
||||
static struct ETCP_LINK* cm_active_direct_link(struct TOPO_GROUP* group, uint64_t node_id) {
|
||||
struct ETCP_CONN* conn = instance_find_conn(group->instance,node_id);
|
||||
if (!conn || conn->state != 1 || conn->close_requested) return NULL;
|
||||
for (struct ETCP_LINK* link = conn->links; link; link = link->next)
|
||||
if (link->initialized && link->link_status) return link;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* ═══════ bg_ping ═══════ */
|
||||
|
||||
static void cm_bg_ping_noop_cb(int s, uint16_t r, void* a, uint64_t n, const uint8_t* d, size_t l) { (void)s;(void)r;(void)a;(void)n;(void)d;(void)l; }
|
||||
@@ -124,9 +133,7 @@ void cm_bg_ping_timer_cb(void* arg) {
|
||||
if (e) {
|
||||
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
|
||||
if (nq->node_id != mgr->instance->node_id) {
|
||||
int has_active = 0;
|
||||
struct ETCP_CONN* dc = topo_group_find_conn_for_node(group, nq->node_id);
|
||||
if (dc && dc->links) { struct ETCP_LINK* l = dc->links; while (l) { if (l->initialized && l->link_status) { has_active = 1; break; } l = l->next; } }
|
||||
int has_active = cm_active_direct_link(group,nq->node_id) != NULL;
|
||||
if (!has_active) {
|
||||
struct TOPO_NODE* bg_ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id);
|
||||
if (bg_ni) cm_bg_ping_to_node(mgr, bg_ni);
|
||||
@@ -141,10 +148,7 @@ rearm:
|
||||
/* ═══════ candidate_ping ═══════ */
|
||||
|
||||
static int cm_has_active_conn(struct TOPO_GROUP* g, uint64_t nid) {
|
||||
struct ETCP_CONN* c = topo_group_find_conn_for_node(g, nid);
|
||||
if (!c || !c->links) return 0;
|
||||
struct ETCP_LINK* l = c->links; while (l) { if (l->initialized && l->link_status) return 1; l = l->next; }
|
||||
return 0;
|
||||
return cm_active_direct_link(g,nid) != NULL;
|
||||
}
|
||||
|
||||
/* Обновление кандидатов-посредников: удаляет протухших (>30с без свежих RTT),
|
||||
@@ -182,18 +186,20 @@ void cm_candidate_ping_timer_cb(void* arg) {
|
||||
&& (now - cm_get_node_max_probe_time(nq)) >= CONN_MGR_CANDIDATE_STALE_TB)
|
||||
{ DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: candidate 0x%016llx stale, removing", (unsigned long long)nid);
|
||||
memmove(&mgr->best_candidates[i], &mgr->best_candidates[i+1], (--mgr->best_candidate_count - i) * sizeof(mgr->best_candidates[0])); i--; continue; }
|
||||
if (cm_has_active_conn(mgr->group, nid)) {
|
||||
struct ETCP_CONN* dc = topo_group_find_conn_for_node(mgr->group, nid);
|
||||
struct ETCP_LINK* link = cm_active_direct_link(mgr->group,nid);
|
||||
if (link) {
|
||||
struct ETCP_CONN* dc = link->etcp;
|
||||
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id);
|
||||
/* пир в SLEEP-фазе — прямое соединение живо, пинг будит его: пропускаем */
|
||||
if (ni && dc && dc->links && !dc->peer_sleep_phase) {
|
||||
if (dc->links->is_tcp) {
|
||||
if (ni && !dc->peer_sleep_phase) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP,"candidate probe: direct peer=%016llx link=%u",(unsigned long long)nid,link->local_link_id);
|
||||
if (link->is_tcp) {
|
||||
struct stcp_client** slot = NULL;
|
||||
for (uint8_t i = 0; i < CONN_MGR_MAX_CANDIDATES; i++)
|
||||
if (!mgr->cand_tcp_pings[i]) { slot = &mgr->cand_tcp_pings[i]; break; }
|
||||
if (slot) etcp_send_tcp_ping(mgr->instance, ni->public_key, &dc->links->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_cand_tcp_ping_cb, slot, slot);
|
||||
if (slot) etcp_send_tcp_ping(mgr->instance, ni->public_key, &link->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_cand_tcp_ping_cb, slot, slot);
|
||||
} else
|
||||
etcp_send_ping(mgr->instance, ni->public_key, &dc->links->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_bg_ping_noop_cb, NULL, NULL, 0);
|
||||
etcp_send_ping(mgr->instance, ni->public_key, &link->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_bg_ping_noop_cb, NULL, NULL, 0);
|
||||
}
|
||||
} else route_connectivity_probe_node(mgr->instance, mgr->group, nq);
|
||||
}
|
||||
|
||||
@@ -42,6 +42,7 @@ enum CONN_MGR_STATE {
|
||||
CONN_MGR_STATE_DISCONNECTED = 0,
|
||||
CONN_MGR_STATE_CONNECTING = 1,
|
||||
CONN_MGR_STATE_CONNECTED = 2,
|
||||
CONN_MGR_STATE_FAILED = 3,
|
||||
};
|
||||
|
||||
enum CM_TRY_STATE {
|
||||
@@ -103,6 +104,7 @@ struct CONN_MGR_ENTRY {
|
||||
uint8_t state, conn_type;
|
||||
uint64_t last_traffic_tb;
|
||||
struct CONN_MGR_HANDLE* handles; struct NODE_CONN_DIRECT* ncd_handle;
|
||||
uint8_t ncd_down_notified;
|
||||
uint64_t intermediaries[3]; uint8_t intermediariy_count, rr_idx;
|
||||
uint8_t local_scan_state, main_connect_state, db_loaded;
|
||||
uint8_t reverse_active, indirect_active, indirect_ready;
|
||||
|
||||
+326
-428
File diff suppressed because it is too large.
Load diff
@@ -4,9 +4,11 @@
|
||||
// Подпись/шифрование вынесены в автономный модуль 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] [reset_id:8] [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]
|
||||
// [peer_reset_id:8] [challenge:8] [payload...]
|
||||
// flags: bit7=START, bit6=RST, bit3=SIGNED, bit2=ENCRYPTED, bit1=CLOSE
|
||||
// reset_id — эпоха сеанса: пишется только в START/RST, в data/ACK = 0
|
||||
// reset_id/peer_reset_id — идентификаторы отправителя/получателя во всех пакетах.
|
||||
// START=HELLO, RST=CHALLENGE, START|RST=CONFIRM; challenge используется только в handshake.
|
||||
// ACK-пакет: тот же заголовок, seq=rx_seq, payload_len=0
|
||||
#ifndef ETCP_ROUTER_H
|
||||
#define ETCP_ROUTER_H
|
||||
@@ -33,10 +35,12 @@ 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
|
||||
uint64_t reset_id; // случайный идентификатор локального экземпляра router_conn
|
||||
uint64_t peer_reset_id; // идентификатор экземпляра получателя (0 только в HELLO)
|
||||
uint64_t challenge; // случайный запрос/эхо для подтверждения актуальности пира
|
||||
};
|
||||
#pragma pack(pop)
|
||||
#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 41
|
||||
#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 57
|
||||
#define SVC_ROUTE_MAX_BINDINGS 256
|
||||
|
||||
// Единый формат доставки сервисной кодограммы (router → сервис):
|
||||
@@ -103,6 +107,7 @@ struct ETCP_ROUTER_CONN {
|
||||
uint8_t last_recv_updated; // флаг: есть данные для эха в ACK
|
||||
|
||||
uint32_t tx_seq; // следующий seq для отправки
|
||||
uint32_t tx_sent; // граница фактически отправленных пакетов (ACK <= tx_sent)
|
||||
uint32_t rx_seq; // ожидаемый seq для сборки (next expected)
|
||||
uint32_t tx_acked; // сколько наших пакетов подтвердил remote (для inflight)
|
||||
uint32_t last_sent_ack_seq; // последний отправленный ACK (= rx_seq на момент отправки)
|
||||
@@ -116,17 +121,23 @@ struct ETCP_ROUTER_CONN {
|
||||
|
||||
struct ll_queue* send_q; // очередь ожидающих отправки (inflight полон)
|
||||
struct queue_waiter_handle send_waiter; // waiter на send_input_q (по 1 пакету, round-robin)
|
||||
struct ll_queue* send_waiter_q; // очередь регистрации, не результат повторного поиска маршрута
|
||||
uint32_t retrans_seq; // следующий seq текущего прохода ретрансмитов
|
||||
uint32_t retrans_end; // конец прохода (исключительно)
|
||||
void* watchdog_timer; // DEBUG-watchdog: медленная проверка инварианта drain
|
||||
uint8_t send_blocked; // 1 = inflight полон, ждём ack
|
||||
uint8_t no_route; // 1 = нет BGP-маршрута, передача приостановлена
|
||||
void* no_route_timer; // таймер 20ms проверки появления маршрута
|
||||
uint8_t no_ack_count; // счётчик последовательных ретрансмиссий без ACK
|
||||
uint8_t closed; // 1 = в процессе закрытия, таймеры игнорируют
|
||||
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 send_restart_pending; // 1 = RST принят и отправка переинициализирована (игнор повторных RST до первого ACK)
|
||||
uint64_t reset_id; // собственный идентификатор; не меняется при рестарте пира
|
||||
uint64_t peer_reset_id; // подтверждённый идентификатор пира (0 = handshake не завершён)
|
||||
uint64_t pending_peer_id; // кандидат, ещё не имеющий права менять состояние сессии
|
||||
uint64_t pending_challenge;
|
||||
uint64_t challenge_sent_tb;
|
||||
void* handshake_timer;
|
||||
uint8_t start_sent; // получен первый ACK данных
|
||||
uint8_t peer_sync_done; // 1 = завершено подтверждение peer ID
|
||||
|
||||
// Ретрансмиты: inflight очередь
|
||||
struct ll_queue* inflight_q; // хеш по seq (4 байта), непрерывный блок без SACK
|
||||
@@ -200,6 +211,7 @@ int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn c
|
||||
int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id);
|
||||
|
||||
// Отправить сервисный пакет (авто-conn, seq, inflight-контроль через send_q).
|
||||
// Принимает владение entry во всех исходах, включая ошибку.
|
||||
// mode — битовая маска ROUTE_CRYPTO_SIGN / ROUTE_CRYPTO_ENCRYPT (см. route_crypto.h), 0 = обычный пакет.
|
||||
int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, struct ll_entry* entry, int force, int mode);
|
||||
|
||||
@@ -236,14 +248,14 @@ int etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group_id, u
|
||||
// 4. Служебные функции (для каркаса: etcp.c, topo_group.c)
|
||||
// ====================================================================
|
||||
|
||||
// Сбросить ретрансмиты для всех router_conn к узлу (без колбэков, без CLOSE)
|
||||
// Отвязать отправку от сбрасываемого транспорта; сохранить inflight и ретрансмиты
|
||||
// Используется при etcp_conn_reinit чтобы избежать гонки с очисткой ETCP-очередей
|
||||
void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id);
|
||||
|
||||
// Закрыть все router_conn для указанного (group_id, remote_node_id) (peer умер)
|
||||
void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id);
|
||||
|
||||
// Сбросить состояние роутера для конкретного peer+svc в группе (перезапуск удалённой стороны).
|
||||
// Начать новую локальную сессию для конкретного peer+svc в группе и уведомить сервис.
|
||||
void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id, uint8_t svc_id);
|
||||
|
||||
// Удалить все транзитные очереди соединения (вызывается при закрытии ETCP_CONN)
|
||||
|
||||
@@ -1,141 +1,102 @@
|
||||
# ETCP Router (etcp_router)
|
||||
# ETCP Router: протокол и проверка транспорта
|
||||
|
||||
## 1. Назначение
|
||||
Сервисный слой маршрутизации поверх ETCP — упрощённый TCP с восстановлением порядка доставки, дедупликацией и контролем inflight. Мультиплексирует до 256 сервисов (`svc_id`) на одном ETCP-соединении. Поддерживает transit (многошаговую маршрутизацию через промежуточные узлы), ретрансмиты, minRTT-измерения и подписи Ed25519.
|
||||
Логическое соединение определяется `(group_id, remote_node_id, svc_id)`. Роутер обеспечивает порядок, дедупликацию и повторную отправку между конечными узлами; физический ETCP и транзитные узлы могут меняться. Формат изменён без обратной совместимости.
|
||||
|
||||
## 2. Как пользоваться
|
||||
## Wire-формат
|
||||
|
||||
### Инициализация
|
||||
```c
|
||||
// В utun_instance_init():
|
||||
etcp_router_init(inst);
|
||||
Packed `SVC_ROUTE_HDR`, 57 байт:
|
||||
|
||||
// Зарегистрировать обработчик сервиса (svc_id 0..255)
|
||||
etcp_router_bind(inst, svc_id, my_service_callback);
|
||||
```
|
||||
`cmd:1 | group:8 | dst:8 | src:8 | seq:4 | svc:1 | flags:1 | timestamp:2 | reset_id:8 | peer_reset_id:8 | challenge:8`
|
||||
|
||||
### Отправка
|
||||
```c
|
||||
// Простой способ: сформировать ll_entry с svc_id в первом байте
|
||||
struct ll_entry* e = queue_entry_new(0);
|
||||
e->dgram = u_malloc(1 + payload_len);
|
||||
e->dgram[0] = svc_id;
|
||||
memcpy(e->dgram + 1, payload, payload_len);
|
||||
e->len = 1 + payload_len;
|
||||
etcp_route_send(inst, group_id, dst_node_id, e, force, mode); // mode: 0 / ROUTE_CRYPTO_SIGN / ROUTE_CRYPTO_ENCRYPT
|
||||
Числовое представление полей сохраняет существующий формат проекта (native endian). Отдельной версии/согласования формата нет; узлы должны обновляться совместно.
|
||||
|
||||
// Или через существующий ROUTER_CONN:
|
||||
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, group_id, remote_node_id, svc_id);
|
||||
etcp_router_conn_send(rconn, data, len, mode);
|
||||
```
|
||||
`reset_id` — случайный ненулевой ID локального экземпляра соединения; `peer_reset_id` — ID получателя. DATA, ACK и CLOSE принимаются только при совпадении обоих ID с текущей парой. Смена физического транспорта пару не меняет.
|
||||
|
||||
### Приём
|
||||
```c
|
||||
void my_service_callback(struct ETCP_CONN* conn, struct ll_entry* entry) {
|
||||
if (!entry) { /* соединение закрыто */ return; }
|
||||
uint8_t svc_id = entry->dgram[0];
|
||||
uint64_t from = *(uint64_t*)(entry->dgram + ROUTER_SVC_SRC_OFF); // remote_node_id
|
||||
uint8_t rx_flags = entry->dgram[ROUTER_SVC_FLAGS_OFF]; // был ли SIGNED/ENCRYPTED на wire
|
||||
// ... данные начиная с entry->dgram[ROUTER_SVC_PAYLOAD_OFF] ...
|
||||
queue_dgram_free(entry);
|
||||
queue_entry_free(entry);
|
||||
}
|
||||
```
|
||||
| Пакет | Флаги | Payload | Поля |
|
||||
|---|---|---|---|
|
||||
| HELLO | START (0x80) | нет | собственный ID, peer=0, challenge=0 |
|
||||
| CHALLENGE | RST (0x40) | нет | оба ID, случайный ненулевой challenge |
|
||||
| CONFIRM | START|RST | нет | оба ID, эхо challenge |
|
||||
| DATA | 0 либо SIGNED/ENCRYPTED | непустой | seq, оба ID, challenge=0 |
|
||||
| ACK | 0 | нет | seq=следующий ожидаемый, оба ID, challenge=0 |
|
||||
| CLOSE | CLOSE (0x02) | нет | оба ID, challenge=0 |
|
||||
|
||||
### Backpressure
|
||||
```c
|
||||
// На send_q (inflight переполнен):
|
||||
struct queue_waiter_handle h;
|
||||
etcp_router_on_send_ready(inst, group_id, node_id, svc_id, &h, my_ready_cb, my_arg);
|
||||
// Когда send_q освободится — вызовется my_ready_cb
|
||||
```
|
||||
Незнакомый ID сам по себе не сбрасывает рабочую сессию. Получатель выдаёт свежий challenge и меняет peer ID только после соответствующего CONFIRM. CHALLENGE не меняет рабочую сессию: отправляется CONFIRM и, если нужно, встречный challenge. Потерянные сообщения handshake восстанавливаются повторением HELLO через 300 ms; выдача новых challenge ограничена тем же интервалом.
|
||||
|
||||
### Подпись / шифрование (модуль route_crypto)
|
||||
```c
|
||||
// mode — битовая маска, применяется в начале функции отправки (route_crypto_encode),
|
||||
// при приёме проверяется/расшифровывается перед вызовом коллбэка (route_crypto_decode).
|
||||
#include "route_crypto.h"
|
||||
etcp_route_send(inst, group_id, dst, e, force, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||||
```
|
||||
Первичное согласование сохраняет накопленные исходящие данные. Подтверждённый рестарт пира сбрасывает seq/очереди и уведомляет сервис; собственный ID сохраняется. Явный `etcp_router_conn_restart` создаёт новый собственный ID. Сброс выполняется до уведомления, поэтому данные, отправленные сервисом из callback, сохраняются. Сервис должен заново выполнить прикладную синхронизацию после уведомления; гарантии доставки не продолжаются через явный сброс сессии.
|
||||
|
||||
## 3. API
|
||||
Challenge защищает от случайных старых/переставленных управляющих пакетов, но не является end-to-end аутентификацией против злонамеренного транзитного узла. Управление опирается на защиту физических ETCP-соединений. Подпись/шифрование DATA включаются вызывающим сервисом через `ROUTE_CRYPTO_SIGN` / `ROUTE_CRYPTO_ENCRYPT`.
|
||||
|
||||
### Ключевые структуры
|
||||
## Очереди и подтверждения
|
||||
|
||||
**SVC_ROUTE_HDR** (33 байта) — заголовок пакета:
|
||||
`cmd(1) + group_id(8) + dst_node_id(8) + src_node_id(8) + seq(4) + svc_id(1) + flags(1) + timestamp(2)`
|
||||
- `send_q`: plaintext + seq + crypto mode, до 64 пакетов при обычной отправке. Кодирование происходит при отправке после handshake, с актуальной парой ID.
|
||||
- `tx_seq`: следующий выделяемый seq; `tx_sent`: граница фактически отправленного; `tx_acked`: граница подтверждённого.
|
||||
- `inflight_q`: финальные wire-копии, обычно до 256 пакетов. Копия создаётся до передачи в ETCP; ошибка не теряет pending-пакет.
|
||||
- ACK принимается только в модульном интервале `(tx_acked, tx_sent]`. Дубликаты, старые и выходящие за границу ACK не продлевают ожидание прогресса. Удаляются существующие inflight-записи, а не перебираются все номера до произвольного ACK.
|
||||
- Новые DATA и ретрансмиты идут по одному через waiter физической `send_input_q`. При смене пути отменяется waiter на очереди, где он действительно был зарегистрирован, включая уже запланированный callback.
|
||||
- Каждые 300 ms без прогресса планируется проход повторной отправки. После 17 попыток без ACK соединение закрывается. Отсутствующий маршрут проверяется каждые 20 ms; очередь сохраняется. Пока первичный handshake не завершён, HELLO повторяется без отдельного таймаута.
|
||||
- Watchdog 500 ms восстанавливает отправку и переносит waiter при смене заблокированного пути.
|
||||
- Сброс физического ETCP сохраняет router inflight и seq. Восстановление самого ETCP-handshake остаётся обязанностью транспортного слоя.
|
||||
- Crypto decode/проверка подписи выполняются до помещения в reorder и изменения rx_seq. Повреждённый пакет не подтверждается как доставленный.
|
||||
- DATA выдаются сервису строго по seq; входящие дубликаты повторно проверяются после извлечения из очереди.
|
||||
- Wire-пакет ограничен `PKTNORM_MAX_DGRAM_SIZE` (16384), включая заголовок и crypto overhead. Пустой payload запрещён: он используется для управления.
|
||||
|
||||
Флаги: `ROUTER_FLAG_START` (0x80), `ROUTER_FLAG_RST` (0x40), `ROUTER_FLAG_SIGNED` (0x08), `ROUTER_FLAG_ENCRYPTED` (0x04), `ROUTER_FLAG_CLOSE` (0x02).
|
||||
## Сервисный API
|
||||
|
||||
**Формат доставки сервису** (router → callback):
|
||||
`[svc_id:1][src_node_id:8][dst_node_id:8][rx_flags:1][payload...]` (ROUTER_SVC_HDR_SIZE = 18).
|
||||
`etcp_router_conn_send(rconn, data, len, mode)` копирует payload; 0 означает принятие в очередь, а не end-to-end доставку. При backpressure производитель ждёт через `etcp_router_on_send_ready`; отменяет ожидание через `etcp_router_cancel_send_ready`. Handle должен жить до callback/отмены. Режим `force` низкоуровневого `etcp_route_send` обходит порог очереди и не предназначен для непрерывного потока.
|
||||
|
||||
**ETCP_ROUTER_CONN** — состояние логического подключения (group_id + remote_node_id + svc_id):
|
||||
- `tx_seq`, `rx_seq`, `tx_acked` — seq-нумерация для порядка и inflight
|
||||
- `recv_q` — reorder-очередь с хеш-индексом по seq (восстановление порядка)
|
||||
- `send_q` — очередь ожидания при полном inflight (backpressure)
|
||||
- `send_waiter` — waiter на send_input_q (по 1 пакету, round-robin между сервисами)
|
||||
- `watchdog_timer` — медленная проверка инварианта drain (страховка от заклинивания)
|
||||
- `inflight_q` — копии финальных отправленных пакетов для ретрансмита (хеш по seq)
|
||||
- `incoming_q` — FIFO между сетевым приёмом и recv_q (защита от гонок)
|
||||
- `rtt`, `rtt_jitter`, `minrtt` — измерения задержки
|
||||
- `inflight_limit` — текущий лимит пакетов в полёте (minrtt_probe снижает до 4)
|
||||
- `no_ack_count` — последовательные ретрансмиты без ACK (при 17 — закрытие)
|
||||
- `start_sent`, `peer_sync_done` — синхронизация после (пере)подключения
|
||||
- `closed` — флаг закрытия (игнорирование таймеров)
|
||||
`etcp_router_bind` регистрирует callback, владеющий переданным ll_entry. Формат доставки — 26 байт заголовка:
|
||||
|
||||
**ROUTER_INFLIGHT** — запись в inflight_q: `seq, last_sent_tb, send_count, dgram*` (финальный wire-пакет)
|
||||
`svc:1 | src:8 | dst:8 | rx_flags:1 | group:8 | payload...`
|
||||
|
||||
**TRANSIT_QUEUE** — per (group_id, src, dst) пара на промежуточном узле:
|
||||
- `q` — FIFO транзитных пакетов
|
||||
- `waiter` — backpressure на send_input_q next_hop'а
|
||||
- `conn` — ETCP_CONN следующего шага
|
||||
Читать невыравненные числовые поля нужно через `memcpy`. Заголовок без payload означает уведомление о закрытии/сбросе. После обработки освобождаются `queue_dgram_free(entry)` и `queue_entry_free(entry)`.
|
||||
|
||||
### Внутренняя архитектура
|
||||
```
|
||||
Приём: сеть → incoming_q (FIFO) → router_incoming_q_cb → recv_q (хеш по seq)
|
||||
↓
|
||||
router_try_assembly → deliver → route_crypto_decode → cb
|
||||
Callback может закрыть соединение. Освобождение send_q отложено за пределы callback производителя; отправка проверяет, не закрылся/перезапустился ли rconn. При уничтожении реестр удаляется ровно один раз для каждого соединения.
|
||||
|
||||
Отправка: router_enqueue_send → router_build_packet (seq + route_crypto_encode) → send_q (FIFO, backpressure)
|
||||
router_send_kick → waiter на send_input_q → router_send_drain_cb (по 1 пакету, round-robin)
|
||||
→ etcp_send + inflight_q (копия финального пакета)
|
||||
↓
|
||||
router_track_inflight_state + retrans_schedule
|
||||
ACK: периодический (10ms) + idle (500ms) → router_send_ack(rx_seq)
|
||||
```
|
||||
## Проверки и диагностика
|
||||
|
||||
### Константы
|
||||
| Константа | Значение | Описание |
|
||||
|---|---|---|
|
||||
| `ROUTER_MAX_INFLIGHT` | 256 | Макс. пакетов в полёте |
|
||||
| `ROUTER_ACK_INTERVAL_TB` | 100 | ACK интервал 10ms (0.1ms timebase) |
|
||||
| `ROUTER_ACK_IDLE_TB` | 5000 | Idle таймаут 500ms |
|
||||
| `ROUTER_RETRANS_TIMEOUT_TB` | 3000 | Таймаут ретрансмита 300ms |
|
||||
| `ROUTER_NO_ACK_MAX_RETRANS` | 17 | Макс. ретрансмитов без ACK (≈5s) |
|
||||
| `ROUTER_MAX_SEND_Q_PACKETS` | 64 | Порог backpressure send_q |
|
||||
| `ROUTER_SEND_WATCHDOG_TB` | 5000 | Watchdog проверка инварианта drain (500ms) |
|
||||
`tests/test_etcp_router_faults` перехватывает SVC_ROUTE между настоящими роутерами, очередями и UASYNC-таймерами. Физическая очередь тестовая: нижележащий ETCP не скрывает потери от роутера. Проверяются:
|
||||
|
||||
### Функции
|
||||
| Функция | Описание |
|
||||
|---|---|
|
||||
| `etcp_router_init(inst)` | Инициализация: bind ETCP_ID_SVC_ROUTE, создание router_conns |
|
||||
| `etcp_router_destroy(inst)` | Деинициализация: unbind + close_all + free |
|
||||
| `etcp_router_bind(inst, svc_id, cb)` | Зарегистрировать обработчик сервиса |
|
||||
| `etcp_router_unbind(inst, svc_id)` | Удалить обработчик сервиса |
|
||||
| `etcp_route_send(inst, group, dst, entry, force, mode)` | Отправить пакет (loopback/transit/direct); mode = ROUTE_CRYPTO_SIGN/ENCRYPT |
|
||||
| `etcp_router_conn_get(inst, group, remote, svc_id)` | Найти или создать ROUTER_CONN |
|
||||
| `etcp_router_conn_send(rconn, data, len, mode)` | Отправить данные с авто-seq и inflight-контролем |
|
||||
| `etcp_router_conn_close(rconn)` | Закрыть seq-подключение (CLOSE + уведомление сервиса) |
|
||||
| `etcp_router_conn_close_all_for_node(inst, group, node_id)` | Закрыть все conn к узлу |
|
||||
| `etcp_router_conn_restart(inst, group, remote, svc_id)` | Сброс состояния (перезапуск удалённой стороны) |
|
||||
| `etcp_router_pause_retrans_for_node(inst, node_id)` | Сбросить ретрансмиты без закрытия (для conn reinit) |
|
||||
| `etcp_router_set_max_inflight(rconn, new_max)` | Установить рабочий max_inflight |
|
||||
| `etcp_router_on_send_ready/cancel_send_ready(inst, group, node, svc, h, cb, arg)` | Backpressure на send_q |
|
||||
| `etcp_router_transit_queues_destroy(conn)` | Удалить все транзитные очереди ETCP_CONN |
|
||||
1. Потери DATA/ACK, перестановка, дубликаты, четыре потока (два направления × два сервиса), backpressure и ограниченность физической очереди.
|
||||
2. Потеря HELLO/CHALLENGE/CONFIRM в обоих направлениях.
|
||||
3. Старые, повторные, слишком большие ACK и ACK ещё не отправленного пакета.
|
||||
4. Сохранение inflight при сбросе транспорта.
|
||||
5. Старые DATA/ACK/CLOSE/HELLO/CONFIRM после смены сессии.
|
||||
6. Переполнение uint32 seq.
|
||||
7. Закрытие после смены пути при зарегистрированном и отложенном waiter.
|
||||
8. Закрытие из callback приёма.
|
||||
9. Смена маршрута при занятой старой очереди без новой отправки.
|
||||
10. Исчезновение и восстановление маршрута.
|
||||
11. Таймаут при непрерывных повторных ACK без прогресса.
|
||||
12. Одновременный рестарт обоих концов.
|
||||
13. Граничные размеры payload, crypto overhead, SIZE_MAX.
|
||||
14. Закрытие из callback производителя при освобождении send_q.
|
||||
|
||||
### Модуль route_crypto (пред/пост-обработка пакета)
|
||||
| Функция | Описание |
|
||||
|---|---|
|
||||
| `route_crypto_encode(inst, peer, base, len, mode, &out, &len)` | Шифрование/подпись в начале отправки → финальный пакет |
|
||||
| `route_crypto_decode(inst, wire, len, &out, &len, &rx_flags)` | Проверка подписи/расшифровка перед коллбэком; ошибка → дроп |
|
||||
Успех передачи требует совпадения содержимого, порядка и количества, финальных ACK, пустых send/inflight. Завершение проверяет освобождение таймеров. Unit-набор включает 34 проверки, в том числе криптографию и повторную доставку после неверной подписи.
|
||||
|
||||
Сетевой `test_etcp_router` использует UDP/dummynet, backpressure и сброс с восстановлением ETCP-handshake. `test_etcp_router_reconnect` проверяет a–b–c и перезапуск b, по 500 пакетов до/после, с ожиданием последних ACK в каждой фазе и фиксированным seed.
|
||||
|
||||
Переменная `ROUTER_TEST_LOG=/tmp/router.log` включает подробный файл. Рабочая категория — `etcp_route`; DEBUG показывает handshake, пару эпох, ретрансмиты и смену состояния; поштучные ACK/DATA оставлены на TRACE. Ошибочные ветки не скрыты. В fault-тесте WARN/ERROR для намеренно неверных ACK/размеров и искусственного таймаута ожидаемы.
|
||||
|
||||
Предыдущий этап проверки роутера: 14 fault-сценариев и 34 unit-проверки прошли; fault-набор также прошёл ASan/UBSan/LeakSanitizer (инструментированы router, route_crypto и ll_queue, остальная библиотека обычная). UDP-тест: 440/440 доставлено и подтверждено. Транзитный: 1000/1000 доставлено и подтверждено. Это не доказательство отсутствия всех ошибок стека.
|
||||
|
||||
## Lifecycle транспорта и восстановление
|
||||
|
||||
ETCP имеет единую точку финального освобождения после DELETE, завершения callback и освобождения последней ссылки. Запрос закрытия идемпотентен и сразу запрещает новые send/ref_take.
|
||||
|
||||
NCD-entry удерживает одну ссылку ETCP. При DELETE запись сначала исключается из реестра, затем владельцы получают CLOSED; старые handles безопасно закрываются после создания нового соединения с тем же node_id. Закрытые callbacks/handles удаляются после внешней рассылки. Отложенный начальный UP отменяется при DOWN/close. При удалении сокета обход использует снимок ссылок ETCP, поскольку DOWN может удалить NCD-entry.
|
||||
|
||||
Recovery удерживает TOPO_NODE для каждого уникального кандидата и освобождает ссылку при любом завершении. При успехе BGP сначала получает собственный NCD-handle. `test_topo_recovery` удаляет B и проверяет самостоятельное подключение A↔C и доставку 1000 сообщений без тестового reconnect.
|
||||
|
||||
`etcp_session.c` согласует эпохи физического транспорта через HELLO/CHALLENGE/CONFIRM поверх аутентифицированного UDP/STCP. Каждый DATA/ACK содержит пару эпох; старые кадры отбрасываются до обработки. Сброс потока не генерирует ложный DOWN. REINIT повторно запускает обмен таблицами BGP. Формат и таймауты описаны в `doc/etcp_protocol.txt`.
|
||||
|
||||
Дополнительные проверки:
|
||||
- `test_etcp_lifecycle`: DELETE/refcount, повторное закрытие, закрытие из callback, FIN_WAIT, отмена начального UP, удаление сокета, владение recovery.
|
||||
- `test_etcp_session`: реальные crypto/ingress/таймеры; reset любой стороны, одновременный reset, потери всех видов control, повторный reset, отказ одного из двух линков, timeout, старые DATA/ACK/control, неверные размеры.
|
||||
- `test_etcp_router_tcp`: тот же контроль содержимого, порядка, дубликатов, backpressure и финальных ACK по STCP; обе стороны сбрасываются одновременно.
|
||||
|
||||
Диагностика по запросу: `SESSION_TEST_LOG=/tmp/session.log`, `LIFECYCLE_TEST_LOG=/tmp/lifecycle.log`, `ROUTER_TEST_LOG=/tmp/router.log`, `BGP_TEST_LOG=/tmp/bgp.log`. `ROUTER_TEST_RESET=client|server|both` выбирает сторону reset; `ROUTER_TEST_TCP=1` включает STCP. Ожидаемые ошибки отрицательных сценариев не скрываются.
|
||||
|
||||
Дополнительно исправлен монитор conn_mgr: наличие маршрута через посредника не означает прямого соединения. Фоновый PING выбирает транспорт по node_id конечного peer и живой линк именно этого транспорта. Проверка BGP-треугольника выдерживает полный период фоновых проб и требует отсутствия ошибок входящих пакетов до внесения потерь. На старом monitor эта проверка воспроизводимо падает; с исправлением проходит.
|
||||
|
||||
Итоговая проверка текущих исправлений: 16 регрессионных executable-тестов прошли (router unit/faults/lifecycle/session, UDP, TCP, reconnect, recovery, STCP traffic, NCD, три conn_mgr, normalizer/ETCP, BGP triangle). Семь наборов дополнительно прошли ASan/UBSan/LeakSanitizer: lifecycle, session, router faults, UDP router, TCP router, autonomous recovery, BGP triangle. Инструментированы ETCP core/API/connections/session, NCD, router, recovery, topo_group/connect, conn_mgr core/monitor и ll_queue; остальные библиотечные объекты обычные. Отрицательные тесты намеренно вызывают ошибки; после принудительного удаления транспорта возможен отказ приёма запоздалой UDP-датаграммы без существующего линка. Финальный BGP-прогон, включая фоновые пробы и cleanup таймеров, завершился без диагностических ошибок.
|
||||
@@ -95,7 +95,7 @@ static void topo_group_send_table_complete(struct TOPO_GROUP* group, struct ETCP
|
||||
if (etcp_send(conn, e) != 0) { u_free(req); queue_entry_free(e); }
|
||||
}
|
||||
|
||||
static void topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
static bool topo_group_should_send_to(const struct TOPO_GROUP_NODE* nq, uint64_t target_id);
|
||||
static void topo_group_send_full_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
@@ -262,7 +262,8 @@ static void topo_group_receive_cbk(struct ETCP_CONN* from_conn, struct ll_entry*
|
||||
queue_dgram_free(entry); queue_entry_free(entry); return;
|
||||
}
|
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "BGP recv NODEINFO from=%s nid=%016llx ver=%d grp=%016llx", from_conn->log_name, (unsigned long long)((struct TOPOMSG_NODEINFO_PKT*)data)->node.node_id, ((struct TOPOMSG_NODEINFO_PKT*)data)->node.ver, (unsigned long long)((struct TOPOMSG_NODEINFO_PKT*)data)->node.group_id);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP,"BGP recv %s from=%s group=%016llx len=%u",group_subcmd_name(subcmd),
|
||||
from_conn->log_name,(unsigned long long)pkt_group_id,entry->len);
|
||||
|
||||
if (subcmd == TOPO_SUBCMD_NODEINFO) { nodeinfo_dump_log(data, entry->len); topo_group_process_nodeinfo(group, from_conn, data, entry->len); }
|
||||
else if (subcmd == TOPO_SUBCMD_WITHDRAW) topo_group_process_withdraw(group, from_conn, data, entry->len);
|
||||
@@ -290,6 +291,20 @@ static void topo_group_conn_status(struct ETCP_CONN* conn, int status, void* arg
|
||||
struct TOPO_GROUPS* groups = (struct TOPO_GROUPS*)arg;
|
||||
if (!conn || !groups) return;
|
||||
|
||||
if (status == ETCP_CONN_STATUS_REINIT) {
|
||||
// Сброс normalizer мог уничтожить TABLE_REQUEST/NODEINFO. Повторяем обмен, сохраняя членство и пути.
|
||||
for (struct ll_entry* e = groups->group_list->head; e; e = e->next) {
|
||||
struct TOPO_GROUP* g = (struct TOPO_GROUP*)e;
|
||||
for (struct ll_entry* se = g->senders_list->head; se; se = se->next) {
|
||||
if (((struct TOPO_GROUP_CONN_ITEM*)se->data)->conn != conn) continue;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP,"BGP session resync: peer=%016llx group=%016llx",
|
||||
(unsigned long long)conn->peer_node_id,(unsigned long long)g->group_id);
|
||||
topo_group_send_join_group(g,conn); topo_group_send_table_request(g,conn);
|
||||
break;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (status == ETCP_CONN_STATUS_UP && conn->peer_node_id) {
|
||||
int is_client = conn->instance->config && config_peer_in_clients(conn->instance->config, conn->crypto_ctx.peer_public_key);
|
||||
int explicit_conn = conn->bgp_ready_cbk != NULL;
|
||||
@@ -680,10 +695,10 @@ void topo_groups_set_node_updated_cb(struct TOPO_GROUPS* groups, topo_node_updat
|
||||
}
|
||||
|
||||
/* Добавляет пира в группу при UP: дедуп, добавление в senders, запрос таблицы. */
|
||||
void topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return; }
|
||||
if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance is NULL"); return; }
|
||||
if (!conn->instance->rt) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance->rt is NULL"); return; }
|
||||
int topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return -1; }
|
||||
if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance is NULL"); return -1; }
|
||||
if (!conn->instance->rt) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance->rt is NULL"); return -1; }
|
||||
|
||||
/* дедуп: тот же conn стреляет ETCP_CONN_STATUS_UP дважды (UDP-линк, затем TCP-линк).
|
||||
* Если conn уже в senders_list — повторно не обрабатываем, иначе active_conn_count
|
||||
@@ -692,13 +707,12 @@ void topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
while (se) {
|
||||
if (((struct TOPO_GROUP_CONN_ITEM*)se->data)->conn == conn) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: conn already in senders_list, skip double UP (%s)", conn->log_name);
|
||||
return;
|
||||
return 0;
|
||||
}
|
||||
se = se->next;
|
||||
}
|
||||
}
|
||||
|
||||
topo_recovery_cancel_for_node(group, conn->peer_node_id);
|
||||
|
||||
struct TOPO_GROUP_NODE* peer_nq = topo_node_find_by_id(group, conn->peer_node_id);
|
||||
if (peer_nq) {
|
||||
@@ -710,30 +724,14 @@ void topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: ADD conn=%p peer=%016llx state=%d grp=%016llx ch=%s",
|
||||
(void*)conn, (unsigned long long)conn->peer_node_id, conn->state,
|
||||
(unsigned long long)group->group_id, group->channel_id);
|
||||
topo_group_add_to_senders(group, conn);
|
||||
|
||||
/* BGP — владелец conn: держим NCD-handle (если ещё нет), чтобы соединение
|
||||
* не рвалось другими владельцами (conn_mgr) при их закрытии. cb=NULL —
|
||||
* события BGP уже получает через conn-status callback. */
|
||||
{ struct ll_entry* se = group->senders_list->head;
|
||||
while (se) {
|
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data;
|
||||
if (item->conn == conn) {
|
||||
if (item->handle == NULL) {
|
||||
node_conn_direct_open(conn->instance, conn->peer_node_id, NULL, NULL, &item->handle, NULL);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "topo_group_new_conn: BGP handle acquired node=0x%016llx h=%p",
|
||||
(unsigned long long)conn->peer_node_id, (void*)item->handle);
|
||||
}
|
||||
break;
|
||||
}
|
||||
se = se->next;
|
||||
}
|
||||
}
|
||||
if (topo_group_add_to_senders(group, conn) != 0) return -1;
|
||||
topo_recovery_cancel_for_node(group, conn->peer_node_id);
|
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "topo_group_new_conn: peer=%016llx group=%016llx type=%d ch=%s", (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id);
|
||||
|
||||
topo_group_send_table_request(group, conn);
|
||||
topo_group_connect_on_up(group, conn);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Обрабатывает DOWN пира: чистит пути, каскадно удаляет узлы, шлёт withdraw. */
|
||||
@@ -772,7 +770,8 @@ void topo_group_remove_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
nodes_removed++;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "node removed from group: node=%016llx grp=%016llx type=%u ch=%s (link_down)",
|
||||
(unsigned long long)key, (unsigned long long)group->group_id, group->group_type, group->channel_id);
|
||||
etcp_router_conn_close_all_for_node(group->instance, group->group_id, key);
|
||||
// REINIT временно убирает путь, но end-to-end данные роутера остаются неподтверждёнными.
|
||||
if (!conn->reinit_pending) etcp_router_conn_close_all_for_node(group->instance, group->group_id, key);
|
||||
uint64_t wd_src = (key == conn->peer_node_id) ? conn->instance->node_id : conn->peer_node_id;
|
||||
topo_group_broadcast_withdraw(group, key, wd_src, NULL);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "Removed node %016llx after link down", (unsigned long long)key);
|
||||
@@ -1169,23 +1168,22 @@ void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE*
|
||||
}
|
||||
|
||||
/* Добавляет conn в senders_list (дедуп), если его там ещё нет. */
|
||||
static void topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
if (!group || !conn || !group->senders_list) return;
|
||||
bool already = false;
|
||||
struct ll_entry* e = group->senders_list->head;
|
||||
while (e) { if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) { already = true; break; } e = e->next; }
|
||||
if (!already) {
|
||||
struct ll_entry* item_entry = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM));
|
||||
if (item_entry) {
|
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)item_entry->data;
|
||||
item->conn = conn;
|
||||
item->handle = NULL;
|
||||
queue_data_put(group->senders_list, item_entry);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "senders_add conn=%p peer=%016llx name='%s' grp=%016llx",
|
||||
(void*)conn, (unsigned long long)conn->peer_node_id, conn->log_name,
|
||||
(unsigned long long)group->group_id);
|
||||
}
|
||||
static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
|
||||
if (!group || !conn || !group->senders_list) return -1;
|
||||
for (struct ll_entry* e = group->senders_list->head; e; e = e->next)
|
||||
if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) return 0;
|
||||
struct ll_entry* e = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM));
|
||||
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: allocation failed"); return -1; }
|
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
|
||||
item->conn = conn; item->handle = NULL;
|
||||
if (node_conn_direct_open(conn->instance, conn->peer_node_id, NULL, NULL, &item->handle, NULL) == NCD_ERR) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: cannot acquire NCD handle peer=%016llx", (unsigned long long)conn->peer_node_id);
|
||||
queue_entry_free(e); return -1;
|
||||
}
|
||||
queue_data_put(group->senders_list, e);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "senders_add: acquired peer=%016llx group=%016llx",
|
||||
(unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Решает, надо ли слать узел пиру (не шлём, если пир уже есть в hop_list пути). */
|
||||
|
||||
@@ -281,7 +281,7 @@ void topo_groups_remove_group(struct TOPO_GROUPS* g, uint64_t group_id);
|
||||
*
|
||||
* Вызывается при ETCP on_up.
|
||||
*/
|
||||
void topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
int topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
|
||||
|
||||
/**
|
||||
* @brief Удаляет conn из senders_list, очищает paths во всех nodes, отправляет withdraw если node unreachable.
|
||||
|
||||
@@ -358,7 +358,7 @@ static void tgc_release_handle(struct TOPO_GROUP_CONNECT* gc, struct NODE_CONN_D
|
||||
static void tgc_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
|
||||
struct TOPO_GROUP_CONNECT* gc = (struct TOPO_GROUP_CONNECT*)arg;
|
||||
struct ETCP_CONN* conn = node_conn_direct_get_conn(h);
|
||||
uint64_t node_id = conn ? conn->peer_node_id : 0;
|
||||
uint64_t node_id = node_conn_direct_node_id(h);
|
||||
|
||||
int ok = (event == NCD_EVENT_UP);
|
||||
|
||||
@@ -366,9 +366,11 @@ static void tgc_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void*
|
||||
/* запуск BGP: узел соединён с группой (topo_group_new_conn берёт владение NCD-handle) */
|
||||
if (conn) topo_group_new_conn(gc->group, conn);
|
||||
tgc_release_handle(gc, h);
|
||||
} else if (event == NCD_EVENT_TIMEOUT) {
|
||||
} else {
|
||||
// Каждая попытка даёт один результат. Поздние DOWN/CLOSED не должны уменьшать pending повторно.
|
||||
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(gc->group, node_id);
|
||||
if (nq) nq->handle = NULL;
|
||||
tgc_release_handle(gc, h);
|
||||
}
|
||||
|
||||
switch (gc->phase) {
|
||||
@@ -645,7 +647,7 @@ static void tgc_once_cb(struct NODE_CONN_DIRECT* h, enum ncd_event ev, void* arg
|
||||
DEBUG_WARN(DEBUG_CATEGORY_BGP, "%s: once-connect TIMEOUT node=0x%016llx ch=%s",
|
||||
TGC_ID, (unsigned long long)c->node_id, c->group->channel_id);
|
||||
node_conn_direct_close(h);
|
||||
} else if (ev == NCD_EVENT_DOWN) {
|
||||
} else if (ev == NCD_EVENT_DOWN || ev == NCD_EVENT_CLOSED) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_BGP, "%s: once-connect DOWN node=0x%016llx ch=%s",
|
||||
TGC_ID, (unsigned long long)c->node_id, c->group->channel_id);
|
||||
node_conn_direct_close(h);
|
||||
|
||||
@@ -380,7 +380,7 @@ static void tgi_ncd_cb(struct NODE_CONN_DIRECT* h, enum ncd_event ev, void* arg)
|
||||
if (!qe->dgram) { queue_entry_free(qe); tgi_fail(p); return; }
|
||||
memcpy(qe->dgram, &req, sizeof(req)); qe->len = sizeof(req);
|
||||
if (etcp_send(conn, qe) != 0) { u_free(qe->dgram); queue_entry_free(qe); tgi_fail(p); }
|
||||
} else if (ev == NCD_EVENT_TIMEOUT) {
|
||||
} else if (ev == NCD_EVENT_TIMEOUT || ev == NCD_EVENT_CLOSED) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "invite: ncd TIMEOUT connecting to 0x%016llx", (unsigned long long)p->node_id);
|
||||
tgi_fail(p);
|
||||
}
|
||||
@@ -520,7 +520,7 @@ static void tgi_to_channel_cb(struct NODE_CONN_DIRECT* h, enum ncd_event ev, voi
|
||||
if (g && conn) topo_group_new_conn(g, conn);
|
||||
node_conn_direct_close(h);
|
||||
u_free(c);
|
||||
} else if (ev == NCD_EVENT_TIMEOUT) {
|
||||
} else if (ev == NCD_EVENT_TIMEOUT || ev == NCD_EVENT_CLOSED) {
|
||||
node_conn_direct_force_close(h);
|
||||
u_free(c);
|
||||
}
|
||||
|
||||
@@ -31,12 +31,20 @@ static struct TOPO_RECOVERY_CTX* topo_recovery_ctx_alloc(struct TOPO_GROUP* grou
|
||||
ctx->instance = group->instance; ctx->group = group; ctx->next_hop_id = next_hop_id;
|
||||
ctx->capacity = RECOVERY_CAPACITY_INIT;
|
||||
ctx->nodes = u_calloc(ctx->capacity, sizeof(struct TOPO_RECOVERY_NODE));
|
||||
if (!ctx->nodes) { u_free(ctx); return NULL; }
|
||||
if (!ctx->nodes) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "recovery: nodes allocation failed"); u_free(ctx); return NULL; }
|
||||
return ctx;
|
||||
}
|
||||
|
||||
static void topo_recovery_remove_candidate(struct TOPO_RECOVERY_CTX* ctx, size_t index) {
|
||||
uint64_t id = ctx->nodes[index].node_id;
|
||||
ctx->nodes[index] = ctx->nodes[--ctx->count];
|
||||
topo_node_registry_unref(ctx->instance->topo_groups, id);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "recovery: release candidate=%016llx remaining=%zu", (unsigned long long)id, ctx->count);
|
||||
}
|
||||
|
||||
static void topo_recovery_ctx_free(struct TOPO_RECOVERY_CTX* ctx) {
|
||||
if (!ctx) return;
|
||||
while (ctx->count) topo_recovery_remove_candidate(ctx, ctx->count - 1);
|
||||
u_free(ctx->nodes);
|
||||
u_free(ctx);
|
||||
}
|
||||
@@ -60,22 +68,27 @@ static void topo_recovery_try_next(struct TOPO_RECOVERY_CTX* ctx);
|
||||
static void topo_recovery_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
|
||||
struct TOPO_RECOVERY_CTX* ctx = (struct TOPO_RECOVERY_CTX*)arg;
|
||||
struct ETCP_CONN* conn = node_conn_direct_get_conn(h);
|
||||
uint64_t node_id = conn ? conn->peer_node_id : 0;
|
||||
uint64_t node_id = node_conn_direct_node_id(h);
|
||||
if (node_id != ctx->current_node_id) return;
|
||||
if (ctx->connect_timer) { uasync_cancel_timeout(ctx->instance->ua, ctx->connect_timer); ctx->connect_timer = NULL; }
|
||||
for (size_t i = 0; i < ctx->count; i++)
|
||||
if (ctx->nodes[i].node_id == node_id) { ctx->nodes[i] = ctx->nodes[ctx->count - 1]; ctx->count--; break; }
|
||||
ctx->current_node_id = 0;
|
||||
ctx->current_handle = NULL;
|
||||
if (event == NCD_EVENT_UP) {
|
||||
/* владение переходит к topo_group_new_conn (item->handle), connect-фазу закрываем */
|
||||
if (conn) topo_group_new_conn(ctx->group, conn);
|
||||
node_conn_direct_close(h);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "recovery: next=%016llx reconnected to %016llx", (unsigned long long)ctx->next_hop_id, (unsigned long long)node_id);
|
||||
} else {
|
||||
node_conn_direct_close(h);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "recovery: next=%016llx connect to %016llx failed: %d", (unsigned long long)ctx->next_hop_id, (unsigned long long)node_id, event);
|
||||
if (event == NCD_EVENT_UP && conn) {
|
||||
// Передача владения может отменить другие recovery. Свой ctx исключаем из списка заранее.
|
||||
topo_recovery_ctx_remove(ctx->group, ctx);
|
||||
if (topo_group_new_conn(ctx->group, conn) == 0) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "recovery: reconnected node=%016llx", (unsigned long long)node_id);
|
||||
node_conn_direct_close(h);
|
||||
topo_recovery_ctx_free(ctx);
|
||||
return;
|
||||
}
|
||||
DEBUG_WARN(DEBUG_CATEGORY_BGP, "recovery: BGP ownership transfer failed node=%016llx", (unsigned long long)node_id);
|
||||
ctx->next = ctx->group->recovery_list; ctx->group->recovery_list = ctx;
|
||||
}
|
||||
node_conn_direct_close(h);
|
||||
for (size_t i = 0; i < ctx->count; i++)
|
||||
if (ctx->nodes[i].node_id == node_id) { topo_recovery_remove_candidate(ctx, i); break; }
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "recovery: attempt finished node=%016llx event=%d", (unsigned long long)node_id, event);
|
||||
topo_recovery_try_next(ctx);
|
||||
}
|
||||
|
||||
@@ -90,7 +103,7 @@ static void topo_recovery_timeout_cb(void* arg) {
|
||||
ctx->current_handle = NULL;
|
||||
}
|
||||
for (size_t i = 0; i < ctx->count; i++)
|
||||
if (ctx->nodes[i].node_id == node_id) { ctx->nodes[i] = ctx->nodes[ctx->count - 1]; ctx->count--; break; }
|
||||
if (ctx->nodes[i].node_id == node_id) { topo_recovery_remove_candidate(ctx, i); break; }
|
||||
ctx->current_node_id = 0;
|
||||
topo_recovery_try_next(ctx);
|
||||
}
|
||||
@@ -125,7 +138,7 @@ static void topo_recovery_try_next(struct TOPO_RECOVERY_CTX* ctx) {
|
||||
if (node_conn_direct_open(ctx->instance, node_id, topo_recovery_callback, ctx,
|
||||
&ctx->current_handle, NULL) == NCD_ERR) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_BGP, "recovery: node_conn_direct_open %016llx failed", (unsigned long long)node_id);
|
||||
*best = ctx->nodes[ctx->count - 1]; ctx->count--;
|
||||
topo_recovery_remove_candidate(ctx, (size_t)(best - ctx->nodes));
|
||||
ctx->current_node_id = 0;
|
||||
continue;
|
||||
}
|
||||
@@ -143,7 +156,7 @@ static void topo_recovery_try_next(struct TOPO_RECOVERY_CTX* ctx) {
|
||||
void topo_recovery_add_node(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* nq, uint64_t next_hop) {
|
||||
if (!group || !nq) return;
|
||||
struct TOPO_NODE* ni = topo_node_registry_find(group->instance->topo_groups, nq->node_id);
|
||||
if (!ni) return;
|
||||
if (!ni) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "recovery: missing candidate=%016llx", (unsigned long long)nq->node_id); return; }
|
||||
uint64_t node_id = nq->node_id;
|
||||
|
||||
struct TOPO_RECOVERY_CTX* ctx = topo_recovery_find_by_next_hop(group, next_hop);
|
||||
@@ -154,15 +167,17 @@ void topo_recovery_add_node(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* nq
|
||||
group->recovery_list = ctx;
|
||||
}
|
||||
|
||||
for (size_t i = 0; i < ctx->count; i++) if (ctx->nodes[i].node_id == node_id) return;
|
||||
uint16_t rtt = topo_get_chain_rtt(nq);
|
||||
|
||||
if (ctx->count >= ctx->capacity) {
|
||||
size_t new_cap = ctx->capacity * 2;
|
||||
struct TOPO_RECOVERY_NODE* new_nodes = u_realloc(ctx->nodes, new_cap * sizeof(struct TOPO_RECOVERY_NODE));
|
||||
if (!new_nodes) return;
|
||||
if (!new_nodes) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "recovery: candidate allocation failed"); return; }
|
||||
ctx->nodes = new_nodes; ctx->capacity = new_cap;
|
||||
}
|
||||
|
||||
topo_node_registry_ref(group->instance->topo_groups, node_id);
|
||||
ctx->nodes[ctx->count].node_id = node_id;
|
||||
ctx->nodes[ctx->count].min_rtt = rtt;
|
||||
ctx->nodes[ctx->count].is_next_hop = (uint8_t)(node_id == next_hop ? 1 : 0);
|
||||
|
||||
Reference in new issue
Block a user