jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Close expired nine-byte stream INFO requests with overload reason

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 12:04:31 +01:00
1 parent 3956711c16
commit c148351259
2 files changed
+63 -3
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_network.c
+4 -3
View File
@@ -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);
tests/test_stream_control_expiry.c
+59
View File
@@ -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;
}