diff --git a/src/stream/stream.c b/src/stream/stream.c index cd90cf66..a3da0b4f 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -298,7 +298,7 @@ static void session_deliver(void *arg) { if (result == STREAM_OK) { if (!s->decoded) { s->decoded = 1; - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u", + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u signature=verified", (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq, (unsigned)(s->descriptor.n - s->descriptor.r)); diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c index 80d4d53d..2c371315 100644 --- a/src/stream/stream_buffer.c +++ b/src/stream/stream_buffer.c @@ -182,9 +182,10 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip /* Прекращение владельцем всех обращений предшествует destroy; кольцо ничего не знает о чужих NCD handles. */ void stream_buffer_destroy(struct stream_buffer *b) { if (!b) return; - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu", + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu rejected=%llu", (unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes, - (unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates); + (unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates, + (unsigned long long)b->stats.rejected_batches); for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]); if (b->slots) for (uint32_t i = 0; i < b->capacity; i++) u_free(b->slots[i].path); rs_free(b->rs); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 2fc2cc89..c3494e89 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -24,6 +24,7 @@ #define GROUP1 UINT64_C(710001) #define GROUP2 UINT64_C(710002) +#define SINGLE_INPUT_CAPACITY (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE - STREAM_BATCH_OVERHEAD) #define CHECK(expr) do { if (!(expr)) { \ DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \ } } while (0) @@ -84,7 +85,7 @@ err_t __wrap_uasync_get_load(struct UASYNC *async, struct uasync_load *load) { static void on_global_pressure(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { (void)arg; unsigned node = inst == nodes[1] ? 1 : 2; CHECK(event->descriptor.source_node_id == nodes[3]->node_id && event->descriptor.group_id == GROUP2); - if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == 1180); global_data[node]++; } + if (event->type == STREAM_EVENT_DATA) { CHECK(event->length == SINGLE_INPUT_CAPACITY); global_data[node]++; } else CHECK(event->type == STREAM_EVENT_GAP || event->type == STREAM_EVENT_CLOSED || event->type == STREAM_EVENT_END); } @@ -622,6 +623,21 @@ static void test_network(void) { /* Снять два лишних проверочных субпотока: шесть оставшихся восстанавливают новые пачки. */ for (unsigned i = 6; i < 8; i++) { stream_receive_close(up[i]); up[i] = NULL; stream_receive_close(down[i]); down[i] = NULL; } network_write(publisher, 3); run_until(&network_data[1], 2); run_until(&network_data[2], 2); run_until(&network_gaps[2], 1); + /* Relay пересылает первую часть, пока у него и downstream ещё нет k частей для проверки общей подписи. */ + for (unsigned sub = 1; sub < 6; sub++) { stream_receive_close(up[sub]); up[sub] = NULL; } + CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.blocks == 2); + network_write(publisher, 4); deadline = get_time_tb() + 100000; + do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK); } + while (info.blocks < 3 && get_time_tb() < deadline); + CHECK(info.blocks == 3 && network_data[1] == 2 && network_data[2] == 2); + uint8_t pending_output[sizeof(expected)]; size_t pending_length = 0; + for (unsigned node = 1; node < 3; node++) { + struct stream_local pending; CHECK(stream_local_get(nodes[node], &origin.descriptor, &pending) == STREAM_OK); + CHECK(pending.next == 4 && stream_buffer_decode(pending.buffer, 4, pending_output, sizeof(pending_output), &pending_length) == STREAM_WAIT); + } + for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, sub, 4, &up[sub]) == STREAM_OK); + run_until(&network_data[1], 3); run_until(&network_data[2], 3); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live relay forwards before batch decode; consumers wait for k parts and source signature"); CHECK(stream_publish_finish(publisher) == STREAM_OK); run_until(&network_end[1], 1); run_until(&network_end[2], 1); /* RENEW каждые 2с удерживает даже окончившуюся выдачу; затем имитируем отсутствие продлений. */ @@ -791,7 +807,7 @@ static void test_global_pressure(void) { CHECK(stream_subscribe_manual_open(nodes[i], GROUP2, nodes[3]->node_id, &subscribers[i]) == STREAM_OK); stream_set_event_callback(nodes[i], on_global_pressure, NULL); } - uint8_t payload[1180]; memset(payload, 3, sizeof(payload)); size_t accepted; + uint8_t payload[SINGLE_INPUT_CAPACITY]; memset(payload, 3, sizeof(payload)); size_t accepted; CHECK(stream_publish_write(publisher, payload, sizeof(payload), &accepted) == STREAM_OK && accepted == sizeof(payload)); for (unsigned i = 1; i < 3; i++) for (unsigned sub = 0; sub < 2; sub++) CHECK(stream_receive_open(subscribers[i], nodes[3]->node_id, sub, UINT64_MAX, &receives[i][sub]) == STREAM_OK); @@ -902,10 +918,12 @@ static void test_automatic(void) { } test_clock_run_for(ua, 3000); CHECK(!automatic_data[1] && !automatic_data[2]); - memset(pressure_payload, 0xa9, 1180); size_t accepted; + size_t capacity = stream_publish_capacity(publisher); + CHECK(capacity == SINGLE_INPUT_CAPACITY && capacity <= sizeof(pressure_payload)); + memset(pressure_payload, 0xa9, capacity); size_t accepted; uint64_t initial_probes[3] = {0}; struct stream_selection_info state[3]; for (unsigned frame = 0; frame < 350; frame++) { - CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK && accepted == 1180); + CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK && accepted == capacity); test_clock_run_for(ua, 1200); if (frame == 80 || frame == 180) for (unsigned i = 1; i < 3; i++) { CHECK(stream_selection_get_info(subscribers[i], &state[i]) == STREAM_OK); @@ -925,7 +943,7 @@ static void test_automatic(void) { CHECK(same.probes_started == state[1].probes_started); stream_close(second); second = NULL; uint64_t migrations = state[2].migrations; automatic_drop_relay = 1; for (unsigned frame = 0; frame < 30; frame++) { - CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK); + CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK); test_clock_run_for(ua, 1200); } CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK); @@ -937,7 +955,7 @@ static void test_automatic(void) { memcpy(cancel->dgram, automatic_request, sizeof(automatic_request)); cancel->len = sizeof(automatic_request); automatic_handlers[3](source_connection, cancel); for (unsigned frame = 0; frame < 20; frame++) { - CHECK(stream_publish_write(publisher, pressure_payload, 1180, &accepted) == STREAM_OK); + CHECK(stream_publish_write(publisher, pressure_payload, capacity, &accepted) == STREAM_OK); test_clock_run_for(ua, 1200); } CHECK(stream_selection_get_info(subscribers[2], &state[2]) == STREAM_OK);