feat: [DSM-91] Metric for misrouted messages in streams (#8338)

During a subnet split, export a metric to track misrouted (according to
the current routing table, presumably correctly routed according to an
earlier routing table) messages in streams. We only scan streams to/from
subnets involved in canister migrations, which should make this a no-op
most of the time.

This metric should make it trivial to ascertain when it is safe to
remove a given `canister_migrations` entry (when the registry version
has moved past the routing table update and there are no more misrouted
messages to or from the involved subnets; for a while, to smooth over
any race conditions).

Author: alin-at-dfinity

rs/messaging/src/routing/stream_builder.rs

@@ -17,7 +17,7 @@ use ic_types::{CountBytes, Cycles, SubnetId};
 #[cfg(test)]
 use mockall::automock;
 use prometheus::{Histogram, IntCounter, IntCounterVec, IntGaugeVec};
-use std::collections::{BTreeMap, btree_map};
+use std::collections::{BTreeMap, BTreeSet, btree_map};
 use std::sync::{Arc, Mutex};
 
 #[cfg(test)]
@@ -36,6 +36,8 @@ struct StreamBuilderMetrics {
     pub routed_messages: IntCounterVec,
     /// Successfully routed XNet messages' total payload size.
     pub routed_payload_sizes: Histogram,
+    /// Misrouted messages currently in streams, by remote subnet.
+    pub stream_misrouted_messages: IntGaugeVec,
     /// Critical error counter for detected infinite loops while routing.
     pub critical_error_infinite_loops: IntCounter,
     /// Critical error for payloads above the maximum supported size.
@@ -53,6 +55,7 @@ const METRIC_STREAM_BEGIN: &str = "mr_stream_begin";
 const METRIC_SIGNALS_END: &str = "mr_signals_end";
 const METRIC_ROUTED_MESSAGES: &str = "mr_routed_message_count";
 const METRIC_ROUTED_PAYLOAD_SIZES: &str = "mr_routed_payload_size_bytes";
+const METRIC_STREAM_MISROUTED_MESSAGES: &str = "mr_stream_misrouted_messages";
 
 const LABEL_TYPE: &str = "type";
 const LABEL_STATUS: &str = "status";
@@ -106,6 +109,11 @@ impl StreamBuilderMetrics {
             // 10 B - 5 MB
             decimal_buckets(1, 6),
         );
+        let stream_misrouted_messages= metrics_registry.int_gauge_vec(
+            METRIC_STREAM_MISROUTED_MESSAGES,
+            "Count of misrouted messages in streams, by remote subnet. Only populated for subnets currently involved in a canister migration.",
+            &[LABEL_REMOTE],
+        );
         let critical_error_infinite_loops =
             metrics_registry.error_counter(CRITICAL_ERROR_INFINITE_LOOP);
         let critical_error_payload_too_large =
@@ -144,6 +152,7 @@ impl StreamBuilderMetrics {
             signals_end,
             routed_messages,
             routed_payload_sizes,
+            stream_misrouted_messages,
             critical_error_infinite_loops,
             critical_error_payload_too_large,
             critical_error_response_destination_not_found,
@@ -266,6 +275,94 @@ impl StreamBuilderImpl {
             .observe(msg.payload_size_bytes().get() as f64);
     }
 
+    /// Iterates over all messages in potentially relevant streams and counts how
+    /// many are misrouted (mismatched source or destination subnet according to the
+    /// current routing table).
+    ///
+    /// Only streams to or from subnets involved in migrations may enqueue misrouted
+    /// messages. If this subnet is involved in a migration, we scan all its
+    /// streams. Otherwise, we only scan streams to subnets involved in migrations.
+    fn observe_misrouted_messages(&self, state: &ReplicatedState) {
+        // Reset all gauges to zero before recounting.
+        //
+        // This may lead to a race condition where some keys are temporarily missing,
+        // but we already have a race condition between this metric and the registry
+        // version metric. We work around both by (1) aggregating over all replicas on
+        // the subnet and (2) requiring the condition (no misrouted messages) to hold
+        // for a while before acting on it.
+        self.metrics.stream_misrouted_messages.reset();
+
+        let canister_migrations = state.metadata.network_topology.canister_migrations.as_ref();
+        if canister_migrations.is_empty() {
+            return;
+        }
+
+        // Collect all subnets involved in migrations (source or destination).
+        //
+        // It may be sufficient to only look at "source" subnets of migrations because
+        // we are looking for messages stuck in streams to OLD host subnets. But, just
+        // to be safe, we collect all subnets appearing in migration traces.
+        let mut subnets_with_canister_migrations = BTreeSet::new();
+        for (_, trace) in canister_migrations.iter() {
+            for subnet in trace {
+                subnets_with_canister_migrations.insert(*subnet);
+            }
+        }
+        let relevant_subnets = if subnets_with_canister_migrations.contains(&self.subnet_id) {
+            // This subnet is the source or target of a migration, scan all its streams.
+            state
+                .metadata
+                .streams()
+                .keys()
+                .cloned()
+                .collect::<BTreeSet<_>>()
+        } else {
+            // This is a third-party subnet, only scan streams to subnets involved in
+            // canister migrations.
+            subnets_with_canister_migrations
+        };
+
+        for remote_subnet in &relevant_subnets {
+            let Some(stream) = state.metadata.streams().get(remote_subnet) else {
+                continue;
+            };
+
+            let mut misrouted_messages = 0;
+            // Iterate over all messages in the stream
+            for (_, msg) in stream.messages().iter() {
+                // Check for receiver subnet mismatch.
+                let receiver_host_subnet =
+                    state.metadata.network_topology.route(msg.receiver().get());
+                if receiver_host_subnet != Some(*remote_subnet) {
+                    misrouted_messages += 1;
+                    continue;
+                }
+
+                // Check for sender subnet mismatch.
+                let sender_host_subnet = match msg {
+                    StreamMessage::Request(req) => {
+                        state.metadata.network_topology.route(req.sender.get())
+                    }
+                    StreamMessage::Response(resp) => {
+                        state.metadata.network_topology.route(resp.originator.get())
+                    }
+                    StreamMessage::Refund(_) => {
+                        // Refunds don't have explicit senders. Always assume they are local.
+                        Some(self.subnet_id)
+                    }
+                };
+                if sender_host_subnet != Some(self.subnet_id) {
+                    misrouted_messages += 1;
+                }
+            }
+
+            self.metrics
+                .stream_misrouted_messages
+                .with_label_values(&[&remote_subnet.to_string()])
+                .set(misrouted_messages);
+        }
+    }
+
     /// Implementation of `StreamBuilder::build_streams()`.
     fn build_streams_impl(&self, mut state: ReplicatedState) -> ReplicatedState {
         /// Pops the previously peeked message.
@@ -535,6 +632,7 @@ impl StreamBuilderImpl {
                     .with_label_values(&[&subnet])
                     .set(signals_end.get() as i64);
             });
+        self.observe_misrouted_messages(&state);
 
         {
             // Record the enqueuing time of any messages newly enqueued into `streams`.

rs/messaging/src/routing/stream_builder/tests.rs

@@ -6,7 +6,7 @@ use ic_config::message_routing::{MAX_STREAM_MESSAGES, TARGET_STREAM_SIZE_BYTES};
 use ic_error_types::RejectCode;
 use ic_limits::SYSTEM_SUBNET_STREAM_MSG_LIMIT;
 use ic_management_canister_types_private::Method;
-use ic_registry_routing_table::{CanisterIdRange, RoutingTable};
+use ic_registry_routing_table::{CanisterIdRange, CanisterMigrations, RoutingTable};
 use ic_registry_subnet_type::SubnetType;
 use ic_replicated_state::testing::{
     CanisterQueuesTesting, ReplicatedStateTesting, SystemStateTesting,
@@ -18,7 +18,9 @@ use ic_test_utilities_metrics::{
     nonzero_values,
 };
 use ic_test_utilities_state::{new_canister_state, register_callback};
-use ic_test_utilities_types::ids::{SUBNET_27, SUBNET_42, canister_test_id, user_test_id};
+use ic_test_utilities_types::ids::{
+    SUBNET_3, SUBNET_4, SUBNET_5, SUBNET_27, SUBNET_42, canister_test_id, user_test_id,
+};
 use ic_test_utilities_types::messages::RequestBuilder;
 use ic_types::messages::{
     CallbackId, MAX_INTER_CANISTER_PAYLOAD_IN_BYTES_U64, NO_DEADLINE, Payload, Refund,
@@ -1216,6 +1218,211 @@ fn build_streams_with_oversized_payloads() {
     });
 }
 
+/// Local subnet is splitting: a canister is migrating from local subnet to
+/// subnet B.
+#[test]
+fn test_observe_misrouted_messages_on_splitting_subnet() {
+    with_test_replica_logger(|log| {
+        let (stream_builder, mut state, metrics_registry) = new_fixture(&log);
+
+        // Subnets and canisters.
+        const REMOTE_SUBNET_B: SubnetId = SUBNET_4; // Destination of migration.
+        const REMOTE_SUBNET_Z: SubnetId = SUBNET_5; // Other subnet, no migration.
+
+        let local_canister = canister_test_id(100);
+        // Migrating but not yet migrated canister.
+        let migrating_canister = canister_test_id(200);
+        // Already migrated canister.
+        let migrated_canister = canister_test_id(300);
+        let canister_on_b = canister_test_id(400);
+        let canister_on_z = canister_test_id(500);
+
+        // Routing table: `migrating_canister` is still hosted by the local subnet;
+        // `migrated_canister` has already migrated from the local subnet to B.
+        state.metadata.network_topology.routing_table = Arc::new(
+            RoutingTable::try_from(btreemap! {
+                CanisterIdRange{ start: local_canister, end: local_canister } => LOCAL_SUBNET,
+                CanisterIdRange{ start: migrating_canister, end: migrating_canister } => LOCAL_SUBNET,
+                CanisterIdRange{ start: migrated_canister, end: migrated_canister } => REMOTE_SUBNET_B,
+                CanisterIdRange{ start: canister_on_b, end: canister_on_b } => REMOTE_SUBNET_B,
+                CanisterIdRange{ start: canister_on_z, end: canister_on_z } => REMOTE_SUBNET_Z,
+            })
+            .unwrap(),
+        );
+
+        // Canister migrations: both `migrating_canister` and `migrated_canister` are
+        // migrating from the local subnet to B.
+        state.metadata.network_topology.canister_migrations = Arc::new(
+            CanisterMigrations::try_from(btreemap! {
+                CanisterIdRange{ start: migrating_canister, end: migrating_canister } => vec![LOCAL_SUBNET, REMOTE_SUBNET_B],
+                CanisterIdRange{ start: migrated_canister, end: migrated_canister } => vec![LOCAL_SUBNET, REMOTE_SUBNET_B],
+            })
+            .unwrap(),
+        );
+
+        let message_to = |receiver: CanisterId| {
+            RequestBuilder::default()
+                .sender(local_canister)
+                .receiver(receiver)
+                .build()
+                .into()
+        };
+        let message = |sender: CanisterId, receiver: CanisterId| {
+            RequestBuilder::default()
+                .sender(sender)
+                .receiver(receiver)
+                .build()
+                .into()
+        };
+
+        // Loopback stream, with messages to and from all canisters hosted by subnet A
+        // at any given time. The 3 messages to/from `migrated_canister` are misrouted.
+        let mut loopback_stream = Stream::default();
+        loopback_stream.push(message_to(local_canister));
+        loopback_stream.push(message_to(migrated_canister));
+        loopback_stream.push(message_to(migrating_canister));
+        loopback_stream.push(message(migrated_canister, local_canister));
+        loopback_stream.push(message(migrated_canister, migrated_canister));
+        loopback_stream.push(message(migrating_canister, local_canister));
+        loopback_stream.push(message(migrating_canister, migrating_canister));
+
+        // Stream to subnet B, with messages to all canisters potentially hosted by
+        // subnet B at any given time.
+        //
+        // The 2 messages from the migrated canister and the 2 messages to the migrating
+        // canister are misrouted.
+        let mut stream_to_subnet_b = Stream::default();
+        stream_to_subnet_b.push(message_to(canister_on_b));
+        stream_to_subnet_b.push(message_to(migrated_canister));
+        stream_to_subnet_b.push(message_to(migrating_canister));
+        stream_to_subnet_b.push(message(migrated_canister, canister_on_b));
+        stream_to_subnet_b.push(message(migrated_canister, migrated_canister));
+        stream_to_subnet_b.push(message(migrating_canister, canister_on_b));
+        stream_to_subnet_b.push(message(migrating_canister, migrating_canister));
+
+        // Stream to subnet Z: one message from each currently or previously hosted
+        // canister. The message from `migrated_canister` is misrouted.
+        let mut stream_to_subnet_z = Stream::default();
+        stream_to_subnet_z.push(message_to(canister_on_z));
+        stream_to_subnet_z.push(message(migrated_canister, canister_on_z));
+        stream_to_subnet_z.push(message(migrating_canister, canister_on_z));
+
+        state.modify_streams(|streams| {
+            *streams = btreemap! {
+                LOCAL_SUBNET => loopback_stream,
+                REMOTE_SUBNET_B => stream_to_subnet_b,
+                REMOTE_SUBNET_Z => stream_to_subnet_z,
+            }
+        });
+
+        // Act.
+        stream_builder.observe_misrouted_messages(&state);
+
+        // Assert.
+        assert_eq!(
+            metric_vec(&[
+                (&[(LABEL_REMOTE, &LOCAL_SUBNET.to_string())], 3),
+                (&[(LABEL_REMOTE, &REMOTE_SUBNET_B.to_string())], 4),
+                (&[(LABEL_REMOTE, &REMOTE_SUBNET_Z.to_string())], 1)
+            ]),
+            fetch_int_gauge_vec(&metrics_registry, METRIC_STREAM_MISROUTED_MESSAGES)
+        );
+    });
+}
+
+/// A canister is migrating between remote subnets A and B.
+#[test]
+fn test_observe_misrouted_messages_on_third_party_subnet() {
+    with_test_replica_logger(|log| {
+        let (stream_builder, mut state, metrics_registry) = new_fixture(&log);
+
+        // Subnets and canisters.
+        const REMOTE_SUBNET_A: SubnetId = SUBNET_3; // Source of migration.
+        const REMOTE_SUBNET_B: SubnetId = SUBNET_4; // Destination of migration.
+        const REMOTE_SUBNET_Z: SubnetId = SUBNET_5; // Other subnet, no migration.
+
+        let local_canister = canister_test_id(100);
+        let canister_on_a = canister_test_id(200);
+        let migrated_canister = canister_test_id(300);
+        let migrating_canister = canister_test_id(400);
+        let canister_on_b = canister_test_id(500);
+        let canister_on_z = canister_test_id(600);
+
+        // Routing table: `migrating_canister` is still hosted by subnet A;
+        // `migrated_canister` has already migrated from A to B.
+        state.metadata.network_topology.routing_table = Arc::new(
+            RoutingTable::try_from(btreemap! {
+                CanisterIdRange{ start: local_canister, end: local_canister } => LOCAL_SUBNET,
+                CanisterIdRange{ start: canister_on_a, end: canister_on_a } => REMOTE_SUBNET_A,
+                CanisterIdRange{ start: migrating_canister, end: migrating_canister } => REMOTE_SUBNET_A,
+                CanisterIdRange{ start: migrated_canister, end: migrated_canister } => REMOTE_SUBNET_B,
+                CanisterIdRange{ start: canister_on_b, end: canister_on_b } => REMOTE_SUBNET_B,
+                CanisterIdRange{ start: canister_on_z, end: canister_on_z } => REMOTE_SUBNET_Z,
+            })
+            .unwrap(),
+        );
+
+        // Canister migrations: both `migrating_canister` and `migrated_canister` are
+        // migrating from A to B.
+        state.metadata.network_topology.canister_migrations = Arc::new(
+            CanisterMigrations::try_from(btreemap! {
+                CanisterIdRange{ start: migrated_canister, end: migrated_canister } => vec![REMOTE_SUBNET_A, REMOTE_SUBNET_B],
+                CanisterIdRange{ start: migrating_canister, end: migrating_canister } => vec![REMOTE_SUBNET_A, REMOTE_SUBNET_B],
+            })
+            .unwrap(),
+        );
+
+        let message_to = |receiver: CanisterId| {
+            RequestBuilder::default()
+                .sender(local_canister)
+                .receiver(receiver)
+                .build()
+                .into()
+        };
+
+        // Stream to subnet A with messages to all the canisters it hosted at one time
+        // or another. Only the message to `migrated_canister` is misrouted.
+        let mut stream_to_subnet_a = Stream::default();
+        stream_to_subnet_a.push(message_to(canister_on_a));
+        stream_to_subnet_a.push(message_to(migrated_canister));
+        stream_to_subnet_a.push(message_to(migrating_canister));
+
+        // Stream to subnet B with messages to all the canisters it could potentially
+        // have hosted. Only the message to `migrating_canister` is misrouted.
+        let mut stream_to_subnet_b = Stream::default();
+        stream_to_subnet_b.push(message_to(canister_on_b));
+        stream_to_subnet_b.push(message_to(migrated_canister));
+        stream_to_subnet_b.push(message_to(migrating_canister));
+
+        // Contents of both the loopback stream and the stream to subnet Z. It enqueues
+        // one misrouted message (e.g. due to a manual canister migration). Not counted
+        // because neither the local subnet nor Z are on any migration trace.
+        let mut other_stream = Stream::default();
+        other_stream.push(message_to(canister_on_a));
+
+        state.modify_streams(|streams| {
+            *streams = btreemap! {
+                REMOTE_SUBNET_A => stream_to_subnet_a,
+                REMOTE_SUBNET_B => stream_to_subnet_b,
+                REMOTE_SUBNET_Z => other_stream.clone(),
+                LOCAL_SUBNET => other_stream,
+            }
+        });
+
+        // Act.
+        stream_builder.observe_misrouted_messages(&state);
+
+        // Assert.
+        assert_eq!(
+            metric_vec(&[
+                (&[(LABEL_REMOTE, &REMOTE_SUBNET_A.to_string())], 1),
+                (&[(LABEL_REMOTE, &REMOTE_SUBNET_B.to_string())], 1)
+            ]),
+            fetch_int_gauge_vec(&metrics_registry, METRIC_STREAM_MISROUTED_MESSAGES)
+        );
+    });
+}
+
 /// Sets up the `StreamHandlerImpl`, `ReplicatedState` and `MetricsRegistry` to
 /// be used by a test using specific stream limits.
 fn new_fixture_with_limits(