diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 0fdaafa7..d0bd7b47 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -736,6 +736,15 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; peer_schedule(p); uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms; f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt; + for (struct STREAM_RECEIVE *active = p->flows; active; active = active->next) { + if (active == f || !active->outgoing || active->info.probe_only || active->info.state != STREAM_RECEIVE_ACTIVE + || active->subblock != f->subblock || !same_stream(&active->descriptor, &f->descriptor) + || active->info.path.count != f->info.path.count + || memcmp(active->info.path.nodes, f->info.path.nodes, f->info.path.count * sizeof(uint64_t))) continue; + active->info.rtt_ms = f->info.rtt_ms; active->info.rtt_valid = f->info.rtt_valid; + active->info.path_rtt_ms = f->info.path_rtt_ms; active->info.path_rtt_valid = f->info.path_rtt_valid; + active->info.info_ms = f->info.info_ms; + } } else flow_close(f, STREAM_REASON_PROTOCOL, 0); } else if (m.command == STREAM_CMD_SUBSCRIBE) { if (!authorized(net, sender, &m, &local)) { @@ -766,6 +775,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) { stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2), block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10); + if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t))) + f->info.path_rtt_valid = 0; f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB; } else flow_close(f, STREAM_REASON_PROTOCOL, 1); } diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 94506206..37b273ed 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -223,6 +223,8 @@ static void put_member(unsigned destination, unsigned member, uint64_t group) { static void run_until(unsigned *value, unsigned expected_value) { uint64_t deadline = get_time_tb() + 100000; while (*value < expected_value && get_time_tb() < deadline) test_clock_poll(ua, 100); + if (*value != expected_value) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream test wait expected=%u actual=%u data=%u/%u end=%u/%u closed=%u/%u", + expected_value, *value, data_events[1], data_events[2], end_events[1], end_events[2], closed_events[1], closed_events[2]); CHECK(*value == expected_value); } @@ -611,6 +613,7 @@ static void test_automatic(void) { CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); for (unsigned i = 1; i < 3; i++) wait_state(i, 3, GROUP2, origin.descriptor.source_epoch, 0); + stream_set_event_callback(nodes[1], on_automatic, NULL); stream_set_event_callback(nodes[2], on_automatic, NULL); for (unsigned i = 1; i < 4; i++) { automatic_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, automatic_raw) == 0);