conn_mgr: exercise persistent retries with valid signed node updates

tests/test_conn_mgr_phases.c

@@ -143,13 +143,15 @@ static int learned(struct UTUN_INSTANCE* inst, uint64_t node_id) {
     return defgrp(inst) && topo_node_find_by_id(defgrp(inst), node_id) != NULL;
 }
 
-/* Меняем порт адресов цели на «неверный» (закрытый) — DIRECT создаст линк,
+/* Имитируем свежий подписанный адрес цели с закрытым портом. Подпись обязательна:
- * но handshake не получит ответа и фаза провалится по таймауту NCD. */
+ * длительные сценарии повторяют BGP-обмен и должны сохранять исправные маршруты. */
 static void mangle_addrs(struct UTUN_INSTANCE* inst, uint64_t node_id, uint16_t wrong_port) {
     struct TOPO_NODE* ni = topo_node_registry_find(inst->topo_groups, node_id);
-    if (!ni) return;
+    struct UTUN_INSTANCE* owner = node_id == nid_a ? inst_a : node_id == nid_b ? inst_b : inst_c;
+    if (!ni || !owner || owner->node_id != node_id) { fail("ADDR: missing node/owner"); return; }
     for (struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) a->port = wrong_port;
     for (struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) a6->port = wrong_port;
+    if (topo_node_sign_self(owner, ni) < 0 || topo_node_verify(ni) < 0) fail("ADDR: cannot sign simulated address");
 }
 
 static int best_candidates_ready(struct UTUN_INSTANCE* inst) {
@@ -464,6 +466,8 @@ static int run_upgrade(void) {
     poll_ms(1100);
     if (call_connection_events != events) { fail("CALL PATH: duplicate unchanged status"); return -1; }
     // Потеря прямого транспорта должна вернуть передачу на проверенный резерв без DOWN.
+    mangle_addrs(inst_a, nid_b, (uint16_t)(base_port + 1000));
+    mangle_addrs(inst_b, nid_a, (uint16_t)(base_port + 1000));
     struct ETCP_CONN* direct = node_conn_direct_get_conn(original);
     if (etcp_conn_ref_take(direct)) { fail("UPGRADE: cannot hold direct transport"); return -1; }
     while (direct->links) etcp_link_close(direct->links);
@@ -534,6 +538,95 @@ static int run_invalid_candidates(void) {
     return 0;
 }
 
+/* Реальный обрыв обоих плеч прямого транспорта без закрытия CM-владения. */
+static int drop_direct(struct UTUN_INSTANCE* inst, uint64_t nid) {
+    struct ETCP_CONN* conn = instance_find_conn(inst, nid);
+    if (!conn || etcp_conn_ref_take(conn)) { fail("RECOVERY: missing direct transport"); return -1; }
+    while (conn->links) etcp_link_close(conn->links);
+    etcp_conn_ref_free(conn);
+    return 0;
+}
+
+/* Поздний обрыв, несколько неудачных циклов, свежие адреса и возврат на DIRECT. */
+static int run_recovery(void) {
+    fprintf(stderr, "\n=== Scenario 8: persistent recovery and address change ===\n");
+    if (create_configs("nat", "nat") || scenario_setup()) return -1;
+    if (!poll_until(candidates_a_and_b, 8000)) return -1;
+    uint16_t bad_port = (uint16_t)(base_port + 1000);
+    mangle_addrs(inst_a, nid_b, bad_port); mangle_addrs(inst_b, nid_a, bad_port);
+    if (conn_mgr_open(inst_a, TOPO_GROUP_UTUN, nid_b, conn_cb_a, NULL, &h_a) || !poll_until(ev_a_up, 3000)) return -1;
+    struct CONN_MGR_ENTRY* entry = h_a->entry;
+    uint32_t stale_request = entry->exchange_request_id;
+    struct NODE_CONN_DIRECT* original = entry->ncd_handle;
+    if (etcp_router_bind(inst_b, TEST_CM_SERVICE, data_recv) < 0 || send_data(1, nid_c)) return -1;
+    mangle_addrs(inst_a, nid_b, (uint16_t)(base_port + 2));
+    node_conn_direct_update_node(inst_a, nid_b);
+    if (!poll_until(direct_a_ready, 4000)) { fail("RECOVERY: no initial direct path"); return -1; }
+    // Истёк первоначальный бюджет. Работающее соединение не требует новых проб.
+    poll_ms(CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS + 100);
+    if (entry->attempt_active || entry->timer || timeout_count) { fail("RECOVERY: direct path is still probing"); return -1; }
+    mangle_addrs(inst_a, nid_b, bad_port); mangle_addrs(inst_b, nid_a, bad_port);
+    if (drop_direct(inst_a, nid_b) || drop_direct(inst_b, nid_a)) return -1;
+    poll_ms(20);
+    if (!entry->attempt_active || entry->start_tb + 1000 < get_time_tb() || entry->conn_type != CONN_TYPE_INDIRECT ||
+        entry->ncd_handle != original || down_count || up_count != 1) {
+        fail("RECOVERY: late loss did not start recovery with reserve"); return -1;
+    }
+    if (send_data(2, nid_c)) return -1;
+    // Сокращаем только срок тестовой пробы; интервалы backoff проверяем настоящими часами.
+    for (int cycle = 0; cycle < 3; cycle++) {
+        uint64_t start = entry->start_tb;
+        unsigned delay = entry->retry_delay_ms;
+        entry->deadline_tb = get_time_tb() + 100;
+        cm_arm_timer(entry); poll_ms(30);
+        if (entry->attempt_active || !entry->retry_tb || entry->retry_tb <= get_time_tb()) {
+            fail("RECOVERY: exhausted round did not enter backoff"); return -1;
+        }
+        if (send_data((uint8_t)(3 + cycle), nid_c)) return -1;
+        while (entry->start_tb == start && get_time_tb() < entry->retry_tb + 3000) uasync_poll(ua, POLL_MS);
+        if (entry->start_tb == start || !entry->attempt_active || entry->start_tb - start < delay * 10 || timeout_count) {
+            fail("RECOVERY: retry/backoff lost demand or repeated TIMEOUT"); return -1;
+        }
+    }
+    entry->deadline_tb = get_time_tb() + 100; cm_arm_timer(entry); poll_ms(30);
+    uint64_t start = entry->start_tb;
+    mangle_addrs(inst_a, nid_b, (uint16_t)(base_port + 2));
+    conn_mgr_routes_changed(defgrp(inst_a)->conn_mgr);
+    if (!poll_until(direct_a_ready, 2000) || entry->start_tb == start || entry->ncd_handle != original ||
+        up_count != 1 || down_count || timeout_count || entry->attempt_active || entry->timer) {
+        fail("RECOVERY: address update did not bypass backoff and restore DIRECT"); return -1;
+    }
+    struct CM_INTERM_SEL stale = {0}; stale.request_id = stale_request; stale.count = 1;
+    stale.selected[0] = (struct CONN_MGR_CANDIDATE){nid_c, 1};
+    cm_handle_interm_ack(defgrp(inst_a)->conn_mgr, nid_b, (const uint8_t*)&stale, sizeof(stale));
+    if (send_data(6, nid_a) || entry->conn_type != CONN_TYPE_DIRECT) return -1;
+    // Смена локальных сокетов тоже должна обходить длинный backoff.
+    mangle_addrs(inst_a, nid_b, bad_port); mangle_addrs(inst_b, nid_a, bad_port);
+    if (drop_direct(inst_a, nid_b) || drop_direct(inst_b, nid_a)) return -1;
+    poll_ms(20);
+    entry->deadline_tb = get_time_tb() + 100; cm_arm_timer(entry); poll_ms(30);
+    entry->retry_tb = get_time_tb() + 150000; cm_arm_timer(entry);
+    start = entry->start_tb;
+    mangle_addrs(inst_a, nid_b, (uint16_t)(base_port + 2));
+    entry->node_timestamp = topo_node_registry_find(inst_a->topo_groups, nid_b)->timestamp;
+    conn_mgr_network_changed(defgrp(inst_a)->conn_mgr);
+    if (!poll_until(direct_a_ready, 2000) || entry->start_tb == start) { fail("RECOVERY: local network change did not retry"); return -1; }
+    if (send_data(7, nid_a)) return -1;
+    // Закрытие последнего handle отменяет восстановление даже во время backoff.
+    mangle_addrs(inst_a, nid_b, bad_port); mangle_addrs(inst_b, nid_a, bad_port);
+    if (drop_direct(inst_a, nid_b) || drop_direct(inst_b, nid_a)) return -1;
+    poll_ms(20);
+    entry->deadline_tb = get_time_tb() + 100; cm_arm_timer(entry); poll_ms(30);
+    int ups = up_count, downs = down_count, timeouts = timeout_count;
+    conn_mgr_close(h_a); h_a = NULL;
+    poll_ms(1200);
+    if (cm_find_entry(defgrp(inst_a)->conn_mgr, nid_b) || up_count != ups || down_count != downs || timeout_count != timeouts) {
+        fail("RECOVERY: callback or retry after last close"); return -1;
+    }
+    fprintf(stderr, "  OK: late loss, 1/2/4s retries, unchanged reserve, address/socket wakeup and close cancellation\n");
+    return 0;
+}
+
 /* ─── main ─── */
 
 int main(void) {
@@ -569,6 +662,8 @@ int main(void) {
     scenario_teardown();
     if (run_invalid_candidates() != 0) g_failed = 1;
     scenario_teardown();
+    if (run_recovery() != 0) g_failed = 1;
+    scenario_teardown();
 
     if (ua) { uasync_poll(ua, 0); uasync_destroy(ua, 0); ua = NULL; }