Exercise history retry after a lost initial exchange

tests/test_dm_invite.c

@@ -16,12 +16,31 @@
 #include "../src/dm/dm_core.h"
 #include "../src/dm/dm_crypto.h"
 #include "../src/routing_layer/topo_node_sqlite.h"
+#include "../src/routing_layer/topo_group.h"
 #include "../src/transport_layer/secure_channel.h"
 #include "test_utils.h"
 
 static struct UTUN_INSTANCE* nodes[2];
 static int results[16], calls[16], joined;
 static uint64_t joining_channel;
+static uint64_t dropped_group;
+static int dropped_requests;
+static etcp_recv_fn history_receivers[2];
+
+/* Потеря прикладных пакетов после надёжной ETCP-доставки: первый обмен обязан истечь и повториться. */
+static void drop_history(struct ETCP_CONN* conn, struct ll_entry* entry) {
+    uint64_t gid = 0;
+    if (entry->len >= 10) { memcpy(&gid, entry->dgram + 1, 8); gid = be64toh(gid); }
+    int index = conn->instance == nodes[0] ? 0 : 1;
+    if (gid != dropped_group) { history_receivers[index](conn, entry); return; }
+    if (entry->dgram[9] == DB_MSG_REQUEST_SYNC) {
+        dropped_requests++;
+        DEBUG_INFO(DEBUG_CATEGORY_DM, "invite test: dropped history request=%d group=%016llx", dropped_requests, gid);
+        if (dropped_requests == 2)
+            for (int i = 0; i < 2; i++) etcp_bind(nodes[i], ETCP_RT_ID_DB_SYNC, history_receivers[i]);
+    }
+    queue_dgram_free(entry); queue_entry_free(entry);
+}
 
 static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) {
     if (type == CHAT_EVT_DM_INVITE_RESULT && len == 28) {
@@ -195,6 +214,13 @@ int main(void) {
           "PM invites target group rather than source group");
     CHECK(JOIN_KEY_TTL_SECONDS == 600 && chat_join_lookup_inviter(nodes[0], target, received.join_key) == nodes[0]->node_id,
           "delivered link is registered with ten minute lifetime");
+    if (getenv("UTUN_TEST_HISTORY_TIMEOUT")) {
+        dropped_group = target;
+        for (int i = 0; i < 2; i++) {
+            history_receivers[i] = nodes[i]->api_bindings.callbacks[ETCP_RT_ID_DB_SYNC];
+            etcp_bind(nodes[i], ETCP_RT_ID_DB_SYNC, drop_history);
+        }
+    }
     CHECK(!join_link(ua, link), "B joins target group using received PM link");
     chat_core_connect_channel(nodes[1], target_id);
     deadline = get_time_tb() + 100000;
@@ -205,9 +231,10 @@ int main(void) {
                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() + 300000;
+    deadline = get_time_tb() + 400000;
     while (chat_core_count(nodes[1], target_id) != 5 && get_time_tb() < deadline) uasync_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");
     for (int i = 0; i < 5; i++) {
         char text[64]; snprintf(text, sizeof(text), "history before invitation %d", i);