Refresh matching active path RTT through probes and isolate integration event counters

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);
             }

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, &parameters, &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);