tests: accelerate router fault timers with real I/O clock driver

tests/Makefile.am

@@ -176,6 +176,13 @@ CRYPTO_LIBS = -lcrypto
 # Common libraries (lib/dns.c собирается с -DDNS_RANDOM=RAND_bytes → нужен -lcrypto)
 COMMON_LIBS = $(top_builddir)/lib/libuasync.a $(CRYPTO_LIBS) -lpthread $(WIN_LIBS) @FFMPEG_LIBS@ $(top_builddir)/lib/libopus/libopus_internal.a -lm
 
+test_clock_sources = test_clock.c test_clock.h
+if OS_LINUX
+test_clock_ldflags = -Wl,--wrap=clock_gettime -Wl,--wrap=epoll_wait -Wl,--wrap=poll
+else
+test_clock_ldflags =
+endif
+
 # Вариант общего ядра с Android duty-cycle, без JNI и устройства.
 if OS_LINUX
 test_stcp_LDFLAGS = -Wl,--wrap=send -Wl,--wrap=recv -Wl,--wrap=random_bytes -Wl,--wrap=u_malloc_impl -Wl,--wrap=u_calloc_impl
@@ -303,8 +310,9 @@ test_lwip_tcp_LDADD = $(LIBUTUN) $(COMMON_LIBS)
 test_etcp_router_SOURCES = test_etcp_router.c
 test_etcp_router_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 
-test_etcp_router_faults_SOURCES = test_etcp_router_faults.c
+test_etcp_router_faults_SOURCES = test_etcp_router_faults.c $(test_clock_sources)
 test_etcp_router_faults_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_etcp_router_faults_LDFLAGS = $(test_clock_ldflags)
 
 test_etcp_router_unit_SOURCES = test_etcp_router_unit.c
 test_etcp_router_unit_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)

tests/test_clock.c

@@ -0,0 +1,100 @@
+#include "test_clock.h"
+#include "../lib/debug_config.h"
+#include <stdlib.h>
+#include <time.h>
+
+#ifdef __linux__
+#include <sys/epoll.h>
+#include <poll.h>
+
+static int accelerated;
+static uint64_t offset_us, limit_us, advances, io_events;
+static struct UASYNC* polling;
+static uint64_t step_us;
+
+int __real_clock_gettime(clockid_t id, struct timespec* ts);
+int __wrap_clock_gettime(clockid_t id, struct timespec* ts) {
+    int result = __real_clock_gettime(id, ts);
+    if (!result && (id == CLOCK_BOOTTIME || id == CLOCK_MONOTONIC)) {
+        uint64_t ns = (uint64_t)ts->tv_nsec + __atomic_load_n(&offset_us, __ATOMIC_RELAXED) * 1000;
+        ts->tv_sec += ns / 1000000000; ts->tv_nsec = ns % 1000000000;
+    }
+    return result;
+}
+
+/* Продвигаем время внутри wait: uasync видит нормальное ожидание, а не искусственный stall между poll. */
+static void advance_idle(int result, int timeout_ms) {
+    if (!polling) return;
+    if (result > 0) { io_events += result; return; }
+    if (result < 0 || !timeout_ms || polling->immediate_queue_head) return;
+    uint64_t now = get_time_us(), end = now + step_us;
+    TimeoutEntry next;
+    if (!timeout_heap_peek(polling->timeout_heap, &next) && next.expiration * 1000 < end) end = next.expiration * 1000;
+    if (limit_us && limit_us < end) end = limit_us;
+    if (end <= now) return;
+    __atomic_fetch_add(&offset_us, end - now, __ATOMIC_RELAXED); advances++;
+}
+
+int __real_epoll_wait(int fd, struct epoll_event* events, int count, int timeout);
+int __wrap_epoll_wait(int fd, struct epoll_event* events, int count, int timeout) {
+    int wait_ms = polling ? (timeout && polling->socket_alloc_count - polling->socket_free_count > 1 ? 1 : 0) : timeout;
+    int result = __real_epoll_wait(fd, events, count, wait_ms);
+    advance_idle(result, timeout);
+    return result;
+}
+
+int __real_poll(struct pollfd* fds, nfds_t count, int timeout);
+int __wrap_poll(struct pollfd* fds, nfds_t count, int timeout) {
+    int wait_ms = polling ? (timeout && count > 1 ? 1 : 0) : timeout;
+    int result = __real_poll(fds, count, wait_ms);
+    advance_idle(result, timeout);
+    return result;
+}
+#endif
+
+void test_clock_init(int max_step_tb) {
+    if (max_step_tb <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test clock: invalid step=%d", max_step_tb); abort(); }
+    debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
+#ifdef __linux__
+    accelerated = getenv("UTUN_TEST_REAL_TIME") == NULL;
+    step_us = (uint64_t)max_step_tb * 100;
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "test clock: mode=%s max_step_tb=%d; native timers and real I/O",
+               accelerated ? "accelerated" : "real", max_step_tb);
+#else
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "test clock: real time on this platform");
+#endif
+}
+
+void test_clock_poll(struct UASYNC* ua, int timeout_tb) {
+#ifdef __linux__
+    if (accelerated && timeout_tb > 0) {
+        if (polling) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test clock: nested poll is unsupported"); abort(); }
+        polling = ua;
+        uasync_poll(ua, timeout_tb);
+        polling = NULL;
+        return;
+    }
+#endif
+    uasync_poll(ua, timeout_tb);
+}
+
+void test_clock_run_for(struct UASYNC* ua, int duration_tb) {
+    if (duration_tb < 0) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test clock: negative duration=%d", duration_tb); abort(); }
+    uint64_t end = get_time_us() + (uint64_t)duration_tb * 100;
+#ifdef __linux__
+    if (limit_us) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test clock: nested time interval is unsupported"); abort(); }
+    limit_us = end;
+#endif
+    while (get_time_us() < end) test_clock_poll(ua, 5);
+#ifdef __linux__
+    limit_us = 0;
+#endif
+}
+
+void test_clock_finish(void) {
+#ifdef __linux__
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "test clock: completed mode=%s advanced_tb=%llu steps=%llu real_io_events=%llu",
+               accelerated ? "accelerated" : "real", (unsigned long long)(offset_us / 100),
+               (unsigned long long)advances, (unsigned long long)io_events);
+#endif
+}

tests/test_clock.h

@@ -0,0 +1,14 @@
+#ifndef TEST_CLOCK_H
+#define TEST_CLOCK_H
+
+#include "../lib/u_async.h"
+
+/* Linux: часы продвигаются при пустом I/O до ближайшего таймера, с сохранением реальных сокетов.
+ * UTUN_TEST_REAL_TIME=1 и остальные платформы используют обычные часы/ожидание.
+ * init — до создания uasync; finish — после уничтожения. Вызовы только из потока теста, не из callbacks. */
+void test_clock_init(int max_step_tb);
+void test_clock_poll(struct UASYNC* ua, int timeout_tb);
+void test_clock_run_for(struct UASYNC* ua, int duration_tb);
+void test_clock_finish(void);
+
+#endif

tests/test_etcp_router_faults.c

@@ -9,6 +9,7 @@
 #include "utun_instance.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
+#include "test_clock.h"
 
 #define SVC 0x42
 static uint8_t svc_base = SVC;
@@ -99,7 +100,7 @@ static void wire(int from, struct ll_queue* q) {
     inject(1-from, e);
 }
 static void step(void) {
-    uasync_poll(ua, 1);
+    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); }
 }
 static int setup(void) {
@@ -173,6 +174,12 @@ static int transfer(void) {
         }
         step();
     }
+    if (!complete) for (int side = 0; side < 2; side++) for (int svc = 0; svc < 2; svc++) {
+        struct ETCP_ROUTER_CONN* c = rc(side, svc);
+        DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "router transfer incomplete: side=%d svc=%d sent=%u received=%u acked=%u tx=%u queued=%u inflight=%u",
+                    side, svc, ep[side].sent[svc], ep[1-side].received[svc], c->tx_acked, c->tx_seq,
+                    c->send_q->count, c->inflight_q->count);
+    }
     REQUIRE(!failed); REQUIRE(complete); REQUIRE(backpressure);
     for (int side = 0; side < 2; side++) for (int svc = 0; svc < 2; svc++) {
         REQUIRE(ep[side].received[svc] == COUNT); REQUIRE(ep[side].notified[svc] == 0);
@@ -292,7 +299,7 @@ static int route_switch(void) {
     ((struct conn_queue_entry*)ep[0].route->data)->conn = &ep[0].alternate;
     uint64_t deadline = get_time_tb()+20000;
     while (!settled(0,0) && get_time_tb()<deadline) {
-        uasync_poll(ua,1); wire(0,ep[0].alternate.send_input_q); wire(1,ep[1].conn.send_input_q);
+        test_clock_poll(ua,1); wire(0,ep[0].alternate.send_input_q); wire(1,ep[1].conn.send_input_q);
     }
     REQUIRE(settled(0,0) && ep[1].received[0] == 2); REQUIRE(old->waiter_head == NULL);
     cleanup(); return 0;
@@ -309,17 +316,22 @@ static int no_route_restore(void) {
 }
 static int timeout_with_duplicate_ack(void) {
     REQUIRE(setup() == 0); drop_all_ack = 1; REQUIRE(send_one(0,0) == 0);
-    uint64_t deadline = get_time_tb()+70000, last_dup = 0;
+    uint64_t started = get_time_tb(), deadline = started+70000, last_dup = 0;
+    unsigned duplicates = 0;
     while (!ep[0].notified[0] && get_time_tb()<deadline) {
         if (saved_ack && get_time_tb()-last_dup >= 500) {
             struct ll_entry* e = copy_packet(saved_ack); ((struct SVC_ROUTE_HDR*)e->dgram)->seq = 0;
-            inject(0,e); last_dup = get_time_tb();
+            inject(0,e); last_dup = get_time_tb(); duplicates++;
         }
         step();
     }
     REQUIRE(ep[0].notified[0] == 1); REQUIRE(ep[1].received[0] == 1);
     REQUIRE(dropped_ack >= ROUTER_NO_ACK_MAX_RETRANS-1);
+    REQUIRE(duplicates > ROUTER_NO_ACK_MAX_RETRANS);
+    REQUIRE(get_time_tb() - started >= (ROUTER_NO_ACK_MAX_RETRANS - 1) * ROUTER_RETRANS_TIMEOUT_TB);
     REQUIRE(ep[0].inst.router_conns->count == 0);
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "router timeout: elapsed_tb=%llu duplicate_ACKs=%u dropped_ACKs=%d retrans_limit=%u",
+               (unsigned long long)(get_time_tb() - started), duplicates, dropped_ack, ROUTER_NO_ACK_MAX_RETRANS);
     cleanup(); return 0;
 }
 static int simultaneous_restart(void) {
@@ -337,10 +349,14 @@ static int call_ten_second_gap(void) {
     uint64_t epoch = rc(0,0)->reset_id;
     drop_wire = 1;
     REQUIRE(send_one(0,0) == 0);
-    uint64_t deadline = get_time_tb() + 100000;
+    uint64_t started = get_time_tb(), deadline = started + 100000;
     while (get_time_tb() < deadline && !failed) step();
     REQUIRE(!failed); REQUIRE(ep[0].notified[0] == 0);
     REQUIRE(rc(0,0)->reset_id == epoch && rc(0,0)->inflight_q->count == 1);
+    REQUIRE(rc(0,0)->c_retrans_done >= 100000 / ROUTER_RETRANS_TIMEOUT_TB);
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "router call outage: elapsed_tb=%llu retransmits=%u session=%llu inflight=%u",
+               (unsigned long long)(get_time_tb() - started), rc(0,0)->c_retrans_done,
+               (unsigned long long)epoch, rc(0,0)->inflight_q->count);
     drop_wire = 0;
     REQUIRE(wait_settled(0,0) == 0);
     REQUIRE(ep[1].received[0] == 2 && ep[1].notified[0] == 0);
@@ -364,6 +380,7 @@ static int size_bounds(void) {
 
 int main(void) {
     debug_config_init(); debug_set_categories(DEBUG_CATEGORY_ALL); debug_set_level(DEBUG_LEVEL_WARN);
+    test_clock_init(10); // Один шаг wire сохраняет прежние 1мс моделируемой задержки.
     const char* log = getenv("ROUTER_TEST_LOG");
     if (log) { debug_enable_file_output(log, 1); debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG); }
     struct { const char* name; int (*run)(void); } tests[] = {
@@ -379,5 +396,6 @@ int main(void) {
         if (tests[i].run() != 0 || failed) return 1;
         puts("PASS");
     }
+    test_clock_finish();
     puts("All router fault scenarios passed"); return 0;
 }