From df474a33d01a7104a91bb4e700fa12ad29fb933b Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 08:34:10 +0200 Subject: [PATCH] Check stream pressure timing, accounting and wide metric fields --- tests/test_stream_core.c | 46 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index fb548983..3d30ff6f 100644 --- a/tests/test_stream_core.c +++ b/tests/test_stream_core.c @@ -5,6 +5,7 @@ #include "../src/stream/stream_source.h" #include "../src/stream/stream_metrics.h" +#include "../src/stream/stream_load.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include "test_clock.h" @@ -18,6 +19,45 @@ static const struct stream_descriptor descriptor = { .source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2 }; static uint8_t private_key[32]; +static unsigned cache_added, cache_removed; +static void cache_observer(void *arg, const uint8_t *wire, size_t length, int added) { + (void)arg; CHECK(length == 1300 && stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) == descriptor.group_id); + if (added) cache_added++; else cache_removed++; +} + +/* Рост до X, ровно 500мс подтверждения, короткий глитч, смена состава и фактические задержанные интервалы. */ +static void test_load(void) { + struct stream_load load; struct stream_load_snapshot snapshot; + stream_load_init(&load, 0); + CHECK(stream_load_add(&load, 10, 10000) == STREAM_OK); + CHECK(stream_load_remove(&load, 6, 6000, 1) == STREAM_OK); + CHECK(stream_load_poll(&load, 200, 250, &snapshot) == STREAM_OK && snapshot.growth_rate == 20000 && !snapshot.confirmed); + CHECK(stream_load_add(&load, 2, 2000) == STREAM_OK && stream_load_remove(&load, 1, 1000, 0) == STREAM_OK); + CHECK(stream_load_poll(&load, 400, 450, &snapshot) == STREAM_OK && snapshot.growth_rate == 15000); + CHECK(snapshot.queued.bytes == 5000 && snapshot.arrivals.bytes == snapshot.sent.bytes + snapshot.removed.bytes + snapshot.queued.bytes); + CHECK(stream_load_poll(&load, 699, 749, &snapshot) == STREAM_OK && !snapshot.confirmed); + CHECK(stream_load_poll(&load, 700, 750, &snapshot) == STREAM_WAIT && snapshot.confirmed); + CHECK(stream_load_remove(&load, 5, 5000, 0) == STREAM_OK); + CHECK(stream_load_poll(&load, 701, 0, &snapshot) == STREAM_WAIT && !snapshot.pressure && !snapshot.confirmed); + CHECK(stream_load_remove(&load, 1, 1000, 1) == STREAM_ERR_STATE); + stream_load_init(&load, 1000); + CHECK(stream_load_add(&load, 2, UINT64_C(1) << 40) == STREAM_OK); + CHECK(stream_load_poll(&load, 1300, 300, &snapshot) == STREAM_OK && !snapshot.confirmed && snapshot.interval_ms == 300); + CHECK(stream_load_remove(&load, 2, UINT64_C(1) << 40, 1) == STREAM_OK); + CHECK(stream_load_poll(&load, 1400, 0, &snapshot) == STREAM_WAIT && !snapshot.confirmed); + CHECK(stream_load_poll(&load, 1500, 0, &snapshot) == STREAM_OK && snapshot.arrivals.bytes == snapshot.sent.bytes); + CHECK(stream_load_add(&load, 3, 3900) == STREAM_OK); stream_load_changed(&load); + CHECK(stream_load_poll(&load, 1700, 300, &snapshot) == STREAM_OK && snapshot.growth_rate == 0 && !snapshot.confirmed); + CHECK(stream_load_poll(&load, 13000, 300, &snapshot) == STREAM_OK && snapshot.input_rate == 390); + struct stream_metric_snapshot input = {.lag_ms = UINT64_C(1) << 42, .blocks = UINT64_MAX, .late_blocks = 2, .stops = 3, + .source_rate_bytes_sec = UINT64_C(1) << 40, .receive_rate_bytes_sec = 10000, .source_rate_valid = 1, .receive_rate_valid = 1, .stopped = 1}; + uint8_t wire[STREAM_METRIC_WIRE_SIZE]; stream_metric_write(wire, &input, 1); + struct stream_metric_snapshot output; int valid = 0; + CHECK(stream_metric_read(wire, &output, &valid) == STREAM_OK && valid && output.blocks == UINT64_MAX + && output.lag_ms == input.lag_ms && output.source_rate_bytes_sec == input.source_rate_bytes_sec && output.stopped); + wire[0] |= 128; CHECK(stream_metric_read(wire, &output, &valid) == STREAM_ERR_FORMAT); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] load: A/D/X/Q, growth before drops, 500ms boundary, glitches, revisions, interval/window and wide metrics"); +} /* Воспроизводимые данные, не совпадающие с zero-padding и проверочными кодограммами. */ static void pattern(uint8_t *data, size_t length) { @@ -174,6 +214,7 @@ static void test_ring_and_conflicts(void) { struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK); + cache_added = cache_removed = 0; stream_buffer_set_observer(buffer, cache_observer, NULL); const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = 100; CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_WAIT); put_batch(encoder, buffer, 10, 255); put_batch(encoder, buffer, 12, 255); @@ -184,6 +225,10 @@ static void test_ring_and_conflicts(void) { CHECK(sc_ed25519_sign(private_key, bad, sizeof(bad) - 64, bad + sizeof(bad) - 64) == SC_OK); CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_CONFLICT); put_batch(encoder, buffer, 11, 1); + CHECK(cache_added == 17 && cache_removed == 8); + uint64_t pending = 0, oldest = 0; + CHECK(stream_buffer_pending(buffer, 11, 0, &pending, &oldest) == STREAM_OK && pending == 2 && oldest); + CHECK(stream_buffer_pending(buffer, 12, 0, &pending, &oldest) == STREAM_OK && pending == 1); CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_EXPIRED); CHECK(stream_buffer_read(buffer, 11, 1, &wire, &length, &actual) == STREAM_GAP); CHECK(stream_buffer_read(buffer, 13, 0, &wire, &length, &actual) == STREAM_WAIT); @@ -346,6 +391,7 @@ int main(void) { size_t allocations = u_get_allocated_count(); test_signed_blocks(); CHECK(u_get_allocated_count() == allocations); test_wire(); CHECK(u_get_allocated_count() == allocations); + test_load(); CHECK(u_get_allocated_count() == allocations); test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations); test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations); test_relative_metrics(); CHECK(u_get_allocated_count() == allocations);