test(xnet): exercise silently dropped responses on subnet deletion (#10793)

The subnet-deletion XNet tests only checked that in-flight requests
towards a deleted subnet are rejected. This PR also exercises that
in-flight responses towards a deleted subnet are silently dropped (a
response can never be turned into a reject).

- Add a universal-canister op `LoopUntilGlobalDataSet` that loops on
`canister_status` calls (requesting its own status) until its global
data is set to a fixed blob, then replies — used to delay a response
until a chosen moment.
- Extend both the system test
(`rs/tests/message_routing/xnet/subnet_delete_test.rs`) and its
`StateMachine` analogue
(`rs/state_machine_tests/tests/subnet_delete.rs`): the deleted subnet's
canister calls two canisters on a surviving subnet whose replies are
released so that one response ends up in the stream and the other in the
sender's output queue when the subnet is deleted.
- Verify the drops: the queued response via
`mr_routed_message_count{type="response",status="canister_not_found"}`;
the streamed response via the stream being discarded (metric-silent bulk
drop).

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 3 people

Cargo.lock

@@ -148,6 +148,7 @@ system_test(
     runtime_deps = UNIVERSAL_CANISTER_RUNTIME_DEPS | NNS_CANISTER_RUNTIME_DEPS,
     deps = [
         # Keep sorted.
+        "//rs/config",
         "//rs/nns/constants",
         "//rs/registry/canister",
         "//rs/registry/subnet_type",

rs/state_machine_tests/tests/subnet_delete.rs

@@ -12,6 +12,7 @@ candid = { workspace = true }
 canister-test = { path = "../../../rust_canisters/canister_test" }
 futures = { workspace = true }
 ic-agent = { workspace = true }
+ic-config = { path = "../../../config" }
 ic-error-types = { path = "../../../../packages/ic-error-types" }
 ic-management-canister-types-private = { path = "../../../types/management_canister_types" }
 ic-nns-constants = { path = "../../../nns/constants" }

rs/tests/message_routing/xnet/BUILD.bazel

@@ -126,5 +126,6 @@ try_from_u8!(
         CostHttpRequestV2 = 97,
         MsgCallerInfoData = 98,
         MsgCallerInfoSigner = 99,
+        LoopUntilGlobalDataSet = 100,
     }
 );

rs/tests/message_routing/xnet/Cargo.toml

@@ -28,6 +28,20 @@ pub struct TransformArg {
     pub context: Vec<u8>,
 }
 
+/// Candid argument of the management canister's `canister_status` method.
+#[derive(CandidType, Deserialize)]
+pub struct CanisterIdRecord {
+    pub canister_id: Principal,
+}
+
+/// Encodes a `PushBytes` op (opcode byte, little-endian `u32` length, bytes)
+/// into `out`, matching the encoding produced by the `PayloadBuilder`.
+fn encode_push_bytes(out: &mut Vec<u8>, data: &[u8]) {
+    out.push(Ops::PushBytes as u8);
+    out.extend_from_slice(&(data.len() as u32).to_le_bytes());
+    out.extend_from_slice(data);
+}
+
 fn http_reply_with_body(body: &[u8]) -> Vec<u8> {
     Encode!(&HttpResponse {
         status: 200_u128,
@@ -513,6 +527,47 @@ fn eval(ops_bytes: OpsBytes) {
                     core::arch::wasm32::memory_grow(0, pages as usize);
                 }
             }
+            Ops::LoopUntilGlobalDataSet => {
+                // Pop in the reverse of the builder's push order (trigger then reply).
+                let reply_data = stack.pop_blob();
+                let trigger = stack.pop_blob();
+                if get_global() == trigger {
+                    // The awaited condition holds: reply and stop looping.
+                    api::reply_data_append(&reply_data);
+                    api::reply();
+                } else {
+                    // The condition does not hold yet: make a management-canister
+                    // `canister_status` call for this canister itself, and re-run
+                    // this same op from both its reply and reject callbacks. This
+                    // forms a loop across message boundaries that only ends once
+                    // some other message sets the global data to `trigger`,
+                    // thereby delaying the reply to the original caller.
+                    let mut cb = Vec::new();
+                    encode_push_bytes(&mut cb, &trigger);
+                    encode_push_bytes(&mut cb, &reply_data);
+                    cb.push(Ops::LoopUntilGlobalDataSet as u8);
+                    // A single call yields either a reply or a reject (never
+                    // both), so the same callback env can back both handlers.
+                    let env = add_callback(cb);
+                    let arg = Encode!(&CanisterIdRecord {
+                        canister_id: Principal::from_slice(&api::id()),
+                    })
+                    .unwrap();
+                    api::call_new(
+                        Principal::management_canister().as_slice(),
+                        b"canister_status",
+                        callback,
+                        env,
+                        callback,
+                        env,
+                    );
+                    api::call_data_append(&arg);
+                    let err_code = api::call_perform();
+                    if err_code != 0 {
+                        api::trap_with("call_perform failed in LoopUntilGlobalDataSet")
+                    }
+                }
+            }
         }
     }
 }

rs/tests/message_routing/xnet/subnet_delete_test.rs

@@ -847,6 +847,21 @@ impl PayloadBuilder {
         self
     }
 
+    /// Loops indefinitely — each iteration performing a management-canister
+    /// `canister_status` call for the executing canister — until the canister's
+    /// global data equals `trigger`, and then replies with `reply`.
+    ///
+    /// Because each loop iteration is an inter-canister call, the reply to the
+    /// caller is delayed across message boundaries until some other message
+    /// sets the global data to `trigger` (e.g. via `set_global_data`). This is
+    /// useful for delaying a response until an external condition is met.
+    pub fn loop_until_global_data_set(mut self, trigger: &[u8], reply: &[u8]) -> Self {
+        self = self.push_bytes(trigger);
+        self = self.push_bytes(reply);
+        self.0.push(Ops::LoopUntilGlobalDataSet as u8);
+        self
+    }
+
     pub fn build(self) -> Vec<u8> {
         self.0
     }