diff --git a/Cargo.lock b/Cargo.lock index 011c971a5386..f1c74c39e741 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -18050,6 +18050,7 @@ dependencies = [ "canister-test", "futures", "ic-agent 0.49.0", + "ic-config", "ic-error-types 0.2.0", "ic-management-canister-types-private", "ic-nns-constants", diff --git a/rs/state_machine_tests/tests/subnet_delete.rs b/rs/state_machine_tests/tests/subnet_delete.rs index b6676ed1b482..5ce6695a823e 100644 --- a/rs/state_machine_tests/tests/subnet_delete.rs +++ b/rs/state_machine_tests/tests/subnet_delete.rs @@ -1,13 +1,15 @@ //! `StateMachine` analogue of the system test //! `rs/tests/message_routing/xnet/subnet_delete_test.rs`. //! -//! It verifies that deleting a subnet correctly causes in-flight XNet requests -//! to be rejected, and that messages from the deleted subnet that are still in -//! its stream are not pulled after subnet deletion. The scenario is run as a -//! matrix over the type of the deleted subnet: it must behave the same whether -//! the deleted subnet is a CloudEngine subnet or a regular application subnet -//! (subnet deletion is handled by the routing/topology layer, which is -//! subnet-type agnostic). +//! It verifies that deleting a subnet correctly handles in-flight XNet messages: +//! requests towards the deleted subnet are rejected, responses towards the +//! deleted subnet are silently dropped (a response is never turned into a +//! reject), and messages from the deleted subnet that are still in its stream +//! are not pulled after subnet deletion. The scenario is run as a matrix over +//! the type of the deleted subnet: it must behave the same whether the deleted +//! subnet is a CloudEngine subnet or a regular application subnet (subnet +//! deletion is handled by the routing/topology layer, which is subnet-type +//! agnostic). //! //! The scenario wires together three `StateMachine`s sharing a single registry //! (see `rs/state_machine_tests/tests/multi_subnet.rs` for the basic wiring): @@ -21,25 +23,47 @@ //! implements subnet deletion). //! //! Runbook: -//! 1. Install universal canisters `US` on `S`, `UT` on `T`, `UC` on `C`. -//! 2. Halt `T` (stop executing rounds on it). -//! 3. From `UC` fire a bounded-wait call to `UT` that would set `UT`'s -//! global data to a fixed blob. The call is fire-and-forget (`UC` replies -//! to its ingress immediately) and remains stuck in the `C -> T` stream. -//! 4. Halt `C`. -//! 5. From `US` submit 10 bounded-wait calls to `UC` with a 2 MB payload each -//! (generated at runtime, so the ingress stays small). Each call's -//! `on_reject` handler replies with the reject code as a 4-byte LE integer. -//! The 2 MB payloads fill the `S -> C` stream (`TARGET_STREAM_SIZE_BYTES`), -//! so some calls reach the stream while the rest stay in `US`'s output queue. -//! 6. Delete `C`, unhalt `T`, check `UT`'s global data is still empty, and wait -//! for all 10 calls from `US` to complete. -//! 7. Assert at least one call from `US` was rejected with `DestinationInvalid` -//! (call did not reach the stream: no route after deletion) and at least one -//! with `CanisterReject` (call reached the stream, so its callback was still -//! open when `C` was deleted; `generate_reject_responses_for_deleted_subnets` -//! then synthesizes an immediate reject for it, rather than waiting for the -//! bounded-wait callback to time out). +//! ```text +//! 1. Install universal canisters US on S, UT on T, UC on C, plus two more +//! canisters US2 and US3 on S (for the response-dropping part). +//! 2. Halt T (stop executing rounds on it). +//! 3. From UC fire a bounded-wait call to UT that would set UT's global data to +//! a fixed blob. The call is fire-and-forget (UC replies to its ingress +//! immediately) and remains stuck in the C -> T stream. Also from UC fire two +//! bounded-wait calls to US2 and US3 on S. Each callee loops (making +//! `canister_status` calls) holding its reply back until its global data is +//! set. Both calls are fire-and-forget. +//! 4. Halt C. +//! 5. Release US2's reply (set its global data). Its response is destined for UC +//! on the halted subnet C; C does not consume it and, since the S -> C stream +//! is still empty, it is inducted into that stream. +//! 6. From US submit 10 bounded-wait calls to UC with a 2 MB payload each +//! (generated at runtime, so the ingress stays small). Each call's on_reject +//! handler replies with the reject code as a 4-byte LE integer. The 2 MB +//! payloads fill the S -> C stream (TARGET_STREAM_SIZE_BYTES), so some calls +//! reach the stream while the rest stay in US's output queue. +//! 7. Drive S until all 10 calls are in flight; at this point the S -> C stream +//! is full (TARGET_STREAM_SIZE_BYTES) and the remaining calls stay in US's +//! output queue. +//! 8. Release US3's reply. As the S -> C stream is now full, its response cannot +//! be inducted and stays in US3's output queue. +//! 9. Delete C, unhalt T, check UT's global data is still empty (T must not pull +//! messages from the deleted C's stream), and wait for all 10 calls from US to +//! complete. (Unlike the system test, T observes the post-deletion registry +//! version synchronously — applied in Step 9a — so there is no separate wait +//! for it.) +//! 10. Assert at least one call from US was rejected with DestinationInvalid +//! (call did not reach the stream: no route after deletion) and at least one +//! with CanisterReject (call reached the stream, so its callback was still +//! open when C was deleted; generate_reject_responses_for_deleted_subnets +//! then synthesizes an immediate reject for it, rather than waiting for the +//! bounded-wait callback to time out). +//! 11. Assert both responses towards C were silently dropped: the one that sat in +//! US3's output queue is dropped by the stream builder (counted by +//! mr_routed_message_count{type="response",status="canister_not_found"}), and +//! the one that sat in the S -> C stream is dropped together with the whole +//! stream (metric-silent, asserted by the stream being gone). +//! ``` //! //! Bounded-wait (best-effort) calls with no cycles are used throughout because //! they are the only cross-subnet calls allowed to/from a CloudEngine subnet; @@ -58,6 +82,7 @@ use ic_state_machine_tests::{ add_initial_registry_records, remove_subnet_local_registry_records, update_global_registry_records, }; +use ic_test_utilities_metrics::fetch_int_counter_vec; use ic_types::ingress::{IngressState, IngressStatus, WasmResult}; use ic_types::messages::{MessageId, StreamMessage}; use ic_types::{CanisterId, PrincipalId, RegistryVersion, SubnetId}; @@ -70,11 +95,28 @@ const NUM_CALLS: usize = 10; /// (`2 MiB`) and `6` such payloads exceed `TARGET_STREAM_SIZE_BYTES` (`10 MiB`), /// so the `S -> C` stream fills up and only some of the calls reach it. const PAYLOAD_SIZE: u32 = 2 * 1000 * 1000; -/// Best-effort call timeout. Chosen comfortably larger than the number of rounds -/// executed before deleting `C`, so that calls do not time out prematurely, yet -/// small enough to keep the post-deletion round count modest. -const CALL_TIMEOUT_SECS: u32 = 30; +/// Best-effort call timeout. `with_execute_round_time_increment` advances each +/// subnet's clock by 1s per round, so a best-effort call/response times out after +/// this many rounds. It is chosen comfortably larger than the number of rounds +/// executed while any best-effort call or response is still pending (in +/// particular the two responses towards `C`, which are produced early yet must +/// survive until `C` is deleted), so that nothing times out prematurely; it is +/// still smaller than the post-deletion round budget, so a regression that fails +/// to reject/drop in-flight messages lets them time out within that budget: the +/// calls complete (via a timeout reject) and the failure surfaces as a mismatched +/// reject code in the Step 10 assertions, rather than as the (bounded) +/// completion-wait loop exhausting its rounds with calls still pending. +const CALL_TIMEOUT_SECS: u32 = 120; const FIXED_BLOB: &[u8] = b"cloud-engine-test-fixed-blob"; +/// Blobs used by the response-dropping scenario. A looping canister on `S` holds +/// its reply back until its global data is set to `RESPONSE_TRIGGER_BLOB`, then +/// replies with the given reply blob (the payload of the response that must be +/// silently dropped once `C` is deleted). Two responses are produced: one that +/// ends up in `S`'s stream towards `C`, and one that stays in the sender +/// canister's output queue (because the stream is already full). +const RESPONSE_TRIGGER_BLOB: &[u8] = b"subnet-delete-response-trigger"; +const RESPONSE_STREAM_REPLY_BLOB: &[u8] = b"subnet-delete-response-in-stream"; +const RESPONSE_QUEUE_REPLY_BLOB: &[u8] = b"subnet-delete-response-in-queue"; /// Reject codes (see `ic_error_types::RejectCode`). const DESTINATION_INVALID: u32 = 3; @@ -147,6 +189,88 @@ fn ingress_result(sm: &StateMachine, msg_id: &MessageId) -> Option { } } +/// Fires a fire-and-forget bounded-wait call from `uc` on subnet `c` to +/// `callee`'s `update` method running the looping op: `callee` holds its reply +/// back (looping on `canister_status` calls) until its global data equals +/// `RESPONSE_TRIGGER_BLOB`, then replies with `reply_blob`. `uc` replies to its +/// own ingress immediately; `c` is driven until that ingress completes, so the +/// resulting request is placed in `c`'s outgoing stream towards `callee`'s subnet. +fn fire_looping_call( + c: &StateMachine, + uc: CanisterId, + callee: CanisterId, + user_id: PrincipalId, + reply_blob: &[u8], +) { + let payload = wasm() + .call_simple_with_cycles_and_best_effort_response( + callee, + "update", + call_args() + .other_side(wasm().loop_until_global_data_set(RESPONSE_TRIGGER_BLOB, reply_blob)) + .on_reply(wasm().noop()) + .on_reject(wasm().noop()), + 0_u128, + CALL_TIMEOUT_SECS, + ) + .reply_data(&[]) + .build(); + let msg_id = c.submit_ingress_as(user_id, uc, "update", payload).unwrap(); + let mut completed = false; + for _ in 0..10 { + c.execute_round(); + if ingress_result(c, &msg_id).is_some() { + completed = true; + break; + } + } + assert!( + completed, + "UC fire-and-forget looping ingress did not complete" + ); +} + +/// Releases a looping canister's held-back reply by setting its global data to +/// `RESPONSE_TRIGGER_BLOB` (fire-and-forget; the canister's loop then replies on +/// its next iteration). +fn release_looping_reply(sm: &StateMachine, canister: CanisterId, user_id: PrincipalId) { + let payload = wasm() + .set_global_data(RESPONSE_TRIGGER_BLOB) + .reply_data(&[]) + .build(); + sm.submit_ingress_as(user_id, canister, "update", payload) + .unwrap(); +} + +/// Counts the `Response` messages in `sm`'s stream towards `remote` (0 if there +/// is no such stream). +fn stream_response_count(sm: &StateMachine, remote: SubnetId) -> usize { + sm.get_latest_state() + .get_stream(&remote) + .map(|stream| { + stream + .messages() + .iter() + .filter(|(_, m)| matches!(m, StreamMessage::Response(_))) + .count() + }) + .unwrap_or(0) +} + +/// Returns the number of responses that the stream builder on `sm` has silently +/// dropped because their destination canister had no known route, i.e. the value +/// of `mr_routed_message_count{type="response",status="canister_not_found"}`. +fn dropped_no_route_response_count(sm: &StateMachine) -> u64 { + fetch_int_counter_vec(sm.metrics_registry(), "mr_routed_message_count") + .into_iter() + .filter(|(labels, _)| { + labels.get("type").map(String::as_str) == Some("response") + && labels.get("status").map(String::as_str) == Some("canister_not_found") + }) + .map(|(_, value)| value) + .sum() +} + // The subnet-deletion scenario is run once per supported type of the deleted // subnet `C`; the observable behavior must be identical. Each type gets its own // test so they run in parallel and failures are easy to attribute. @@ -211,8 +335,13 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet t.reload_registry(); c.reload_registry(); - // Step 1: Install universal canisters US on S, UT on T, UC on C. + // Step 1: Install universal canisters US on S, UT on T, UC on C, plus two + // more canisters US2 and US3 on S dedicated to the response-dropping + // scenario: US2 produces the response that ends up in S's stream towards C, + // US3 the response that stays in US3's output queue. let us = install_universal_canister(&s); + let us2 = install_universal_canister(&s); + let us3 = install_universal_canister(&s); let ut = install_universal_canister(&t); let uc = install_universal_canister(&c); @@ -275,9 +404,65 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet } drop(c_state); + // Also fire two more fire-and-forget bounded-wait calls from UC to US2 + // and US3 on the surviving subnet S. Each callee loops (making + // `canister_status` calls) holding its reply back until its global data is + // set (Steps 5, 8). + // Both requests are placed in the C -> S stream while C is still running, so + // S can induct them (and start the loops) before C is deleted. Their eventual + // replies are *responses* destined for UC on the deleted subnet C. + fire_looping_call(&c, uc, us2, user_id, RESPONSE_STREAM_REPLY_BLOB); + fire_looping_call(&c, uc, us3, user_id, RESPONSE_QUEUE_REPLY_BLOB); + // Step 4: Halt subnet C by not executing any more rounds on it. - // Step 5: Submit 10 bounded-wait calls from US to UC, each producing a 2 MB + // Step 5: Release US2's reply by setting its global data, and drive S until + // the resulting response reaches the S -> C stream. This is the first time S + // runs since Step 3, so the drive inducts the two C -> S requests and starts + // US2's and US3's loops, letting US2 observe its global data and reply. C is + // halted, so it does not consume the response; and the S -> C stream is still + // empty (Step 6 has not run yet), so the response is inducted into it. + release_looping_reply(&s, us2, user_id); + let mut response_in_stream = false; + for _ in 0..50 { + s.execute_round(); + if stream_response_count(&s, c_id) >= 1 { + response_in_stream = true; + break; + } + } + assert!( + response_in_stream, + "US2's response did not reach the S -> C stream" + ); + // The S -> C stream must hold exactly US2's response and nothing else yet. + let state = s.get_latest_state(); + let stream = state + .get_stream(&c_id) + .expect("expected an S -> C stream holding US2's response"); + let stream_messages: Vec<&StreamMessage> = stream.messages().iter().map(|(_, m)| m).collect(); + assert_eq!( + stream_messages.len(), + 1, + "expected exactly US2's response in the S -> C stream, got {} messages", + stream_messages.len() + ); + match stream_messages[0] { + StreamMessage::Response(resp) => { + assert_eq!( + resp.respondent, us2, + "unexpected respondent of the streamed response" + ); + assert_eq!( + resp.originator, uc, + "unexpected originator of the streamed response" + ); + } + other => panic!("expected a Response in the S -> C stream, got {other:?}"), + } + drop(state); + + // Step 6: Submit 10 bounded-wait calls from US to UC, each producing a 2 MB // payload at runtime (the ingress itself stays small). The on_reject handler // replies with the reject code as a 4-byte LE integer. let us_uc_payload = wasm() @@ -305,10 +490,14 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet }) .collect(); - // Execute rounds on S until all calls have been performed by US (i.e. all - // ingress messages are being processed, waiting for their callbacks). At this - // point the S -> C stream has filled up to TARGET_STREAM_SIZE_BYTES and the - // remaining calls stay in US's output queue. + // Step 7: Execute rounds on S until all calls have been performed by US (i.e. + // all ingress messages are being processed, waiting for their callbacks). Each + // round the stream builder moves queued messages into the S -> C stream but + // stops adding to it once its byte size reaches the soft limit + // TARGET_STREAM_SIZE_BYTES (the check is `count_bytes() >= + // TARGET_STREAM_SIZE_BYTES`, evaluated before each message, so the last one may + // push it slightly over). By the time all calls are processing the stream is + // full to that target, so the remaining calls stay in US's output queue. let mut all_processing = false; for _ in 0..50 { s.execute_round(); @@ -332,8 +521,9 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet // Assert that the S -> C stream is partially filled: some (but not all) // US -> UC calls reached the stream and the rest are still in US's output - // queue. This partition is what yields both reject codes in step 7 (streamed - // calls -> CanisterReject, queued calls -> DestinationInvalid). + // queue. This partition is what yields both reject codes in step 10 (streamed + // calls -> CanisterReject, queued calls -> DestinationInvalid). The stream + // also still holds the single US2 response from Step 5. let state = s.get_latest_state(); let stream = state .get_stream(&c_id) @@ -349,10 +539,19 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet ) }) .count(); + let response_count = stream + .messages() + .iter() + .filter(|(_, m)| matches!(m, StreamMessage::Response(_))) + .count(); assert_eq!( - stream_request_count, + response_count, 1, + "expected exactly US2's response among the S -> C stream messages" + ); + assert_eq!( + stream_request_count + response_count, stream.messages().len(), - "expected only US -> UC requests in the S -> C stream" + "expected only US -> UC requests and US2's response in the S -> C stream" ); assert!( (1..NUM_CALLS).contains(&stream_request_count), @@ -365,7 +564,29 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet ); drop(state); - // Step 6a: Delete subnet C. We remove it from the shared pool of subnets + // Step 8: Release US3's reply, now that the S -> C stream is full. US3's + // response towards UC on C therefore cannot be inducted into the stream and + // stays in US3's output queue. We drive S enough rounds for US3's loop to + // reply, then assert the stream is unchanged (still exactly one response, the + // US2 one from Step 5), confirming US3's response did not enter the stream. + release_looping_reply(&s, us3, user_id); + for _ in 0..20 { + s.execute_round(); + } + assert_eq!( + stream_response_count(&s, c_id), + 1, + "US3's response must stay in US3's output queue, not enter the full S -> C stream" + ); + assert!( + s.get_latest_state() + .canister_state(&us3) + .unwrap() + .has_output(), + "expected US3's response in US3's output queue before deleting C" + ); + + // Step 9a: Delete subnet C. We remove it from the shared pool of subnets // (making it unreachable for S and T) and tombstone its registry records, // matching how PocketIC implements subnet deletion. let next_version = RegistryVersion::new(registry_data_provider.latest_version().get() + 1); @@ -387,11 +608,18 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet s.reload_registry(); t.reload_registry(); - // Step 6b: Unhalt subnet T and verify that it does not pull the messages from - // the deleted subnet C: UT's global data must stay empty. + // Step 9b: Unhalt subnet T by resuming round execution on it. + // + // Step 9c: No analog here. The system test waits for T to observe the registry + // version at which C was deleted; in this test `reload_registry()` in Step 9a + // applied that version to T synchronously, so T already sees C as deleted by + // the time it runs. for _ in 0..5 { t.execute_round(); } + + // Step 9d: Verify T does not pull the messages from the deleted subnet C: + // UT's global data must stay empty. let global_data = t .query( ut, @@ -408,7 +636,7 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet WasmResult::Reject(reject) => panic!("unexpected reject querying UT: {reject}"), } - // Step 6c: Drive S until all 10 calls from US complete. The calls still in + // Step 9e: Drive S until all 10 calls from US complete. The calls still in // US's output queue are rejected with DestinationInvalid (no route to C), // while the calls already in the S -> C stream (whose callback is still // open) get an immediate synthetic CanisterReject once C disappears from @@ -429,7 +657,7 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet } assert!(done, "not all US -> UC calls completed"); - // Step 7: Assert that all calls were rejected and that both reject codes occur. + // Step 10: Assert that all calls were rejected and that both reject codes occur. let mut destination_invalid_count = 0_usize; let mut canister_reject_count = 0_usize; for result in results { @@ -461,4 +689,29 @@ fn xnet_messages_rejected_after_subnet_deletion_impl(deleted_subnet_type: Subnet "expected at least one CanisterReject rejection, got {canister_reject_count} \ (deleted subnet type: {deleted_subnet_type:?})" ); + + // Step 11: Verify the two responses towards C were silently dropped. + // + // The response that sat in US3's output queue (Step 8) is dropped by the + // stream builder when it finds no route to C, incrementing + // `mr_routed_message_count{type="response",status="canister_not_found"}`. + // (The queued US -> UC requests increment the same counter with + // `type="request"` and are additionally rejected, see Step 10.) Unlike a + // request, a response is never turned into a reject. + assert_eq!( + dropped_no_route_response_count(&s), + 1, + "expected exactly the queued US3 response to be silently dropped as no_route \ + (deleted subnet type: {deleted_subnet_type:?})" + ); + + // The response that sat in the S -> C stream (Step 5) is dropped together + // with the whole stream when C is deleted; that bulk discard is intentionally + // metric-silent, so we assert it directly: S no longer has a stream towards + // the deleted subnet C. + assert!( + s.get_latest_state().get_stream(&c_id).is_none(), + "expected S's stream towards the deleted subnet C to be discarded \ + (deleted subnet type: {deleted_subnet_type:?})" + ); } diff --git a/rs/tests/message_routing/xnet/BUILD.bazel b/rs/tests/message_routing/xnet/BUILD.bazel index 30da303f61ae..212f55359429 100644 --- a/rs/tests/message_routing/xnet/BUILD.bazel +++ b/rs/tests/message_routing/xnet/BUILD.bazel @@ -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", diff --git a/rs/tests/message_routing/xnet/Cargo.toml b/rs/tests/message_routing/xnet/Cargo.toml index 08845f953a32..532ff0f8e711 100644 --- a/rs/tests/message_routing/xnet/Cargo.toml +++ b/rs/tests/message_routing/xnet/Cargo.toml @@ -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" } diff --git a/rs/tests/message_routing/xnet/subnet_delete_test.rs b/rs/tests/message_routing/xnet/subnet_delete_test.rs index 392ad38fff81..e9720ce4155c 100644 --- a/rs/tests/message_routing/xnet/subnet_delete_test.rs +++ b/rs/tests/message_routing/xnet/subnet_delete_test.rs @@ -1,39 +1,61 @@ /* tag::catalog[] Title:: Subnet deletion with in-flight XNet messages. -Goal:: Verify that deleting a subnet correctly causes in-flight XNet messages to -be rejected, and that messages from the deleted subnet that are still in its -stream are not pulled after subnet deletion. The scenario is run once per type of -the deleted subnet (as two separate tests sharing the same IC): it must behave -the same whether the deleted subnet is a CloudEngine subnet or a regular -Application subnet (subnet deletion is handled by the routing/topology layer, -which is subnet-type agnostic). +Goal:: Verify that deleting a subnet correctly handles in-flight XNet messages: +requests towards the deleted subnet are rejected, responses towards the deleted +subnet are silently dropped (a response can never be rejected), and messages +from the deleted subnet that are still in its stream are not pulled after subnet +deletion. The scenario is run once per type of the deleted subnet (as two +separate tests sharing the same IC): it must behave the same whether the deleted +subnet is a CloudEngine subnet or a regular Application subnet (subnet deletion +is handled by the routing/topology layer, which is subnet-type agnostic). Runbook:: 0. Set up an IC with an NNS subnet, three Application subnets S, T and C_app, and one CloudEngine subnet C_cloud. Then, in each of the two tests, pick the subnet matching that test's deleted-subnet type as the deleted subnet C (C_cloud or C_app) and run the following scenario: -1. Install universal canisters US on S, UT on T, UC on C. +1. Install universal canisters US on S, UT on T, UC on C, plus two more + canisters US2 and US3 on S (used for the response-dropping scenario). 2. Halt T. Wait until T makes no progress. 3. From UC fire a bounded-wait call to UT that would set UT's global data to a fixed blob. The call is fire-and-forget (UC replies to its ingress - immediately) and remains stuck in the C -> T stream (T is halted). + immediately) and remains stuck in the C -> T stream (T is halted). Also from + UC fire two bounded-wait calls to US2 and US3 (on the surviving subnet S). + Each callee holds its reply back (looping on `canister_status` calls) until + its global data is set. Both calls are fire-and-forget; US2 and US3 start + looping. 4. Halt C. Wait until C makes no progress. -5. From US submit 10 bounded-wait calls to UC with 2 MB payload each +5. Set US2's global data, releasing its reply. The reply is a response + destined for UC on the halted subnet C; since C is halted it is not consumed + and, as the S -> C stream is still empty, it is inducted into that stream. + Wait until it shows up in the stream. +6. From US submit 10 bounded-wait calls to UC with 2 MB payload each (generated at runtime, ingress stays small). Each call's on_reject handler replies with the reject code as a 4-byte LE integer. The 2 MB payloads fill the S -> C stream, so some calls reach the stream while the rest stay in US's output queue. -6. Delete C, unhalt T, verify T's registry version is the version at which C +7. Wait until the S -> C stream is full (its byte size reaches + TARGET_STREAM_SIZE_BYTES); C is halted so the stream never drains. +8. Set US3's global data, releasing its reply. As the stream is now full, this + second response towards C cannot be inducted and stays in US3's output queue. +9. Delete C, unhalt T, verify T's registry version is the version at which C was deleted, check UT's global data is still empty (T must not pull messages from the deleted C's stream), and wait for all 10 calls from US to complete. -7. Assert at least one call from US was rejected with DestinationInvalid +10. Assert at least one call from US was rejected with DestinationInvalid (call did not reach the stream: no route after deletion) and at least one with CanisterReject (call reached the stream, so its callback was still open when C was deleted; the destination subnet's disappearance triggers an immediate synthetic reject for it, rather than waiting for the bounded-wait callback to time out). +11. Assert (via the `mr_routed_message_count{type="response", + status="canister_not_found"}` metric on S) that at least one response towards + C was silently dropped: this is the response that sat in US3's output queue + (Step 8), dropped by the stream builder when the route to C disappeared, + without producing any reject. The response that sat in the S -> C stream + (Step 5) is dropped together with the whole stream on deletion, which is + intentionally metric-silent; it is covered indirectly (any panic on S would + fail the test via the unallowed "panicked" log pattern). Note:: Bounded-wait (best-effort) calls with no cycles are used throughout because they are the only cross-subnet calls allowed to/from a CloudEngine @@ -48,6 +70,7 @@ end::catalog[] */ use anyhow::{Result, bail}; use candid::{Decode, Encode}; use futures::future::join_all; +use ic_config::message_routing::TARGET_STREAM_SIZE_BYTES; use ic_consensus_system_test_utils::rw_message::cert_state_makes_no_progress_with_retries; use ic_management_canister_types_private::{ IC_00, Method, Payload, SubnetInfoArgs, SubnetInfoResponse, @@ -58,12 +81,15 @@ use ic_system_test_driver::driver::group::SystemTestGroup; use ic_system_test_driver::driver::ic::{InternetComputer, Subnet}; use ic_system_test_driver::driver::test_env::TestEnv; use ic_system_test_driver::driver::test_env_api::{ - HasPublicApiUrl, HasRegistryVersion, HasTopologySnapshot, IcNodeContainer, READY_WAIT_TIMEOUT, - RETRY_BACKOFF, SubnetSnapshot, install_registry_canister_with_testnet_topology, + HasPublicApiUrl, HasRegistryVersion, HasTopologySnapshot, IcNodeContainer, IcNodeSnapshot, + READY_WAIT_TIMEOUT, RETRY_BACKOFF, SubnetSnapshot, + install_registry_canister_with_testnet_topology, }; use ic_system_test_driver::retry_with_msg_async; use ic_system_test_driver::systest; -use ic_system_test_driver::util::{UniversalCanister, assert_create_agent, block_on}; +use ic_system_test_driver::util::{ + MetricsFetcher, UniversalCanister, assert_create_agent, block_on, +}; use ic_types::SubnetId; use ic_universal_canister::{call_args, wasm}; use registry_canister::init::RegistryCanisterInitPayloadBuilder; @@ -74,11 +100,22 @@ use std::time::Duration; const NUM_NODES: usize = 4; const CALL_TIMEOUT_SECS: u32 = 300; const FIXED_BLOB: &[u8] = b"cloud-engine-test-fixed-blob"; +// Blobs used by the response-dropping scenario. A looping canister on S holds +// back its reply until its global data is set to `RESPONSE_TRIGGER_BLOB`, then +// replies with the given reply blob (the payload of the response that must be +// silently dropped once C is deleted). Two responses are produced: one that ends +// up in S's *stream* towards C, and one that stays in the sender canister's +// *output queue* (because the stream is already full). +const RESPONSE_TRIGGER_BLOB: &[u8] = b"subnet-delete-response-trigger"; +const RESPONSE_STREAM_REPLY_BLOB: &[u8] = b"subnet-delete-response-in-stream"; +const RESPONSE_QUEUE_REPLY_BLOB: &[u8] = b"subnet-delete-response-in-queue"; // The two tests run the scenario once each (per deleted-subnet type). A single -// run takes ~115s in practice; the per-test timeout leaves ample headroom while -// staying below the bazel default `test_timeout` ("long" = 900s), so an in-test -// timeout fires with a clean error before bazel hard-kills the target. The -// overall timeout covers the shared setup plus both tests running sequentially. +// run takes on the order of a couple of minutes in practice (halting subnets, +// filling the S->C stream and scraping metrics); the per-test timeout leaves +// ample headroom while staying below the bazel default `test_timeout` ("long" = +// 900s), so an in-test timeout fires with a clean error before bazel hard-kills +// the target. The overall timeout covers the shared setup plus both tests +// running sequentially. const PER_TEST_TIMEOUT: Duration = Duration::from_secs(300); const OVERALL_TIMEOUT: Duration = Duration::from_secs(600); @@ -240,18 +277,27 @@ async fn run_scenario( let t_agent = assert_create_agent(t_node.get_public_url().as_str()).await; let c_agent = assert_create_agent(c_node.get_public_url().as_str()).await; - // Step 1: Install universal canisters US on S, UT on T, UC on C. + // Step 1: Install universal canisters US on S, UT on T, UC on C, plus two + // more canisters US2 and US3 on S dedicated to the response-dropping + // scenario: US2 produces the response that ends up in S's stream towards C, + // US3 the response that stays in US3's output queue. slog::info!(logger, "Step 1: Installing universal canisters on S, T, C"); let us = UniversalCanister::new_with_retries(&s_agent, s_node.effective_canister_id(), logger).await; + let us2 = + UniversalCanister::new_with_retries(&s_agent, s_node.effective_canister_id(), logger).await; + let us3 = + UniversalCanister::new_with_retries(&s_agent, s_node.effective_canister_id(), logger).await; let ut = UniversalCanister::new_with_retries(&t_agent, t_node.effective_canister_id(), logger).await; let uc = UniversalCanister::new_with_retries(&c_agent, c_node.effective_canister_id(), logger).await; slog::info!( logger, - "Step 1 done: US={}, UT={}, UC={}", + "Step 1 done: US={}, US2={}, US3={}, UT={}, UC={}", us.canister_id(), + us2.canister_id(), + us3.canister_id(), ut.canister_id(), uc.canister_id(), ); @@ -290,7 +336,24 @@ async fn run_scenario( ) .await .expect("UC fire-and-forget call to UT should succeed"); - slog::info!(logger, "Step 3 done: UC->UT fire-and-forget call fired"); + + // Also fire two bounded-wait calls from UC to US2 and US3 (on the surviving + // sender subnet S). Each callee holds back its reply (by looping on + // `canister_status` calls) until its global data is set to + // RESPONSE_TRIGGER_BLOB. UC replies to its ingress immediately + // (fire-and-forget). Both C and S are still running, so US2/US3 receive the + // requests and start looping; their eventual replies will be *responses* + // destined for UC on C (see Steps 5, 8, 9a). + slog::info!( + logger, + "Step 3: Firing bounded-wait UC->US2 and UC->US3 looping calls (fire-and-forget)" + ); + fire_looping_call(&uc, us2.canister_id(), RESPONSE_STREAM_REPLY_BLOB).await; + fire_looping_call(&uc, us3.canister_id(), RESPONSE_QUEUE_REPLY_BLOB).await; + slog::info!( + logger, + "Step 3 done: UC->UT, UC->US2 and UC->US3 fire-and-forget calls fired" + ); // Step 4: Halt subnet C and wait until C makes no progress. slog::info!(logger, "Step 4: Halting subnet C ({})", c_subnet.subnet_id); @@ -304,14 +367,43 @@ async fn run_scenario( ); slog::info!(logger, "Step 4 done: subnet C is halted"); - // Step 5: Submit 10 bounded-wait calls from US to UC each with 2 MB payload + // Step 5: Now that C is halted, release US2's held-back reply by setting its + // global data to RESPONSE_TRIGGER_BLOB. US2's loop then replies; that reply is + // a *response* destined for UC on the (halted, soon to be deleted) subnet C. + // Because C is halted it does not consume the response, and because the S->C + // stream is still empty (Step 6 has not run yet) the response is inducted into + // S's stream towards C. When C is deleted (Step 9a) this streamed response + // must be silently dropped (its whole stream is discarded): unlike the + // requests in Steps 6/10 (which are rejected), a response can never be + // rejected. + slog::info!( + logger, + "Step 5: Releasing US2's reply (response into the S->C stream)" + ); + us2.update( + wasm() + .set_global_data(RESPONSE_TRIGGER_BLOB) + .reply_data(&[]), + ) + .await + .expect("setting US2 global data should succeed"); + // Wait until the response is actually inducted into S's stream towards C + // (before any of the large Step 6 payloads fill that stream). + wait_for_stream_gauge_at_least(&s_node, "mr_stream_messages", c_subnet.subnet_id, 1, logger) + .await; + slog::info!( + logger, + "Step 5 done: US2 reply is in the S->C stream (>= 1 message)" + ); + + // Step 6: Submit 10 bounded-wait calls from US to UC each with 2 MB payload // (generated at runtime on S; the ingress itself is small). // The on_reject handler replies with the reject code as a 4-byte LE integer. - // We submit all 10 calls before step 6 so they are in-flight in the S->C + // We submit all 10 calls before step 9 so they are in-flight in the S->C // stream before C is deleted. slog::info!( logger, - "Step 5: Submitting 10 bounded-wait US->UC calls (2 MB payload each)" + "Step 6: Submitting 10 bounded-wait US->UC calls (2 MB payload each)" ); let us_uc_wasm: Vec = wasm() .call_simple_with_cycles_and_best_effort_response( @@ -354,18 +446,68 @@ async fn run_scenario( } })) .await; - slog::info!(logger, "Step 5 done: {} pending", us_uc_request_ids.len()); + slog::info!(logger, "Step 6 done: {} pending", us_uc_request_ids.len()); + + // Step 7: Wait until S's stream towards C is full. Each round the stream + // builder moves queued messages into the S -> C stream, but stops adding to it + // once its byte size reaches the soft limit TARGET_STREAM_SIZE_BYTES (the + // check is `count_bytes() >= TARGET_STREAM_SIZE_BYTES`, evaluated before each + // message, so the last one may push it slightly over). We detect this via the + // `mr_stream_bytes` gauge. Once the stream is full the remaining large + // requests stay in US's output queue, and any further message (the Step 8 + // response) will stay in its sender's output queue too. C is halted, so it + // never garbage-collects the stream and the stream never drains below the + // target. + slog::info!( + logger, + "Step 7: Waiting for the S->C stream to fill ({} bytes)", + TARGET_STREAM_SIZE_BYTES, + ); + wait_for_stream_gauge_at_least( + &s_node, + "mr_stream_bytes", + c_subnet.subnet_id, + TARGET_STREAM_SIZE_BYTES as u64, + logger, + ) + .await; + slog::info!(logger, "Step 7 done: S->C stream is full"); + + // Step 8: Now that the stream is full, release US3's held-back reply. Its + // response towards UC on C cannot be inducted into the full S->C stream, so it + // stays in US3's output queue. When C is deleted (Step 9a) this queued + // response must be silently dropped by the stream builder (route to C gone), + // which — unlike the queued requests — happens without producing any reject. + slog::info!( + logger, + "Step 8: Releasing US3's reply (response into US3's output queue)" + ); + us3.update( + wasm() + .set_global_data(RESPONSE_TRIGGER_BLOB) + .reply_data(&[]), + ) + .await + .expect("setting US3 global data should succeed"); + slog::info!(logger, "Step 8 done: US3 reply released"); - // Step 6a: Delete subnet C. Snapshot the topology right before the deletion: + // Step 9a: Delete subnet C. Snapshot the topology right before the deletion: // `block_for_newer_registry_version` derives its target version from the // snapshot's registry version + 1, so the baseline must be captured before // the deletion, otherwise it could block forever waiting for a version that // only appears after the (later) unhalting of T. slog::info!( logger, - "Step 6a: Deleting subnet C ({})", + "Step 9a: Deleting subnet C ({})", c_subnet.subnet_id ); + // Baseline for the cumulative "silently dropped response" counter, captured + // before the deletion that triggers the drop. This counter carries over + // between matrix entries (shared IC and sender subnet S), so Step 11 asserts + // an increase relative to this baseline rather than an absolute threshold. + let dropped_responses_before = read_dropped_response_count(&s_node) + .await + .expect("failed to read the dropped response counter before deletion"); let topo_before_delete = env.topology_snapshot(); let delete_arg = DeleteSubnetPayload { subnet_id: c_subnet.subnet_id.get().into(), @@ -387,38 +529,38 @@ async fn run_scenario( let c_delete_registry_version = topo_after_delete.get_registry_version().get(); slog::info!( logger, - "Step 6a done: subnet C deleted at registry version {}", + "Step 9a done: subnet C deleted at registry version {}", c_delete_registry_version, ); - // Step 6b: Unhalt subnet T. + // Step 9b: Unhalt subnet T. slog::info!( logger, - "Step 6b: Unhalting subnet T ({})", + "Step 9b: Unhalting subnet T ({})", t_subnet.subnet_id ); set_subnet_halted(governance, t_subnet.subnet_id, false).await; - slog::info!(logger, "Step 6b done: subnet T unhalted"); + slog::info!(logger, "Step 9b done: subnet T unhalted"); - // Step 6c: Wait for T to observe the registry version at which C was deleted. + // Step 9c: Wait for T to observe the registry version at which C was deleted. slog::info!( logger, - "Step 6c: Waiting for subnet T to observe registry version {}", + "Step 9c: Waiting for subnet T to observe registry version {}", c_delete_registry_version, ); wait_for_subnet_registry_version(&ut, t_subnet.subnet_id, c_delete_registry_version, logger) .await; slog::info!( logger, - "Step 6c done: subnet T has observed registry version {}", + "Step 9c done: subnet T has observed registry version {}", c_delete_registry_version, ); - // Step 6d: Execute multiple update rounds on T, checking after each that UT's + // Step 9d: Execute multiple update rounds on T, checking after each that UT's // global data is still empty (T must not pull messages from the deleted C's stream). slog::info!( logger, - "Step 6d: Checking that UT global data remains empty over multiple rounds" + "Step 9d: Checking that UT global data remains empty over multiple rounds" ); for i in 0..5_usize { let global_data = ut @@ -432,12 +574,12 @@ async fn run_scenario( global_data.len() ); } - slog::info!(logger, "Step 6d done: UT global data is empty as expected"); + slog::info!(logger, "Step 9d done: UT global data is empty as expected"); - // Step 6e: Wait for all 10 calls from US to UC to complete. + // Step 9e: Wait for all 10 calls from US to UC to complete. slog::info!( logger, - "Step 6e: Waiting for all 10 US->UC calls to complete" + "Step 9e: Waiting for all 10 US->UC calls to complete" ); let us_uc_results = join_all( us_uc_request_ids @@ -445,13 +587,13 @@ async fn run_scenario( .map(|req_id| s_agent.wait(req_id, us.canister_id())), ) .await; - slog::info!(logger, "Step 6e done: all 10 US->UC calls completed"); + slog::info!(logger, "Step 9e done: all 10 US->UC calls completed"); - // Step 7: Assert at least one call from US was rejected with DestinationInvalid + // Step 10: Assert at least one call from US was rejected with DestinationInvalid // and at least one with CanisterReject. slog::info!( logger, - "Step 7: Asserting rejection codes for 10 US->UC calls" + "Step 10: Asserting rejection codes for 10 US->UC calls" ); let mut dest_invalid_count = 0_usize; let mut canister_reject_count = 0_usize; @@ -464,7 +606,7 @@ async fn run_scenario( bytes.len() ); let code = u32::from_le_bytes(bytes.try_into().unwrap()); - slog::info!(logger, "Step 7: US->UC reject code {}", code); + slog::info!(logger, "Step 10: US->UC reject code {}", code); match code { 3 => dest_invalid_count += 1, 4 => canister_reject_count += 1, @@ -473,7 +615,7 @@ async fn run_scenario( } slog::info!( logger, - "Step 7 done: DestinationInvalid={}, CanisterReject={}", + "Step 10 done: DestinationInvalid={}, CanisterReject={}", dest_invalid_count, canister_reject_count, ); @@ -487,6 +629,38 @@ async fn run_scenario( "Expected at least one CanisterReject rejection, got {canister_reject_count} \ (deleted subnet type: {deleted_subnet_type:?})" ); + // Step 11: Verify the two responses towards C were silently dropped. + // + // The response that was in US3's *output queue* (Step 8) is dropped by the + // stream builder when it finds no route to C; this increments + // `mr_routed_message_count{type="response",status="canister_not_found"}`, so + // we can assert on it directly. (The queued requests from Step 6 increment the + // same counter with `type="request"` and are additionally rejected, see Step + // 10.) + // + // The response that was already in the S->C *stream* (Step 5) is dropped + // together with the whole stream when C is deleted; that bulk discard is + // intentionally metric-silent, so it cannot be asserted via a counter. It is + // covered indirectly: a panic on S (e.g. from attempting to reject a response) + // would fail the test via the unallowed "panicked" log pattern. + slog::info!( + logger, + "Step 11: Verifying via metrics that the queued response towards C was dropped" + ); + // Assert the cumulative counter grew by at least 1 relative to the baseline + // captured before deletion (Step 9a): a bare `>= 1` check would spuriously + // pass in the second matrix entry off the first entry's leftover count. + let dropped_responses = + wait_for_dropped_response_count_at_least(&s_node, dropped_responses_before + 1, logger) + .await; + slog::info!( + logger, + "Step 11 done: {} response(s) towards C dropped as no_route (canister_not_found) \ + (was {} before deletion)", + dropped_responses, + dropped_responses_before, + ); + slog::info!( logger, "Scenario passed for deleted subnet type {:?}", @@ -494,6 +668,140 @@ async fn run_scenario( ); } +/// Fires a fire-and-forget, bounded-wait call from `uc` to `callee`'s `update` +/// method running the looping op: `callee` holds back its reply until its global +/// data equals `RESPONSE_TRIGGER_BLOB`, then replies with `reply_blob`. `uc` +/// replies to its own ingress immediately. Bounded-wait with no cycles is used +/// as elsewhere in the scenario (required to/from a CloudEngine subnet). +async fn fire_looping_call( + uc: &UniversalCanister<'_>, + callee: impl AsRef<[u8]>, + reply_blob: &[u8], +) { + uc.update( + wasm() + .call_simple_with_cycles_and_best_effort_response( + callee, + "update", + call_args() + .other_side( + wasm().loop_until_global_data_set(RESPONSE_TRIGGER_BLOB, reply_blob), + ) + .on_reply(wasm().noop()) + .on_reject(wasm().noop()), + 0_u128, + CALL_TIMEOUT_SECS, + ) + .reply_data(&[]), + ) + .await + .expect("UC fire-and-forget looping call should succeed"); +} + +/// Reads the value of a per-stream gauge (e.g. `mr_stream_messages` or +/// `mr_stream_bytes`) for the stream towards `remote` on `node`, returning 0 if +/// the corresponding labelled series does not exist yet (no such stream). +async fn read_stream_gauge( + node: &IcNodeSnapshot, + metric: &str, + remote: &str, +) -> anyhow::Result { + let map = MetricsFetcher::new(std::iter::once(node.clone()), vec![metric.to_string()]) + .fetch::() + .await + .map_err(|e| anyhow::anyhow!("failed to fetch {metric}: {e}"))?; + // The stream gauges are labelled with the destination subnet id (`remote`); + // match on the full `remote=""` label to avoid selecting a different + // series whose subnet id happens to contain `remote` as a substring. + let label_match = format!("remote=\"{remote}\""); + Ok(map + .iter() + .find(|(key, _)| key.contains(&label_match)) + .and_then(|(_, values)| values.first().copied()) + .unwrap_or(0)) +} + +/// Waits until the per-stream gauge `metric` for the stream towards `remote` on +/// `node` reaches at least `at_least`. +async fn wait_for_stream_gauge_at_least( + node: &IcNodeSnapshot, + metric: &str, + remote: SubnetId, + at_least: u64, + logger: &slog::Logger, +) { + let remote = remote.to_string(); + retry_with_msg_async!( + format!("waiting for {metric} towards {remote} to reach {at_least}"), + logger, + READY_WAIT_TIMEOUT, + RETRY_BACKOFF, + || async { + let value = read_stream_gauge(node, metric, &remote).await?; + if value < at_least { + bail!("{metric} towards {remote} is {value} (target {at_least})"); + } + Ok(()) + } + ) + .await + .unwrap_or_else(|e| panic!("{metric} towards {remote} did not reach {at_least}: {e}")); +} + +/// Reads the current value of the cumulative "silently dropped response" counter +/// on `node`, i.e. the sum of +/// `mr_routed_message_count{type="response",status="canister_not_found"}` across +/// all destination-subnet labels. Returns 0 if the counter is not present yet. +/// +/// This counter is cumulative and, because all matrix entries share the same IC +/// and the same sender subnet S, it carries over between scenario runs. Callers +/// must therefore compare against a baseline captured before the deletion rather +/// than against an absolute threshold. +async fn read_dropped_response_count(node: &IcNodeSnapshot) -> anyhow::Result { + let map = MetricsFetcher::new( + std::iter::once(node.clone()), + vec!["mr_routed_message_count".to_string()], + ) + .fetch::() + .await + .map_err(|e| anyhow::anyhow!("failed to fetch mr_routed_message_count: {e}"))?; + Ok(map + .iter() + .filter(|(key, _)| { + key.contains("type=\"response\"") && key.contains("status=\"canister_not_found\"") + }) + .filter_map(|(_, values)| values.first().copied()) + .sum()) +} + +/// Waits until the "silently dropped response" counter on `node` reaches at least +/// `at_least`, and returns the observed count. See [`read_dropped_response_count`] +/// for why callers pass a baseline-relative target rather than an absolute one. +async fn wait_for_dropped_response_count_at_least( + node: &IcNodeSnapshot, + at_least: u64, + logger: &slog::Logger, +) -> u64 { + retry_with_msg_async!( + format!( + "waiting for the silently dropped response counter (mr_routed_message_count \ + type=response status=canister_not_found) to reach {at_least}" + ), + logger, + READY_WAIT_TIMEOUT, + RETRY_BACKOFF, + || async { + let dropped = read_dropped_response_count(node).await?; + if dropped < at_least { + bail!("dropped response counter is {dropped} (target {at_least})"); + } + Ok(dropped) + } + ) + .await + .unwrap_or_else(|e| panic!("silently dropped response counter did not reach {at_least}: {e}")) +} + async fn set_subnet_halted( governance: &UniversalCanister<'_>, subnet_id: ic_types::SubnetId, diff --git a/rs/universal_canister/impl/src/lib.rs b/rs/universal_canister/impl/src/lib.rs index 44c13dafeacb..1e6b9a8c22a9 100644 --- a/rs/universal_canister/impl/src/lib.rs +++ b/rs/universal_canister/impl/src/lib.rs @@ -126,5 +126,6 @@ try_from_u8!( CostHttpRequestV2 = 97, MsgCallerInfoData = 98, MsgCallerInfoSigner = 99, + LoopUntilGlobalDataSet = 100, } ); diff --git a/rs/universal_canister/impl/src/main.rs b/rs/universal_canister/impl/src/main.rs index 181bfccf7c29..fb1f56c56199 100644 --- a/rs/universal_canister/impl/src/main.rs +++ b/rs/universal_canister/impl/src/main.rs @@ -28,6 +28,20 @@ pub struct TransformArg { pub context: Vec, } +/// 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, 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 { 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") + } + } + } } } } diff --git a/rs/universal_canister/lib/src/lib.rs b/rs/universal_canister/lib/src/lib.rs index 6468cb8ba83f..7af34af0b22f 100644 --- a/rs/universal_canister/lib/src/lib.rs +++ b/rs/universal_canister/lib/src/lib.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 { self.0 }