Validate partial overload cuts preserve six FEC data streams

tests/test_stream_service.c

@@ -399,12 +399,47 @@ static void test_pressure(void) {
     queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
     run_until(&pressure_data, 5);
     test_clock_run_for(ua, 1000); info = source_load(); CHECK(!info.load.queued.bytes && !info.request_id);
+    /* Вход 8 пакетов/интервал, выход 7: небольшой, но устойчивый избыток. Клиент снимает две parity-выдачи и сохраняет k=6. */
+    while (queue->count) test_clock_poll(ua, 100);
+    queue_set_callback(queue, NULL, NULL);
+    for (unsigned round = 0; round < 60; round++) {
+        CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
+        uint64_t target = source_load().load.sent.packets + 7;
+        deadline = get_time_tb() + 100000;
+        while (source_load().load.sent.packets < target && !source_load().released_count && get_time_tb() < deadline) {
+            if (queue->count) callback(queue, callback_arg);
+            test_clock_poll(ua, 100);
+        }
+        info = source_load();
+        if (info.released_count) break;
+        CHECK(info.load.sent.packets >= target);
+        test_clock_run_for(ua, 2000);
+    }
+    info = source_load(); CHECK(info.released_count == 2 && info.released_rate >= info.required_rate && selects_seen);
+    for (unsigned i = 0; i < 8; i++) {
+        struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
+        CHECK(i >= 6 ? receive.state == STREAM_RECEIVE_CLOSED && receive.reason == STREAM_REASON_OVERLOAD
+                     : receive.state == STREAM_RECEIVE_ACTIVE);
+    }
+    queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
+    test_clock_run_for(ua, 10000); CHECK(!source_load().load.queued.bytes && conn->links_up);
+    unsigned delivered = pressure_data;
+    CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
+    run_until(&pressure_data, delivered + 1);
+    CHECK(stream_receive_restart(subscriber) == STREAM_OK);
+    for (unsigned i = 0; i < 8; i++) {
+        stream_receive_close(receives[i]); receives[i] = NULL;
+        CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK);
+    }
+    run_until(&pressure_data, delivered + 2); test_clock_run_for(ua, 1000);
+    uint64_t previous_request = source_load().request_id;
+    overload_seen = selects_seen = 0; stale_select_length = 0;
     /* Тот же транспорт остаётся UP. Теперь держим очередь дольше подтверждения и возвращаем управление получателю. */
     while (queue->count) test_clock_poll(ua, 100);
     queue_set_callback(queue, NULL, NULL);
     for (unsigned i = 0; i < 24; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
     deadline = get_time_tb() + 15000;
-    do { test_clock_poll(ua, 100); info = source_load(); } while (!info.request_active && get_time_tb() < deadline);
+    do { test_clock_poll(ua, 100); info = source_load(); } while ((!info.request_active || info.request_id == previous_request) && get_time_tb() < deadline);
     CHECK(info.request_active && info.required_count > 1 && info.required_rate && !info.released_count);
     uint64_t request = info.request_id, expires = info.deadline_ms;
     test_clock_run_for(ua, 3000); info = source_load(); CHECK(info.request_id == request && info.deadline_ms == expires);
@@ -433,7 +468,7 @@ static void test_pressure(void) {
     for (unsigned i = 1; i <= 3; i += 2) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_handlers[i]) == 0);
     stream_set_event_callback(nodes[1], on_stream, NULL);
     CHECK(stream_handle_active(reader1) && stream_handle_active(reader2));
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: short glitch, multi-flow frozen rates, fixed 2s deadline, fallback, late SELECT and shared ETCP");
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: glitch, parity-only cooperation, frozen rates, fixed deadline, fallback and stale SELECT");
 }
 
 /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */