Fix delayed router epoch confirmation and collapse duplicate BLE links

doc/ble_etcp.txt

@@ -27,6 +27,14 @@ Write Without Response/Notify, HIGH и PHY2M включены. BLE version3 до
 подтверждённым исправлением всей ETCP-нагрузки; аппаратные результаты ниже.
 
 Общий код: tools/cgathui_common/ble_etcp.c/h.
+Дубликаты одного ETCP_CONN сворачиваются после аутентифицированного UP победителя:
+остаётся GATT с большим node_id central (исходящий для central, входящий для peripheral).
+Порядок появления встречных линков не влияет на выбор. Одиночный линк с любым
+направлением сохраняется. При одинаковом central только он выбирает ранее созданный
+локальный handle; peripheral следует закрытию с его стороны, не сравнивает свои ID.
+Закрывается только лишний ETCP_LINK штатным путём: неподтверждённые DATA возвращаются
+в очередь, общий conn/его эпоха и прочие транспорты сохраняются. INFO discovery
+содержит peer, keep/close BLE handles, ETCP path IDs, master node IDs и направления.
 Платформенные backend остаются в lib/ble и android_ble.c; Qt не участвует в протоколе.
 UTUN binary этот модуль не собирает. При активном UTUN-сервисе общего instance
 BLE-драйвер не передаёт ETCP: CHAT и UTUN используют общий conn/normalizer,

doc/etcp_router_arch.md

@@ -72,7 +72,13 @@ sequenceDiagram
 
 Получение HELLO/CHALLENGE с неизвестной эпохой само по себе не сбрасывает работающий канал.
 Переставленные/старые CONFIRM не совпадают с pending ID/challenge и игнорируются.
-Новые challenge ограничены интервалом 300 ms; первичный HELLO повторяется без отдельного timeout.
+Для одного pending peer ID повторяется тот же challenge не чаще300мс; срок nonce —2с
+от первой отправки, повторы его не продлевают. Для другого peer ID создаётся новый nonce
+с тем же ограничением частоты300мс: перезапуск пира во время handshake не ждёт истечения старого nonce.
+После истечения срока следующий запрос получает новый nonce; старый CONFIRM отклоняется
+также по возрасту. Первичный HELLO повторяется каждые300мс. Интервал повтора не меняет nonce:
+ответ800мс должен подтверждать сессию и при нескольких повторных START/RST.
+DEBUG etcp_route показывает candidate/nonce, repeat, возраст и received/expected для отвергнутого CONFIRM.
 
 | Событие | Что сохраняется | Что меняется |
 | --- | --- | --- |

doc/tasks.md

@@ -4,6 +4,30 @@
 
 ## Текущие задачи
 
+[+] **Router: сохранять challenge при повторном согласовании медленного аудиоканала** — 31.10.2026.
+    Неудачный SM→Windows звонок e3af74baab3b7f52: ACCEPT и ACTIVE с обеих сторон,
+    Windows11:29:37.106..57.109 rx=0, tx=416, затем no traffic20s. CALL_AUDIO svc63
+    сохранил старую peer epoch d0a70699c433725e, кандидат SM e52a71415f15d163 не подтверждается:
+    за20с46 новых challenge одного кандидата и34 stale response; входящих аудио DATA нет, только control.
+    router_challenge_peer после300мс заменяет pending_challenge. Windows37.918 выдал7b3812be4117d019,
+    38.234 заменил на44063429e8d6b8c3, первое confirm пришло38.698: ответ не успевает до замены nonce.
+    Повторы START/RST поддерживают цикл. Разделить интервал повтора и срок жизни challenge;
+    для того же pending_peer_id повторять прежний nonce, сохранив проверки пары эпох/echo и ограниченный срок.
+    Дополнить stale-диагностику received/expected epoch+nonce и возрастом challenge; на SM включить
+    etcp_route=debug при повторе (native logcat начала звонка уже вытеснен, Telecom подтвердил ACTIVE).
+    Проверка: повторный звонок после смены Router epoch, RTT>300мс, дубли/переупорядочение START/confirm,
+    защита от старого/чужого ответа. Артефакты: chatgui.log и /tmp/utun-call-failed-20261031/sm-logcat.txt.
+    Дополнительно: у Windows одновременно2 BLE UP-пути к SM, поэтому single_path=0 и увеличенный
+    DATA RTO не применяется; Router handshake300мс от DATA RTO независим.
+    Исправлено: nonce повторяется каждые300мс и живёт2с без продления при повторах;
+    другой кандидат получает новый nonce после общего ограничения частоты300мс. Confirm проверяет срок, пару эпох и nonce,
+    DEBUG содержит received/expected и возраст. Тесты:800мс confirm, повторный рестарт,
+    сохранение nonce, конкурирующий кандидат, просроченный/старый/неверный ответ — PASS.
+    BLE-дубликаты сворачиваются по большему node_id central после UP победителя;
+    оба порядка появления, одинаковый central, DATA в полёте, неизменность эпох/отсутствие утечек — PASS.
+    Итоговые проверки с ожиданием2с: ядро233/233, proxy/burst/load PASS, desktop116/116,
+    Android308/308, build.sh44/44. Звонок на физических устройствах после обновления ещё не проверен.
+
 [ ] **ETCP/BLE: уведомление о сбросе принятых драйвером буферов** — 31.10.2026, проработать отдельно.
     При reconnect или другом сбросе драйвер сообщает, какие принятые отправки больше не будут доставлены.
     ETCP снимает соответствующие физические попытки и немедленно ставит неподтверждённые DATA на повтор

src/routing_layer/etcp_router.c

@@ -423,24 +423,37 @@ static void router_handshake_timer_cb(void* arg) {
 static void router_challenge_peer(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint64_t peer_id) {
     uint64_t now = get_time_tb();
     if (rconn->pending_peer_id && now - rconn->challenge_sent_tb < ROUTER_HANDSHAKE_TIMEOUT_TB) return;
-    if (router_random_id(&rconn->pending_challenge) != 0) return;
-    rconn->pending_peer_id = peer_id;
+    int pending = rconn->pending_peer_id == peer_id && now - rconn->challenge_started_tb < ROUTER_CHALLENGE_LIFETIME_TB;
+    if (!pending) {
+        if (router_random_id(&rconn->pending_challenge) != 0) return;
+        rconn->pending_peer_id = peer_id; rconn->challenge_started_tb = now;
+    }
     rconn->challenge_sent_tb = now;
+    DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router challenge: peer=%016llx svc=%u candidate=%016llx nonce=%016llx "
+        "repeat=%d age_ms=%llu lifetime_ms=%u",
+        (unsigned long long)rconn->remote_node_id, rconn->svc_id, (unsigned long long)peer_id,
+        (unsigned long long)rconn->pending_challenge, pending,
+        (unsigned long long)((now - rconn->challenge_started_tb) / 10), ROUTER_CHALLENGE_LIFETIME_TB / 10);
     router_control(rconn, conn, ROUTER_FLAG_RST, peer_id, rconn->pending_challenge);
 }
 
 // Только ответ на наш свежий challenge разрешает сменить пару эпох. Свой id сохраняем:
 // иначе два конца бесконечно перезапускали бы друг друга.
 static void router_confirm_peer(struct ETCP_ROUTER_CONN* rconn, const struct SVC_ROUTE_HDR* h) {
+    uint64_t age = get_time_tb() - rconn->challenge_started_tb;
     if (h->peer_reset_id != rconn->reset_id || !h->challenge || h->challenge != rconn->pending_challenge
-        || h->reset_id != rconn->pending_peer_id) {
-        DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router confirm: stale response peer=%016llx svc=%u",
-                    (unsigned long long)rconn->remote_node_id, rconn->svc_id);
+        || h->reset_id != rconn->pending_peer_id || age >= ROUTER_CHALLENGE_LIFETIME_TB) {
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router confirm: stale response peer=%016llx svc=%u "
+            "received=%016llx/%016llx/%016llx expected=%016llx/%016llx/%016llx age_ms=%llu expired=%d",
+            (unsigned long long)rconn->remote_node_id, rconn->svc_id,
+            (unsigned long long)h->reset_id, (unsigned long long)h->peer_reset_id, (unsigned long long)h->challenge,
+            (unsigned long long)rconn->pending_peer_id, (unsigned long long)rconn->reset_id,
+            (unsigned long long)rconn->pending_challenge, (unsigned long long)(age / 10), age >= ROUTER_CHALLENGE_LIFETIME_TB);
         return;
     }
     uint64_t previous = rconn->peer_reset_id;
     uint64_t peer_id = h->reset_id;
-    rconn->pending_peer_id = 0; rconn->pending_challenge = 0;
+    rconn->pending_peer_id = 0; rconn->pending_challenge = 0; rconn->challenge_started_tb = rconn->challenge_sent_tb = 0;
     if (previous == peer_id) return;
     if (previous && router_conn_reset(rconn) != 0) { router_close_and_notify(rconn); return; }
     if (rconn->handshake_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; }
@@ -1671,7 +1684,7 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uin
     if (router_random_id(&new_id) != 0) { router_close_and_notify(rconn); return; }
     if (router_conn_reset(rconn) != 0) { router_close_and_notify(rconn); return; }
     rconn->reset_id = new_id;
-    rconn->pending_peer_id = 0; rconn->pending_challenge = 0;
+    rconn->pending_peer_id = 0; rconn->pending_challenge = 0; rconn->challenge_started_tb = rconn->challenge_sent_tb = 0;
     router_send_close_to_service(rconn);
     if (!rconn->closed) router_handshake_start(rconn);
 }

src/routing_layer/etcp_router.h

@@ -167,7 +167,8 @@ struct ETCP_ROUTER_CONN {
     uint64_t peer_reset_id;      // подтверждённый идентификатор пира (0 = handshake не завершён)
     uint64_t pending_peer_id;    // кандидат, ещё не имеющий права менять состояние сессии
     uint64_t pending_challenge; // случайный запрос подтверждения кандидату
-    uint64_t challenge_sent_tb; // время отправки запроса, 0.1 ms
+    uint64_t challenge_started_tb; // начало ограниченного срока nonce; повторы не продлевают, 0.1ms
+    uint64_t challenge_sent_tb; // последняя отправка того же nonce; ограничивает частоту повторов, 0.1ms
     void* handshake_timer;     // таймер повторов согласования
     uint8_t  start_sent;         // получен первый ACK данных
 
@@ -228,6 +229,7 @@ int etcp_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t pe
 
 // Ретрансмиты
 #define ROUTER_HANDSHAKE_TIMEOUT_TB 3000 // 300ms: согласование сессии, независимо от DATA
+#define ROUTER_CHALLENGE_LIFETIME_TB 20000 // 2с: срок одного nonce, независимо от интервала повторов
 #define ROUTER_RETRANS_MIN_TB     20000  // 2с: нижний транспорт уже обеспечивает надёжную доставку
 #define ROUTER_RETRANS_BACKOFF_TB 300000 // 30с: предел backoff, но не меньше измеренного базового RTO
 #define ROUTER_INFLIGHT_HASH_SIZE  1024

tests/test_etcp_router_faults.c

@@ -29,6 +29,9 @@ static int failed, drop_data, drop_ack, drop_hello[2], drop_challenge[2], drop_c
 static int reorder, duplicated, dropped_data, dropped_ack, handshake_drops, closed_in_handler;
 static unsigned max_wire;
 static struct ll_entry *held, *saved_data, *saved_ack, *saved_hello, *saved_confirm;
+static struct ll_entry* delayed_confirm[2];
+static uint64_t confirm_due_tb[2];
+static int delay_confirm;
 
 static void release(struct ll_entry* e) { if (e) { queue_dgram_free(e); queue_entry_free(e); } }
 static struct ll_entry* copy_packet(const struct ll_entry* e) {
@@ -83,6 +86,11 @@ static void wire(int from, struct ll_queue* q) {
     if (!data && h->flags == ROUTER_FLAG_RST) drop = &drop_challenge[from];
     if (!data && h->flags == (ROUTER_FLAG_START | ROUTER_FLAG_RST)) drop = &drop_confirm[from];
     if (drop && *drop) { --*drop; handshake_drops++; release(e); return; }
+    if (delay_confirm && !data && h->flags == (ROUTER_FLAG_START | ROUTER_FLAG_RST)) {
+        if (!delayed_confirm[from]) { delayed_confirm[from] = e; confirm_due_tb[from] = get_time_tb() + 8000; }
+        else release(e);
+        return;
+    }
     if (data && from == 0 && h->svc_id == svc_base) {
         if (!saved_data) saved_data = copy_packet(e);
         if (drop_data && h->seq == 2) { drop_data--; dropped_data++; release(e); return; }
@@ -102,12 +110,16 @@ static void wire(int from, struct ll_queue* q) {
 static void step(void) {
     test_clock_poll(ua, 1);
     for (int i = 0; i < 2; i++) { wire(i, ep[i].conn.send_input_q); wire(i, ep[i].alternate.send_input_q); }
+    for (int i = 0; i < 2; i++) if (delayed_confirm[i] && get_time_tb() >= confirm_due_tb[i]) {
+        struct ll_entry* e = delayed_confirm[i]; delayed_confirm[i] = NULL; inject(1-i, e);
+    }
 }
 static int setup(void) {
     memset(ep, 0, sizeof(ep)); failed = 0; ua = uasync_create(); REQUIRE(ua);
     drop_data = drop_ack = drop_all_ack = reorder = duplicated = dropped_data = dropped_ack = handshake_drops = closed_in_handler = 0;
     memset(drop_hello, 0, sizeof(drop_hello)); memset(drop_challenge, 0, sizeof(drop_challenge)); memset(drop_confirm, 0, sizeof(drop_confirm));
     drop_wire = 0;
+    delay_confirm = 0; memset(delayed_confirm, 0, sizeof(delayed_confirm));
     max_wire = 0; held = saved_data = saved_ack = saved_hello = saved_confirm = NULL;
     for (int i = 0; i < 2; i++) {
         ep[i].inst.ua = ua; ep[i].inst.node_id = i + 1;
@@ -138,6 +150,7 @@ static void cleanup(void) {
         release(queue_data_get(ep[i].inst.connections)); queue_free(ep[i].inst.connections);
     }
     release(held); release(saved_data); release(saved_ack); release(saved_hello); release(saved_confirm);
+    for (int i = 0; i < 2; ++i) release(delayed_confirm[i]);
     uasync_poll(ua, 0);
     if (ua->timer_alloc_count != ua->timer_free_count) {
         fprintf(stderr, "FAIL teardown: live timers=%llu\n", (unsigned long long)(ua->timer_alloc_count - ua->timer_free_count));
@@ -225,6 +238,70 @@ static int reinit_retains(void) {
     REQUIRE(ep[0].notified[0] == 0 && ep[1].notified[0] == 0);
     cleanup(); return 0;
 }
+/* Confirm800мс при повторах START300мс воспроизводит зацикливание из Windows/SM лога. */
+static int delayed_handshake(void) {
+    REQUIRE(setup() == 0); delay_confirm = 1;
+    REQUIRE(send_one(0,0) == 0); REQUIRE(send_one(1,0) == 0);
+    REQUIRE(wait_settled(0,0) == 0); REQUIRE(wait_settled(1,0) == 0);
+    REQUIRE(rc(0,0)->peer_reset_id == rc(1,0)->reset_id && rc(1,0)->peer_reset_id == rc(0,0)->reset_id);
+    REQUIRE(!ep[0].notified[0] && !ep[1].notified[0]);
+    delay_confirm = 0;
+    for (int i = 0; i < 2; ++i) if (delayed_confirm[i]) {
+        struct ll_entry* e = delayed_confirm[i]; delayed_confirm[i] = NULL; inject(1-i, e);
+    }
+    // Перезапуск одной стороны после работавшего звонка: вторая должна принять новую эпоху с тем же поздним confirm.
+    etcp_router_conn_restart(&ep[0].inst, 0, ep[1].inst.node_id, svc_base);
+    delay_confirm = 1;
+    REQUIRE(send_one(0,0) == 0); REQUIRE(wait_settled(0,0) == 0);
+    REQUIRE(rc(1,0)->peer_reset_id == rc(0,0)->reset_id);
+    REQUIRE(ep[0].notified[0] == 1 && ep[1].notified[0] == 1);
+    cleanup(); return 0;
+}
+
+static struct ll_entry* candidate_packet(uint64_t peer_epoch, uint8_t flags, uint64_t nonce) {
+    struct ll_entry* e = copy_packet(saved_hello);
+    struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)e->dgram;
+    h->reset_id = peer_epoch; h->peer_reset_id = rc(1,0)->reset_id; h->flags = flags; h->challenge = nonce;
+    return e;
+}
+
+static int challenge_lifetime(void) {
+    REQUIRE(setup() == 0); REQUIRE(send_one(0,0) == 0); REQUIRE(wait_settled(0,0) == 0);
+    struct ETCP_ROUTER_CONN* c = rc(1,0);
+    uint64_t previous = c->peer_reset_id, candidate = previous ^ 1;
+    if (!candidate) candidate = 2;
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START, 0));
+    uint64_t nonce = c->pending_challenge, started = c->challenge_started_tb;
+    REQUIRE(nonce && c->pending_peer_id == candidate);
+    test_clock_run_for(ua, 3500);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START, 0));
+    REQUIRE(c->pending_challenge == nonce && c->challenge_started_tb == started && c->challenge_sent_tb > started);
+    inject(1, candidate_packet(candidate ^ 4, ROUTER_FLAG_START, 0));
+    REQUIRE(c->pending_peer_id == candidate && c->pending_challenge == nonce);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START | ROUTER_FLAG_RST, nonce ^ 1));
+    REQUIRE(c->peer_reset_id == previous && !ep[1].notified[0]);
+    // Пир может перезапуститься до завершения проверки: другой кандидат не ждёт весь срок старого nonce.
+    test_clock_run_for(ua, 3500);
+    inject(1, candidate_packet(candidate ^ 4, ROUTER_FLAG_START, 0));
+    REQUIRE(c->pending_peer_id == (candidate ^ 4) && c->pending_challenge && c->pending_challenge != nonce);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START | ROUTER_FLAG_RST, nonce));
+    REQUIRE(c->peer_reset_id == previous && !ep[1].notified[0]);
+    test_clock_run_for(ua, 3500);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START, 0));
+    REQUIRE(c->pending_peer_id == candidate && c->pending_challenge && c->pending_challenge != nonce);
+    nonce = c->pending_challenge; started = c->challenge_started_tb;
+    test_clock_run_for(ua, ROUTER_CHALLENGE_LIFETIME_TB);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START | ROUTER_FLAG_RST, nonce));
+    REQUIRE(c->peer_reset_id == previous); // Подходящий nonce после истечения срока тоже не принимается.
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START, 0));
+    REQUIRE(c->pending_challenge && c->pending_challenge != nonce && c->challenge_started_tb > started);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START | ROUTER_FLAG_RST, nonce));
+    REQUIRE(c->peer_reset_id == previous);
+    inject(1, candidate_packet(candidate, ROUTER_FLAG_START | ROUTER_FLAG_RST, c->pending_challenge));
+    REQUIRE(c->peer_reset_id == candidate && ep[1].notified[0] == 1 && !c->pending_challenge);
+    cleanup(); return 0;
+}
+
 static int stale_session(void) {
     REQUIRE(setup() == 0); REQUIRE(send_one(0,0) == 0); REQUIRE(wait_settled(0,0) == 0);
     uint64_t previous = rc(0,0)->reset_id;
@@ -588,6 +665,8 @@ int main(void) {
     struct { const char* name; int (*run)(void); } tests[] = {
         {"lazy producer readiness, route switch, cancellation, restart and close", lazy_send_requests},
         {"loss/reorder/duplicate and duplex multiplexing", loss_reorder}, {"lost HELLO/challenge/confirm", handshake_loss},
+        {"800ms confirm survives 300ms retries and peer restart", delayed_handshake},
+        {"challenge nonce reuse, competing candidate, expiry and old echo", challenge_lifetime},
         {"ACK bounds and no false progress", ack_bounds}, {"transport reinit retains inflight", reinit_retains},
         {"restart rejects old DATA/ACK/CLOSE/HELLO/confirm", stale_session}, {"sequence wraparound", wraparound},
         {"waiter lifetime after route switch", waiter_lifetime}, {"service closes during delivery", close_from_handler}, {"route switch while old next hop blocked", route_switch},

tools/cgathui_common/ble_etcp.c

@@ -25,7 +25,7 @@ struct ble_etcp_link {
     struct ETCP_LINK* link; /* Заимствован до stop; ETCP владеет ctx после bind. */
     struct packet_link_events events;
     void* arg;
-    int ready, closed, stopped, bound, rx_enabled, key_sent, key_received;
+    int ready, closed, stopped, bound, rx_enabled, key_sent, key_received, outgoing;
     int invitation_only, rejected_data_logged;
     uint8_t peer_key[32]; /* Bootstrap ещё не доказывает владение; хранится для повышения INVITE-only после JOIN. */
     uint64_t admission_retry_us;
@@ -142,6 +142,7 @@ static int attach(struct ble_etcp_link* c, const uint8_t* key) {
     if (sc_derive_node_id_from_pubkey(key) != c->node_id) { fail(c, "public key does not match BLE node ID"); return -1; }
     int outgoing;
     if (ble_get_outgoing(d->radio, c->id, &outgoing)) { fail(c, "GATT direction unavailable"); return -1; }
+    c->outgoing = outgoing;
     struct ETCP_CONN* conn = instance_find_conn(d->instance, c->node_id);
     int created = 0;
     if (conn && (conn->close_requested || conn->state == 2)) { fail(c, "ETCP connection is closing"); return -1; }
@@ -572,6 +573,33 @@ static void groups(struct ble_etcp* d) {
         if (wanted && ble_etcp_connect(d, node)) DEBUG_WARN(CAT, "BLE group connect deferred peer=%016llx", (unsigned long long)node);
     }
 }
+/* Мастер — GATT central. Закрываем проигравший путь лишь после аутентифицированного UP победителя.
+ * При одном мастере выбор делает только central (меньший локальный handle): peripheral следует его закрытию. */
+static void collapse_duplicates(struct ble_etcp* d) {
+    if (!d->instance) return;
+    uint64_t self = d->instance->node_id;
+    for (int i = 0; i < BLE_LINKS_MAX; i++) {
+        struct ble_etcp_link* keep = d->links[i];
+        if (!keep || keep->closed || keep->stopped || !keep->link || keep->link->closing ||
+            !keep->link->initialized || !keep->link->link_status) continue;
+        uint64_t master = keep->outgoing ? self : keep->node_id;
+        for (int j = 0; j < BLE_LINKS_MAX; j++) {
+            struct ble_etcp_link* drop = d->links[j];
+            if (!drop || drop == keep || drop->closed || drop->stopped || !drop->link || drop->link->closing ||
+                drop->link->etcp != keep->link->etcp) continue;
+            uint64_t other_master = drop->outgoing ? self : drop->node_id;
+            if (master < other_master || (master == other_master && (master != self || keep->id > drop->id))) continue;
+            DEBUG_INFO(CAT, "BLE duplicate: peer=%016llx keep=%llu path=%u master=%016llx outgoing=%d "
+                "close=%llu path=%u master=%016llx outgoing=%d winner_authenticated_up=1",
+                (unsigned long long)keep->node_id, (unsigned long long)keep->id, keep->link->local_link_id,
+                (unsigned long long)master, keep->outgoing, (unsigned long long)drop->id, drop->link->local_link_id,
+                (unsigned long long)other_master, drop->outgoing);
+            etcp_link_close(drop->link);
+            // Status callbacks могут закрыть соседей: следующий выбор делаем заново после выхода из их стека.
+            schedule(d); return;
+        }
+    }
+}
 static void run(void* arg) {
     struct ble_etcp* d = arg; d->work = NULL;
     for (int i = 0; i < BLE_LINKS_MAX; i++) if (d->rejected[i]) {
@@ -582,6 +610,7 @@ static void run(void* arg) {
     ble_join_run(d->bootstrap);
     if (d->check_groups && d->instance) { d->check_groups = 0; groups(d); }
     for (int i = 0; i < BLE_LINKS_MAX; i++) if (d->links[i]) service_link(d->links[i]);
+    collapse_duplicates(d);
 }
 static void tick(void* arg) {
     struct ble_etcp* d = arg; d->timer = NULL;

tools/cgathui_common/ble_etcp.h

@@ -5,6 +5,7 @@
  * Таблица соседей принадлежит этому объекту, объявления не удостоверяют личность.
  * Все операции в ua, destroy до instance и вне callbacks. При instance=NULL — только discovery/сырой BLE (пробы).
  * CHAT fallback: обнаруженный член группы без живого BGP-пути; UTUN запрещает ETCP поверх BLE.
+ * Дубликаты одного ETCP_CONN: остаётся UP-линк с большим node_id GATT central; одиночный путь не заменяется.
  */
 #ifndef CHAT_BLE_ETCP_H
 #define CHAT_BLE_ETCP_H

tools/cgathui_common/tests/test_ble_etcp.c

@@ -32,7 +32,13 @@ struct endpoint {
 };
 static struct endpoint endpoints[2];
 static struct UASYNC* ua;
-struct fake_link { ble_link_id id, other; struct ble_backend* peer; uint8_t data[512]; size_t len; int pending; enum ble_wire_channel channel; };
+struct fake_link {
+    ble_link_id id, other;
+    struct ble_backend* peer;
+    uint8_t data[512]; size_t len;
+    int pending, disconnected;
+    enum ble_wire_channel channel;
+};
 struct ble_backend { struct ble_transport* owner; struct UASYNC* ua; void* task; int index, started; struct fake_link links[16]; };
 static struct ble_backend* backends[2];
 static unsigned int mtu = 23;
@@ -56,7 +62,7 @@ static struct fake_link* slot(struct ble_backend* b, ble_link_id id) {
 int ble_backend_connect(struct ble_backend* b, ble_link_id id, const struct ble_endpoint* ep) {
     struct ble_backend* p = backends[atoi(ep->id)]; CHECK(p);
     struct fake_link *a = slot(b, 0), *c = slot(p, 0); CHECK(a && c);
-    struct ble_endpoint remote = {0}; snprintf(remote.id, sizeof(remote.id), "%d", b->index);
+    struct ble_endpoint remote = {0}; snprintf(remote.id, sizeof(remote.id), "%d/incoming/%llu", b->index, (unsigned long long)id);
     ble_link_id other = ble_backend_incoming(p->owner, &remote); CHECK(other);
     *a = (struct fake_link){.id=id, .other=other, .peer=p};
     *c = (struct fake_link){.id=other, .other=id, .peer=b};
@@ -64,7 +70,13 @@ int ble_backend_connect(struct ble_backend* b, ble_link_id id, const struct ble_
     schedule(b); schedule(p); return 0;
 }
 void ble_backend_close(struct ble_backend* b, ble_link_id id) {
-    struct fake_link* f = slot(b, id); if (f) memset(f, 0, sizeof(*f));
+    struct fake_link* f = slot(b, id);
+    if (!f) return;
+    for (int i = 0; i < 2; i++) if (backends[i] && backends[i] == f->peer) {
+        struct fake_link* other = slot(backends[i], f->other);
+        if (other) { other->disconnected = 1; schedule(backends[i]); }
+    }
+    memset(f, 0, sizeof(*f));
 }
 int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_wire_channel channel, const uint8_t* data, size_t len) {
     struct fake_link* f = slot(b, id); CHECK(f && !f->pending && len <= sizeof(f->data));
@@ -76,6 +88,7 @@ static void run(void* arg) {
     if (!b->started) { b->started = 1; ble_backend_status(b->owner, 1, "fake ready"); }
     for (int i = 0; i < 16; i++) {
         struct fake_link* f = &b->links[i]; if (!f->id) continue;
+        if (f->disconnected) { ble_backend_closed(b->owner, f->id, "peer GATT closed"); continue; }
         if (!f->len) { f->len = 1; ble_backend_opened(b->owner, f->id, mtu); }
         if (!f->pending || hold) continue;
         struct fake_link* target = f->peer ? slot(f->peer, f->other) : NULL;
@@ -144,6 +157,65 @@ static void exchange(unsigned seq) {
 
 static void drain(void) { for (int i = 0; i < 1000; i++) uasync_poll(ua, 0); }
 
+static void duplicate_links(int first_preferred, int same_master) {
+    size_t baseline = u_get_allocated_count();
+    memset(endpoints, 0, sizeof(endpoints)); created = hold = drop = 0; mtu = 517; setup();
+    for (int i = 0; i < 2; ++i) {
+        struct ble_config config = {.node_id = endpoints[i].inst.node_id, .company_id = 0xffff};
+        endpoints[i].driver = ble_etcp_create(ua, &endpoints[i].inst, &config, 0); assert(endpoints[i].driver);
+        assert(!ble_start(ble_etcp_radio(endpoints[i].driver)));
+    }
+    drain();
+    int larger = endpoints[0].inst.node_id > endpoints[1].inst.node_id ? 0 : 1;
+    int first = first_preferred ? larger : 1 - larger;
+    struct ble_endpoint remote = {0}; snprintf(remote.id, sizeof(remote.id), "%d", 1 - first);
+    assert(ble_connect(ble_etcp_radio(endpoints[first].driver), &remote, endpoints[1-first].inst.node_id));
+    wait_ready();
+    struct ETCP_CONN* original[2] = {endpoints[0].conn, endpoints[1].conn};
+    uint64_t epochs[2] = {original[0]->reset_id, original[1]->reset_id};
+    unsigned reinit[2] = {original[0]->reinit_count, original[1]->reinit_count};
+    hold = 1; exchange(0); drain(); // Принятые нижним драйвером DATA ещё не доставлены.
+    int second = same_master ? first : 1 - first;
+    snprintf(remote.id, sizeof(remote.id), "%d/second", 1 - second);
+    assert(ble_connect(ble_etcp_radio(endpoints[second].driver), &remote, endpoints[1-second].inst.node_id));
+    drain();
+    for (int i = 0; i < 2; ++i) assert(original[i]->links_up && original[i]->links && !original[i]->links->next);
+    hold = 0; schedule(backends[0]); schedule(backends[1]);
+    uint64_t end = get_time_us() + 5000000;
+    int ready = 0;
+    while (!ready && get_time_us() < end) {
+        uasync_poll(ua, 10); ready = 1;
+        for (int i = 0; i < 2; ++i) {
+            struct ETCP_LINK* link = original[i]->links;
+            if (!link || link->next || !link->initialized || !link->link_status || endpoints[i].received != 1) ready = 0;
+        }
+    }
+    assert(ready);
+    int master = same_master ? first : larger;
+    for (int i = 0; i < 2; ++i) {
+        struct etcp_link_snapshot snapshot;
+        assert(instance_find_conn(&endpoints[i].inst, endpoints[1-i].inst.node_id) == original[i]);
+        assert(original[i]->reset_id == epochs[i] && original[i]->reinit_count == reinit[i]);
+        assert(!etcp_link_get_snapshot(original[i]->links, &snapshot));
+        assert(snapshot.direction == (i == master ? PACKET_LINK_OUTGOING : PACKET_LINK_INCOMING));
+        assert(original[i]->links->single_path_rto_active);
+    }
+    exchange(1); end = get_time_us() + 5000000;
+    while ((endpoints[0].received != 2 || endpoints[1].received != 2) && get_time_us() < end) uasync_poll(ua, 10);
+    assert(endpoints[0].received == 2 && endpoints[1].received == 2);
+    for (int i = 0; i < 2; ++i) ble_etcp_destroy(endpoints[i].driver);
+    drain();
+    for (int i = 0; i < 2; ++i) etcp_connection_close(original[i]);
+    uasync_drain_immediate(ua);
+    for (int i = 0; i < 2; ++i) {
+        queue_free(endpoints[i].inst.connections); memory_pool_destroy(endpoints[i].inst.pkt_pool);
+        memory_pool_destroy(endpoints[i].inst.data_pool); memory_pool_destroy(endpoints[i].inst.ack_pool);
+    }
+    uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline);
+    printf("PASS BLE duplicates: first_preferred=%d same_master=%d, winner UP, pending DATA retained, no epoch reset/leaks\n",
+        first_preferred, same_master);
+}
+
 /* Подмена ключа после HELLO не должна создать ETCP с заявленным node_id. */
 static void reject_key(void) {
     memset(endpoints, 0, sizeof(endpoints)); created = hold = drop = 0; corrupt_keys = 2; mtu = 23; setup();
@@ -493,6 +565,7 @@ int main(void) {
         memory_pool_destroy(endpoints[i].inst.data_pool); memory_pool_destroy(endpoints[i].inst.ack_pool);
     }
     uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline);
+    duplicate_links(0, 0); duplicate_links(1, 0); duplicate_links(1, 1);
     reject_key(); assert(u_get_allocated_count() == baseline);
     invitation_without_etcp(); assert(u_get_allocated_count() == baseline);
     group_fallback();