Check permission round teardown, blocked error handoff and safe restart callbacks

src/chat/merkle_sync.c

@@ -377,7 +377,7 @@ void merkle_sync_revalidate_peers(struct UTUN_INSTANCE* inst, uint8_t dataset, c
                 && (!selected || s->peer < selected->peer)) selected = s;
         if (!selected) return;
         after = selected->peer; after_valid = 1;
-        if (selected->state != MS_FAILED && selected->state != MS_FAILING
+        if (selected->state != MS_FAILED
             && selected->dataset->ops->validate_peer(selected->dataset->data_ctx, owned_ns, after) != 1)
             ms_fail(selected, MT_ERR_DATA, "model permission revoked", 0);
         /* Callback мог удалить selected и соседние сессии: следующий поиск начинается заново. */

tests/test_merkle_protocol.c

@@ -6,7 +6,7 @@
 struct fixture {
     struct UTUN_INSTANCE inst;
     struct ETCP_CONN conn;
-    int callbacks, result, apply_error;
+    int callbacks, result, apply_error, authorized;
 };
 
 static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
@@ -296,6 +296,45 @@ static void test_dataset_isolation(void) {
     puts("PASS: dataset roots, sessions, changes and cancellation are independent");
 }
 
+/* Два разных полных ключа с одинаковыми первыми 64 битами проходят пагинацию. */
+static int t_authorized(void* ctx, const char* ns, uint64_t peer) {
+    (void)ns; (void)peer; return ((struct fixture*)ctx)->authorized;
+}
+
+/* Отзыв закрывает даже заблокированный ERROR; возврат начинает новый раунд той же подписки. */
+static void test_permission_rounds(void) {
+    for (int failing = 0; failing < 2; failing++) {
+        struct fixture f; t_init(&f);
+        struct merkle_sync_data_ops ops = t_ops; ops.validate_peer = t_authorized;
+        f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
+        struct ms_session* s = t_session(&f);
+        if (failing) ms_fail(s, MT_ERR_DATA, "injected queued error", 1);
+        else assert(ms_send(s, MS_QUERY, 1, NULL, 0) == 0);
+        assert(s->tx && s->deadline && !f.callbacks);
+        merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
+        merkle_sync_revalidate_peers(&f.inst, MT_DATA_MEMBERS, "1001");
+        assert(s->state == MS_FAILED && !s->tx && !s->deadline && !s->work && !s->waiting);
+        assert(f.callbacks == 1 && f.result == MT_ERR_DATA && f.conn.ref_count == 1);
+        assert(!f.conn.send_input_q->waiter_head);
+        f.authorized = 1;
+        merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001");
+        merkle_sync_revalidate_peers(&f.inst, MT_DATA_MEMBERS, "1001");
+        assert(s->state == MS_IDLE && s->dirty && s->work && f.callbacks == 1);
+        uasync_call_soon_cancel(f.inst.ua, s->work); s->work = NULL;
+        ms_work(s);
+        assert(s->state == MS_BEGIN_WAIT && s->round > 7 && s->tx && s->deadline);
+        t_destroy(&f);
+    }
+    struct fixture f; t_init(&f);
+    struct merkle_sync_data_ops ops = t_ops; ops.validate_peer = t_authorized;
+    f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops;
+    struct ms_session* s = t_session(&f); s->requests->cb = t_cancel_done;
+    merkle_sync_revalidate_peers(&f.inst, MT_DATA_MEMBERS, "1001");
+    assert(!f.inst.msync->sessions && !f.conn.ref_count && f.callbacks == 1 && f.result == MT_ERR_DATA);
+    t_destroy(&f);
+    puts("PASS: permission revocation releases blocked rounds; restoration restarts watches; callback cancellation is safe");
+}
+
 /* Два разных полных ключа с одинаковыми первыми 64 битами проходят пагинацию. */
 static int t_full_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after,
         uint8_t* data, size_t* len, merkle_key_t* next, int* more) {
@@ -340,7 +379,7 @@ int main(void) {
     test_page_size_and_cursor(); test_readonly_tree(); test_backpressure_and_disconnect(); test_peer_restart();
     test_send_rejection();
     test_error_handoff(); test_blocked_error_timeout(); test_pull_lifecycle();
-    test_dataset_isolation(); test_full_cursor();
+    test_dataset_isolation(); test_full_cursor(); test_permission_rounds();
     assert(u_get_allocated_count() == baseline);
     puts("ALL PASS: protocol regressions, no tracked allocations leaked");
     return 0;