From c1483512597ab4508a3b02a9839cb04cfdb17a12 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 12:01:51 +0100 Subject: [PATCH] Close expired nine-byte stream INFO requests with overload reason --- src/stream/stream_network.c | 7 ++-- tests/test_stream_control_expiry.c | 59 ++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) create mode 100644 tests/test_stream_control_expiry.c diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index b836d404..0830ba65 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -1144,10 +1144,11 @@ static void expire_controls(struct stream_peer *p, uint64_t now) { if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) { int outgoing = m.command == STREAM_CMD_INFO_REQ || m.command == STREAM_CMD_SUBSCRIBE; struct STREAM_RECEIVE *f = flow_find(p, &m, outgoing); - if (f && m.command == STREAM_CMD_INFO_REQ && (m.length != 8 || stream_read_be(m.payload, 8) != f->probe)) f = NULL; + if (f && m.command == STREAM_CMD_INFO_REQ && (m.length != 9 || stream_read_be(m.payload, 8) != f->probe)) f = NULL; if (f && (outgoing || m.command == STREAM_CMD_ACCEPT)) flow_close(f, STREAM_REASON_OVERLOAD, 0); - DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream control expired peer=%llx command=%u handle=%llu age=%llu", - (unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, (unsigned long long)(now - queued_ms)); + DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream control expired peer=%llx command=%u handle=%llu bytes=%zu age_ms=%llu flow_closed=%d", + (unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, m.length, + (unsigned long long)(now - queued_ms), f && f->info.state == STREAM_RECEIVE_CLOSED); } p->load.rejected++; p->network->load.rejected++; queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(p->control); diff --git a/tests/test_stream_control_expiry.c b/tests/test_stream_control_expiry.c new file mode 100644 index 00000000..9098f919 --- /dev/null +++ b/tests/test_stream_control_expiry.c @@ -0,0 +1,59 @@ +/* Настоящая control FIFO: границы TTL, обе формы INFO_REQ9, чужой probe_id и SUBSCRIBE8. */ +#include "../src/stream/stream_network.c" + +#define CHECK(x) do { if (!(x)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stream control expiry line=%d: %s", __LINE__, #x); exit(1); \ +} } while (0) + +/* Минимальный контекст изолирует expiry; ID/тип/длина проходят штатные control_put и wire parser. */ +static unsigned check_expiry(struct stream_peer *peer, uint8_t command, int media, int mismatch, int expect_closed) { + struct STREAM_RECEIVE flow = {.peer = peer, .descriptor = {.group_id = 1, .source_node_id = 2, .source_epoch = 3, + .n = 8, .r = 0, .block_size = STREAM_RS_DEFAULT_SIZE}, .subblock = 0, .outgoing = 1, .owned = 1, + .probe = UINT64_C(0x12345678), .info_sent = 1}; + stream_metrics_init(&flow.metrics); stream_metrics_init(&flow.offered_metrics); + flow.info = (struct stream_receive_info){.peer = peer->node_id, .state = STREAM_RECEIVE_INFO, .probe_only = 1}; + uint8_t payload[9] = {0}; stream_write_be(payload, flow.probe + mismatch, 8); payload[8] = media; + size_t bytes = sizeof(payload); + if (command == STREAM_CMD_SUBSCRIBE) { + flow.info.handle = 1; flow.info.state = STREAM_RECEIVE_REQUESTED; flow.info.probe_only = 0; bytes = 8; + } + peer->flows = &flow; + struct stream_message message = {.command = command, .descriptor = flow.descriptor, .handle = flow.info.handle}; + CHECK(control_put(peer, message, payload, bytes) == STREAM_OK && peer->control->head); + struct stream_message parsed; + CHECK(stream_wire_parse(peer->control->head->dgram, peer->control->head->len, &parsed) == STREAM_OK); + CHECK(parsed.command == command && parsed.length == bytes); + uint64_t queued_ms; memcpy(&queued_ms, peer->control->head->data, sizeof(queued_ms)); + expire_controls(peer, queued_ms + STREAM_LOAD_MAX_AGE_MS - 1); + CHECK(peer->control->count == 1 && flow.info.state != STREAM_RECEIVE_CLOSED); + expire_controls(peer, queued_ms + STREAM_LOAD_MAX_AGE_MS); + CHECK(peer->control->count == 1 && flow.info.state != STREAM_RECEIVE_CLOSED); + expire_controls(peer, queued_ms + STREAM_LOAD_MAX_AGE_MS + 1); + int closed = flow.info.state == STREAM_RECEIVE_CLOSED; + int correct = closed == expect_closed && !peer->control->count && (!closed || flow.info.reason == STREAM_REASON_OVERLOAD); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "control expiry command=%u bytes=%zu media=%d mismatch=%d state=%d reason=%d expected_closed=%d pass=%d", + command, parsed.length, media, mismatch, flow.info.state, flow.info.reason, expect_closed, correct); + if (peer->pump) uasync_call_soon_cancel(peer->network->inst->ua, peer->pump); + peer->pump = NULL; peer->flows = NULL; + return !correct; +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); + debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO); + size_t allocations = u_get_allocated_count(); + struct UASYNC *ua = uasync_create(); CHECK(ua); + struct UTUN_INSTANCE instance = {.ua = ua}; + struct stream_network network = {.inst = &instance}; + struct stream_peer peer = {.network = &network, .node_id = 2, .own_epoch = 1, .other_epoch = 2}; + peer.control = queue_new(ua, 0, 0, 0, "control_expiry"); CHECK(peer.control); + unsigned failed = check_expiry(&peer, STREAM_CMD_INFO_REQ, 0, 0, 1); + failed += check_expiry(&peer, STREAM_CMD_INFO_REQ, 1, 0, 1); + failed += check_expiry(&peer, STREAM_CMD_INFO_REQ, 1, 1, 0); + failed += check_expiry(&peer, STREAM_CMD_SUBSCRIBE, 0, 0, 1); + queue_free(peer.control); uasync_destroy(ua, 0); + CHECK(u_get_allocated_count() == allocations); + CHECK(!failed); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS stream control expiry: 4 cases, before/at/after TTL, no tracked leaks"); + return 0; +}