Verify history recovery when a group session loses readiness

tests/test_dm_invite.c

@@ -27,6 +27,12 @@ static uint64_t joining_channel;
 static uint64_t dropped_group;
 static int dropped_requests;
 static etcp_recv_fn history_receivers[2];
+static int ready_events, unavailable_events;
+
+static void peer_ready(struct TOPO_GROUP* group, uint64_t peer, int ready, void* arg) {
+    (void)group; (void)peer; (void)arg;
+    if (ready) ready_events++; else unavailable_events++;
+}
 
 /* Потеря прикладных пакетов после надёжной ETCP-доставки: первый обмен обязан истечь и повториться. */
 static void drop_history(struct ETCP_CONN* conn, struct ll_entry* entry) {
@@ -247,6 +253,26 @@ int main(void) {
     CHECK(history_a && history_b && !db_sync_last_chain_hash8(history_a, &hash_a) &&
           !db_sync_last_chain_hash8(history_b, &hash_b) && hash_a == hash_b &&
           !db_sync_chain_verify(history_a) && !db_sync_chain_verify(history_b), "received history signatures and chains agree");
+    struct TOPO_GROUP* target_group = topo_groups_find(nodes[1]->topo_groups, target);
+    struct ETCP_CONN* shared_conn = instance_find_conn(nodes[1], nodes[0]->node_id);
+    CHECK(target_group && shared_conn && topo_group_peer_ready(target_group, nodes[0]->node_id), "target session is READY");
+    CHECK(!topo_group_add_peer_ready_cbk(target_group, peer_ready, NULL), "observe target session transitions");
+    topo_group_peer_leave(target_group, nodes[0]->node_id);
+    CHECK(unavailable_events == 1 && !topo_group_peer_ready(target_group, nodes[0]->node_id) && shared_conn->links_up,
+          "leaving target publishes readiness loss while source group's transport stays UP");
+    const char* missed = "message while target session is unavailable";
+    struct chat_msg_submit offline = {0};
+    snprintf(offline.channel_id, sizeof(offline.channel_id), "%s", target_id);
+    snprintf(offline.content_type, sizeof(offline.content_type), "text");
+    offline.data = (uint8_t*)missed; offline.data_len = (uint32_t)strlen(missed);
+    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);
+    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");
+    topo_group_remove_peer_ready_cbk(target_group, peer_ready, NULL);
     const char* text = "joined through PM invitation";
     struct chat_msg_submit message = {0};
     snprintf(message.channel_id, sizeof(message.channel_id), "%s", target_id);
@@ -254,7 +280,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) != 6 && get_time_tb() < deadline) uasync_poll(ua, 100);
+    while (chat_core_count(nodes[0], target_id) != 7 && get_time_tb() < deadline) uasync_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);