Add multicast branching queues and bounded loss reports

src/routing_layer/multicast.c

@@ -0,0 +1,643 @@
+#include <stdlib.h>
+#include <string.h>
+#include <limits.h>
+#include "../lib/debug_config.h"
+#include "../lib/mem.h"
+#include "../lib/u_async.h"
+#include "multicast.h"
+#include "etcp.h"
+#include "etcp_api.h"
+#include "etcp_router.h"
+#include "pkt_normalizer.h"
+#include "topo_group.h"
+#include "utun_instance.h"
+
+#define MC_CAT DEBUG_CATEGORY_ETCPROUTE
+#define MC_MAX_RECIPIENTS ((PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE) / 8)
+#define MC_REPORT_CHUNK ((PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE - ROUTER_PATH_MAX_SIZE - MULTICAST_REPORT_HEADER_SIZE) / 8)
+
+struct mc_packet {
+    uint64_t group, source;
+    uint16_t descriptor, count;
+    const uint8_t *recipients, *payload, *visited;
+    size_t payload_len;
+    uint8_t hops;
+};
+
+struct mc_out {
+    struct mc_out* next;
+    struct multicast_ctx* ctx;
+    struct ETCP_CONN* conn;       /* заимствован до DELETE; затем NULL до опустошения FIFO */
+    struct ll_queue* q;
+    struct queue_waiter_handle waiter;
+};
+
+struct mc_window {
+    struct mc_window* next;
+    uint64_t group, source, dst, deadline;
+    uint16_t descriptor;
+    int dirty;                   /* есть потери после первого немедленного отчёта */
+};
+
+struct mc_report {
+    struct mc_report* next;
+    struct multicast_ctx* ctx;
+    struct multicast_loss loss;
+    uint64_t due, retry;          /* неизменный конец окна / следующая попытка передачи */
+    uint16_t cursor;              /* уже переданные router получатели длинного отчёта */
+    struct queue_waiter_handle waiter;
+    uint64_t ids[];
+};
+
+struct multicast_ctx {
+    struct UTUN_INSTANCE* inst;
+    struct mc_out* outputs;
+    struct mc_window* windows;
+    struct mc_report* reports;
+    void *report_timer, *route_wake;
+    uint64_t eviction_warn;
+    struct multicast_stats stats;
+    multicast_recv_fn recv;
+    multicast_loss_fn loss;
+    void* arg;
+};
+
+struct mc_target { struct ETCP_CONN* conn; uint64_t id; };
+
+static void mc_report_tick(void* arg);
+static void mc_report_schedule(struct multicast_ctx* ctx);
+static void mc_forward(struct multicast_ctx* ctx, struct ll_entry* e, struct mc_out* previous);
+static void mc_out_drain(struct ll_queue* q, void* arg);
+
+static uint64_t mc_read(const uint8_t* p, unsigned bytes) {
+    uint64_t v = 0;
+    for (unsigned i = 0; i < bytes; i++) v |= (uint64_t)p[i] << (i * 8);
+    return v;
+}
+
+static void mc_write(uint8_t* p, uint64_t v, unsigned bytes) {
+    for (unsigned i = 0; i < bytes; i++) p[i] = (uint8_t)(v >> (i * 8));
+}
+
+static void mc_free(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); }
+
+/* Разбирает только границы; полный контроль идентичностей выполняется на входе. */
+static int mc_parse(const struct ll_entry* e, struct mc_packet* p) {
+    if (!e || !e->dgram || e->len < MULTICAST_HEADER_SIZE + 9 || e->len > PKTNORM_MAX_DGRAM_SIZE) return -1;
+    const uint8_t* w = e->dgram;
+    p->group = mc_read(w + 1, 8); p->descriptor = mc_read(w + 9, 2);
+    p->count = mc_read(w + 11, 2); p->source = mc_read(w + 13, 8); p->hops = w[e->len - 1];
+    size_t head = MULTICAST_HEADER_SIZE + (size_t)p->count * 8, tail = (size_t)p->hops * 8 + 1;
+    if (w[0] != ETCP_ID_MULTICAST || !p->source || !p->count || p->count > MC_MAX_RECIPIENTS ||
+        !p->hops || p->hops > MULTICAST_MAX_VISITED || head + tail > e->len) return -1;
+    p->recipients = w + MULTICAST_HEADER_SIZE; p->payload = w + head;
+    p->payload_len = e->len - head - tail; p->visited = w + e->len - tail;
+    if (head + p->payload_len + MULTICAST_PATH_RESERVE > PKTNORM_MAX_DGRAM_SIZE) return -1;
+    return 0;
+}
+
+static int mc_has_visited(const struct mc_packet* p, uint64_t id) {
+    for (unsigned i = 0; i < p->hops; i++) if (mc_read(p->visited + i * 8, 8) == id) return 1;
+    return 0;
+}
+
+/* Проверка канонического списка не требует выделений и не допускает повторной доставки. */
+static int mc_ids_valid(const uint8_t* ids, uint16_t count) {
+    uint64_t previous = 0;
+    for (unsigned i = 0; i < count; i++) {
+        uint64_t id = mc_read(ids + i * 8, 8);
+        if (id <= previous) return 0;
+        previous = id;
+    }
+    return count != 0;
+}
+
+/* Собирает ветку с тем же payload и уже дополненным путём. */
+static struct ll_entry* mc_build(const struct mc_packet* p, const struct mc_target* targets, unsigned count) {
+    size_t len = MULTICAST_HEADER_SIZE + count * 8 + p->payload_len + p->hops * 8 + 1;
+    struct ll_entry* e = ll_alloc_lldgram((uint16_t)len);
+    if (!e) { DEBUG_ERROR(MC_CAT, "multicast packet allocation failed bytes=%zu recipients=%u", len, count); return NULL; }
+    uint8_t* w = e->dgram;
+    w[0] = ETCP_ID_MULTICAST; mc_write(w + 1, p->group, 8); mc_write(w + 9, p->descriptor, 2);
+    mc_write(w + 11, count, 2); mc_write(w + 13, p->source, 8);
+    for (unsigned i = 0; i < count; i++) mc_write(w + MULTICAST_HEADER_SIZE + i * 8, targets[i].id, 8);
+    size_t off = MULTICAST_HEADER_SIZE + count * 8;
+    if (p->payload_len) memcpy(w + off, p->payload, p->payload_len);
+    memcpy(w + off + p->payload_len, p->visited, p->hops * 8); w[len - 1] = p->hops;
+    return e;
+}
+
+/* Запись всегда удаляется вместе с waiter, в том числе уже запланированным. */
+static void mc_report_remove(struct multicast_ctx* ctx, struct mc_report* r) {
+    struct mc_report** pp = &ctx->reports;
+    while (*pp != r) pp = &(*pp)->next;
+    *pp = r->next;
+    etcp_router_cancel_send_ready(ctx->inst, r->loss.group_id, r->loss.source_node_id, ETCP_RT_ID_MULTICAST_REPORT, &r->waiter);
+    ctx->stats.report_bytes -= sizeof(*r) + (size_t)r->loss.recipient_count * 8;
+    u_free(r);
+}
+
+static void mc_window_remove(struct multicast_ctx* ctx, struct mc_window** pp) {
+    struct mc_window* w = *pp;
+    *pp = w->next; ctx->stats.report_bytes -= sizeof(*w); u_free(w);
+}
+
+/* Общий жёсткий бюджет включает сами записи, их ID и состояние троттлинга. */
+static int mc_report_room(struct multicast_ctx* ctx, size_t bytes) {
+    if (bytes > MULTICAST_REPORT_BYTES) {
+        DEBUG_ERROR(MC_CAT, "multicast report exceeds budget bytes=%zu limit=%u", bytes, MULTICAST_REPORT_BYTES); return -1;
+    }
+    while (ctx->stats.report_bytes + bytes > MULTICAST_REPORT_BYTES) {
+        if (ctx->reports) {
+            ctx->stats.reports_evicted++;
+            mc_report_remove(ctx, ctx->reports);
+        } else if (ctx->windows) mc_window_remove(ctx, &ctx->windows);
+        else break;
+        uint64_t now = get_time_tb();
+        if (now >= ctx->eviction_warn) {
+            DEBUG_WARN(MC_CAT, "multicast report budget exhausted bytes=%zu need=%zu evicted=%llu",
+                       ctx->stats.report_bytes, bytes, (unsigned long long)ctx->stats.reports_evicted);
+            ctx->eviction_warn = now + MULTICAST_REPORT_INTERVAL_TB;
+        }
+    }
+    return 0;
+}
+
+/* Сохраняет потерю до освобождения пакета. Удалённому источнику сообщается только OVERLOAD. */
+static void mc_loss(struct multicast_ctx* ctx, const struct mc_packet* p, uint64_t dst,
+                    enum multicast_loss_reason reason, const struct mc_target* targets, unsigned count) {
+    if (reason == MULTICAST_OVERLOAD) ctx->stats.overload_packets++;
+    else if (reason == MULTICAST_NO_ROUTE) ctx->stats.no_route_packets++;
+    else if (reason == MULTICAST_PATH_LOOP) ctx->stats.loop_packets++;
+    else ctx->stats.internal_errors++;
+    if (reason != MULTICAST_OVERLOAD) {
+        DEBUG_WARN(MC_CAT, "multicast drop group=%016llx source=%016llx descriptor=%u dst=%016llx recipients=%u reason=%d",
+                   (unsigned long long)p->group, (unsigned long long)p->source, p->descriptor,
+                   (unsigned long long)dst, count, reason);
+        if (p->source != ctx->inst->node_id) return;
+    }
+    uint64_t now = get_time_tb(), due = now;
+    struct mc_window* window = NULL;
+    if (reason == MULTICAST_OVERLOAD) {
+        for (window = ctx->windows; window; window = window->next)
+            if (window->group == p->group && window->source == p->source &&
+                window->descriptor == p->descriptor && window->dst == dst) break;
+        if (window) {
+            /* Просроченное окно сначала завершит таймер; поздний пакет относится к новому окну. */
+            if (window->deadline <= now) window->deadline = now + MULTICAST_REPORT_INTERVAL_TB;
+            due = window->deadline; window->dirty = 1;
+        }
+    }
+    for (struct mc_report* r = ctx->reports; r; r = r->next) {
+        if (r->due != due || r->cursor || r->loss.group_id != p->group || r->loss.source_node_id != p->source ||
+            r->loss.dst != dst || r->loss.descriptor != p->descriptor || r->loss.reason != reason ||
+            r->loss.recipient_count != count || r->loss.packets == UINT32_MAX) continue;
+        unsigned i = 0;
+        while (i < count && r->ids[i] == targets[i].id) i++;
+        if (i == count) { r->loss.packets++; mc_report_schedule(ctx); return; }
+    }
+    size_t bytes = sizeof(struct mc_report) + count * 8;
+    int new_window = reason == MULTICAST_OVERLOAD && !window;
+    if (mc_report_room(ctx, bytes + (new_window ? sizeof(struct mc_window) : 0)) < 0) return;
+    /* room может вытеснить окно; после него старый указатель больше не используется. */
+    if (new_window) {
+        window = u_calloc(1, sizeof(*window));
+        if (!window) { DEBUG_ERROR(MC_CAT, "multicast throttle allocation failed"); return; }
+        window->group = p->group; window->source = p->source; window->dst = dst; window->descriptor = p->descriptor;
+        window->deadline = now + MULTICAST_REPORT_INTERVAL_TB;
+        struct mc_window** tail = &ctx->windows;
+        while (*tail) tail = &(*tail)->next;
+        *tail = window; ctx->stats.report_bytes += sizeof(*window);
+    }
+    struct mc_report* r = u_calloc(1, bytes);
+    if (!r) { DEBUG_ERROR(MC_CAT, "multicast loss allocation failed bytes=%zu", bytes); return; }
+    r->ctx = ctx; r->due = due; r->retry = due;
+    r->loss = (struct multicast_loss){ .group_id = p->group, .source_node_id = p->source,
+        .src = ctx->inst->node_id, .dst = dst, .descriptor = p->descriptor, .recipient_count = count,
+        .recipients = r->ids, .packets = 1, .reason = reason, .prefix_len = p->payload_len < 16 ? p->payload_len : 16 };
+    if (r->loss.prefix_len) memcpy(r->loss.prefix, p->payload, r->loss.prefix_len);
+    for (unsigned i = 0; i < count; i++) r->ids[i] = targets[i].id;
+    struct mc_report** tail = &ctx->reports;
+    while (*tail) tail = &(*tail)->next;
+    *tail = r; ctx->stats.report_bytes += bytes;
+    mc_report_schedule(ctx);
+}
+
+/* Освобождение router-очереди только будит общий таймер; нет рекурсивной отправки. */
+static void mc_report_ready(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct mc_report* r = arg;
+    r->retry = get_time_tb(); mc_report_schedule(r->ctx);
+}
+
+/* Возвращает 1 после передачи всех частей, 0 при backpressure, -1 при ошибке отправки. */
+static int mc_report_send(struct multicast_ctx* ctx, struct mc_report* r) {
+    if (r->loss.source_node_id == ctx->inst->node_id) {
+        if (ctx->loss) ctx->loss(ctx->inst, &r->loss, ctx->arg);
+        ctx->stats.reports_sent++; ctx->stats.reports_received++; return 1;
+    }
+    while (r->cursor < r->loss.recipient_count) {
+        if (!etcp_router_send_q_has_room(ctx->inst, r->loss.group_id, r->loss.source_node_id, ETCP_RT_ID_MULTICAST_REPORT)) {
+            etcp_router_on_send_ready(ctx->inst, r->loss.group_id, r->loss.source_node_id,
+                                     ETCP_RT_ID_MULTICAST_REPORT, &r->waiter, mc_report_ready, r);
+            return 0;
+        }
+        unsigned count = r->loss.recipient_count - r->cursor;
+        if (count > MC_REPORT_CHUNK) count = MC_REPORT_CHUNK;
+        struct ll_entry* e = ll_alloc_lldgram(1 + MULTICAST_REPORT_HEADER_SIZE + count * 8);
+        if (!e) { DEBUG_ERROR(MC_CAT, "multicast report packet allocation failed recipients=%u", count); return -1; }
+        uint8_t* w = e->dgram + 1; e->dgram[0] = ETCP_RT_ID_MULTICAST_REPORT;
+        mc_write(w, r->loss.descriptor, 2); mc_write(w + 2, r->loss.src, 8); mc_write(w + 10, r->loss.dst, 8);
+        mc_write(w + 18, r->loss.packets, 4); mc_write(w + 22, count, 2); w[24] = r->loss.prefix_len;
+        memcpy(w + 25, r->loss.prefix, 16);
+        for (unsigned i = 0; i < count; i++) mc_write(w + MULTICAST_REPORT_HEADER_SIZE + i * 8, r->ids[r->cursor + i], 8);
+        if (etcp_route_send(ctx->inst, r->loss.group_id, r->loss.source_node_id, e, 0, 0) < 0) {
+            DEBUG_WARN(MC_CAT, "multicast report send failed source=%016llx descriptor=%u cursor=%u",
+                       (unsigned long long)r->loss.source_node_id, r->loss.descriptor, r->cursor); return -1;
+        }
+        r->cursor += count; ctx->stats.reports_sent++;
+    }
+    return 1;
+}
+
+/* Фиксированные окна заканчиваются независимо от нового трафика; blocked-отчёты не смешиваются с новыми. */
+static void mc_report_tick(void* arg) {
+    struct multicast_ctx* ctx = arg;
+    ctx->report_timer = NULL;
+    uint64_t now = get_time_tb();
+    for (struct mc_window** pp = &ctx->windows; *pp;) {
+        struct mc_window* w = *pp;
+        if (w->deadline > now) { pp = &w->next; continue; }
+        if (!w->dirty) { mc_window_remove(ctx, pp); continue; }
+        w->dirty = 0; w->deadline = now + MULTICAST_REPORT_INTERVAL_TB; pp = &w->next;
+    }
+    unsigned processed = 0;
+    for (;;) {
+        struct mc_report* r = ctx->reports;
+        while (r && r->retry > now) r = r->next;
+        if (!r || processed++ == 16) break;
+        /* Callback приложения может отправлять новые пакеты: не сохраняем курсор списка через него.
+         * Временно снимаем запись с бюджета/списка, чтобы eviction не освободил её из callback. */
+        if (r->loss.source_node_id == ctx->inst->node_id) {
+            struct mc_report** pp = &ctx->reports;
+            while (*pp != r) pp = &(*pp)->next;
+            *pp = r->next; ctx->stats.report_bytes -= sizeof(*r) + r->loss.recipient_count * 8;
+            mc_report_send(ctx, r); u_free(r); continue;
+        }
+        r->retry = now + MULTICAST_REPORT_INTERVAL_TB;
+        int result = mc_report_send(ctx, r);
+        if (result == 1) {
+            DEBUG_WARN(MC_CAT, "multicast overload report group=%016llx source=%016llx descriptor=%u link=%016llx->%016llx recipients=%u packets=%u",
+                       (unsigned long long)r->loss.group_id, (unsigned long long)r->loss.source_node_id, r->loss.descriptor,
+                       (unsigned long long)r->loss.src, (unsigned long long)r->loss.dst, r->loss.recipient_count, r->loss.packets);
+            mc_report_remove(ctx, r);
+        }
+    }
+    mc_report_schedule(ctx);
+}
+
+/* Единственный таймер выбирает ближайшее окно или повтор, в покое таймера нет. */
+static void mc_report_schedule(struct multicast_ctx* ctx) {
+    uint64_t due = UINT64_MAX, now = get_time_tb();
+    for (struct mc_window* w = ctx->windows; w; w = w->next) if (w->deadline < due) due = w->deadline;
+    for (struct mc_report* r = ctx->reports; r; r = r->next) if (r->retry < due) due = r->retry;
+    if (ctx->report_timer) { uasync_cancel_timeout(ctx->inst->ua, ctx->report_timer); ctx->report_timer = NULL; }
+    if (due == UINT64_MAX) return;
+    int delay = due <= now ? 0 : (due - now > INT_MAX ? INT_MAX : (int)(due - now));
+    ctx->report_timer = uasync_set_timeout(ctx->inst->ua, delay, ctx, mc_report_tick, "multicast_reports");
+    if (!ctx->report_timer) DEBUG_ERROR(MC_CAT, "multicast report timer allocation failed pending_bytes=%zu", ctx->stats.report_bytes);
+}
+
+/* Удаление FIFO только после отмены waiter; callbacks очереди отменяет queue_free. */
+static void mc_out_destroy(struct multicast_ctx* ctx, struct mc_out* o) {
+    if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter);
+    struct mc_out** pp = &ctx->outputs;
+    while (*pp != o) pp = &(*pp)->next;
+    *pp = o->next;
+    DEBUG_DEBUG(MC_CAT, "multicast queue destroy peer=%016llx packets=%d bytes=%zu",
+                (unsigned long long)(o->conn ? o->conn->peer_node_id : 0), o->q->count, o->q->total_bytes);
+    queue_set_callback(o->q, NULL, NULL);
+    struct ll_entry* e;
+    while ((e = queue_data_get(o->q))) mc_free(e);
+    queue_free(o->q); u_free(o);
+}
+
+/* Единственная FIFO на транспорт, без групповых или descriptor-подочередей. */
+static struct mc_out* mc_out_get(struct multicast_ctx* ctx, struct ETCP_CONN* conn) {
+    for (struct mc_out* o = ctx->outputs; o; o = o->next) if (o->conn == conn) return o;
+    struct mc_out* o = u_calloc(1, sizeof(*o));
+    if (!o) { DEBUG_ERROR(MC_CAT, "multicast output allocation failed"); return NULL; }
+    o->q = queue_new(ctx->inst->ua, 0, 0, 0, "multicast_out");
+    if (!o->q) { DEBUG_ERROR(MC_CAT, "multicast FIFO allocation failed"); u_free(o); return NULL; }
+    o->ctx = ctx; o->conn = conn; o->next = ctx->outputs; ctx->outputs = o;
+    queue_set_callback_defer(o->q, 1); queue_set_callback(o->q, mc_out_drain, o);
+    DEBUG_DEBUG(MC_CAT, "multicast queue created peer=%016llx limit=%u", (unsigned long long)conn->peer_node_id, MULTICAST_QUEUE_BYTES);
+    return o;
+}
+
+/* Сортировка по next hop, внутри ветки — по ID: канонический wire сохраняется после split. */
+static int mc_target_compare(const void* a, const void* b) {
+    const struct mc_target *x = a, *y = b;
+    uintptr_t xc = (uintptr_t)x->conn, yc = (uintptr_t)y->conn;
+    if (xc != yc) return xc < yc ? -1 : 1;
+    return (x->id > y->id) - (x->id < y->id);
+}
+
+/* Выбирает маршруты один раз на обработку, изолирует потери каждой исходящей ветки. */
+static void mc_forward(struct multicast_ctx* ctx, struct ll_entry* e, struct mc_out* previous) {
+    struct mc_packet p;
+    if (mc_parse(e, &p) < 0) {
+        ctx->stats.invalid_packets++; DEBUG_ERROR(MC_CAT, "multicast invalid internal packet len=%u", e->len); mc_free(e); return;
+    }
+    struct TOPO_GROUP* group = topo_groups_find(ctx->inst->topo_groups, p.group);
+    if (!group || group->stopping) {
+        DEBUG_WARN(MC_CAT, "multicast group unavailable group=%016llx", (unsigned long long)p.group); mc_free(e); return;
+    }
+    struct mc_target targets[MC_MAX_RECIPIENTS];
+    unsigned count = 0;
+    for (unsigned i = 0; i < p.count; i++) {
+        uint64_t id = mc_read(p.recipients + i * 8, 8);
+        if (id == ctx->inst->node_id) {
+            ctx->stats.delivered++;
+            if (ctx->recv) ctx->recv(ctx->inst, p.group, p.source, p.descriptor, p.payload, p.payload_len, ctx->arg);
+            else DEBUG_WARN(MC_CAT, "multicast no receiver group=%016llx descriptor=%u", (unsigned long long)p.group, p.descriptor);
+            continue;
+        }
+        targets[count++] = (struct mc_target){ topo_group_find_conn_for_node(group, id), id };
+    }
+    qsort(targets, count, sizeof(*targets), mc_target_compare);
+    for (unsigned start = 0; start < count;) {
+        unsigned end = start + 1;
+        struct ETCP_CONN* conn = targets[start].conn;
+        while (end < count && targets[end].conn == conn) end++;
+        unsigned n = end - start;
+        if (!conn) mc_loss(ctx, &p, 0, MULTICAST_NO_ROUTE, targets + start, n);
+        else if (mc_has_visited(&p, conn->peer_node_id))
+            mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_PATH_LOOP, targets + start, n);
+        else {
+            struct mc_out* o = mc_out_get(ctx, conn);
+            size_t size = MULTICAST_HEADER_SIZE + n * 8 + p.payload_len + p.hops * 8 + 1;
+            if (!o) mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_INTERNAL_ERROR, targets + start, n);
+            else if (o->q->total_bytes + size > MULTICAST_QUEUE_BYTES) {
+                DEBUG_TRACE(MC_CAT, "multicast overflow source=%016llx descriptor=%u peer=%016llx bytes=%zu packet=%zu recipients=%u",
+                            (unsigned long long)p.source, p.descriptor, (unsigned long long)conn->peer_node_id,
+                            o->q->total_bytes, size, n);
+                mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_OVERLOAD, targets + start, n);
+            } else {
+                struct ll_entry* branch = mc_build(&p, targets + start, n);
+                if (!branch) mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_INTERNAL_ERROR, targets + start, n);
+                else {
+                    DEBUG_TRACE(MC_CAT, "multicast split source=%016llx descriptor=%u peer=%016llx recipients=%u bytes=%u",
+                                (unsigned long long)p.source, p.descriptor, (unsigned long long)conn->peer_node_id, n, branch->len);
+                    if (o == previous) queue_data_put_first(o->q, branch);
+                    else queue_data_put(o->q, branch);
+                }
+            }
+        }
+        start = end;
+    }
+    mc_free(e);
+}
+
+/* Проверяет каждый адрес: изменение части назначений требует повторного разделения. */
+static int mc_same_route(struct mc_out* o, const struct ll_entry* e) {
+    struct mc_packet p;
+    if (!o->conn || o->conn->state == 2 || o->conn->close_requested || !o->conn->links_up || mc_parse(e, &p) < 0) return 0;
+    struct TOPO_GROUP* group = topo_groups_find(o->ctx->inst->topo_groups, p.group);
+    if (!group || group->stopping) return 0;
+    for (unsigned i = 0; i < p.count; i++)
+        if (topo_group_find_conn_for_node(group, mc_read(p.recipients + i * 8, 8)) != o->conn) return 0;
+    return 1;
+}
+
+static void mc_out_ready(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct mc_out* o = arg;
+    queue_resume_callback(o->q);
+}
+
+/* Автозабор FIFO: один пакет на свободную send_input_q, затем снова round-robin waiter. */
+static void mc_out_drain(struct ll_queue* q, void* arg) {
+    struct mc_out* o = arg;
+    struct multicast_ctx* ctx = o->ctx;
+    struct ll_entry* e = queue_data_get(q);
+    if (!e) { queue_resume_callback(q); return; }
+    if (!mc_same_route(o, e)) mc_forward(ctx, e, o);
+    else if (o->conn->send_input_q->count) {
+        queue_data_put_first(q, e);
+        if (queue_waiter_wait(o->conn->send_input_q, &o->waiter, mc_out_ready, o) >= 0) return;
+        DEBUG_ERROR(MC_CAT, "multicast send waiter allocation failed peer=%016llx", (unsigned long long)o->conn->peer_node_id);
+        e = queue_data_get(q); mc_free(e); ctx->stats.internal_errors++;
+    } else if (etcp_send(o->conn, e) < 0) {
+        DEBUG_ERROR(MC_CAT, "multicast transport send failed peer=%016llx", (unsigned long long)o->conn->peer_node_id);
+        mc_free(e); ctx->stats.internal_errors++;
+    } else ctx->stats.sent++;
+    queue_resume_callback(q);
+    if (!o->conn && !q->count) mc_out_destroy(ctx, o);
+}
+
+/* Отмена изменённых назначений идёт через remove, обычный забор FIFO остаётся в drain callback. */
+static void mc_recheck(void* arg) {
+    struct multicast_ctx* ctx = arg;
+    ctx->route_wake = NULL;
+    for (struct mc_out* o = ctx->outputs; o;) {
+        struct mc_out* next = o->next;
+        for (struct ll_entry* e = o->q->head; e;) {
+            struct ll_entry* following = e->next;
+            if (!mc_same_route(o, e)) {
+                queue_remove_data(o->q, e); mc_forward(ctx, e, o);
+            }
+            e = following;
+        }
+        if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter);
+        queue_resume_callback(o->q);
+        if (!o->conn && !o->q->count) mc_out_destroy(ctx, o);
+        o = next;
+    }
+}
+
+void multicast_routes_changed(struct UTUN_INSTANCE* inst) {
+    struct multicast_ctx* ctx = inst ? inst->multicast : NULL;
+    if (!ctx || !ctx->outputs || ctx->route_wake) return;
+    ctx->route_wake = uasync_call_soon(inst->ua, ctx, mc_recheck);
+    if (!ctx->route_wake) DEBUG_ERROR(MC_CAT, "multicast route wake allocation failed");
+}
+
+/* DELETE приходит до уничтожения send_input_q: waiter отменяется, указатель conn забывается. */
+static void mc_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
+    struct multicast_ctx* ctx = arg;
+    if (status == ETCP_CONN_STATUS_NEW) return;
+    if (status == ETCP_CONN_STATUS_DELETE) {
+        for (struct mc_out* o = ctx->outputs; o; o = o->next) if (o->conn == conn) {
+            queue_waiter_cancel(conn->send_input_q, &o->waiter); o->conn = NULL;
+        }
+    }
+    multicast_routes_changed(ctx->inst);
+}
+
+/* Приём DATA: проверяет путь до локальной доставки и добавляет self один раз до очередей. */
+static void mc_recv(struct ETCP_CONN* from, struct ll_entry* e) {
+    struct multicast_ctx* ctx = from && from->instance ? from->instance->multicast : NULL;
+    struct mc_packet p;
+    if (!ctx || mc_parse(e, &p) < 0) goto invalid;
+    struct TOPO_GROUP* group = topo_groups_find(ctx->inst->topo_groups, p.group);
+    if (!group || group->stopping || !mc_ids_valid(p.recipients, p.count)) goto invalid;
+    int member = 0;
+    for (struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL; se; se = se->next) {
+        struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)se->data;
+        if (peer->conn == from && peer->accepted && !peer->transport_down) { member = 1; break; }
+    }
+    if (!member || mc_read(p.visited, 8) != p.source || mc_read(p.visited + (p.hops - 1) * 8, 8) != from->peer_node_id)
+        goto invalid;
+    for (unsigned i = 0; i < p.hops; i++) {
+        uint64_t id = mc_read(p.visited + i * 8, 8);
+        if (!id) goto invalid;
+        for (unsigned j = 0; j < i; j++) if (mc_read(p.visited + j * 8, 8) == id) goto invalid;
+    }
+    if (mc_has_visited(&p, ctx->inst->node_id) || p.hops == MULTICAST_MAX_VISITED) {
+        ctx->stats.loop_packets++;
+        DEBUG_WARN(MC_CAT, "multicast loop/path limit source=%016llx peer=%016llx hops=%u",
+                   (unsigned long long)p.source, (unsigned long long)from->peer_node_id, p.hops);
+        mc_free(e); return;
+    }
+    struct ll_entry* expanded = ll_alloc_lldgram(e->len + 8);
+    if (!expanded) { DEBUG_ERROR(MC_CAT, "multicast path allocation failed"); ctx->stats.internal_errors++; mc_free(e); return; }
+    memcpy(expanded->dgram, e->dgram, e->len - 1);
+    mc_write(expanded->dgram + e->len - 1, ctx->inst->node_id, 8); expanded->dgram[expanded->len - 1] = p.hops + 1;
+    mc_free(e); mc_forward(ctx, expanded, NULL); return;
+invalid:
+    if (ctx) ctx->stats.invalid_packets++;
+    DEBUG_WARN(MC_CAT, "multicast invalid data peer=%016llx len=%u",
+               (unsigned long long)(from ? from->peer_node_id : 0), e ? e->len : 0);
+    if (e) mc_free(e);
+}
+
+/* Router уже доставил пакет адресу ядра; проверяем связность src и каноничность отчёта. */
+static void mc_report_recv(struct ETCP_CONN* conn, struct ll_entry* e) {
+    struct multicast_ctx* ctx = conn && conn->instance ? conn->instance->multicast : NULL;
+    if (!ctx || !e || !e->dgram || e->len < ROUTER_SVC_HDR_SIZE) goto invalid;
+    if (e->len == ROUTER_SVC_HDR_SIZE) { mc_free(e); return; } /* штатный router CLOSE */
+    if (e->len < ROUTER_SVC_HDR_SIZE + MULTICAST_REPORT_HEADER_SIZE) goto invalid;
+    uint64_t source, dst, group;
+    memcpy(&source, e->dgram + ROUTER_SVC_SRC_OFF, 8); memcpy(&dst, e->dgram + ROUTER_SVC_DST_OFF, 8);
+    memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8);
+    const uint8_t* w = e->dgram + ROUTER_SVC_HDR_SIZE;
+    unsigned count = mc_read(w + 22, 2);
+    if (dst != ctx->inst->node_id || mc_read(w + 2, 8) != source || !source || !mc_read(w + 10, 8) ||
+        !mc_read(w + 18, 4) || w[24] > 16 || count > MC_REPORT_CHUNK ||
+        e->len != ROUTER_SVC_HDR_SIZE + MULTICAST_REPORT_HEADER_SIZE + count * 8 ||
+        !mc_ids_valid(w + MULTICAST_REPORT_HEADER_SIZE, count)) goto invalid;
+    struct TOPO_GROUP* g = topo_groups_find(ctx->inst->topo_groups, group);
+    if (!g || g->stopping) goto invalid;
+    uint64_t ids[MC_REPORT_CHUNK];
+    for (unsigned i = 0; i < count; i++) ids[i] = mc_read(w + MULTICAST_REPORT_HEADER_SIZE + i * 8, 8);
+    struct multicast_loss loss = { .group_id = group, .source_node_id = dst, .src = source,
+        .dst = mc_read(w + 10, 8), .descriptor = mc_read(w, 2), .recipient_count = count,
+        .recipients = ids, .packets = mc_read(w + 18, 4), .reason = MULTICAST_OVERLOAD, .prefix_len = w[24] };
+    memcpy(loss.prefix, w + 25, 16); ctx->stats.reports_received++;
+    DEBUG_WARN(MC_CAT, "multicast overload received group=%016llx descriptor=%u link=%016llx->%016llx recipients=%u packets=%u",
+               (unsigned long long)group, loss.descriptor, (unsigned long long)loss.src,
+               (unsigned long long)loss.dst, count, loss.packets);
+    if (ctx->loss) ctx->loss(ctx->inst, &loss, ctx->arg);
+    mc_free(e); return;
+invalid:
+    if (ctx) ctx->stats.invalid_packets++;
+    DEBUG_WARN(MC_CAT, "multicast malformed report len=%u", e ? e->len : 0);
+    if (e) mc_free(e);
+}
+
+int multicast_init(struct UTUN_INSTANCE* inst) {
+    if (!inst || !inst->ua || inst->multicast || inst->api_bindings.callbacks[ETCP_ID_MULTICAST] ||
+        inst->router_bindings.callbacks[ETCP_RT_ID_MULTICAST_REPORT]) {
+        DEBUG_ERROR(MC_CAT, "multicast init invalid instance or occupied binding"); return -1;
+    }
+    struct multicast_ctx* ctx = u_calloc(1, sizeof(*ctx));
+    if (!ctx) { DEBUG_ERROR(MC_CAT, "multicast context allocation failed"); return -1; }
+    ctx->inst = inst; inst->multicast = ctx;
+    etcp_add_conn_status_cbk(inst, mc_conn_status, ctx);
+    if (!inst->conn_status_cbks || inst->conn_status_cbks->fn != mc_conn_status || inst->conn_status_cbks->arg != ctx) {
+        DEBUG_ERROR(MC_CAT, "multicast status subscription allocation failed"); u_free(ctx); inst->multicast = NULL; return -1;
+    }
+    etcp_bind(inst, ETCP_ID_MULTICAST, mc_recv); etcp_router_bind(inst, ETCP_RT_ID_MULTICAST_REPORT, mc_report_recv);
+    DEBUG_INFO(MC_CAT, "multicast initialized data_limit=%u reports_limit=%u interval_ms=300",
+               MULTICAST_QUEUE_BYTES, MULTICAST_REPORT_BYTES);
+    return 0;
+}
+
+void multicast_destroy(struct UTUN_INSTANCE* inst) {
+    struct multicast_ctx* ctx = inst ? inst->multicast : NULL;
+    if (!ctx) return;
+    DEBUG_INFO(MC_CAT, "multicast destroy sent=%llu delivered=%llu overload=%llu reports_evicted=%llu report_bytes=%zu",
+               (unsigned long long)ctx->stats.sent, (unsigned long long)ctx->stats.delivered,
+               (unsigned long long)ctx->stats.overload_packets, (unsigned long long)ctx->stats.reports_evicted, ctx->stats.report_bytes);
+    etcp_unbind(inst, ETCP_ID_MULTICAST); etcp_router_unbind(inst, ETCP_RT_ID_MULTICAST_REPORT);
+    etcp_remove_conn_status_cbk(inst, mc_conn_status, ctx);
+    if (ctx->route_wake) uasync_call_soon_cancel(inst->ua, ctx->route_wake);
+    if (ctx->report_timer) uasync_cancel_timeout(inst->ua, ctx->report_timer);
+    while (ctx->outputs) mc_out_destroy(ctx, ctx->outputs);
+    while (ctx->reports) mc_report_remove(ctx, ctx->reports);
+    while (ctx->windows) mc_window_remove(ctx, &ctx->windows);
+    inst->multicast = NULL; u_free(ctx);
+}
+
+void multicast_set_callbacks(struct UTUN_INSTANCE* inst, multicast_recv_fn recv, multicast_loss_fn loss, void* arg) {
+    if (!inst || !inst->multicast) { DEBUG_ERROR(MC_CAT, "multicast callbacks before init"); return; }
+    inst->multicast->recv = recv; inst->multicast->loss = loss; inst->multicast->arg = arg;
+}
+
+int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients,
+                   uint16_t count, const void* payload, size_t len) {
+    if (!group || group->stopping || !group->instance || !group->instance->multicast || !recipients ||
+        !count || count > MC_MAX_RECIPIENTS || (!payload && len) ||
+        len > PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE - (size_t)count * 8) {
+        DEBUG_ERROR(MC_CAT, "multicast send invalid arguments count=%u len=%zu", count, len); return -1;
+    }
+    struct mc_target targets[MC_MAX_RECIPIENTS];
+    for (unsigned i = 0; i < count; i++) {
+        if (!recipients[i]) { DEBUG_ERROR(MC_CAT, "multicast zero recipient index=%u", i); return -1; }
+        targets[i] = (struct mc_target){ NULL, recipients[i] };
+    }
+    qsort(targets, count, sizeof(*targets), mc_target_compare);
+    unsigned unique = 0;
+    for (unsigned i = 0; i < count; i++) if (!unique || targets[i].id != targets[unique - 1].id) targets[unique++] = targets[i];
+    uint8_t visited[8]; mc_write(visited, group->instance->node_id, 8);
+    struct mc_packet p = { .group = group->group_id, .source = group->instance->node_id, .descriptor = descriptor,
+        .payload = payload, .payload_len = len, .visited = visited, .hops = 1 };
+    struct ll_entry* e = mc_build(&p, targets, unique);
+    if (!e) return -1;
+    mc_forward(group->instance->multicast, e, NULL); return 0;
+}
+
+int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out) {
+    if (!inst || !inst->multicast || !out) { DEBUG_ERROR(MC_CAT, "multicast stats invalid arguments"); return -1; }
+    *out = inst->multicast->stats;
+    for (struct mc_out* o = inst->multicast->outputs; o; o = o->next) { out->queues++; out->queued_bytes += o->q->total_bytes; }
+    return 0;
+}
+
+void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id) {
+    struct multicast_ctx* ctx = inst ? inst->multicast : NULL;
+    if (!ctx) return;
+    unsigned dropped = 0;
+    for (struct mc_out* o = ctx->outputs; o; o = o->next) {
+        for (struct ll_entry* e = o->q->head; e;) {
+            struct ll_entry* next = e->next;
+            if (mc_read(e->dgram + 1, 8) == group_id) { queue_remove_data(o->q, e); mc_free(e); dropped++; }
+            e = next;
+        }
+        if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter);
+        queue_resume_callback(o->q);
+    }
+    for (struct mc_report* r = ctx->reports; r;) {
+        struct mc_report* next = r->next;
+        if (r->loss.group_id == group_id) mc_report_remove(ctx, r);
+        r = next;
+    }
+    for (struct mc_window** pp = &ctx->windows; *pp;) {
+        if ((*pp)->group == group_id) mc_window_remove(ctx, pp);
+        else pp = &(*pp)->next;
+    }
+    DEBUG_INFO(MC_CAT, "multicast group closed group=%016llx discarded=%u reports_bytes=%zu",
+               (unsigned long long)group_id, dropped, ctx->stats.report_bytes);
+    mc_report_schedule(ctx);
+}

src/routing_layer/multicast.h

@@ -0,0 +1,95 @@
+/* multicast — адресная рассылка по маршрутам topo-группы, с разделением по next hop.
+ * send копирует данные; descriptor задаёт приложение источника, это не номер пакета.
+ * Одна FIFO на ETCP-соединение обслуживает все группы/источники/дескрипторы.
+ * При переполнении теряется новая ветка; источник получает OVERLOAD через router.
+ * Все вызовы и callbacks — uasync-поток. В callbacks допустима отправка, но нельзя
+ * уничтожать ядро/группу/модуль. Данные callbacks заимствованы только на время вызова.
+ *
+ * Wire DATA, все числа little-endian:
+ * cmd:1 | group:8 | descriptor:2 | count:2 | source:8 | recipients:8*count |
+ * payload | visited:8*hops | hops:1.
+ * Получатели строго возрастают. Источник начинает visited собой, каждый транзит
+ * добавляет себя ровно один раз. Первый visited=source, последний=ETCP peer.
+ * Полный резерв visited учитывается до отправки, payload на транзите не меняется.
+ *
+ * Wire OVERLOAD — payload отдельного router-сервиса:
+ * descriptor:2 | src:8 | dst:8 | packets:4 | count:2 | prefix_len:1 |
+ * prefix:16 | recipients:8*count.
+ * src/dst — концы перегруженного участка; router dst — источник DATA.
+ * packets относится к КАЖДОМУ указанному получателю. Части длинного отчёта имеют
+ * непересекающиеся списки: складывать их packets как число исходных пакетов нельзя.
+ * Отчёт сохраняет префикс первого пакета; разные наборы получателей не смешиваются.
+ */
+#ifndef UTUN_MULTICAST_H
+#define UTUN_MULTICAST_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct UTUN_INSTANCE;
+struct TOPO_GROUP;
+struct multicast_ctx;
+
+#define MULTICAST_QUEUE_BYTES (256u * 1024u)
+#define MULTICAST_REPORT_BYTES (16u * 1024u)
+#define MULTICAST_REPORT_INTERVAL_TB 3000
+#define MULTICAST_MAX_VISITED 64
+#define MULTICAST_HEADER_SIZE 21
+#define MULTICAST_PATH_RESERVE (8 * MULTICAST_MAX_VISITED + 1)
+#define MULTICAST_REPORT_HEADER_SIZE 41
+
+enum multicast_loss_reason {
+    MULTICAST_OVERLOAD = 1,       /* очередь выбранного исходящего соединения заполнена */
+    MULTICAST_NO_ROUTE,           /* только локальное уведомление источнику */
+    MULTICAST_PATH_LOOP,          /* локальное уведомление; транзит логирует потерю */
+    MULTICAST_INTERNAL_ERROR      /* выделение памяти / отказ транспорта */
+};
+
+struct multicast_loss {
+    uint64_t group_id;            /* группа исходной рассылки */
+    uint64_t source_node_id;      /* источник исходного DATA */
+    uint64_t src, dst;            /* узел потери и выбранный next hop; dst=0 без маршрута */
+    uint16_t descriptor;         /* идентификатор приложения источника */
+    uint16_t recipient_count;    /* число ID в recipients */
+    const uint64_t* recipients;   /* заимствованный массив в host byte order */
+    uint32_t packets;            /* потери для каждого получателя этого списка */
+    enum multicast_loss_reason reason;
+    uint8_t prefix_len;           /* 0..16 значимых байт */
+    uint8_t prefix[16];           /* начало payload первого потерянного пакета */
+};
+
+typedef void (*multicast_recv_fn)(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t source,
+                                  uint16_t descriptor, const uint8_t* payload, size_t len, void* arg);
+typedef void (*multicast_loss_fn)(struct UTUN_INSTANCE* inst, const struct multicast_loss* loss, void* arg);
+
+struct multicast_stats {
+    uint64_t sent, delivered, overload_packets, no_route_packets, loop_packets, invalid_packets, internal_errors;
+    uint64_t reports_sent, reports_received, reports_evicted; /* отправленные части / принятые / вытесненные записи */
+    size_t queued_bytes, report_bytes; /* данные в FIFO / память отчётов и окон, без уже переданного транспорту */
+    unsigned queues;             /* лениво созданные FIFO, в том числе временно опустевшие */
+};
+
+/* Инициализация после etcp_router_init; destroy до router/ETCP. 0/-1; повторный init — ошибка. */
+int multicast_init(struct UTUN_INSTANCE* inst);
+void multicast_destroy(struct UTUN_INSTANCE* inst);
+/* Один приёмник и обработчик потерь на ядро; NULL отключает соответствующий callback. */
+void multicast_set_callbacks(struct UTUN_INSTANCE* inst, multicast_recv_fn recv, multicast_loss_fn loss, void* arg);
+/* count>0, ID ненулевые; повторы удаляются. len может быть 0. 0=принято к обработке,
+ * -1=неверный аргумент/размер или отказ начального выделения. Частичные потери — через loss.
+ * Доставка best effort: повторов/сквозных ACK нет; порядок при смене маршрутов не гарантируется. */
+int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients,
+                   uint16_t count, const void* payload, size_t len);
+/* Снимок диагностики; 0/-1, не создаёт состояния. */
+int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out);
+/* Внутренние hooks топологии: отложенный пересмотр маршрутов / отмена данных удаляемой группы. */
+void multicast_routes_changed(struct UTUN_INSTANCE* inst);
+void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id);
+
+#ifdef __cplusplus
+}
+#endif
+#endif