Exercise stream overload under real ETCP backpressure

2 days ago · f11592be3e
1 changed files with 109 additions and 0 deletions
--- a/tests/test_stream_service.c
+++ b/tests/test_stream_service.c
@ -39,6 +39,38 @@ static uint64_t network_next[4] = {0, 1, 1, 0};
 static uint8_t first_request[STREAM_WIRE_HEADER + 8];
 static etcp_recv_fn saved_raw[4], saved_route;
 static int drop_renew, gap_batch;
+static unsigned pressure_data, overload_seen, selects_seen;
+static int ignore_select;
+static uint8_t pressure_payload[7080], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER];
+static size_t stale_select_length;
+static etcp_recv_fn pressure_handlers[4];
+
+static void pressure_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
+    unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4);
+    if (entry->len >= STREAM_WIRE_HEADER) {
+        if (entry->dgram[2] == STREAM_CMD_OVERLOAD) overload_seen++;
+        if (entry->dgram[2] == STREAM_CMD_OVERLOAD_SELECT) {
+            selects_seen++; CHECK(entry->len <= sizeof(stale_select));
+            memcpy(stale_select, entry->dgram, entry->len); stale_select_length = entry->len;
+            if (ignore_select) { queue_dgram_free(entry); queue_entry_free(entry); return; }
+        }
+    }
+    pressure_handlers[node](conn, entry);
+}
+
+static void on_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
+    (void)arg; CHECK(inst == nodes[1] && event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2);
+    if (event->type == STREAM_EVENT_DATA) {
+        CHECK(event->length == sizeof(pressure_payload) && !memcmp(event->data, pressure_payload, sizeof(pressure_payload))); pressure_data++;
+    } else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END);
+}
+
+static struct stream_peer_load_info source_load(void) {
+    struct stream_peer_load_info info; CHECK(stream_peer_get_load(nodes[3], nodes[1]->node_id, &info) == STREAM_OK);
+    CHECK(info.load.arrivals.bytes == info.load.sent.bytes + info.load.removed.bytes + info.load.queued.bytes);
+    CHECK(info.load.arrivals.packets == info.load.sent.packets + info.load.removed.packets + info.load.queued.packets);
+    return info;
+}

 /* Протокол проходит настоящие ETCP/router обработчики; перехват только считает команды и имитирует потерю RENEW. */
 static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
@ -328,6 +360,82 @@ static void test_network(void) {
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live ETCP chain: common index/restart, FEC, paths/LOOP, GAP, duplicate/reverse, END, leases, REINIT");
 }

+/* Настоящий send_input_q намеренно держим непустым: короткий глитч не режет, устойчивый рост вызывает OVERLOAD. */
+static void test_pressure(void) {
+    memset(pressure_payload, 42, sizeof(pressure_payload));
+    struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL;
+    CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK);
+    struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
+    wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0);
+    CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK);
+    stream_set_event_callback(nodes[1], on_pressure, NULL);
+    for (unsigned i = 1; i <= 3; i += 2) {
+        pressure_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(pressure_handlers[i] != NULL);
+        CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_raw) == 0);
+    }
+    size_t accepted = 0;
+    CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK && accepted == sizeof(pressure_payload));
+    struct STREAM_RECEIVE *receives[8] = {0};
+    for (unsigned i = 0; i < 8; i++) CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK);
+    run_until(&pressure_data, 1);
+    uint64_t deadline = get_time_tb() + 100000;
+    while (get_time_tb() < deadline) {
+        int ready = 1;
+        for (unsigned i = 0; i < 8; i++) {
+            struct stream_receive_info info; CHECK(stream_receive_get_info(receives[i], &info) == STREAM_OK);
+            if (!info.blocks) ready = 0;
+            CHECK(info.rtt_valid && info.supplier_metrics_valid);
+        }
+        if (ready) break; test_clock_poll(ua, 100);
+    }
+    struct ETCP_CONN *conn = instance_find_conn(nodes[3], nodes[1]->node_id); CHECK(conn && conn->links_up);
+    struct ll_queue *queue = conn->send_input_q;
+    while (queue->count) test_clock_poll(ua, 100);
+    queue_callback_fn callback = queue->callback; void *callback_arg = queue->callback_arg;
+    CHECK(callback != NULL); queue_set_callback(queue, NULL, NULL);
+    for (unsigned i = 0; i < 4; i++) CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK);
+    test_clock_run_for(ua, 3000);
+    struct stream_peer_load_info info = source_load(); CHECK(info.load.queued.bytes && !info.request_id && !overload_seen);
+    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);
+    /* Тот же транспорт остаётся 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);
+    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);
+    ignore_select = 1; queue_set_callback(queue, callback, callback_arg); queue_resume_callback(queue);
+    deadline = get_time_tb() + 100000;
+    while (!selects_seen && get_time_tb() < deadline) test_clock_poll(ua, 100);
+    CHECK(overload_seen && selects_seen && stale_select_length);
+    /* Повторы SELECT намеренно не доходят: source исполняет достаточный остаток на исходном deadline. */
+    while (get_time_tb() / 10 < expires + 100) test_clock_poll(ua, 100);
+    info = source_load(); CHECK(!info.request_active && info.released_count > 1 && info.released_rate >= info.required_rate);
+    unsigned closed = 0;
+    for (unsigned i = 0; i < 8; i++) {
+        struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
+        if (receive.state == STREAM_RECEIVE_CLOSED) { CHECK(receive.reason == STREAM_REASON_OVERLOAD); closed++; }
+    }
+    CHECK(closed == info.released_count);
+    /* Поздний выбор старого id не закрывает дополнительную выдачу. */
+    ignore_select = 0;
+    struct ETCP_CONN *back = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(back != NULL);
+    while (back->send_input_q->count) test_clock_poll(ua, 100);
+    struct ll_entry *late = ll_alloc_lldgram((uint16_t)stale_select_length); CHECK(late != NULL);
+    memcpy(late->dgram, stale_select, stale_select_length); late->len = (uint16_t)stale_select_length; CHECK(etcp_send(back, late) == 0);
+    test_clock_run_for(ua, 1000); CHECK(source_load().released_count == closed && conn->links_up);
+    for (unsigned i = 0; i < 8; i++) stream_receive_close(receives[i]);
+    stream_close(subscriber); stream_close(publisher); test_clock_run_for(ua, 1000);
+    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");
+}
+
 /* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */
 static void test_profile_atomicity(void) {
    char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1];
@ -472,6 +580,7 @@ int main(void) {
    u_free(candidates);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] Merkle subscriptions and BGP gating; mobile source allowed, mobile relay excluded");
    test_network();
+    test_pressure();

    /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */
    struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK);