diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index c400d77f..6838ecc8 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -53,6 +53,7 @@ static int automatic_drop_relay; static int automatic_drop_tail; static etcp_recv_fn automatic_handlers[4]; static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; +static void run_until(unsigned *value, unsigned expected_value); err_t __real_uasync_get_load(struct UASYNC *async, struct uasync_load *load); err_t __wrap_uasync_get_load(struct UASYNC *async, struct uasync_load *load) { err_t result = __real_uasync_get_load(async, load); @@ -100,6 +101,45 @@ static void on_automatic(struct UTUN_INSTANCE *inst, const struct stream_event * else CHECK(event->type == STREAM_EVENT_CLOSED); } +struct restart_events { uint64_t next; unsigned data, gaps; }; +static void on_restart(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)inst; struct restart_events *events = arg; + if (event->type == STREAM_EVENT_DATA) { + CHECK(event->first_seq == events->next++ && event->length == sizeof(expected)); + CHECK(!memcmp(event->data, expected, sizeof(expected))); events->data++; + } else if (event->type == STREAM_EVENT_GAP) { + CHECK(event->first_seq == events->next && event->last_seq >= event->first_seq); + events->next = event->last_seq + 1; events->gaps++; + } else CHECK(event->type == STREAM_EVENT_CLOSED); +} + +/* Два restart до отложенного GAP сохраняют ещё не сообщённую приложению границу. */ +static void test_restart_bound(struct STREAM_HANDLE *subscriber, const struct stream_descriptor *descriptor) { + struct restart_events events = {.next = 50}; + struct stream_encoder *encoder = NULL; const uint8_t *blocks[RS_MAX_BLOCKS]; + CHECK(stream_encoder_create(&encoder, descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK); + stream_set_event_callback(nodes[3], on_restart, &events); + CHECK(stream_encoder_encode(encoder, 50, 1, 0, expected, sizeof(expected), blocks) == STREAM_OK); + for (unsigned sub = 0; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == STREAM_OK); + run_until(&events.data, 1); + CHECK(stream_receive_restart(subscriber) == STREAM_OK); + CHECK(stream_encoder_encode(encoder, 60, 2, 0, expected, sizeof(expected), blocks) == STREAM_OK); + CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_OK); + CHECK(stream_receive_restart(subscriber) == STREAM_OK); + CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_DUPLICATE); + for (unsigned sub = 1; sub < 6; sub++) CHECK(stream_ingest_block(nodes[3], blocks[sub], 1300) == STREAM_OK); + run_until(&events.data, 2); + CHECK(events.gaps == 1 && events.next == 61); + CHECK(stream_receive_restart(subscriber) == STREAM_OK); + CHECK(stream_receive_restart(subscriber) == STREAM_OK); + CHECK(stream_ingest_block(nodes[3], blocks[0], 1300) == STREAM_DUPLICATE); + test_clock_run_for(ua, 1000); + struct stream_local local; CHECK(stream_local_get(nodes[3], descriptor, &local) == STREAM_OK); + CHECK(local.anchored && local.start == 61 && local.next == 61 && events.data == 2 && events.gaps == 1); + stream_set_event_callback(nodes[3], NULL, NULL); stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated restart: deferred GAP retained; stale bootstrap never replays DATA"); +} + 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) { @@ -515,6 +555,8 @@ static void test_pressure(void) { stream_receive_close(receives[i]); receives[i] = NULL; CHECK(stream_receive_open(subscriber, nodes[3]->node_id, i, UINT64_MAX, &receives[i]) == STREAM_OK); } + test_clock_run_for(ua, 1000); CHECK(pressure_data == delivered + 1); + CHECK(stream_publish_write(publisher, pressure_payload, sizeof(pressure_payload), &accepted) == STREAM_OK); run_until(&pressure_data, delivered + 2); test_clock_run_for(ua, 1000); uint64_t previous_request = source_load().request_id; overload_seen = selects_seen = 0; stale_select_length = 0; @@ -927,6 +969,7 @@ int main(void) { CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3); + test_restart_bound(desktop, &info.descriptor); struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &info.descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK); const uint8_t *blocks[RS_MAX_BLOCKS];