/* Настоящие multicast/router, BGP-маршруты и uasync; ETCP-пакеты передаются в памяти.
 * Управляемые очереди позволяют проверять точную границу, независимые ветки и смену пути. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "etcp.h"
#include "etcp_router.h"
#include "multicast.h"
#include "pkt_normalizer.h"
#include "topo_group.h"
#include "utun_instance.h"
#include "test_clock.h"

#define CHECK(x) do { if (!(x)) { \
    DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); abort(); \
} } while (0)
#define A 0
#define B 1
#define C 2
#define D 3
#define GROUPS 2

struct wire {
    struct ETCP_CONN conn;
    struct wire* reverse;
    struct ll_entry* blocker;
    unsigned packets, max_recipients;
};

struct fixture {
    struct UASYNC* ua;
    struct UTUN_INSTANCE* inst[4];
    struct TOPO_GROUP* group[4][GROUPS];
    struct wire wire[4][4];
};

struct received {
    unsigned count;
    size_t len;
    uint8_t prefix[16];
    int unsubscribe;
};

struct losses {
    unsigned count;
    struct multicast_loss item[256];
    uint64_t ids[256][8];
    int close;
};

static void release(struct ll_entry* e) {
    if (e) { queue_dgram_free(e); queue_entry_free(e); }
}

static void clear_queue(struct ll_queue* q) {
    if (!q) return;
    struct ll_entry* e;
    while ((e = queue_data_get(q))) release(e);
    queue_free(q);
}

static void receive(struct multicast_subscription* sub, const uint8_t* payload, size_t len, void* arg) {
    struct received* r = arg;
    r->count++; r->len = len;
    memset(r->prefix, 0, sizeof(r->prefix));
    if (len) memcpy(r->prefix, payload, len < 16 ? len : 16);
    if (r->unsubscribe) multicast_unsubscribe(sub);
}

static void loss(struct UTUN_INSTANCE* inst, const struct multicast_loss* event, void* arg) {
    struct losses* l = arg;
    CHECK(l->count < 256);
    unsigned i = l->count++;
    l->item[i] = *event;
    unsigned n = event->recipient_count < 8 ? event->recipient_count : 8;
    memcpy(l->ids[i], event->recipients, n * sizeof(uint64_t));
    l->item[i].recipients = l->ids[i];
    if (l->close) multicast_close(inst, event->descriptor);
}

static void wire_send(struct ll_queue* q, void* arg) {
    struct wire* w = arg;
    struct ll_entry* e = queue_data_get(q);
    if (e) {
        CHECK(e != w->blocker && e->len > 0);
        if (e->dgram[0] == ETCP_ID_MULTICAST) {
            w->packets++;
            unsigned n = e->dgram[11] | (unsigned)e->dgram[12] << 8;
            if (n > w->max_recipients) w->max_recipients = n;
        }
        struct ETCP_CONN* peer = &w->reverse->conn;
        etcp_recv_fn cb = peer->instance->api_bindings.callbacks[e->dgram[0]];
        CHECK(cb);
        cb(peer, e);
    }
    queue_resume_callback(q);
}

static void block(struct wire* w) {
    CHECK(!w->blocker && w->conn.send_input_q->count == 0);
    queue_set_callback(w->conn.send_input_q, NULL, NULL);
    w->blocker = ll_alloc_lldgram(1); CHECK(w->blocker);
    w->blocker->len = 1; w->blocker->dgram[0] = 0xff;
    CHECK(queue_data_put(w->conn.send_input_q, w->blocker) == 0);
}

static void unblock(struct wire* w) {
    CHECK(w->blocker);
    CHECK(queue_data_get(w->conn.send_input_q) == w->blocker);
    release(w->blocker); w->blocker = NULL;
    queue_set_callback(w->conn.send_input_q, wire_send, w);
    queue_resume_callback(w->conn.send_input_q);
}

static void route(struct fixture* f, int from, int group, uint64_t dst, int via) {
    struct TOPO_GROUP* g = f->group[from][group];
    struct TOPO_GROUP_NODE* node = topo_node_find_by_id(g, dst);
    if (!node) {
        node = (struct TOPO_GROUP_NODE*)queue_entry_new(sizeof(*node) - sizeof(struct ll_entry)); CHECK(node);
        node->node_id = dst;
        CHECK(queue_data_put_with_index(g->nodes, &node->ll) == 0);
    }
    topo_nodeq_clear_paths(node);
    if (via >= 0) {
        uint64_t hop = via + 1;
        CHECK(f->wire[from][via].conn.send_input_q);
        CHECK(topo_group_add_path(node, &f->wire[from][via].conn, &hop, 1, 0) == 0);
    }
}

static struct fixture* setup(void) {
    struct fixture* f = u_calloc(1, sizeof(*f)); CHECK(f);
    f->ua = uasync_create(); CHECK(f->ua);
    for (int i = 0; i < 4; i++) {
        struct UTUN_INSTANCE* inst = f->inst[i] = u_calloc(1, sizeof(*inst)); CHECK(inst);
        inst->node_id = i + 1; inst->ua = f->ua;
        inst->connections = queue_new(f->ua, 16, 0, 8, "mc_test_connections"); CHECK(inst->connections);
        inst->topo_groups = u_calloc(1, sizeof(*inst->topo_groups)); CHECK(inst->topo_groups);
        inst->topo_groups->instance = inst;
        inst->topo_groups->group_list = queue_new(f->ua, 16, 0, 8, "mc_test_groups"); CHECK(inst->topo_groups->group_list);
        for (int j = 0; j < GROUPS; j++) {
            struct TOPO_GROUP* g = (struct TOPO_GROUP*)queue_entry_new(sizeof(*g) - sizeof(struct ll_entry)); CHECK(g);
            g->instance = inst; g->group_id = j ? 0x55 : TOPO_GROUP_UTUN;
            g->nodes = queue_new(f->ua, 256, 0, 8, "mc_test_nodes"); CHECK(g->nodes);
            g->senders_list = queue_new(f->ua, 0, 0, 0, "mc_test_peers"); CHECK(g->senders_list);
            CHECK(queue_data_put_with_index(inst->topo_groups->group_list, &g->ll) == 0);
            f->group[i][j] = g;
        }
        CHECK(etcp_router_init(inst) == 0);
        CHECK(multicast_init(inst) == 0);
    }
    const int edges[][2] = {{A, B}, {B, C}, {B, D}, {A, D}, {D, C}};
    for (unsigned i = 0; i < sizeof(edges) / sizeof(edges[0]); i++) {
        for (int dir = 0; dir < 2; dir++) {
            int a = edges[i][dir], b = edges[i][1 - dir];
            struct wire* w = &f->wire[a][b];
            w->reverse = &f->wire[b][a]; w->conn.instance = f->inst[a];
            w->conn.peer_node_id = b + 1; w->conn.state = 1; w->conn.links_up = 1;
            w->conn.send_input_q = queue_new(f->ua, 0, 0, 0, "mc_test_wire"); CHECK(w->conn.send_input_q);
            queue_set_callback_defer(w->conn.send_input_q, 1);
            queue_set_callback(w->conn.send_input_q, wire_send, w);
            struct ll_entry* e = queue_entry_new(sizeof(struct conn_queue_entry)); CHECK(e);
            struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
            ce->peer_node_id = b + 1; ce->conn = &w->conn;
            CHECK(queue_data_put_with_index(f->inst[a]->connections, e) == 0);
            for (int g = 0; g < GROUPS; g++) {
                e = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM)); CHECK(e);
                struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)e->data;
                peer->group = f->group[a][g]; peer->node_id = b + 1; peer->conn = &w->conn; peer->accepted = 1;
                CHECK(queue_data_put(f->group[a][g]->senders_list, e) == 0);
            }
        }
    }
    const int next[4][4] = {{-1, B, B, B}, {A, -1, C, D}, {B, B, -1, D}, {B, B, C, -1}};
    for (int a = 0; a < 4; a++) for (int b = 0; b < 4; b++) if (a != b)
        for (int g = 0; g < GROUPS; g++) route(f, a, g, b + 1, next[a][b]);
    return f;
}

/* Дренируем готовую работу независимо от скорости CPU/Valgrind, не сдвигая будущие окна. */
static void pump(struct fixture* f) {
    for (int i = 0; i < 1024; i++) {
        uasync_drain_immediate(f->ua); uasync_poll(f->ua, 0);
        TimeoutEntry next;
        if (!f->ua->immediate_queue_head &&
            (timeout_heap_peek(f->ua->timeout_heap, &next) || next.expiration > get_time_us() / 1000)) return;
    }
    CHECK(0 && "multicast immediate work did not settle");
}

static struct multicast_stats stats(struct fixture* f, int node) {
    struct multicast_stats s; CHECK(multicast_get_stats(f->inst[node], &s) == 0); return s;
}

static void teardown(struct fixture* f) {
    for (int i = 0; i < 4; i++) for (int j = 0; j < 4; j++)
        if (f->wire[i][j].conn.send_input_q) queue_set_callback(f->wire[i][j].conn.send_input_q, NULL, NULL);
    for (int i = 0; i < 4; i++) multicast_destroy(f->inst[i]);
    for (int i = 0; i < 4; i++) etcp_router_destroy(f->inst[i]);
    for (int i = 0; i < 4; i++) for (int j = 0; j < 4; j++) etcp_router_transit_queues_destroy(&f->wire[i][j].conn);
    uasync_poll(f->ua, 0); /* Освободить помеченные отменёнными heap-таймеры. */
    uasync_drain_immediate(f->ua);
    for (int i = 0; i < 4; i++) {
        for (int j = 0; j < 4; j++) clear_queue(f->wire[i][j].conn.send_input_q);
        for (int g = 0; g < GROUPS; g++) {
            struct TOPO_GROUP* group = f->group[i][g];
            for (struct ll_entry* e = group->nodes->head; e; e = e->next) topo_nodeq_clear_paths((struct TOPO_GROUP_NODE*)e);
            clear_queue(group->nodes); clear_queue(group->senders_list);
        }
        clear_queue(f->inst[i]->topo_groups->group_list);
        u_free(f->inst[i]->topo_groups);
        clear_queue(f->inst[i]->connections);
        CHECK(!f->inst[i]->conn_status_cbks);
        u_free(f->inst[i]);
    }
    CHECK(f->ua->timer_alloc_count == f->ua->timer_free_count);
    uasync_destroy(f->ua, 0); u_free(f);
}

static void test_delivery(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: tree splitting, copying, subscriptions and source isolation");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd == 1);
    struct received rx[4] = {{0}};
    struct multicast_subscription* sub[4];
    for (int i = 0; i < 4; i++) { sub[i] = multicast_subscribe(f->group[i][0], 1, fd, receive, &rx[i]); CHECK(sub[i]); }
    CHECK(!multicast_subscribe(f->group[C][0], 1, fd, receive, &rx[C]));
    uint64_t ids[] = {4, 3, 2, 1, 3};
    char data[] = "copied";
    CHECK(multicast_send(f->inst[A], fd, ids, 5, data, sizeof(data)) == 0);
    CHECK(rx[A].count == 0); memset(data, 'x', sizeof(data)); memset(ids, 0, sizeof(ids));
    pump(f);
    for (int i = 0; i < 4; i++) CHECK(rx[i].count == 1 && memcmp(rx[i].prefix, "copied", 7) == 0);
    CHECK(f->wire[A][B].packets == 1 && f->wire[A][B].max_recipients == 3);
    CHECK(f->wire[B][C].packets == 1 && f->wire[B][D].packets == 1);
    multicast_unsubscribe(sub[B]); sub[B] = NULL;
    uint64_t bcd[] = {2, 3, 4};
    CHECK(multicast_send(f->inst[A], fd, bcd, 3, NULL, 0) == 0);
    pump(f); CHECK(rx[B].count == 1 && rx[C].count == 2 && rx[C].len == 0 && rx[D].count == 2);
    CHECK(stats(f, B).unsubscribed == 1 && l.count == 0);
    int other = multicast_open(f->group[D][0], loss, &l); CHECK(other == fd);
    CHECK(multicast_send(f->inst[D], other, bcd + 1, 1, "other source", 12) == 0);
    pump(f); CHECK(rx[C].count == 2 && stats(f, C).unsubscribed == 1);
    for (int i = 0; i < 4; i++) multicast_unsubscribe(sub[i]);
    teardown(f);
}

static void test_lifetime(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: deferred callbacks, close/unsubscribe, group isolation");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    int other = multicast_open(f->group[A][1], loss, &l); CHECK(fd == 1 && other == 2);
    uint64_t self = 1, remote = 2;
    struct received rx = {0}, rx_other = {0};
    struct multicast_subscription* sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub);
    CHECK(multicast_send(f->inst[A], fd, &self, 1, "pending", 7) == 0);
    multicast_unsubscribe(sub);
    sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub);
    pump(f); CHECK(rx.count == 0);
    rx.unsubscribe = 1;
    CHECK(multicast_send(f->inst[A], fd, &self, 1, "first", 5) == 0);
    CHECK(multicast_send(f->inst[A], fd, &self, 1, "second", 6) == 0);
    pump(f); CHECK(rx.count == 1 && stats(f, A).delivery_bytes == 0);
    sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub);
    CHECK(multicast_send(f->inst[A], fd, &self, 1, "close", 5) == 0);
    multicast_close(f->inst[A], fd);
    pump(f); CHECK(rx.count == 1 && stats(f, A).delivery_bytes == 0);
    multicast_unsubscribe(sub);
    fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd == 3);
    sub = multicast_subscribe(f->group[B][0], 1, fd, receive, &rx); CHECK(sub);
    struct multicast_subscription* sub_other = multicast_subscribe(f->group[B][1], 1, other, receive, &rx_other); CHECK(sub_other);
    block(&f->wire[A][B]);
    CHECK(multicast_send(f->inst[A], fd, &remote, 1, "cancel", 6) == 0);
    CHECK(multicast_send(f->inst[A], other, &remote, 1, "keep", 4) == 0);
    pump(f); CHECK(stats(f, A).queues == 1);
    multicast_close_group(f->inst[A], f->group[A][0]->group_id);
    unblock(&f->wire[A][B]); pump(f);
    CHECK(rx.count == 1 && rx_other.count == 1 && memcmp(rx_other.prefix, "keep", 4) == 0);
    CHECK(stats(f, A).descriptors == 1);
    multicast_close_group(f->inst[B], f->group[B][0]->group_id);
    multicast_unsubscribe(sub);
    /* Handle остаётся пригодным для unsubscribe после уничтожения ядра. */
    teardown(f); multicast_unsubscribe(sub_other);
}

static void test_overload(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: exact 256 KiB boundary, healthy branch and timer-only tail report");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd > 0);
    struct received c = {0}, d = {0};
    struct multicast_subscription* sc = multicast_subscribe(f->group[C][0], 1, fd, receive, &c); CHECK(sc);
    struct multicast_subscription* sd = multicast_subscribe(f->group[D][0], 1, fd, receive, &d); CHECK(sd);
    block(&f->wire[B][C]);
    uint64_t both[] = {3, 4}, one = 3;
    uint8_t payload[8146]; memset(payload, 0xab, sizeof(payload));
    for (int i = 0; i < 32; i++) {
        CHECK(multicast_send(f->inst[A], fd, both, 2, payload, sizeof(payload)) == 0);
        pump(f);
    }
    CHECK(stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES && stats(f, B).overload_packets == 0);
    CHECK(c.count == 0 && d.count == 32);
    uint64_t first_drop = get_time_tb();
    CHECK(multicast_send(f->inst[A], fd, &one, 1, "one", 3) == 0);
    CHECK(l.count == 0); pump(f);
    DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "first report: losses=%u pending=%zu sent=%llu received=%llu overload=%llu",
                l.count, stats(f, B).report_bytes, (unsigned long long)stats(f, B).reports_sent,
                (unsigned long long)stats(f, A).reports_received, (unsigned long long)stats(f, B).overload_packets);
    CHECK(l.count == 1);
    CHECK(l.item[0].src == 2 && l.item[0].dst == 3 && l.item[0].source_node_id == 1 && l.item[0].descriptor == fd);
    CHECK(l.item[0].packets == 1 && l.item[0].recipient_count == 1 && l.ids[0][0] == 3);
    CHECK(l.item[0].prefix_len == 3 && memcmp(l.item[0].prefix, "one", 3) == 0);
    for (int i = 3; i < 16; i++) CHECK(l.item[0].prefix[i] == 0);
    CHECK(multicast_send(f->inst[A], fd, &one, 1, "two", 3) == 0);
    CHECK(multicast_send(f->inst[A], fd, &one, 1, "three", 5) == 0);
    route(f, A, 0, 5, B); route(f, B, 0, 5, C);
    uint64_t different[] = {3, 5};
    CHECK(multicast_send(f->inst[A], fd, different, 2, "set", 3) == 0);
    pump(f); CHECK(l.count == 1);
    uint64_t now = get_time_tb();
    CHECK(now < first_drop + 2500);
    test_clock_run_for(f->ua, (int)(first_drop + 2500 - now)); CHECK(l.count == 1);
    test_clock_run_for(f->ua, 700); pump(f); CHECK(l.count == 3);
    CHECK(l.item[1].packets == 2 && l.item[1].recipient_count == 1 && memcmp(l.item[1].prefix, "two", 3) == 0);
    CHECK(l.item[2].packets == 1 && l.item[2].recipient_count == 2 && l.ids[2][0] == 3 && l.ids[2][1] == 5);
    CHECK(stats(f, B).overload_packets == 4 && stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES);
    unblock(&f->wire[B][C]); pump(f);
    CHECK(c.count == 32 && stats(f, B).queued_bytes == 0);
    test_clock_run_for(f->ua, 3100); CHECK(stats(f, B).report_bytes == 0);
    multicast_unsubscribe(sc); multicast_unsubscribe(sd); teardown(f);
}

static void test_report_budget(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: bounded 16 KiB reports and closing a descriptor inside its loss callback");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    block(&f->wire[A][B]);
    uint64_t dst = 2;
    uint8_t payload[8154] = {0};
    for (int i = 0; i < 32; i++) CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0);
    CHECK(stats(f, A).queued_bytes == MULTICAST_QUEUE_BYTES);
    for (int i = 0; i < 200; i++) {
        int descriptor = multicast_open(f->group[A][0], loss, &l); CHECK(descriptor > 0);
        CHECK(multicast_send(f->inst[A], descriptor, &dst, 1, "loss", 4) == 0);
        CHECK(stats(f, A).report_bytes <= MULTICAST_REPORT_BYTES);
    }
    CHECK(stats(f, A).reports_evicted > 0 && l.count == 0);
    pump(f); CHECK(l.count > 0 && l.count < 200);
    CHECK(stats(f, A).queues == 1);
    /* Два готовых уведомления одного FD; close в первом отменяет второе. */
    struct losses close_loss = { .close = 1 };
    int closing = multicast_open(f->group[A][0], loss, &close_loss);
    uint64_t missing[] = {900, 901};
    CHECK(multicast_send(f->inst[A], closing, missing, 1, "first", 5) == 0);
    CHECK(multicast_send(f->inst[A], closing, missing + 1, 1, "second", 6) == 0);
    pump(f); CHECK(close_loss.count == 1 && close_loss.item[0].reason == MULTICAST_NO_ROUTE);
    multicast_close(f->inst[A], fd); CHECK(stats(f, A).queued_bytes == 0);
    teardown(f);
}

static void test_reroute(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: route change behind blocked head and ETCP DELETE with pending waiter");
    struct fixture* f = setup();
    struct losses l = {0}; struct received c = {0}, d = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    struct multicast_subscription* sc = multicast_subscribe(f->group[C][0], 1, fd, receive, &c); CHECK(sc);
    struct multicast_subscription* sd = multicast_subscribe(f->group[D][0], 1, fd, receive, &d); CHECK(sd);
    uint64_t b = 2, target = 3;
    block(&f->wire[A][B]);
    CHECK(multicast_send(f->inst[A], fd, &b, 1, "blocked head", 12) == 0);
    CHECK(multicast_send(f->inst[A], fd, &target, 1, "reroute", 7) == 0);
    pump(f);
    route(f, A, 0, 3, D); multicast_routes_changed(f->inst[A]); pump(f);
    CHECK(c.count == 1 && memcmp(c.prefix, "reroute", 7) == 0 && f->wire[A][B].packets == 0);
    route(f, A, 0, 2, D);
    struct wire* old = &f->wire[A][B];
    old->conn.links_up = 0; old->conn.state = 2;
    etcp_fire_conn_status(&old->conn, ETCP_CONN_STATUS_DELETE);
    clear_queue(old->conn.send_input_q); old->conn.send_input_q = NULL; old->blocker = NULL;
    pump(f); CHECK(stats(f, A).queued_bytes == 0 && l.count == 0);
    CHECK(stats(f, B).unsubscribed == 1);
    multicast_unsubscribe(sc); multicast_unsubscribe(sd); teardown(f);
}

static void put_le(uint8_t* p, uint64_t value, unsigned n) {
    for (unsigned i = 0; i < n; i++) p[i] = value >> (i * 8);
}

static struct ll_entry* data_packet(uint16_t fd, const uint64_t* ids, unsigned count, const uint64_t* path, unsigned hops) {
    unsigned len = MULTICAST_HEADER_SIZE + count * 8 + hops * 8 + 1;
    struct ll_entry* e = ll_alloc_lldgram(len); CHECK(e); e->len = len;
    e->dgram[0] = ETCP_ID_MULTICAST; put_le(e->dgram + 1, TOPO_GROUP_UTUN, 8);
    put_le(e->dgram + 9, fd, 2); put_le(e->dgram + 11, count, 2); put_le(e->dgram + 13, 1, 8);
    for (unsigned i = 0; i < count; i++) put_le(e->dgram + MULTICAST_HEADER_SIZE + i * 8, ids[i], 8);
    for (unsigned i = 0; i < hops; i++) put_le(e->dgram + MULTICAST_HEADER_SIZE + count * 8 + i * 8, path[i], 8);
    e->dgram[len - 1] = hops; return e;
}

static void test_validation(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: malformed wire, loop prevention and argument rejection without partial send");
    struct fixture* f = setup();
    struct losses l = {0}; struct received r = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    struct multicast_subscription* sub = multicast_subscribe(f->group[B][0], 1, fd, receive, &r); CHECK(sub);
    uint64_t ids[] = {2, 2}, path[] = {1, 2, 4};
    etcp_recv_fn recv = f->inst[B]->api_bindings.callbacks[ETCP_ID_MULTICAST];
    recv(&f->wire[B][A].conn, data_packet(fd, ids, 2, path, 1)); /* повтор адресата */
    recv(&f->wire[B][A].conn, data_packet(fd, ids, 1, path, 2)); /* последний hop не равен пиру */
    recv(&f->wire[B][D].conn, data_packet(fd, ids, 1, path, 3)); /* повторное посещение B */
    struct ll_entry* e = data_packet(fd, ids, 1, path, 1); e->len--;
    recv(&f->wire[B][A].conn, e); /* обрезанный trailer */
    CHECK(stats(f, B).invalid_packets == 3 && stats(f, B).loop_packets == 1);
    uint8_t payload[PKTNORM_MAX_DGRAM_SIZE] = {0};
    CHECK(multicast_send(f->inst[A], fd, ids, 1, payload, sizeof(payload)) < 0);
    ids[1] = 0;
    CHECK(multicast_send(f->inst[A], fd, ids, 2, NULL, 0) < 0);
    CHECK(multicast_send(f->inst[A], 0, ids, 1, NULL, 0) < 0);
    pump(f); CHECK(r.count == 0 && l.count == 0 && f->wire[A][B].packets == 0);
    multicast_unsubscribe(sub); teardown(f);
}

static void test_long_report(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: longest recipient list and loss report split into router-sized parts");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    block(&f->wire[B][C]);
    uint64_t dst = 3;
    uint8_t payload[8146] = {0};
    for (int i = 0; i < 32; i++) {
        CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0); pump(f);
    }
    CHECK(stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES);
    enum { COUNT = (PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE) / 8 };
    uint64_t ids[COUNT];
    for (unsigned i = 0; i < COUNT; i++) {
        ids[i] = i + 100; route(f, A, 0, ids[i], B); route(f, B, 0, ids[i], C);
    }
    CHECK(multicast_send(f->inst[A], fd, ids, COUNT, NULL, 0) == 0); pump(f);
    CHECK(l.count == 2 && l.item[0].recipient_count + l.item[1].recipient_count == COUNT);
    CHECK(l.item[0].packets == 1 && l.item[1].packets == 1 && !l.item[0].prefix_len && !l.item[1].prefix_len);
    CHECK(l.ids[0][0] == 100 && l.ids[1][0] == 100 + l.item[0].recipient_count);
    CHECK(stats(f, B).reports_sent == 2 && stats(f, A).reports_received == 2);
    teardown(f);
}

static void test_report_backpressure(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: blocked report transport, eviction, restart and stale closed descriptor");
    struct fixture* f = setup();
    struct losses l = {0};
    int fd = multicast_open(f->group[A][0], loss, &l);
    block(&f->wire[B][C]); block(&f->wire[B][A]);
    uint64_t dst = 3;
    uint8_t payload[8146] = {0};
    for (int i = 0; i < 32; i++) {
        CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0); pump(f);
    }
    int closed = 0;
    for (int i = 0; i < 200; i++) {
        int descriptor = multicast_open(f->group[A][0], loss, &l); CHECK(descriptor > 0);
        if (!i) closed = descriptor;
        CHECK(multicast_send(f->inst[A], descriptor, &dst, 1, "congested", 9) == 0); pump(f);
        CHECK(stats(f, B).report_bytes <= MULTICAST_REPORT_BYTES);
    }
    CHECK(l.count == 0 && stats(f, B).reports_evicted > 0);
    multicast_close(f->inst[A], closed);
    /* Перезапуск router очищает send_q и отменяет её waiters; таймер должен продолжить отчёты. */
    etcp_router_conn_restart(f->inst[B], f->group[B][0]->group_id, 1, ETCP_RT_ID_MULTICAST_REPORT);
    unblock(&f->wire[B][A]); test_clock_run_for(f->ua, 10000);
    CHECK(l.count > 0 && l.count < 200);
    for (unsigned i = 0; i < l.count; i++) {
        CHECK(l.item[i].descriptor != closed && l.item[i].reason == MULTICAST_OVERLOAD);
        CHECK(l.item[i].src == 2 && l.item[i].dst == 3 && l.item[i].packets == 1);
    }
    CHECK(stats(f, B).report_bytes == 0);
    teardown(f);
}

static void test_descriptor_wrap(void) {
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: sequential 16-bit IDs, exhaustion, wrap and skipping occupied IDs");
    struct fixture* f = setup();
    struct losses l = {0};
    for (int i = 1; i <= UINT16_MAX; i++) CHECK(multicast_open(f->group[A][i % GROUPS], loss, &l) == i);
    CHECK(multicast_open(f->group[A][0], loss, &l) < 0);
    multicast_close(f->inst[A], 2); multicast_close(f->inst[A], 65000);
    CHECK(multicast_open(f->group[A][0], loss, &l) == 2);
    CHECK(multicast_open(f->group[A][0], loss, &l) == 65000);
    CHECK(stats(f, A).descriptors == UINT16_MAX);
    teardown(f);
}

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_categories(DEBUG_CATEGORY_ALL);
    debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO);
    test_clock_init(1000);
    if (getenv("UTUN_TEST_DEBUG")) {
        debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_DEBUG);
        debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_TRACE);
        debug_set_category_level(DEBUG_CATEGORY_TIMERS, DEBUG_LEVEL_DEBUG);
    }
    test_delivery(); test_lifetime(); test_overload(); test_report_budget();
    test_reroute(); test_validation(); test_long_report(); test_report_backpressure();
    debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_WARN);
    test_descriptor_wrap();
    test_clock_finish();
    printf("PASS multicast: delivery, lifetime, overload, reports, reroute, validation, descriptor wrap\n");
    return 0;
}