Test full chat restart and late-peer convergence across member exclusion and restoration

src/chat/member_sync.c

@@ -1336,7 +1336,7 @@ enum ms_author_status member_sync_author_key(struct UTUN_INSTANCE* inst, const c
     sqlite3_stmt* st = NULL; sqlite3* db = inst->topo_sqlite_db;
     enum ms_author_status status = MS_AUTHOR_ERROR;
     const char* reason = "member lookup";
-    uint8_t verified_key[32];
+    uint8_t verified_key[32]; int final;
     int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
     if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)author);
     if (rc == SQLITE_OK) rc = sqlite3_step(st);
@@ -1357,7 +1357,7 @@ enum ms_author_status member_sync_author_key(struct UTUN_INSTANCE* inst, const c
         || (tree_bytes && !member.signature) || (join_bytes && !member.join_sig)) {
         status = MS_AUTHOR_ERROR; reason = "member column allocation"; rc = SQLITE_NOMEM; goto done;
     }
-    if (!tree_bytes || !join_bytes || !member.join_ts
+    if (!tree_bytes || !join_bytes || sqlite3_column_int64(st, 5) <= 0
         || _sig_is_zero64(member.signature) || _sig_is_zero64(member.join_sig)) {
         status = MS_AUTHOR_PENDING; reason = "join or invitation pending"; goto done;
     }
@@ -1394,8 +1394,8 @@ enum ms_author_status member_sync_author_key(struct UTUN_INSTANCE* inst, const c
         goto done;
     }
     memcpy(verified_key, member.ed25519, 32); status = MS_AUTHOR_ACTIVE; reason = "verified";
-done:;
-    int final = sqlite3_finalize(st);
+done:
+    final = sqlite3_finalize(st);
     if (final != SQLITE_OK) { status = MS_AUTHOR_ERROR; reason = "member finalize"; rc = final; }
     if (status == MS_AUTHOR_ACTIVE) memcpy(key, verified_key, 32);
     if (status == MS_AUTHOR_ERROR) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC,

tests/test_chat_poll_network.c

@@ -21,7 +21,7 @@
 #include <stdlib.h>
 
 static struct UASYNC* ua;
-static struct UTUN_INSTANCE* nodes[3];
+static struct UTUN_INSTANCE* nodes[4];
 static char channel[21], message_hex[65];
 static uint8_t message_id[32];
 
@@ -41,8 +41,8 @@ static struct UTUN_INSTANCE* create_node(unsigned port) {
 }
 
 /* Все три заранее принятых участника имеют настоящие подписи root/join. */
-static void seed_members(uint8_t channel_x[32], uint8_t channel_ed[32], uint8_t owner_private[32]) {
-    for (int dst = 0; dst < 3; dst++) for (int author = 0; author < 3; author++) {
+static void seed_members(uint8_t channel_x[32], uint8_t channel_ed[32], uint8_t owner_private[32], int count) {
+    for (int dst = 0; dst < count; dst++) for (int author = 0; author < count; author++) {
         uint8_t body[112], join[64], tree[64];
         assert(member_sync_build_join_msg(channel_x, channel_ed, nodes[author]->node_id,
             nodes[author]->my_keys.public_key, 1, body, sizeof(body)) == 112);
@@ -121,11 +121,32 @@ static void subscribe(int a, int b) {
     db_sync_reinitiate(db_sync_instance_find(nodes[a], strtoull(channel, NULL, 10)), nodes[b]->node_id);
 }
 
+/* Владелец подписывает исключение/возврат; распространяется обычным member-sync. */
+static void owner_deleted(int dst, int author, unsigned ver, int deleted, const uint8_t owner_private[32]) {
+    char tags[96]; uint8_t body[128], signature[64];
+    snprintf(tags, sizeof(tags), "{\"ver\":\"%u\",\"deleted\":\"%s\"}", ver, deleted ? "yes" : "no");
+    int len = member_sync_build_owner_msg(nodes[author]->node_id, tags, body, sizeof(body)); assert(len > 0);
+    assert(sc_ed25519_sign(owner_private, body, len, signature) == SC_OK);
+    assert(member_sync_put(nodes[dst], channel, nodes[author]->node_id, nodes[author]->my_keys.public_key,
+        nodes[author]->my_ed25519_pubkey, NULL, 0, NULL, 0, NULL, tags, signature, 0, 0, NULL) & MS_APPLY_CHANGED);
+}
+
+static int wait_feedback_roots(int a, int b) {
+    uint64_t deadline = get_time_tb() + 15u * 10000u; uint8_t left[32], right[32];
+    while (get_time_tb() < deadline) {
+        assert(merkle_sync_read_hash(nodes[a], MT_DATA_FEEDBACK, channel, 0, 0, left) == 0);
+        assert(merkle_sync_read_hash(nodes[b], MT_DATA_FEEDBACK, channel, 0, 0, right) == 0);
+        if (!memcmp(left, right, 32)) return 1;
+        uasync_poll(ua, 5);
+    }
+    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "poll network: feedback roots differ nodes=%d/%d", a, b); return 0;
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
     debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
     debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO);
-    ua = uasync_create(); assert(ua); unsigned ports[3];
+    ua = uasync_create(); assert(ua); unsigned ports[4];
     for (int i = 0; i < 3; i++) {
         nodes[i] = create_node(0);
         ports[i] = ss_get_port(&test_etcp_bound_socket(nodes[i], "udp")->local_addr);
@@ -141,10 +162,10 @@ int main(void) {
         assert(topo_node_sqlite_channel_put(nodes[i]->topo_sqlite_db, channel, "Poll network", nodes[0]->node_id,
             x, NULL, ed, NULL, signature) == 0); chat_core_ensure_channel_ready(nodes[i], channel);
     }
-    seed_members(x, ed, owner_private);
+    seed_members(x, ed, owner_private, 3);
     struct NODE_CONN_DIRECT* ab = connect_node(0, 1, ports[1]), *bc = connect_node(1, 2, ports[2]);
     assert(wait_transport(ab, bc));
-    struct TOPO_PEER_REQUEST *groups[3] = {0}; uint64_t gid = strtoull(channel, NULL, 10);
+    struct TOPO_PEER_REQUEST *groups[5] = {0}; uint64_t gid = strtoull(channel, NULL, 10);
     assert(topo_group_peer_open(topo_groups_find(nodes[0]->topo_groups, gid), nodes[1]->node_id, &groups[0]) == 0);
     assert(topo_group_peer_open(topo_groups_find(nodes[1]->topo_groups, gid), nodes[2]->node_id, &groups[1]) == 0);
     assert(topo_group_peer_open(topo_groups_find(nodes[2]->topo_groups, gid), nodes[1]->node_id, &groups[2]) == 0);
@@ -169,9 +190,48 @@ int main(void) {
     assert(chat_feedback_get(nodes[2], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &old) == 1);
     assert(old.ver == 2 && old.data_len == 0);
     puts("PASS: returning peer receives new versions; withdrawn votes and reactions stay withdrawn");
-    for (int i = 0; i < 3; i++) topo_group_peer_close(groups[i]);
-    node_conn_direct_close(ab); node_conn_direct_close(bc);
-    for (int i = 0; i < 3; i++) utun_instance_destroy(nodes[i]);
+
+    run_command(0, "react", ",\"emoji\":0,\"on\":true");
+    assert(wait_result(1, 2, 1, 1) && wait_result(2, 2, 1, 1));
+    struct chat_feedback_record saved_reaction;
+    assert(chat_feedback_get(nodes[2], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &saved_reaction) == 1);
+    assert(saved_reaction.ver == 3 && saved_reaction.data_len == 2);
+    owner_deleted(1, 0, 1, 1, owner_private);
+    assert(wait_result(1, 1, 0, 0) && wait_result(2, 1, 0, 0) && wait_feedback_roots(1, 2));
+    topo_group_peer_close(groups[2]); groups[2] = NULL; chat_service_stop(nodes[2]);
+    assert(chat_service_start(nodes[2]) == 0);
+    assert(topo_group_peer_open(topo_groups_find(nodes[2]->topo_groups, gid), nodes[1]->node_id, &groups[2]) == 0);
+    assert(wait_ready(groups[2])); subscribe(1, 2);
+    assert(wait_result(2, 1, 0, 0));
+    assert(chat_feedback_get(nodes[2], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &old) == 1);
+    assert(old.ver == saved_reaction.ver && !memcmp(old.signature, saved_reaction.signature, 64));
+    assert(chat_feedback_get(nodes[2], channel, message_id, nodes[1]->node_id, CHAT_FEEDBACK_ANSWER, &old) == 1);
+    assert(old.ver == 2 && old.data_len == 0);
+    puts("PASS: signed exclusion removes votes/reactions; full chat-service restart preserves inactive history and cancellations");
+
+    /* Новый D впервые догоняет историю, когда A исключён и его feedback не выдаётся. */
+    nodes[3] = create_node(0); ports[3] = ss_get_port(&test_etcp_bound_socket(nodes[3], "udp")->local_addr);
+    assert(topo_node_sqlite_channel_put(nodes[3]->topo_sqlite_db, channel, "Poll network", nodes[0]->node_id,
+        x, NULL, ed, NULL, signature) == 0); chat_core_ensure_channel_ready(nodes[3], channel);
+    seed_members(x, ed, owner_private, 4);
+    struct NODE_CONN_DIRECT* bd = connect_node(1, 3, ports[3]);
+    assert(topo_group_peer_open(topo_groups_find(nodes[1]->topo_groups, gid), nodes[3]->node_id, &groups[3]) == 0);
+    assert(topo_group_peer_open(topo_groups_find(nodes[3]->topo_groups, gid), nodes[1]->node_id, &groups[4]) == 0);
+    assert(wait_ready(groups[3]) && wait_ready(groups[4])); subscribe(1, 3);
+    assert(wait_result(3, 1, 0, 0) && wait_feedback_roots(1, 3));
+    assert(chat_feedback_get(nodes[3], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &old) == 0);
+    assert(chat_feedback_get(nodes[3], channel, message_id, nodes[1]->node_id, CHAT_FEEDBACK_ANSWER, &old) == 1);
+    assert(old.ver == 2 && old.data_len == 0);
+    puts("PASS: new peer receives the active dataset and cancellation without receiving the excluded author's feedback");
+    owner_deleted(1, 0, 2, 0, owner_private);
+    assert(wait_result(2, 2, 1, 1) && wait_result(3, 2, 1, 1));
+    assert(wait_feedback_roots(1, 2) && wait_feedback_roots(1, 3));
+    assert(chat_feedback_get(nodes[3], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &old) == 1);
+    assert(old.ver == saved_reaction.ver && !memcmp(old.signature, saved_reaction.signature, 64));
+    puts("PASS: signed restoration reactivates cached states and delivers them to the new peer with matching Merkle roots");
+    for (int i = 0; i < 5; i++) topo_group_peer_close(groups[i]);
+    node_conn_direct_close(ab); node_conn_direct_close(bc); node_conn_direct_close(bd);
+    for (int i = 0; i < 4; i++) utun_instance_destroy(nodes[i]);
     uasync_poll(ua, 0); assert(ua->timer_alloc_count == ua->timer_free_count && ua->socket_alloc_count == ua->socket_free_count);
     uasync_destroy(ua, 0); puts("ALL PASS: network polls and offline recovery"); return 0;
 }