Cover multicast report backpressure, chunking and timer cleanup

tests/test_multicast.c

@@ -179,7 +179,16 @@ static struct fixture* setup(void) {
     return f;
 }
 
-static void pump(struct fixture* f) { test_clock_run_for(f->ua, 10); }
+/* Дренируем готовую работу независимо от скорости 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;
@@ -191,6 +200,7 @@ static void teardown(struct fixture* f) {
     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);
@@ -411,6 +421,63 @@ static void test_validation(void) {
     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();
@@ -434,7 +501,9 @@ int main(void) {
         debug_set_category_level(DEBUG_CATEGORY_TIMERS, DEBUG_LEVEL_DEBUG);
     }
     test_delivery(); test_lifetime(); test_overload(); test_report_budget();
-    test_reroute(); test_validation(); test_descriptor_wrap();
+    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;