tests: accelerate DM retries and TTL with synchronized clocks

tests/Makefile.am

@@ -178,7 +178,7 @@ COMMON_LIBS = $(top_builddir)/lib/libuasync.a $(CRYPTO_LIBS) -lpthread $(WIN_LIB
 
 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
+test_clock_ldflags = -Wl,--wrap=clock_gettime -Wl,--wrap=gettimeofday -Wl,--wrap=epoll_wait -Wl,--wrap=poll
 else
 test_clock_ldflags =
 endif
@@ -559,17 +559,20 @@ test_chat_join_e2e_SOURCES = test_chat_join_e2e.c
 test_chat_join_e2e_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib
 test_chat_join_e2e_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 
-test_dm_e2e_SOURCES = test_dm_e2e.c
+test_dm_e2e_SOURCES = test_dm_e2e.c $(test_clock_sources)
 test_dm_e2e_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/dm -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib
 test_dm_e2e_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_dm_e2e_LDFLAGS = $(test_clock_ldflags)
 
-test_dm_invite_SOURCES = test_dm_invite.c
+test_dm_invite_SOURCES = test_dm_invite.c $(test_clock_sources)
 test_dm_invite_CFLAGS = $(test_dm_e2e_CFLAGS)
 test_dm_invite_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_dm_invite_LDFLAGS = $(test_clock_ldflags)
 
-test_dm_media_SOURCES = test_dm_media.c
+test_dm_media_SOURCES = test_dm_media.c $(test_clock_sources)
 test_dm_media_CFLAGS = $(test_dm_e2e_CFLAGS)
 test_dm_media_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_dm_media_LDFLAGS = $(test_clock_ldflags)
 
 test_dm_worker_SOURCES = test_dm_worker.c
 test_dm_worker_CFLAGS = $(test_dm_e2e_CFLAGS)

tests/test_clock.c

@@ -7,7 +7,7 @@
 #include <sys/epoll.h>
 #include <poll.h>
 
-static int accelerated;
+static int accelerated, sync_wall_time;
 static uint64_t offset_us, limit_us, advances, io_events;
 static struct UASYNC* polling;
 static uint64_t step_us;
@@ -15,13 +15,23 @@ 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)) {
+    if (!result && (id == CLOCK_BOOTTIME || id == CLOCK_MONOTONIC || (sync_wall_time && id == CLOCK_REALTIME))) {
         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;
 }
 
+int __real_gettimeofday(struct timeval* tv, void* tz);
+int __wrap_gettimeofday(struct timeval* tv, void* tz) {
+    int result = __real_gettimeofday(tv, tz);
+    if (!result && tv && sync_wall_time) {
+        uint64_t us = (uint64_t)tv->tv_usec + __atomic_load_n(&offset_us, __ATOMIC_RELAXED);
+        tv->tv_sec += us / 1000000; tv->tv_usec = us % 1000000;
+    }
+    return result;
+}
+
 /* Продвигаем время внутри wait: uasync видит нормальное ожидание, а не искусственный stall между poll. */
 static void advance_idle(int result, int timeout_ms) {
     if (!polling) return;
@@ -78,6 +88,18 @@ void test_clock_poll(struct UASYNC* ua, int timeout_tb) {
     uasync_poll(ua, timeout_tb);
 }
 
+void test_clock_sync_wall_time(void) {
+#ifdef __linux__
+    sync_wall_time = 1;
+#endif
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "test clock: calendar and monotonic clocks synchronized for TTL/retry_at");
+}
+
+void test_clock_mainloop(struct UASYNC* ua) {
+    if (!ua) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "test clock: null mainloop instance"); abort(); }
+    while (!__atomic_load_n(&ua->stop, __ATOMIC_ACQUIRE)) test_clock_poll(ua, 100);
+}
+
 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;

tests/test_clock.h

@@ -7,7 +7,10 @@
  * UTUN_TEST_REAL_TIME=1 и остальные платформы используют обычные часы/ожидание.
  * init — до создания uasync; finish — после уничтожения. Вызовы только из потока теста, не из callbacks. */
 void test_clock_init(int max_step_tb);
+// Та же прибавка для календарных часов (TTL и durable retry_at); до создания инстансов.
+void test_clock_sync_wall_time(void);
 void test_clock_poll(struct UASYNC* ua, int timeout_tb);
+void test_clock_mainloop(struct UASYNC* ua);
 void test_clock_run_for(struct UASYNC* ua, int duration_tb);
 void test_clock_finish(void);
 

tests/test_dm_e2e.c

@@ -29,6 +29,7 @@
 #include "../lib/mem.h"
 #include "../lib/platform_compat.h"
 #include "test_utils.h"
+#include "test_clock.h"
 
 #include <sqlite3.h>
 #define OPENSSL_API_COMPAT 0x10100000L
@@ -656,6 +657,7 @@ static void dm2_tick(void* arg) {
 int main(int argc, char** argv) {
     (void)argc; (void)argv;
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
+    test_clock_init(1000); test_clock_sync_wall_time();
     if (getenv("UTUN_TEST_DEBUG")) {
         debug_set_category_level_by_name("dm", "trace");
         debug_set_category_level_by_name("chat_sync", "info");
@@ -706,7 +708,7 @@ int main(int argc, char** argv) {
         utun_instance_init(t.inst[IDX_B]);
 
         t.timer = uasync_set_timeout(ua, TICK_TB, &t, dm_tick, "dm_tick");
-        uasync_mainloop(ua);
+        test_clock_mainloop(ua);
 
         int ok = (t.result == 1);
 
@@ -756,7 +758,7 @@ int main(int argc, char** argv) {
         utun_instance_init(t.inst[IDX_B]);
 
         t.timer = uasync_set_timeout(ua, TICK_TB, &t, dm2_tick, "dm2_tick");
-        uasync_mainloop(ua);
+        test_clock_mainloop(ua);
 
         int ok = (t.result == 1);
 
@@ -771,5 +773,6 @@ int main(int argc, char** argv) {
     snprintf(cmd, sizeof(cmd), "rm -rf %s", dir2); (void)!system(cmd);
 
     printf("\n%d/%d passed, %d failed\n", G_PASSED, G_TOTAL, G_FAILED);
+    test_clock_finish();
     return G_FAILED > 0 ? 1 : 0;
 }

tests/test_dm_invite.c

@@ -20,6 +20,7 @@
 #include "../src/transport_layer/secure_channel.h"
 #include "../src/transport_layer/etcp.h"
 #include "test_utils.h"
+#include "test_clock.h"
 
 static struct UTUN_INSTANCE* nodes[2];
 static int results[16], calls[16], joined;
@@ -97,7 +98,7 @@ static void request(uint64_t id, uint64_t group, const char* source, int cancel)
 
 static int wait_result(struct UASYNC* ua, int id, int expected) {
     uint64_t deadline = get_time_tb() + 30000;
-    while (!calls[id] && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (!calls[id] && get_time_tb() < deadline) test_clock_poll(ua, 100);
     return calls[id] == 1 && results[id] == expected;
 }
 
@@ -110,7 +111,7 @@ static int join_link(struct UASYNC* ua, const char* link) {
     chat_sync_connect_from_invite(nodes[1], invite.channelId, invite.nodeId, invite.pubkey,
                                   addresses, invite.addrCount, bytes, NULL, invite.join_key);
     uint64_t deadline = get_time_tb() + 200000;
-    while (!joined && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (!joined && get_time_tb() < deadline) test_clock_poll(ua, 100);
     if (joined != 1) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite test: JOIN failed ch=%llu result=%d",
         (unsigned long long)joining_channel, joined); return -1; }
     return 0;
@@ -146,6 +147,7 @@ static int start_node(struct UASYNC* ua, const char* directory, int index, int p
 
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
+    test_clock_init(1000); test_clock_sync_wall_time();
     if (getenv("UTUN_TEST_DEBUG")) {
         debug_set_category_level(DEBUG_CATEGORY_DM, DEBUG_LEVEL_DEBUG);
         debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
@@ -167,7 +169,7 @@ int main(void) {
     struct link_result first = {0};
     CHECK(chat_invite_build_link(nodes[0], source, 0, NULL, link_ready, &first), "build initial invite");
     uint64_t deadline = get_time_tb() + 30000;
-    while (!first.ready && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (!first.ready && get_time_tb() < deadline) test_clock_poll(ua, 100);
     CHECK(first.ready == 1 && !join_link(ua, first.link), "join common source group");
     chat_core_create_channel_auto(nodes[0], "target \"группа\"\n<>&");
     CHECK(!scalar(nodes[0], "SELECT channel_id FROM channels WHERE name='target \"группа\"\n<>&'", &target),
@@ -212,7 +214,7 @@ int main(void) {
     deadline = get_time_tb() + 100000;
     char* link = NULL;
     do {
-        uasync_poll(ua, 100);
+        test_clock_poll(ua, 100);
         if (!dm_list_messages_json(nodes[1], conv, 10, 0, messages, sizeof(messages), &length)) link = strstr(messages, "utun://");
     } while (!link && get_time_tb() < deadline);
     CHECK(link, "invite delivered to B over source group's UDP route");
@@ -240,14 +242,14 @@ int main(void) {
     chat_core_connect_channel(nodes[1], target_id);
     deadline = get_time_tb() + 100000;
     while (!topo_node_sqlite_member_in_channel(nodes[1]->topo_sqlite_db, target_id, nodes[1]->node_id) &&
-           get_time_tb() < deadline) uasync_poll(ua, 100);
+           get_time_tb() < deadline) test_clock_poll(ua, 100);
     DEBUG_INFO(DEBUG_CATEGORY_DM, "invite test: membership after JOIN ch=%s A_has_B=%d B_has_B=%d", target_id,
                topo_node_sqlite_member_in_channel(nodes[0]->topo_sqlite_db, target_id, nodes[1]->node_id),
                topo_node_sqlite_member_in_channel(nodes[1]->topo_sqlite_db, target_id, nodes[1]->node_id));
     CHECK(topo_node_sqlite_member_in_channel(nodes[0]->topo_sqlite_db, target_id, nodes[1]->node_id) &&
           topo_node_sqlite_member_in_channel(nodes[1]->topo_sqlite_db, target_id, nodes[1]->node_id), "membership committed on both nodes");
     deadline = get_time_tb() + 400000;
-    while (chat_core_count(nodes[1], target_id) != 5 && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (chat_core_count(nodes[1], target_id) != 5 && get_time_tb() < deadline) test_clock_poll(ua, 100);
     CHECK(chat_core_count(nodes[1], target_id) == 5, "history created before invitation arrives after JOIN");
     CHECK(!getenv("UTUN_TEST_HISTORY_TIMEOUT") || dropped_requests == 2, "requester retries after lost initial exchange");
     CHECK(!chat_core_get_messages_json(nodes[1], target_id, 10, 0, messages, sizeof(messages), &length), "read received history");
@@ -276,7 +278,7 @@ int main(void) {
     chat_core_submit_message(nodes[0], &offline);
     CHECK(!topo_group_new_conn(target_group, shared_conn), "restart target session on shared transport");
     deadline = get_time_tb() + 100000;
-    while (chat_core_count(nodes[1], target_id) != 6 && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (chat_core_count(nodes[1], target_id) != 6 && get_time_tb() < deadline) test_clock_poll(ua, 100);
     CHECK(ready_events == 1 && chat_core_count(nodes[1], target_id) == 6 &&
           !chat_core_get_messages_json(nodes[1], target_id, 10, 0, messages, sizeof(messages), &length) && strstr(messages, missed),
           "new READY session resumes history and receives the missed message");
@@ -288,7 +290,7 @@ int main(void) {
     message.data = (uint8_t*)text; message.data_len = (uint32_t)strlen(text);
     chat_core_submit_message(nodes[1], &message);
     deadline = get_time_tb() + 100000;
-    while (chat_core_count(nodes[0], target_id) != 7 && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (chat_core_count(nodes[0], target_id) != 7 && get_time_tb() < deadline) test_clock_poll(ua, 100);
     CHECK(!chat_core_get_messages_json(nodes[0], target_id, 10, 0, messages, sizeof(messages), &length) && strstr(messages, text),
           "joined client can send a group message to the inviter");
     request(6, target, "", 0);
@@ -298,6 +300,7 @@ int main(void) {
 done:
     for (int i = 0; i < 2; i++) if (nodes[i]) utun_instance_destroy(nodes[i]);
     if (ua) uasync_destroy(ua, 0);
+    test_clock_finish();
     printf("PM invite: cancellation, failure, delivery and JOIN: %s\n", failed ? "FAIL" : "OK");
     return failed;
 }

tests/test_dm_media.c

@@ -26,6 +26,7 @@
 #include "../src/transport_layer/etcp.h"
 #include "../src/media_delivery/file_transfer.h"
 #include "test_utils.h"
+#include "test_clock.h"
 
 #define GROUP 4242424242424243ULL
 #define CHANNEL "4242424242424243"
@@ -325,6 +326,7 @@ static void tick(void* arg) {
 int main(void) {
     struct media_test t = {0};
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
+    test_clock_init(1000); test_clock_sync_wall_time();
     if (getenv("UTUN_TEST_DEBUG")) {
         debug_set_category_level_by_name("dm", "debug"); debug_set_category_level_by_name("media", "debug");
         debug_set_category_level_by_name("bgp", "info"); debug_set_category_level_by_name("etcp_route", "info");
@@ -361,9 +363,10 @@ int main(void) {
     int base = 48000 + getpid() % 5000;
     if (!t.ua || start_node(&t, S, base) || start_node(&t, A, base) || start_node(&t, M, base) || start_node(&t, B, base)) return 1;
     t.timer = uasync_set_timeout(t.ua, 100, &t, tick, "dm_media_test");
-    uasync_mainloop(t.ua);
+    test_clock_mainloop(t.ua);
     for (int i = 0; i < NODES; i++) if (t.nodes[i]) utun_instance_destroy(t.nodes[i]);
     uasync_destroy(t.ua, 0);
+    test_clock_finish();
     printf("test_dm_media: phase=%u heartbeats=%u %s artifacts=%s\n", t.phase, t.heartbeats,
            !t.failed && t.phase == 9 ? "PASS" : "FAIL", t.dir);
     return t.failed || t.phase != 9;