Cover publication after callback cancellation and real request authorization

tests/test_member_purge.c

@@ -3,6 +3,7 @@
 #include "chat/chat_core_priv.h"
 #include "utun_instance.h"
 #include "routing_layer/topo_node_sqlite.h"
+#include "routing_layer/topo_group.h"
 #include "transport_layer/etcp.h"
 #include "../lib/mem.h"
 #include "../lib/debug_config.h"
@@ -270,6 +271,12 @@ static void purge_done(uint64_t peer, const char* channel, int result, void* arg
 
 static void test_purge_start(struct UASYNC* ua) {
     struct purge_fixture f; purge_setup(&f, ua, 0); purge_bad(&f, 123);
+    struct TOPO_GROUPS groups = {.instance = &f.inst};
+    groups.group_list = queue_new(ua, 16, 0, 8, "purge groups"); assert(groups.group_list);
+    struct TOPO_GROUP* group = (struct TOPO_GROUP*)queue_entry_new(sizeof(*group) - sizeof(struct ll_entry)); assert(group);
+    group->group_id = 1001; group->group_type = TOPO_GROUP_TYPE_CHAT; group->instance = &f.inst;
+    strcpy(group->channel_id, "1001");
+    assert(queue_data_put_with_index(groups.group_list, &group->ll) == 0); f.inst.topo_groups = &groups;
     uint64_t peer = f.member.node_id;
     f.inst.connections = queue_new(ua, 16, 0, 8, "purge test connections"); assert(f.inst.connections);
     f.conn.instance = &f.inst; f.conn.peer_node_id = peer; f.conn.initialized = 1; f.conn.links_up = 1;
@@ -290,6 +297,7 @@ static void test_purge_start(struct UASYNC* ua) {
     e = queue_data_get(f.inst.connections); assert(e); queue_entry_free(e);
     queue_free(f.inst.connections); f.inst.connections = NULL;
     queue_free(f.conn.send_input_q);
+    f.inst.topo_groups = NULL; queue_entry_free(queue_data_get(groups.group_list)); queue_free(groups.group_list);
     purge_destroy(&f);
     DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS purge: failed start creates no request/callback; retry attaches real session; cancel releases ownership");
 }

tests/test_merkle_protocol.c

@@ -343,6 +343,12 @@ static int t_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round
 }
 static int t_commit_pull(void* ctx, void* pull) { assert(pull); pull_committed++; return ((struct fixture*)ctx)->apply_error; }
 static void t_abort_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_released++; u_free(pull); }
+static int pull_published;
+static void t_publish_pull(void* ctx, void* pull) {
+    struct fixture* f = ctx;
+    assert(pull && f->callbacks == 1 && !f->inst.msync->sessions);
+    merkle_sync_cancel_ns(&f->inst, "1001"); pull_published++;
+}
 static int t_stage_page(void* ctx, void* pull, uint64_t prefix, int av, const merkle_key_t* after,
         const merkle_key_t* next, int more, const uint8_t* data, size_t len) {
     (void)ctx; (void)pull; (void)prefix; (void)av; (void)after; (void)next; (void)more; (void)data; (void)len; return 0;
@@ -493,10 +499,11 @@ static void test_receive_only(void) {
     struct merkle_sync_data_ops ops = t_ops;
     ops.validate_peer = t_authorized; ops.validate_request = t_can_run;
     ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page;
+    ops.publish_pull = t_publish_pull;
     for (int rejected = 0; rejected < 2; rejected++) {
         struct fixture f; t_init(&f); f.inst.node_id = 3; f.running = 1;
         f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
-        int committed = pull_committed, released = pull_released;
+        int committed = pull_committed, released = pull_released, published = pull_published;
         uint8_t id[MS_TOKEN_SIZE] = {8}, accepted[MS_TOKEN_SIZE] = {9};
         t_receive_tokens(&f, MS_BEGIN | MS_READ_ONLY, 8, 0, id, NULL, NULL, 0);
         assert(!f.inst.msync->sessions); /* Право запросить не разрешает выдавать данные. */
@@ -519,6 +526,7 @@ static void test_receive_only(void) {
         t_receive_tokens(&f, MS_DONE | MS_READ_ONLY | MS_RESPONSE, s->round, 0, s->round_id, s->accept_id, NULL, 0);
         if (rejected) uasync_poll(f.inst.ua, 0);
         assert(pull_committed == committed + 1 && pull_released == released + 1);
+        assert(pull_published == published + !rejected);
         assert(f.callbacks == 1 && f.result == (rejected ? MT_ERR_DATA : MT_OK) && !f.inst.msync->sessions);
         t_destroy(&f);
     }
@@ -560,6 +568,9 @@ static void test_receive_only_source_change(void) {
 static void test_receive_only_boundaries(void) {
     struct fixture f; t_init(&f); f.running = 1;
     struct merkle_sync_data_ops ops = t_ops;
+    ops.publish_pull = t_publish_pull;
+    assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &ops, &f) < 0);
+    ops.publish_pull = NULL;
     ops.validate_peer = t_authorized; ops.validate_request = t_can_run;
     assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &ops, &f) < 0); /* Нужна изолированная загрузка. */
     ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page;