diff --git a/Cargo.lock b/Cargo.lock
index d327dd2a187..e1c47ea7762 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3296,6 +3296,18 @@ version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1"
 
+[[package]]
+name = "downstream-calls-test"
+version = "0.1.0"
+dependencies = [
+ "candid",
+ "dfn_core",
+ "ic-base-types",
+ "ic-cdk",
+ "ic-cdk-macros",
+ "serde",
+]
+
 [[package]]
 name = "drawille"
 version = "0.3.0"
@@ -8697,6 +8709,7 @@ dependencies = [
  "assert_matches",
  "candid",
  "canister-test",
+ "downstream-calls-test",
  "ic-base-types",
  "ic-certification-version",
  "ic-config",
@@ -8743,6 +8756,7 @@ dependencies = [
  "prometheus",
  "rand 0.8.5",
  "rand_chacha 0.3.1",
+ "serde",
  "slog",
  "tempfile",
  "xnet-test",
diff --git a/Cargo.toml b/Cargo.toml
index 2988743efc3..51594162177 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -300,6 +300,7 @@ members = [
   "rs/rust_canisters/dfn_json",
   "rs/rust_canisters/dfn_macro",
   "rs/rust_canisters/dfn_protobuf",
+  "rs/rust_canisters/downstream_calls_test",
   "rs/rust_canisters/ecdsa",
   "rs/rust_canisters/http_types",
   "rs/rust_canisters/memory_test",
diff --git a/rs/canister_sandbox/backend_lib/src/sandbox_server.rs b/rs/canister_sandbox/backend_lib/src/sandbox_server.rs
index 1d9a13d7a7d..d82c99cd25b 100644
--- a/rs/canister_sandbox/backend_lib/src/sandbox_server.rs
+++ b/rs/canister_sandbox/backend_lib/src/sandbox_server.rs
@@ -161,7 +161,7 @@ mod tests {
     use ic_test_utilities::types::ids::{canister_test_id, subnet_test_id, user_test_id};
     use ic_types::{
         ingress::WasmResult,
-        messages::CallContextId,
+        messages::{CallContextId, RequestMetadata},
         methods::{FuncRef, WasmMethod},
         time::Time,
         CanisterTimer, ComputeAllocation, Cycles, MemoryAllocation, NumBytes, NumInstructions,
@@ -218,6 +218,7 @@ mod tests {
             CanisterTimer::Inactive,
             0,
             BTreeSet::from([controller]),
+            RequestMetadata::new(0, Time::from_nanos_since_unix_epoch(0)),
         )
     }
 
diff --git a/rs/canonical_state/src/encoding/tests/compatibility.rs b/rs/canonical_state/src/encoding/tests/compatibility.rs
index c10d2c00ea2..0e2d0ad8a09 100644
--- a/rs/canonical_state/src/encoding/tests/compatibility.rs
+++ b/rs/canonical_state/src/encoding/tests/compatibility.rs
@@ -322,10 +322,10 @@ fn canonical_encoding_request_v14_plus() {
             .payment(Cycles::new(4))
             .method_name("test".to_string())
             .method_payload(vec![6])
-            .metadata(Some(RequestMetadata {
-                call_tree_depth: 13,
-                call_tree_start_time: Time::from_nanos_since_unix_epoch(101),
-            }))
+            .metadata(Some(RequestMetadata::new(
+                13,
+                Time::from_nanos_since_unix_epoch(101),
+            )))
             .build()
             .into();
 
diff --git a/rs/canonical_state/src/encoding/tests/test_fixtures.rs b/rs/canonical_state/src/encoding/tests/test_fixtures.rs
index 7f330099652..6a92938d7b4 100644
--- a/rs/canonical_state/src/encoding/tests/test_fixtures.rs
+++ b/rs/canonical_state/src/encoding/tests/test_fixtures.rs
@@ -33,10 +33,10 @@ pub fn request(certification_version: CertificationVersion) -> RequestOrResponse
         .method_name("test".to_string())
         .method_payload(vec![6])
         .metadata(
-            (certification_version >= CertificationVersion::V14).then_some(RequestMetadata {
-                call_tree_depth: 1,
-                call_tree_start_time: Time::from_nanos_since_unix_epoch(100_000),
-            }),
+            (certification_version >= CertificationVersion::V14).then_some(RequestMetadata::new(
+                1,
+                Time::from_nanos_since_unix_epoch(100_000),
+            )),
         )
         .build()
         .into()
diff --git a/rs/canonical_state/src/encoding/types.rs b/rs/canonical_state/src/encoding/types.rs
index 618fb667c95..57ca9e7e585 100644
--- a/rs/canonical_state/src/encoding/types.rs
+++ b/rs/canonical_state/src/encoding/types.rs
@@ -55,9 +55,9 @@ pub struct RequestMetadata {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub call_tree_depth: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub call_tree_start_time: Option<Time>,
+    pub call_tree_start_time_u64: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub call_subtree_deadline: Option<Time>,
+    pub call_subtree_deadline_u64: Option<u64>,
 }
 
 /// Canonical representation of `ic_types::messages::Request`.
@@ -260,9 +260,11 @@ impl TryFrom<RequestOrResponse> for ic_types::messages::RequestOrResponse {
 impl From<&ic_types::messages::RequestMetadata> for RequestMetadata {
     fn from(metadata: &ic_types::messages::RequestMetadata) -> Self {
         RequestMetadata {
-            call_tree_depth: Some(metadata.call_tree_depth),
-            call_tree_start_time: Some(metadata.call_tree_start_time),
-            call_subtree_deadline: None,
+            call_tree_depth: Some(*metadata.call_tree_depth()),
+            call_tree_start_time_u64: Some(
+                metadata.call_tree_start_time().as_nanos_since_unix_epoch(),
+            ),
+            call_subtree_deadline_u64: None,
         }
     }
 }
@@ -293,12 +295,10 @@ impl From<(&ic_types::messages::Request, CertificationVersion)> for Request {
 
 impl From<RequestMetadata> for ic_types::messages::RequestMetadata {
     fn from(metadata: RequestMetadata) -> Self {
-        ic_types::messages::RequestMetadata {
-            call_tree_depth: metadata.call_tree_depth.unwrap_or(0),
-            call_tree_start_time: metadata
-                .call_tree_start_time
-                .unwrap_or(Time::from_nanos_since_unix_epoch(0)),
-        }
+        ic_types::messages::RequestMetadata::new(
+            metadata.call_tree_depth.unwrap_or(0),
+            Time::from_nanos_since_unix_epoch(metadata.call_tree_start_time_u64.unwrap_or(0)),
+        )
     }
 }
 
diff --git a/rs/embedders/fuzz/fuzz_targets/execute_with_wasm_executor.rs b/rs/embedders/fuzz/fuzz_targets/execute_with_wasm_executor.rs
index 8693d31a08e..2da6db0db12 100644
--- a/rs/embedders/fuzz/fuzz_targets/execute_with_wasm_executor.rs
+++ b/rs/embedders/fuzz/fuzz_targets/execute_with_wasm_executor.rs
@@ -27,6 +27,7 @@ use ic_test_utilities::{
     types::ids::user_test_id,
 };
 use ic_types::{
+    messages::RequestMetadata,
     methods::{FuncRef, WasmMethod},
     ComputeAllocation, MemoryAllocation, NumBytes, NumInstructions,
 };
@@ -114,6 +115,7 @@ fn setup_wasm_execution_input(func_ref: FuncRef) -> WasmExecutionInput {
         &network_topology,
         dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     let canister_current_memory_usage = NumBytes::new(0);
diff --git a/rs/embedders/src/wasmtime_embedder/wasmtime_embedder_tests.rs b/rs/embedders/src/wasmtime_embedder/wasmtime_embedder_tests.rs
index 8a405b1440e..922677d270c 100644
--- a/rs/embedders/src/wasmtime_embedder/wasmtime_embedder_tests.rs
+++ b/rs/embedders/src/wasmtime_embedder/wasmtime_embedder_tests.rs
@@ -18,7 +18,9 @@ use ic_system_api::{
 use ic_test_utilities::{
     cycles_account_manager::CyclesAccountManagerBuilder, mock_time, types::ids::canister_test_id,
 };
-use ic_types::{ComputeAllocation, MemoryAllocation, NumBytes, NumInstructions};
+use ic_types::{
+    messages::RequestMetadata, ComputeAllocation, MemoryAllocation, NumBytes, NumInstructions,
+};
 use ic_wasm_types::BinaryEncodedWasm;
 
 use lazy_static::lazy_static;
@@ -50,6 +52,7 @@ fn test_wasmtime_system_api() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
     let canister_memory_limit = NumBytes::from(4 << 30);
     let canister_current_memory_usage = NumBytes::from(0);
diff --git a/rs/embedders/tests/wasmtime_random_memory_writes.rs b/rs/embedders/tests/wasmtime_random_memory_writes.rs
index b61856a9fcf..69d23f1903f 100644
--- a/rs/embedders/tests/wasmtime_random_memory_writes.rs
+++ b/rs/embedders/tests/wasmtime_random_memory_writes.rs
@@ -20,6 +20,7 @@ use ic_test_utilities::{
 use ic_test_utilities_logger::with_test_replica_logger;
 use ic_types::MemoryAllocation;
 use ic_types::{
+    messages::RequestMetadata,
     methods::{FuncRef, WasmMethod},
     ComputeAllocation, Cycles, NumBytes, NumInstructions, PrincipalId,
 };
@@ -69,6 +70,7 @@ fn test_api_for_update(
         }
         .dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
     let canister_memory_limit = NumBytes::from(4 << 30);
     let canister_current_memory_usage = NumBytes::from(0);
diff --git a/rs/execution_environment/benches/lib/src/common.rs b/rs/execution_environment/benches/lib/src/common.rs
index 35e40bdd10b..f3fce0fd63c 100644
--- a/rs/execution_environment/benches/lib/src/common.rs
+++ b/rs/execution_environment/benches/lib/src/common.rs
@@ -32,7 +32,7 @@ use ic_test_utilities::{
 };
 use ic_test_utilities_execution_environment::generate_network_topology;
 use ic_types::{
-    messages::{CallbackId, CanisterMessage, Payload, RejectContext},
+    messages::{CallbackId, CanisterMessage, Payload, RejectContext, RequestMetadata},
     methods::{Callback, WasmClosure},
     Cycles, MemoryAllocation, NumBytes, NumInstructions, Time,
 };
@@ -113,7 +113,12 @@ where
         .system_state
         .call_context_manager_mut()
         .unwrap()
-        .new_call_context(call_origin.clone(), Cycles::new(10), mock_time());
+        .new_call_context(
+            call_origin.clone(),
+            Cycles::new(10),
+            mock_time(),
+            RequestMetadata::new(0, mock_time()),
+        );
     let callback = Callback::new(
         call_context_id,
         canister_test_id(LOCAL_CANISTER_ID),
diff --git a/rs/execution_environment/src/execution/common.rs b/rs/execution_environment/src/execution/common.rs
index 652c5b6d49f..79fae4c27da 100644
--- a/rs/execution_environment/src/execution/common.rs
+++ b/rs/execution_environment/src/execution/common.rs
@@ -6,6 +6,8 @@ use lazy_static::lazy_static;
 use prometheus::IntCounter;
 use std::sync::atomic::{AtomicU64, Ordering};
 
+use crate::execution_environment::ExecutionResponse;
+use crate::{as_round_instructions, metrics::CallTreeMetrics, ExecuteMessageResult, RoundLimits};
 use ic_base_types::{CanisterId, NumBytes, SubnetId};
 use ic_embedders::wasm_executor::{CanisterStateChanges, SliceExecutionOutput};
 use ic_error_types::{ErrorCode, RejectCode, UserError};
@@ -18,7 +20,7 @@ use ic_replicated_state::{
     CallContext, CallContextAction, CallOrigin, CanisterState, ExecutionState, NetworkTopology,
     SystemState,
 };
-use ic_system_api::sandbox_safe_system_state::SystemStateChanges;
+use ic_system_api::sandbox_safe_system_state::{RequestMetadataStats, SystemStateChanges};
 use ic_types::ingress::{IngressState, IngressStatus, WasmResult};
 use ic_types::messages::{
     CallContextId, CallbackId, CanisterCall, CanisterCallOrTask, MessageId, Payload, RejectContext,
@@ -27,9 +29,6 @@ use ic_types::messages::{
 use ic_types::methods::{Callback, WasmMethod};
 use ic_types::{Cycles, NumInstructions, Time, UserId};
 
-use crate::execution_environment::ExecutionResponse;
-use crate::{as_round_instructions, ExecuteMessageResult, RoundLimits};
-
 lazy_static! {
     /// Track how many system task errors have been encountered
     /// so that we can restrict logging to a sample of them.
@@ -414,7 +413,7 @@ fn try_apply_canister_state_changes(
     network_topology: &NetworkTopology,
     subnet_id: SubnetId,
     log: &ReplicaLogger,
-) -> HypervisorResult<()> {
+) -> HypervisorResult<RequestMetadataStats> {
     subnet_available_memory
         .try_decrement(
             output.allocated_bytes,
@@ -446,6 +445,8 @@ pub fn apply_canister_state_changes(
     subnet_id: SubnetId,
     log: &ReplicaLogger,
     state_changes_error: &IntCounter,
+    call_tree_metrics: &dyn CallTreeMetrics,
+    call_context_creation_time: Time,
 ) {
     if let Some(CanisterStateChanges {
         globals,
@@ -468,7 +469,7 @@ pub fn apply_canister_state_changes(
             subnet_id,
             log,
         ) {
-            Ok(()) => {
+            Ok(request_stats) => {
                 execution_state.wasm_memory = wasm_memory;
                 execution_state.stable_memory = stable_memory;
                 execution_state.exported_globals = globals;
@@ -477,6 +478,8 @@ pub fn apply_canister_state_changes(
                 // i.e., `(start)`, `canister_init`, `canister_pre_upgrade`, and `canister_post_upgrade`)
                 // call this `apply_canister_state_change` to finish execution.
                 system_state.canister_version += 1;
+
+                call_tree_metrics.observe(request_stats, call_context_creation_time, time);
             }
             Err(err) => {
                 debug_assert_eq!(err, HypervisorError::OutOfMemory);
diff --git a/rs/execution_environment/src/execution/inspect_message.rs b/rs/execution_environment/src/execution/inspect_message.rs
index 0b45b13ab67..74b86738f2a 100644
--- a/rs/execution_environment/src/execution/inspect_message.rs
+++ b/rs/execution_environment/src/execution/inspect_message.rs
@@ -1,7 +1,7 @@
 use crate::Hypervisor;
 use crate::{
     execution_environment::{as_round_instructions, RoundLimits},
-    metrics::IngressFilterMetrics,
+    metrics::{CallTreeMetricsNoOp, IngressFilterMetrics},
 };
 use ic_error_types::{ErrorCode, UserError};
 use ic_interfaces::execution_environment::SubnetAvailableMemory;
@@ -28,7 +28,7 @@ pub fn execute_inspect_message(
     network_topology: &NetworkTopology,
     logger: &ReplicaLogger,
     state_changes_error: &IntCounter,
-    metrics: &IngressFilterMetrics,
+    ingress_filter_metrics: &IngressFilterMetrics,
 ) -> (NumInstructions, Result<(), UserError>) {
     let canister_id = canister.canister_id();
     let memory_usage = canister.memory_usage();
@@ -69,7 +69,9 @@ pub fn execute_inspect_message(
         // Ignore compute allocation
         compute_allocation_used: 0,
     };
-    let inspect_message_timer = metrics.inspect_message_duration_seconds.start_timer();
+    let inspect_message_timer = ingress_filter_metrics
+        .inspect_message_duration_seconds
+        .start_timer();
     let (output, _output_execution_state, _system_state_accessor) = hypervisor.execute(
         system_api,
         time,
@@ -82,10 +84,12 @@ pub fn execute_inspect_message(
         network_topology,
         &mut round_limits,
         state_changes_error,
+        &CallTreeMetricsNoOp,
+        time,
     );
     drop(inspect_message_timer);
-    metrics.inspect_message_count.inc();
-    metrics
+    ingress_filter_metrics.inspect_message_count.inc();
+    ingress_filter_metrics
         .inspect_message_instructions
         .observe((message_instruction_limit.get() - output.num_instructions_left.get()) as f64);
     match output.wasm_result {
diff --git a/rs/execution_environment/src/execution/install.rs b/rs/execution_environment/src/execution/install.rs
index dee63efd2b8..3fc6064738d 100644
--- a/rs/execution_environment/src/execution/install.rs
+++ b/rs/execution_environment/src/execution/install.rs
@@ -24,7 +24,7 @@ use ic_replicated_state::{
 };
 use ic_system_api::ApiType;
 use ic_types::funds::Cycles;
-use ic_types::messages::CanisterCall;
+use ic_types::messages::{CanisterCall, RequestMetadata};
 use ic_types::methods::{FuncRef, SystemMethod, WasmMethod};
 
 /// Installs a new code in canister. The algorithm consists of five stages:
@@ -156,6 +156,7 @@ pub(crate) fn execute_install(
             helper.canister_message_memory_usage(),
             helper.execution_parameters().clone(),
             FuncRef::Method(method),
+            RequestMetadata::for_new_call_tree(original.time),
             round_limits,
             round.network_topology,
         );
@@ -269,6 +270,7 @@ fn install_stage_2b_continue_install_after_start(
         helper.canister_message_memory_usage(),
         helper.execution_parameters().clone(),
         FuncRef::Method(method),
+        RequestMetadata::for_new_call_tree(original.time),
         round_limits,
         round.network_topology,
     );
diff --git a/rs/execution_environment/src/execution/nonreplicated_query.rs b/rs/execution_environment/src/execution/nonreplicated_query.rs
index 24befe5f5b0..8fa29ea9b8d 100644
--- a/rs/execution_environment/src/execution/nonreplicated_query.rs
+++ b/rs/execution_environment/src/execution/nonreplicated_query.rs
@@ -7,13 +7,13 @@
 
 use crate::execution::common::{validate_canister, validate_method};
 use crate::execution_environment::RoundLimits;
-use crate::{Hypervisor, NonReplicatedQueryKind};
+use crate::{metrics::CallTreeMetricsNoOp, Hypervisor, NonReplicatedQueryKind};
 use ic_error_types::UserError;
 use ic_interfaces::execution_environment::SystemApiCallCounters;
 use ic_replicated_state::{CallOrigin, CanisterState, NetworkTopology};
 use ic_system_api::{ApiType, ExecutionParameters};
 use ic_types::ingress::WasmResult;
-use ic_types::messages::CallContextId;
+use ic_types::messages::{CallContextId, RequestMetadata};
 use ic_types::methods::{FuncRef, WasmMethod};
 use ic_types::{Cycles, NumInstructions, Time};
 use prometheus::IntCounter;
@@ -81,7 +81,12 @@ pub fn execute_non_replicated_query(
                 .system_state
                 .call_context_manager_mut()
                 .unwrap()
-                .new_call_context(call_origin, Cycles::zero(), time);
+                .new_call_context(
+                    call_origin,
+                    Cycles::zero(),
+                    time,
+                    RequestMetadata::for_new_call_tree(time),
+                );
             (
                 ic_system_api::NonReplicatedQueryKind::Stateful {
                     call_context_id,
@@ -101,6 +106,7 @@ pub fn execute_non_replicated_query(
         data_certificate,
         non_replicated_query_kind,
     );
+
     // As we are executing the query in non-replicated mode, we can
     // modify the canister as the caller is not going to be able to
     // commit modifications to the canister anyway.
@@ -116,6 +122,8 @@ pub fn execute_non_replicated_query(
         network_topology,
         round_limits,
         state_changes_error,
+        &CallTreeMetricsNoOp,
+        time,
     );
     canister.system_state = output_system_state;
     if preserve_changes {
diff --git a/rs/execution_environment/src/execution/replicated_query.rs b/rs/execution_environment/src/execution/replicated_query.rs
index e81c808a22e..1494d44d37d 100644
--- a/rs/execution_environment/src/execution/replicated_query.rs
+++ b/rs/execution_environment/src/execution/replicated_query.rs
@@ -8,15 +8,15 @@ use crate::execution::common::{
     finish_call_with_error, validate_message, wasm_result_to_query_response,
 };
 use crate::execution_environment::{ExecuteMessageResult, RoundContext, RoundLimits};
+use crate::metrics::CallTreeMetricsNoOp;
 use ic_error_types::{ErrorCode, UserError};
 use ic_replicated_state::{CallOrigin, CanisterState};
+use ic_system_api::{ApiType, ExecutionParameters};
+use ic_types::methods::{FuncRef, WasmMethod};
 use ic_types::{
     messages::{CanisterCall, CanisterCallOrTask},
     NumBytes, NumInstructions, Time,
 };
-
-use ic_system_api::{ApiType, ExecutionParameters};
-use ic_types::methods::{FuncRef, WasmMethod};
 use prometheus::IntCounter;
 
 // Execute an inter-canister request or an ingress message as a replicated query.
@@ -134,6 +134,8 @@ pub fn execute_replicated_query(
         round.network_topology,
         round_limits,
         state_changes_error,
+        &CallTreeMetricsNoOp,
+        time,
     );
 
     let result = output.wasm_result;
diff --git a/rs/execution_environment/src/execution/response.rs b/rs/execution_environment/src/execution/response.rs
index b8ee4cf005f..5d0b963c3bd 100644
--- a/rs/execution_environment/src/execution/response.rs
+++ b/rs/execution_environment/src/execution/response.rs
@@ -19,7 +19,8 @@ use ic_sys::PAGE_SIZE;
 use ic_system_api::{ApiType, ExecutionParameters};
 use ic_types::ingress::WasmResult;
 use ic_types::messages::{
-    CallContextId, CallbackId, CanisterMessage, CanisterMessageOrTask, Payload, Response,
+    CallContextId, CallbackId, CanisterMessage, CanisterMessageOrTask, Payload, RequestMetadata,
+    Response,
 };
 use ic_types::methods::{Callback, FuncRef, WasmClosure};
 use ic_types::Cycles;
@@ -32,6 +33,7 @@ use crate::execution::common::{
 use crate::execution_environment::{
     ExecuteMessageResult, ExecutionResponse, PausedExecution, RoundContext, RoundLimits,
 };
+use crate::metrics::CallTreeMetrics;
 
 #[cfg(test)]
 mod tests;
@@ -353,6 +355,7 @@ impl ResponseHelper {
         round: &RoundContext,
         round_limits: &mut RoundLimits,
         reserved_cleanup_instructions: NumInstructions,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> Result<ExecuteMessageResult, (Self, HypervisorError, NumInstructions)> {
         self.canister
             .system_state
@@ -402,7 +405,10 @@ impl ResponseHelper {
             round.hypervisor.subnet_id(),
             round.log,
             round.counters.state_changes_error,
+            call_tree_metrics,
+            original.call_context_creation_time,
         );
+
         // Return total instructions: wasm executor leftovers + cleanup reservation.
         let instructions_available = output.num_instructions_left + reserved_cleanup_instructions;
         match output.wasm_result {
@@ -428,6 +434,7 @@ impl ResponseHelper {
         original: &OriginalContext,
         round: &RoundContext,
         round_limits: &mut RoundLimits,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult {
         self.canister
             .system_state
@@ -454,6 +461,8 @@ impl ResponseHelper {
             round.hypervisor.subnet_id(),
             round.log,
             round.counters.state_changes_error,
+            call_tree_metrics,
+            original.call_context_creation_time,
         );
 
         match output.wasm_result {
@@ -631,6 +640,8 @@ struct OriginalContext {
     callback_id: CallbackId,
     call_origin: CallOrigin,
     time: Time,
+    call_context_creation_time: Time,
+    request_metadata: RequestMetadata,
     message_instruction_limit: NumInstructions,
     message: Arc<Response>,
     subnet_size: usize,
@@ -667,6 +678,7 @@ impl PausedExecution for PausedResponseExecution {
         round: RoundContext,
         round_limits: &mut RoundLimits,
         _subnet_size: usize,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult {
         info!(
             round.log,
@@ -714,6 +726,7 @@ impl PausedExecution for PausedResponseExecution {
             self.original,
             round,
             round_limits,
+            call_tree_metrics,
         )
     }
 
@@ -756,6 +769,7 @@ impl PausedExecution for PausedCleanupExecution {
         round: RoundContext,
         round_limits: &mut RoundLimits,
         _subnet_size: usize,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult {
         info!(
             round.log,
@@ -806,6 +820,7 @@ impl PausedExecution for PausedCleanupExecution {
             self.original,
             round,
             round_limits,
+            call_tree_metrics,
         )
     }
 
@@ -838,6 +853,7 @@ pub fn execute_response(
     round_limits: &mut RoundLimits,
     subnet_size: usize,
     subnet_memory_reservation: NumBytes,
+    call_tree_metrics: &dyn CallTreeMetrics,
 ) -> ExecuteMessageResult {
     let (callback, callback_id, call_context, call_context_id) =
         match common::get_call_context_and_callback(
@@ -875,6 +891,8 @@ pub fn execute_response(
         callback_id,
         call_origin: call_context.call_origin().clone(),
         time,
+        call_context_creation_time: call_context.time(),
+        request_metadata: call_context.metadata().clone(),
         message_instruction_limit: execution_parameters.instruction_limits.message(),
         message: Arc::clone(&response),
         subnet_size,
@@ -940,6 +958,7 @@ pub fn execute_response(
         helper.canister().message_memory_usage(),
         execution_parameters.clone(),
         func_ref,
+        original.request_metadata.clone(),
         round_limits,
         round.network_topology,
     );
@@ -953,6 +972,7 @@ pub fn execute_response(
         original,
         round,
         round_limits,
+        call_tree_metrics,
     )
 }
 
@@ -982,6 +1002,7 @@ fn execute_response_cleanup(
     original: OriginalContext,
     round: RoundContext,
     round_limits: &mut RoundLimits,
+    call_tree_metrics: &dyn CallTreeMetrics,
 ) -> ExecuteMessageResult {
     execution_parameters
         .instruction_limits
@@ -1006,6 +1027,7 @@ fn execute_response_cleanup(
         helper.canister().message_memory_usage(),
         execution_parameters.clone(),
         func_ref,
+        original.request_metadata.clone(),
         round_limits,
         round.network_topology,
     );
@@ -1018,6 +1040,7 @@ fn execute_response_cleanup(
         original,
         round,
         round_limits,
+        call_tree_metrics,
     )
 }
 
@@ -1033,6 +1056,7 @@ fn process_response_result(
     original: OriginalContext,
     round: RoundContext,
     round_limits: &mut RoundLimits,
+    call_tree_metrics: &dyn CallTreeMetrics,
 ) -> ExecuteMessageResult {
     match result {
         WasmExecutionResult::Paused(slice, paused_wasm_execution) => {
@@ -1080,6 +1104,7 @@ fn process_response_result(
                 &round,
                 round_limits,
                 reserved_cleanup_instructions,
+                call_tree_metrics,
             ) {
                 Ok(result) => result,
                 Err((helper, err, instructions_left)) => {
@@ -1096,6 +1121,7 @@ fn process_response_result(
                             original,
                             round,
                             round_limits,
+                            call_tree_metrics,
                         ),
                         None => {
                             // No cleanup closure present. Return the callback error as-is.
@@ -1125,6 +1151,7 @@ fn process_cleanup_result(
     original: OriginalContext,
     round: RoundContext,
     round_limits: &mut RoundLimits,
+    call_tree_metrics: &dyn CallTreeMetrics,
 ) -> ExecuteMessageResult {
     match result {
         WasmExecutionResult::Paused(slice, paused_wasm_execution) => {
@@ -1172,6 +1199,7 @@ fn process_cleanup_result(
                 &original,
                 &round,
                 round_limits,
+                call_tree_metrics,
             )
         }
     }
diff --git a/rs/execution_environment/src/execution/update.rs b/rs/execution_environment/src/execution/update.rs
index 27d092ff6eb..25889af8e5a 100644
--- a/rs/execution_environment/src/execution/update.rs
+++ b/rs/execution_environment/src/execution/update.rs
@@ -8,6 +8,7 @@ use crate::execution::common::{
 use crate::execution_environment::{
     ExecuteMessageResult, PausedExecution, RoundContext, RoundLimits,
 };
+use crate::metrics::CallTreeMetrics;
 use ic_base_types::CanisterId;
 use ic_embedders::wasm_executor::{CanisterStateChanges, PausedWasmExecution, WasmExecutionResult};
 use ic_error_types::{ErrorCode, UserError};
@@ -17,16 +18,15 @@ use ic_interfaces::execution_environment::{
 };
 use ic_logger::{info, ReplicaLogger};
 use ic_replicated_state::{CallOrigin, CanisterState};
+use ic_system_api::{ApiType, ExecutionParameters};
 use ic_types::messages::{
     CallContextId, CanisterCall, CanisterCallOrTask, CanisterMessage, CanisterMessageOrTask,
-    CanisterTask,
+    CanisterTask, RequestMetadata,
 };
+use ic_types::methods::{FuncRef, SystemMethod, WasmMethod};
 use ic_types::{CanisterTimer, Cycles, NumBytes, NumInstructions, Time};
 use ic_wasm_types::WasmEngineError::FailedToApplySystemChanges;
 
-use ic_system_api::{ApiType, ExecutionParameters};
-use ic_types::methods::{FuncRef, SystemMethod, WasmMethod};
-
 #[cfg(test)]
 mod tests;
 
@@ -42,6 +42,7 @@ pub fn execute_update(
     round: RoundContext,
     round_limits: &mut RoundLimits,
     subnet_size: usize,
+    call_tree_metrics: &dyn CallTreeMetrics,
 ) -> ExecuteMessageResult {
     let (clean_canister, prepaid_execution_cycles, resuming_aborted) =
         match prepaid_execution_cycles {
@@ -86,6 +87,14 @@ pub fn execute_update(
         clean_canister.system_state.reserved_balance(),
     );
 
+    let request_metadata = match &call_or_task {
+        CanisterCallOrTask::Call(CanisterCall::Request(request)) => match &request.metadata {
+            Some(metadata) => metadata.for_downstream_call(),
+            None => RequestMetadata::for_new_call_tree(time),
+        },
+        _ => RequestMetadata::for_new_call_tree(time),
+    };
+
     let original = OriginalContext {
         call_origin: CallOrigin::from(&call_or_task),
         method,
@@ -94,6 +103,7 @@ pub fn execute_update(
         execution_parameters,
         subnet_size,
         time,
+        request_metadata,
         freezing_threshold,
         canister_id: clean_canister.canister_id(),
     };
@@ -143,6 +153,7 @@ pub fn execute_update(
         message_memory_usage,
         original.execution_parameters.clone(),
         FuncRef::Method(original.method.clone()),
+        original.request_metadata.clone(),
         round_limits,
         round.network_topology,
     );
@@ -191,6 +202,7 @@ pub fn execute_update(
                 original,
                 round,
                 round_limits,
+                call_tree_metrics,
             )
         }
     }
@@ -247,6 +259,7 @@ struct OriginalContext {
     execution_parameters: ExecutionParameters,
     subnet_size: usize,
     time: Time,
+    request_metadata: RequestMetadata,
     freezing_threshold: Cycles,
     canister_id: CanisterId,
 }
@@ -282,6 +295,7 @@ impl UpdateHelper {
                 original.call_origin.clone(),
                 original.call_or_task.cycles(),
                 original.time,
+                original.request_metadata.clone(),
             );
 
         let initial_cycles_balance = canister.system_state.balance();
@@ -342,6 +356,7 @@ impl UpdateHelper {
         original: OriginalContext,
         round: RoundContext,
         round_limits: &mut RoundLimits,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult {
         self.canister
             .system_state
@@ -393,7 +408,10 @@ impl UpdateHelper {
             round.hypervisor.subnet_id(),
             round.log,
             round.counters.state_changes_error,
+            call_tree_metrics,
+            original.time,
         );
+
         let heap_delta = if output.wasm_result.is_ok() {
             NumBytes::from((output.instance_stats.dirty_pages * ic_sys::PAGE_SIZE) as u64)
         } else {
@@ -468,6 +486,7 @@ impl PausedExecution for PausedCallExecution {
         round: RoundContext,
         round_limits: &mut RoundLimits,
         _subnet_size: usize,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult {
         info!(
             round.log,
@@ -546,6 +565,7 @@ impl PausedExecution for PausedCallExecution {
                     self.original,
                     round,
                     round_limits,
+                    call_tree_metrics,
                 )
             }
         }
diff --git a/rs/execution_environment/src/execution/update/tests.rs b/rs/execution_environment/src/execution/update/tests.rs
index f0dd72971e1..78587255d69 100644
--- a/rs/execution_environment/src/execution/update/tests.rs
+++ b/rs/execution_environment/src/execution/update/tests.rs
@@ -15,7 +15,7 @@ use ic_replicated_state::{
 };
 use ic_state_machine_tests::{Cycles, IngressStatus, WasmResult};
 use ic_sys::PAGE_SIZE;
-use ic_types::messages::CallbackId;
+use ic_types::messages::{CallbackId, RequestMetadata};
 use ic_types::{NumInstructions, NumPages};
 use ic_universal_canister::{call_args, wasm};
 
@@ -438,7 +438,12 @@ fn dts_update_resume_fails_due_to_call_context_change() {
         .system_state
         .call_context_manager_mut()
         .unwrap()
-        .new_call_context(CallOrigin::SystemTask, Cycles::new(0), time);
+        .new_call_context(
+            CallOrigin::SystemTask,
+            Cycles::new(0),
+            time,
+            RequestMetadata::for_new_call_tree(time),
+        );
 
     test.execute_slice(a_id);
 
diff --git a/rs/execution_environment/src/execution/upgrade.rs b/rs/execution_environment/src/execution/upgrade.rs
index 16a8e2bcb6d..d58b2936cd7 100644
--- a/rs/execution_environment/src/execution/upgrade.rs
+++ b/rs/execution_environment/src/execution/upgrade.rs
@@ -26,7 +26,10 @@ use ic_replicated_state::{
 };
 use ic_system_api::ApiType;
 use ic_types::methods::{FuncRef, SystemMethod, WasmMethod};
-use ic_types::{funds::Cycles, messages::CanisterCall};
+use ic_types::{
+    funds::Cycles,
+    messages::{CanisterCall, RequestMetadata},
+};
 
 #[cfg(test)]
 mod tests;
@@ -149,6 +152,7 @@ pub(crate) fn execute_upgrade(
             helper.canister_message_memory_usage(),
             helper.execution_parameters().clone(),
             FuncRef::Method(method),
+            RequestMetadata::for_new_call_tree(original.time),
             round_limits,
             round.network_topology,
         );
@@ -316,6 +320,7 @@ fn upgrade_stage_2_and_3a_create_execution_state_and_call_start(
             helper.canister_message_memory_usage(),
             helper.execution_parameters().clone(),
             FuncRef::Method(method),
+            RequestMetadata::for_new_call_tree(original.time),
             round_limits,
             round.network_topology,
         );
@@ -432,6 +437,7 @@ fn upgrade_stage_4a_call_post_upgrade(
         helper.canister_message_memory_usage(),
         helper.execution_parameters().clone(),
         FuncRef::Method(method),
+        RequestMetadata::for_new_call_tree(original.time),
         round_limits,
         round.network_topology,
     );
diff --git a/rs/execution_environment/src/execution_environment.rs b/rs/execution_environment/src/execution_environment.rs
index a1505e936d0..5b7cfb45a1e 100644
--- a/rs/execution_environment/src/execution_environment.rs
+++ b/rs/execution_environment/src/execution_environment.rs
@@ -15,7 +15,7 @@ use crate::{
     },
     hypervisor::Hypervisor,
     ic00_permissions::Ic00MethodPermissions,
-    metrics::IngressFilterMetrics,
+    metrics::{CallTreeMetrics, CallTreeMetricsImpl, IngressFilterMetrics},
     NonReplicatedQueryKind,
 };
 use candid::Encode;
@@ -245,6 +245,7 @@ pub trait PausedExecution: std::fmt::Debug + Send {
         round_context: RoundContext,
         round_limits: &mut RoundLimits,
         subnet_size: usize,
+        call_tree_metrics: &dyn CallTreeMetrics,
     ) -> ExecuteMessageResult;
 
     /// Aborts the paused execution.
@@ -285,6 +286,7 @@ pub struct ExecutionEnvironment {
     canister_manager: CanisterManager,
     ingress_history_writer: Arc<dyn IngressHistoryWriter<State = ReplicatedState>>,
     metrics: ExecutionEnvironmentMetrics,
+    call_tree_metrics: CallTreeMetricsImpl,
     config: ExecutionConfig,
     cycles_account_manager: Arc<CyclesAccountManager>,
     own_subnet_id: SubnetId,
@@ -369,6 +371,7 @@ impl ExecutionEnvironment {
             canister_manager,
             ingress_history_writer,
             metrics,
+            call_tree_metrics: CallTreeMetricsImpl::new(metrics_registry),
             config,
             cycles_account_manager,
             own_subnet_id,
@@ -1367,6 +1370,7 @@ impl ExecutionEnvironment {
                     round,
                     round_limits,
                     subnet_size,
+                    &self.call_tree_metrics,
                 )
             }
             WasmMethod::System(_) => {
@@ -1403,6 +1407,7 @@ impl ExecutionEnvironment {
             round,
             round_limits,
             subnet_size,
+            &self.call_tree_metrics,
         )
     }
 
@@ -1774,6 +1779,7 @@ impl ExecutionEnvironment {
             round_limits,
             subnet_size,
             scaled_subnet_memory_reservation,
+            &self.call_tree_metrics,
         )
     }
 
@@ -3193,7 +3199,13 @@ pub fn execute_canister(
                     log: &exec_env.log,
                     time,
                 };
-                let result = paused.resume(canister, round_context, round_limits, subnet_size);
+                let result = paused.resume(
+                    canister,
+                    round_context,
+                    round_limits,
+                    subnet_size,
+                    &exec_env.call_tree_metrics,
+                );
                 let (canister, instructions_used, heap_delta, ingress_status) =
                     exec_env.process_result(result);
                 return ExecuteCanisterResult {
diff --git a/rs/execution_environment/src/execution_environment_metrics.rs b/rs/execution_environment/src/execution_environment_metrics.rs
index 9aab0583b2c..6d22245533c 100644
--- a/rs/execution_environment/src/execution_environment_metrics.rs
+++ b/rs/execution_environment/src/execution_environment_metrics.rs
@@ -52,6 +52,7 @@ pub(crate) struct ExecutionEnvironmentMetrics {
     /// Critical error for failed canister creation.
     pub(crate) canister_creation_error: IntCounter,
 }
+
 impl ExecutionEnvironmentMetrics {
     pub fn new(metrics_registry: &MetricsRegistry) -> Self {
         Self {
diff --git a/rs/execution_environment/src/hypervisor.rs b/rs/execution_environment/src/hypervisor.rs
index 9020e2a9f89..658a35b3fd6 100644
--- a/rs/execution_environment/src/hypervisor.rs
+++ b/rs/execution_environment/src/hypervisor.rs
@@ -15,13 +15,17 @@ use ic_replicated_state::NetworkTopology;
 use ic_replicated_state::{page_map::allocated_pages_count, ExecutionState, SystemState};
 use ic_system_api::ExecutionParameters;
 use ic_system_api::{sandbox_safe_system_state::SandboxSafeSystemState, ApiType};
-use ic_types::{methods::FuncRef, CanisterId, NumBytes, NumInstructions, SubnetId, Time};
+use ic_types::{
+    messages::RequestMetadata, methods::FuncRef, CanisterId, NumBytes, NumInstructions, SubnetId,
+    Time,
+};
 use ic_wasm_types::CanisterModule;
 use prometheus::{Histogram, IntCounter, IntGauge};
 use std::{path::PathBuf, sync::Arc};
 
 use crate::execution::common::{apply_canister_state_changes, update_round_limits};
 use crate::execution_environment::{as_round_instructions, CompilationCostHandling, RoundLimits};
+use crate::metrics::CallTreeMetrics;
 use ic_replicated_state::page_map::PageAllocatorFileDescriptor;
 
 #[cfg(test)]
@@ -317,6 +321,8 @@ impl Hypervisor {
         network_topology: &NetworkTopology,
         round_limits: &mut RoundLimits,
         state_changes_error: &IntCounter,
+        call_tree_metrics: &dyn CallTreeMetrics,
+        call_context_creation_time: Time,
     ) -> (WasmExecutionOutput, ExecutionState, SystemState) {
         assert_eq!(
             execution_parameters.instruction_limits.message(),
@@ -330,6 +336,7 @@ impl Hypervisor {
             canister_current_message_memory_usage,
             execution_parameters,
             func_ref,
+            RequestMetadata::for_new_call_tree(time),
             round_limits,
             network_topology,
         );
@@ -353,6 +360,8 @@ impl Hypervisor {
             self.own_subnet_id,
             &self.log,
             state_changes_error,
+            call_tree_metrics,
+            call_context_creation_time,
         );
         (output, execution_state, system_state)
     }
@@ -368,6 +377,7 @@ impl Hypervisor {
         canister_current_message_memory_usage: NumBytes,
         execution_parameters: ExecutionParameters,
         func_ref: FuncRef,
+        request_metadata: RequestMetadata,
         round_limits: &mut RoundLimits,
         network_topology: &NetworkTopology,
     ) -> WasmExecutionResult {
@@ -387,6 +397,7 @@ impl Hypervisor {
             network_topology,
             self.dirty_page_overhead,
             execution_parameters.compute_allocation,
+            request_metadata,
         );
         let (compilation_result, execution_result) = Arc::clone(&self.wasm_executor).execute(
             WasmExecutionInput {
diff --git a/rs/execution_environment/src/metrics.rs b/rs/execution_environment/src/metrics.rs
index ab066efb35f..c6c1d8c1b7e 100644
--- a/rs/execution_environment/src/metrics.rs
+++ b/rs/execution_environment/src/metrics.rs
@@ -2,8 +2,10 @@ use ic_metrics::{
     buckets::{decimal_buckets, decimal_buckets_with_zero},
     MetricsRegistry,
 };
+use ic_system_api::sandbox_safe_system_state::RequestMetadataStats;
 use ic_types::{
-    NumInstructions, NumMessages, NumSlices, MAX_STABLE_MEMORY_IN_BYTES, MAX_WASM_MEMORY_IN_BYTES,
+    NumInstructions, NumMessages, NumSlices, Time, MAX_STABLE_MEMORY_IN_BYTES,
+    MAX_WASM_MEMORY_IN_BYTES,
 };
 use prometheus::{Histogram, IntCounter, IntCounterVec};
 use std::{cell::RefCell, rc::Rc, time::Instant};
@@ -42,6 +44,96 @@ impl IngressFilterMetrics {
     }
 }
 
+/// Trait for observing metrics concerning call trees.
+///
+/// New call trees are created by ingress messages or canister tasks; canister requests are found
+/// on each branch in the call tree.
+///
+/// The age and depth of a call tree is measured relative to the root.
+pub trait CallTreeMetrics {
+    fn observe(
+        &self,
+        request_stats: RequestMetadataStats,
+        call_context_creation_time: Time,
+        time: Time,
+    );
+}
+
+/// Implementation of `CallTreeMetrics` that doesn't record anything.
+pub(crate) struct CallTreeMetricsNoOp;
+
+impl CallTreeMetrics for CallTreeMetricsNoOp {
+    fn observe(
+        &self,
+        _request_stats: RequestMetadataStats,
+        _call_context_creation_time: Time,
+        _time: Time,
+    ) {
+    }
+}
+
+#[derive(Clone)]
+pub struct CallTreeMetricsImpl {
+    /// The depth down the call tree requests were created at (starting at 0).
+    pub(crate) request_call_tree_depth: Histogram,
+    /// Call tree age at the point when each new request was created.
+    pub(crate) request_call_tree_age_seconds: Histogram,
+    /// Call context age at the point when each new request was created.
+    pub(crate) request_call_context_age_seconds: Histogram,
+}
+
+impl CallTreeMetricsImpl {
+    pub fn new(metrics_registry: &MetricsRegistry) -> Self {
+        Self {
+            request_call_tree_depth: metrics_registry.histogram(
+                "execution_environment_request_call_tree_depth",
+                "The depth down the call tree that new requests were created at (0 based).",
+                decimal_buckets_with_zero(0, 2),
+            ),
+            request_call_tree_age_seconds: metrics_registry.histogram(
+                "execution_environment_request_call_tree_age_seconds",
+                "Call tree age at the point when each new request was created.",
+                decimal_buckets_with_zero(0, 6),
+            ),
+            request_call_context_age_seconds: metrics_registry.histogram(
+                "execution_environment_request_call_context_age_seconds",
+                "Call context age at the point when each new request was created.",
+                decimal_buckets_with_zero(0, 6),
+            ),
+        }
+    }
+}
+
+impl CallTreeMetrics for CallTreeMetricsImpl {
+    fn observe(
+        &self,
+        request_stats: RequestMetadataStats,
+        call_context_creation_time: Time,
+        time: Time,
+    ) {
+        // Observe call-tree related metrics.
+        if let Some(ref metadata) = request_stats.metadata {
+            for _ in 0..request_stats.count {
+                self.request_call_tree_depth
+                    .observe(*metadata.call_tree_depth() as f64);
+            }
+            let duration = time.saturating_sub(*metadata.call_tree_start_time());
+            for _ in 0..request_stats.count {
+                self.request_call_tree_age_seconds
+                    .observe(duration.as_secs_f64());
+            }
+        }
+
+        // Observe new requests vs. original context.
+        for _ in 0..request_stats.count {
+            self.request_call_context_age_seconds.observe(
+                time.saturating_sub(call_context_creation_time)
+                    .as_secs_f64(),
+            );
+        }
+    }
+}
+
 pub(crate) struct QueryHandlerMetrics {
     pub query: ScopedMetrics,
     pub query_initial_call: ScopedMetrics,
diff --git a/rs/execution_environment/src/query_handler/query_context.rs b/rs/execution_environment/src/query_handler/query_context.rs
index ae59f427785..f008d0c9cb5 100644
--- a/rs/execution_environment/src/query_handler/query_context.rs
+++ b/rs/execution_environment/src/query_handler/query_context.rs
@@ -5,7 +5,7 @@ use crate::{
     execution_environment::{as_round_instructions, RoundLimits},
     hypervisor::Hypervisor,
     metrics::{
-        MeasurementScope, QueryHandlerMetrics, QUERY_HANDLER_CRITICAL_ERROR,
+        CallTreeMetricsNoOp, MeasurementScope, QueryHandlerMetrics, QUERY_HANDLER_CRITICAL_ERROR,
         SYSTEM_API_CALL_PERFORM, SYSTEM_API_CANISTER_CYCLE_BALANCE,
         SYSTEM_API_CANISTER_CYCLE_BALANCE128, SYSTEM_API_TIME,
     },
@@ -585,6 +585,8 @@ impl<'a> QueryContext<'a> {
             &self.network_topology,
             &mut self.round_limits,
             self.query_critical_error,
+            &CallTreeMetricsNoOp,
+            call_context.time(),
         );
 
         self.system_api_call_counters
@@ -685,6 +687,8 @@ impl<'a> QueryContext<'a> {
                 &self.network_topology,
                 &mut self.round_limits,
                 self.query_critical_error,
+                &CallTreeMetricsNoOp,
+                time,
             );
 
         self.system_api_call_counters
diff --git a/rs/messaging/BUILD.bazel b/rs/messaging/BUILD.bazel
index f6442516187..5817c0b9485 100644
--- a/rs/messaging/BUILD.bazel
+++ b/rs/messaging/BUILD.bazel
@@ -74,8 +74,12 @@ rust_test(
 rust_test_suite(
     name = "messaging_integration",
     srcs = glob(["tests/**/*.rs"]),
-    data = ["//rs/rust_canisters/xnet_test:xnet-test-canister"],
+    data = [
+        "//rs/rust_canisters/downstream_calls_test:downstream-calls-test-canister",
+        "//rs/rust_canisters/xnet_test:xnet-test-canister",
+    ],
     env = {
+        "DOWNSTREAM_CALLS_TEST_CANISTER_WASM_PATH": "$(rootpath //rs/rust_canisters/downstream_calls_test:downstream-calls-test-canister)",
         "XNET_TEST_CANISTER_WASM_PATH": "$(rootpath //rs/rust_canisters/xnet_test:xnet-test-canister)",
     },
     deps = [
@@ -85,6 +89,7 @@ rust_test_suite(
         "//rs/registry/subnet_type",
         "//rs/replicated_state",
         "//rs/rust_canisters/canister_test",
+        "//rs/rust_canisters/downstream_calls_test",
         "//rs/rust_canisters/xnet_test",
         "//rs/state_machine_tests",
         "//rs/test_utilities",
@@ -95,5 +100,6 @@ rust_test_suite(
         "@crate_index//:assert_matches",
         "@crate_index//:candid",
         "@crate_index//:maplit",
+        "@crate_index//:serde",
     ],
 )
diff --git a/rs/messaging/Cargo.toml b/rs/messaging/Cargo.toml
index 94eb3cb5004..a03b58ee836 100644
--- a/rs/messaging/Cargo.toml
+++ b/rs/messaging/Cargo.toml
@@ -46,6 +46,7 @@ ic-interfaces-state-manager-mocks = { path = "../interfaces/state_manager/mocks"
 ic-state-machine-tests = { path = "../state_machine_tests" }
 xnet-test = { path = "../rust_canisters/xnet_test" }
 canister-test = { path = "../rust_canisters/canister_test" }
+downstream-calls-test = { path = "../rust_canisters/downstream_calls_test" }
 ic-registry-local-registry = { path = "../registry/local_registry" }
 ic-registry-local-store = { path = "../registry/local_store" }
 ic-registry-local-store-artifacts = { path = "../registry/local_store/artifacts" }
@@ -60,6 +61,7 @@ maplit = "1.0.2"
 mockall = { workspace = true }
 rand = "0.8"
 rand_chacha = "0.3"
+serde = "1.0"
 tempfile = "3.0"
 
 [features]
diff --git a/rs/messaging/tests/call_tree_tests.rs b/rs/messaging/tests/call_tree_tests.rs
new file mode 100644
index 00000000000..b959be3ef25
--- /dev/null
+++ b/rs/messaging/tests/call_tree_tests.rs
@@ -0,0 +1,168 @@
+use candid::{Decode, Encode};
+use canister_test::Project;
+use downstream_calls_test::{CallOrResponse, State};
+use ic_base_types::CanisterId;
+use ic_base_types::PrincipalId;
+use ic_registry_routing_table::{routing_table_insert_subnet, RoutingTable};
+use ic_registry_subnet_type::SubnetType;
+use ic_state_machine_tests::{StateMachine, StateMachineBuilder, WasmResult};
+use ic_test_utilities::types::ids::subnet_test_id;
+use ic_test_utilities_metrics::fetch_histogram_stats;
+use ic_types::Cycles;
+use std::collections::VecDeque;
+
+struct CallTreeTestFixture(StateMachine);
+
+impl CallTreeTestFixture {
+    /// Creates a new state machine with a specific number of 'downstream-calls-test-canisters`
+    /// installed and started.
+    fn with_num_canisters(num_canisters: u64) -> (Self, Vec<CanisterId>) {
+        let subnet_id = subnet_test_id(1);
+        let env = StateMachineBuilder::new()
+            .with_subnet_id(subnet_id)
+            .with_subnet_type(SubnetType::Application)
+            .with_routing_table({
+                let mut routing_table = RoutingTable::new();
+                routing_table_insert_subnet(&mut routing_table, subnet_id).unwrap();
+                routing_table
+            })
+            .build();
+
+        // Install `num_canisters` canisters.
+        let wasm = Project::cargo_bin_maybe_from_env("downstream-calls-test-canister", &[]).bytes();
+        let canister_ids = (0..num_canisters)
+            .map(|_| {
+                env.install_canister_with_cycles(
+                    wasm.clone(),
+                    Vec::new(),
+                    None,
+                    Cycles::new(u128::MAX / 2),
+                )
+                .expect("Installing downstream-calls-test-canister failed")
+            })
+            .collect::<Vec<_>>();
+
+        // Start canisters.
+        for canister_id in canister_ids.iter() {
+            env.start_canister(*canister_id)
+                .expect("failed to start canister");
+        }
+
+        (Self(env), canister_ids)
+    }
+
+    /// Runs a test using an action sequence (a list of call and response commands) and an entry
+    /// canister Id.
+    ///
+    /// The test is iniated by an ingress message to the entry canister. From there the actions
+    /// list is passed from canister to canister until it is exhausted, after which responses are
+    /// handed down until they eventually form the response to the initiating ingress message.
+    ///
+    /// While traversing the call tree corresponding to the actions list, the number of calls made
+    /// and the sum of the call depth at which they are made is kept track of. After the fact,
+    /// these counters should match the stats tracked in the metrics for
+    /// 'execution_environment_request_call_tree_depth'.
+    fn run_actions_sequence_test(
+        &self,
+        entry_canister_id: CanisterId,
+        actions: VecDeque<CallOrResponse>,
+    ) {
+        let payload = Encode!(&State {
+            actions,
+            call_count: 0,
+            current_depth: 0,
+            depth_total: 0,
+        })
+        .unwrap();
+
+        let result = self
+            .0
+            .execute_ingress_as(
+                PrincipalId::new_anonymous(),
+                entry_canister_id,
+                "reply_or_defer",
+                payload,
+            )
+            .unwrap();
+
+        if let WasmResult::Reply(msg) = result {
+            let state = Decode!(&msg, State).unwrap();
+            let stats = fetch_histogram_stats(
+                self.0.metrics_registry(),
+                "execution_environment_request_call_tree_depth",
+            )
+            .unwrap();
+
+            assert_eq!(state.call_count, stats.count);
+            assert_eq!(state.depth_total as f64, stats.sum);
+        } else {
+            unreachable!();
+        }
+    }
+}
+
+/// Test the metrics are recorded as expected for a sequence of canisters.
+/// ```text
+/// tree:              call tree depth:
+///
+/// ingress              _
+///    \
+///     A                0
+///      \
+///       B              1
+///        \
+///         C            2
+///          \
+///           D          3
+///            \
+///             E        4
+/// ```
+#[test]
+fn test_linear_sequence_call_tree_depth() {
+    use CallOrResponse::*;
+    let (fixture, mut canister_ids) = CallTreeTestFixture::with_num_canisters(5);
+    let entry_canister_id = canister_ids.pop().unwrap();
+
+    fixture.run_actions_sequence_test(
+        entry_canister_id, // Call A.
+        [
+            Call(canister_ids[0]), // Call B from A.
+            Call(canister_ids[1]), // Call C from B.
+            Call(canister_ids[2]), // Call D from C.
+            Call(canister_ids[3]), // Call E from D.
+        ]
+        .into(),
+    );
+}
+
+/// Test the metrics are recorded as expected for a call tree with multiple branches.
+/// ```text
+/// tree:            call tree depth:
+///
+///    ingress         _
+///       |
+///       A            0
+///      / \
+///     B   C          1
+///        / \
+///       D   E        2
+/// ```
+#[test]
+fn test_multiple_branches_call_tree_depth() {
+    use CallOrResponse::*;
+    let (fixture, mut canister_ids) = CallTreeTestFixture::with_num_canisters(5);
+    let entry_canister_id = canister_ids.pop().unwrap();
+
+    fixture.run_actions_sequence_test(
+        entry_canister_id, // Call A.
+        [
+            Call(canister_ids[0]), // Call B from A.
+            Response,              // Response from B, back to A.
+            Call(canister_ids[1]), // Call C from A.
+            Call(canister_ids[2]), // Call D from C.
+            Response,              // Response from D, back to C.
+            Call(canister_ids[3]), // Call E from C.
+        ]
+        .into(),
+    );
+}
diff --git a/rs/protobuf/def/state/canister_state_bits/v1/canister_state_bits.proto b/rs/protobuf/def/state/canister_state_bits/v1/canister_state_bits.proto
index 6ad1844f76f..7caf022134e 100644
--- a/rs/protobuf/def/state/canister_state_bits/v1/canister_state_bits.proto
+++ b/rs/protobuf/def/state/canister_state_bits/v1/canister_state_bits.proto
@@ -32,6 +32,7 @@ message CallContext {
   bool deleted = 8;
   uint64 time_nanos = 9;
   uint64 instructions_executed = 10;
+  state.queues.v1.RequestMetadata metadata = 11;
 }
 
 message CallContextEntry {
diff --git a/rs/protobuf/src/gen/state/state.canister_state_bits.v1.rs b/rs/protobuf/src/gen/state/state.canister_state_bits.v1.rs
index d596b7f848e..8b83f48f755 100644
--- a/rs/protobuf/src/gen/state/state.canister_state_bits.v1.rs
+++ b/rs/protobuf/src/gen/state/state.canister_state_bits.v1.rs
@@ -11,6 +11,8 @@ pub struct CallContext {
     pub time_nanos: u64,
     #[prost(uint64, tag = "10")]
     pub instructions_executed: u64,
+    #[prost(message, optional, tag = "11")]
+    pub metadata: ::core::option::Option<super::super::queues::v1::RequestMetadata>,
     #[prost(oneof = "call_context::CallOrigin", tags = "1, 2, 3, 4, 7")]
     pub call_origin: ::core::option::Option<call_context::CallOrigin>,
 }
diff --git a/rs/replicated_state/src/canister_state/system_state/call_context_manager.rs b/rs/replicated_state/src/canister_state/system_state/call_context_manager.rs
index 32f27b39cfc..7d01df9782c 100644
--- a/rs/replicated_state/src/canister_state/system_state/call_context_manager.rs
+++ b/rs/replicated_state/src/canister_state/system_state/call_context_manager.rs
@@ -10,7 +10,10 @@ use ic_protobuf::types::v1 as pb_types;
 use ic_types::NumInstructions;
 use ic_types::{
     ingress::WasmResult,
-    messages::{CallContextId, CallbackId, CanisterCall, CanisterCallOrTask, MessageId, Response},
+    messages::{
+        CallContextId, CallbackId, CanisterCall, CanisterCallOrTask, MessageId, RequestMetadata,
+        Response,
+    },
     methods::Callback,
     user_id_into_protobuf, user_id_try_from_protobuf, CanisterId, Cycles, Funds, PrincipalId, Time,
     UserId,
@@ -43,6 +46,9 @@ pub struct CallContext {
     /// Point in time at which the `CallContext` was created.
     time: Time,
 
+    /// Metadata for requests generated within this `CallContext`.
+    metadata: RequestMetadata,
+
     /// The total number of instructions executed in the given call context.
     /// This value is used for the `ic0.performance_counter` type 1.
     instructions_executed: NumInstructions,
@@ -55,6 +61,7 @@ impl CallContext {
         deleted: bool,
         available_cycles: Cycles,
         time: Time,
+        metadata: RequestMetadata,
     ) -> Self {
         Self {
             call_origin,
@@ -62,6 +69,7 @@ impl CallContext {
             deleted,
             available_cycles,
             time,
+            metadata,
             instructions_executed: NumInstructions::default(),
         }
     }
@@ -115,6 +123,11 @@ impl CallContext {
         self.time
     }
 
+    /// Metadata for requests generated within this `CallContext`.
+    pub fn metadata(&self) -> &RequestMetadata {
+        &self.metadata
+    }
+
     /// Return the total number of instructions executed in the given call context.
     /// This value is used for the `ic0.performance_counter` type 1.
     pub fn instructions_executed(&self) -> NumInstructions {
@@ -131,6 +144,7 @@ impl From<&CallContext> for pb::CallContext {
             deleted: item.deleted,
             available_funds: Some((&funds).into()),
             time_nanos: item.time.as_nanos_since_unix_epoch(),
+            metadata: Some((&item.metadata).into()),
             instructions_executed: item.instructions_executed.get(),
         }
     }
@@ -148,6 +162,12 @@ impl TryFrom<pb::CallContext> for CallContext {
             deleted: value.deleted,
             available_cycles: funds.cycles(),
             time: Time::from_nanos_since_unix_epoch(value.time_nanos),
+            metadata: value
+                .metadata
+                .map(From::from)
+                .unwrap_or(RequestMetadata::for_new_call_tree(
+                    Time::from_nanos_since_unix_epoch(0),
+                )),
             instructions_executed: value.instructions_executed.into(),
         })
     }
@@ -323,6 +343,7 @@ impl CallContextManager {
         call_origin: CallOrigin,
         cycles: Cycles,
         time: Time,
+        metadata: RequestMetadata,
     ) -> CallContextId {
         self.next_call_context_id += 1;
         let id = CallContextId::from(self.next_call_context_id);
@@ -334,6 +355,7 @@ impl CallContextManager {
                 deleted: false,
                 available_cycles: cycles,
                 time,
+                metadata,
                 instructions_executed: NumInstructions::default(),
             },
         );
diff --git a/rs/replicated_state/src/canister_state/system_state/call_context_manager/tests.rs b/rs/replicated_state/src/canister_state/system_state/call_context_manager/tests.rs
index fdff0220f19..7d4b26d58bc 100644
--- a/rs/replicated_state/src/canister_state/system_state/call_context_manager/tests.rs
+++ b/rs/replicated_state/src/canister_state/system_state/call_context_manager/tests.rs
@@ -1,6 +1,9 @@
 use super::*;
-use ic_test_utilities::types::ids::canister_test_id;
-use ic_types::methods::{WasmClosure, UNKNOWN_CANISTER_ID};
+use ic_test_utilities::{mock_time, types::ids::canister_test_id};
+use ic_types::{
+    messages::RequestMetadata,
+    methods::{WasmClosure, UNKNOWN_CANISTER_ID},
+};
 
 #[test]
 fn call_context_origin() {
@@ -11,6 +14,7 @@ fn call_context_origin() {
         CallOrigin::CanisterUpdate(id, cb_id),
         Cycles::new(10),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
     assert_eq!(
         ccm.call_contexts().get(&cc_id).unwrap().call_origin,
@@ -29,17 +33,20 @@ fn call_context_handling() {
         CallOrigin::CanisterUpdate(canister_test_id(123), CallbackId::from(1)),
         Cycles::zero(),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
     let call_context_id2 = call_context_manager.new_call_context(
         CallOrigin::CanisterUpdate(canister_test_id(123), CallbackId::from(2)),
         Cycles::zero(),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
 
     let call_context_id3 = call_context_manager.new_call_context(
         CallOrigin::CanisterUpdate(canister_test_id(123), CallbackId::from(3)),
         Cycles::zero(),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
 
     // Call context 3 was not responded and does not have outstanding calls,
@@ -255,6 +262,7 @@ fn withdraw_cycles_fails_when_not_enough_available_cycles() {
         CallOrigin::CanisterUpdate(id, cb_id),
         Cycles::new(30),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
 
     assert_eq!(
@@ -277,6 +285,7 @@ fn withdraw_cycles_succeeds_when_enough_available_cycles() {
         CallOrigin::CanisterUpdate(id, cb_id),
         Cycles::new(30),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
 
     assert_eq!(
@@ -294,6 +303,7 @@ fn test_call_context_instructions_executed_is_updated() {
         CallOrigin::CanisterUpdate(canister_test_id(123), CallbackId::from(1)),
         Cycles::zero(),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
     // Register a callback, so the call context is not deleted in `on_canister_result()` later.
     let _callback_id = call_context_manager.register_callback(Callback::new(
diff --git a/rs/replicated_state/src/canister_state/tests.rs b/rs/replicated_state/src/canister_state/tests.rs
index db0e4f5a6df..54580a08ff8 100644
--- a/rs/replicated_state/src/canister_state/tests.rs
+++ b/rs/replicated_state/src/canister_state/tests.rs
@@ -25,8 +25,8 @@ use ic_test_utilities::types::{
 };
 use ic_types::{
     messages::{
-        CallContextId, CallbackId, CanisterCall, StopCanisterCallId, StopCanisterContext,
-        MAX_RESPONSE_COUNT_BYTES,
+        CallContextId, CallbackId, CanisterCall, RequestMetadata, StopCanisterCallId,
+        StopCanisterContext, MAX_RESPONSE_COUNT_BYTES,
     },
     methods::{Callback, WasmClosure},
     nominal_cycles::NominalCycles,
@@ -98,6 +98,7 @@ impl CanisterStateFixture {
                 CallOrigin::CanisterUpdate(CANISTER_ID, CallbackId::from(1)),
                 Cycles::zero(),
                 Time::from_nanos_since_unix_epoch(0),
+                RequestMetadata::new(0, mock_time()),
             );
         self.canister_state
             .system_state
@@ -813,6 +814,7 @@ fn reverts_stopping_status_after_split() {
         CallOrigin::Ingress(user_test_id(1), message_test_id(2)),
         Cycles::from(0u128),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
     canister_state.system_state.status = CanisterStatus::Stopping {
         call_context_manager: call_context_manager.clone(),
diff --git a/rs/rust_canisters/downstream_calls_test/BUILD.bazel b/rs/rust_canisters/downstream_calls_test/BUILD.bazel
new file mode 100644
index 00000000000..e3e2e3bca16
--- /dev/null
+++ b/rs/rust_canisters/downstream_calls_test/BUILD.bazel
@@ -0,0 +1,32 @@
+load("@rules_rust//rust:defs.bzl", "rust_library")
+load("//bazel:canisters.bzl", "rust_canister")
+
+package(default_visibility = ["//visibility:public"])
+
+DEPENDENCIES = [
+    "//rs/rust_canisters/dfn_core",
+    "//rs/types/base_types",
+    "@crate_index//:candid",
+    "@crate_index//:ic-cdk",
+    "@crate_index//:serde",
+]
+
+MACRO_DEPENDENCIES = [
+    "@crate_index//:ic-cdk-macros",
+]
+
+rust_library(
+    name = "downstream_calls_test",
+    srcs = ["src/lib.rs"],
+    crate_name = "downstream_calls_test",
+    version = "0.1.0",
+    deps = DEPENDENCIES,
+)
+
+rust_canister(
+    name = "downstream-calls-test-canister",
+    srcs = ["src/main.rs"],
+    proc_macro_deps = MACRO_DEPENDENCIES,
+    service_file = ":downstream_calls_test.did",
+    deps = DEPENDENCIES + [":downstream_calls_test"],
+)
diff --git a/rs/rust_canisters/downstream_calls_test/Cargo.toml b/rs/rust_canisters/downstream_calls_test/Cargo.toml
new file mode 100644
index 00000000000..ab9591cfc11
--- /dev/null
+++ b/rs/rust_canisters/downstream_calls_test/Cargo.toml
@@ -0,0 +1,16 @@
+[package]
+name = "downstream-calls-test"
+version = "0.1.0"
+edition = "2021"
+
+[[bin]]
+name = "downstream-calls-test-canister"
+path = "src/main.rs"
+
+[dependencies]
+candid = { workspace = true }
+dfn_core = { path = "../dfn_core" }
+ic-base-types = { path = "../../types/base_types" }
+ic-cdk = { workspace = true }
+ic-cdk-macros = { workspace = true }
+serde = "1.0"
diff --git a/rs/rust_canisters/downstream_calls_test/downstream_calls_test.did b/rs/rust_canisters/downstream_calls_test/downstream_calls_test.did
new file mode 100644
index 00000000000..8266120ade4
--- /dev/null
+++ b/rs/rust_canisters/downstream_calls_test/downstream_calls_test.did
@@ -0,0 +1,3 @@
+service : {
+  reply_or_defer : (vec blob, nat64, nat64, nat64) -> (text);
+}
diff --git a/rs/rust_canisters/downstream_calls_test/src/lib.rs b/rs/rust_canisters/downstream_calls_test/src/lib.rs
new file mode 100644
index 00000000000..43399aa9f28
--- /dev/null
+++ b/rs/rust_canisters/downstream_calls_test/src/lib.rs
@@ -0,0 +1,25 @@
+use candid::CandidType;
+use ic_base_types::CanisterId;
+use serde::{Deserialize, Serialize};
+use std::collections::VecDeque;
+
+/// An action for a 'downstream_calls_test' canister, i.e. either call another such
+/// canister or respond to the calling canister.
+#[derive(Serialize, Deserialize, Debug, CandidType)]
+pub enum CallOrResponse {
+    Call(CanisterId),
+    Response,
+}
+
+/// The state that is passed from ('downstream_calls_test') canister to canister.
+/// It keeps track of call tree information and contains a sequence of actions. At each
+/// step the action at the front is popped from the queue and executed, while the rest
+/// is passed on to the next canister. We are using a queue over a vector so that it is
+/// straightforward to execute the list of actions front to back.
+#[derive(Serialize, Deserialize, Debug, CandidType)]
+pub struct State {
+    pub actions: VecDeque<CallOrResponse>,
+    pub call_count: u64,
+    pub current_depth: u64,
+    pub depth_total: u64,
+}
diff --git a/rs/rust_canisters/downstream_calls_test/src/main.rs b/rs/rust_canisters/downstream_calls_test/src/main.rs
new file mode 100644
index 00000000000..15e52ef44a1
--- /dev/null
+++ b/rs/rust_canisters/downstream_calls_test/src/main.rs
@@ -0,0 +1,63 @@
+use candid::{candid_method, Decode, Encode};
+use dfn_core::api::{call_bytes, Funds};
+use downstream_calls_test::{CallOrResponse, State};
+use ic_cdk_macros::update;
+
+fn main() {}
+
+/// Replies to or defers to another canister a list of actions (call or response commands),
+/// along with counters keeping track of:
+///  - The current call depth in the call tree.
+///  - The total number of calls made.
+///  - The sum of call tree depths at which these calls were made.
+///
+/// At each step, the first command in the actions list is extracted and executed. Therefore
+/// the actions list becomes shorter at each step and traverses the call tree that it's elements
+/// describe.
+///
+/// Once the list has exhausted all commands anywhere in the call tree, response commands are
+/// executed until the tree is concluded and a response to the initiating ingress message is
+/// formed.
+///
+/// The tree may also be concluded by sufficiently many response commands in a sequence in the
+/// action list itself (even if the list is not empty at the time of conclusion).
+///
+/// Any action list will produce a finite call tree, provided the canister IDs it contains
+/// correspond to running 'downstream-call-test-canisters'.
+///
+/// See `test_linear_sequence_call_tree_depth` and `test_multiple_branches_call_tree_depth` in
+/// 'rs/messaging/tests/call_tree_tests.rs' for examples on how to use this canister.
+#[candid_method(update)]
+#[update]
+async fn reply_or_defer(mut state: State) -> State {
+    loop {
+        match state.actions.pop_front() {
+            Some(CallOrResponse::Call(canister_id)) => {
+                let response = call_bytes(
+                    canister_id,
+                    "reply_or_defer",
+                    &Encode!(&State {
+                        actions: state.actions,
+                        call_count: state.call_count + 1,
+                        current_depth: state.current_depth + 1,
+                        depth_total: state.depth_total + state.current_depth,
+                    })
+                    .unwrap(),
+                    Funds::zero(),
+                )
+                .await
+                .expect("calling other canister failed");
+
+                state = Decode!(&response, State).expect("decoding response failed");
+            }
+            Some(CallOrResponse::Response) | None => {
+                return State {
+                    actions: state.actions,
+                    call_count: state.call_count,
+                    current_depth: state.current_depth.saturating_sub(1),
+                    depth_total: state.depth_total,
+                }
+            }
+        }
+    }
+}
diff --git a/rs/state_manager/src/tree_hash.rs b/rs/state_manager/src/tree_hash.rs
index bb1e1df3aa6..826f6b8d980 100644
--- a/rs/state_manager/src/tree_hash.rs
+++ b/rs/state_manager/src/tree_hash.rs
@@ -210,10 +210,10 @@ mod tests {
                     RequestBuilder::new()
                         .metadata(
                             (certification_version >= CertificationVersion::V14 && i % 5 != 0)
-                                .then_some(RequestMetadata {
-                                    call_tree_depth: i % 3,
-                                    call_tree_start_time: Time::from_nanos_since_unix_epoch(i % 2),
-                                }),
+                                .then_some(RequestMetadata::new(
+                                    i % 3,
+                                    Time::from_nanos_since_unix_epoch(i % 2),
+                                )),
                         )
                         .build()
                         .into(),
diff --git a/rs/system_api/src/request_in_prep.rs b/rs/system_api/src/request_in_prep.rs
index 2b519fe7a84..b6bf34fd0bf 100644
--- a/rs/system_api/src/request_in_prep.rs
+++ b/rs/system_api/src/request_in_prep.rs
@@ -222,7 +222,7 @@ pub(crate) fn into_request(
         method_payload,
         sender_reply_callback: callback_id,
         payment: cycles,
-        metadata: None,
+        metadata: Some(sandbox_safe_system_state.request_metadata.clone()),
     };
     // We cannot call `Request::payload_size_bytes()` before constructing the
     // request, so ensure our separate calculation matches the actual size.
diff --git a/rs/system_api/src/sandbox_safe_system_state.rs b/rs/system_api/src/sandbox_safe_system_state.rs
index 9c37167fb28..3a51a83330b 100644
--- a/rs/system_api/src/sandbox_safe_system_state.rs
+++ b/rs/system_api/src/sandbox_safe_system_state.rs
@@ -20,7 +20,7 @@ use ic_replicated_state::{
     CallOrigin, CanisterStatus, NetworkTopology, SystemState,
 };
 use ic_types::{
-    messages::{CallContextId, CallbackId, RejectContext, Request},
+    messages::{CallContextId, CallbackId, RejectContext, Request, RequestMetadata},
     methods::Callback,
     CanisterTimer, ComputeAllocation, Cycles, MemoryAllocation, NumInstructions, NumPages, Time,
 };
@@ -286,13 +286,13 @@ impl SystemStateChanges {
         network_topology: &NetworkTopology,
         own_subnet_id: SubnetId,
         logger: &ReplicaLogger,
-    ) -> HypervisorResult<()> {
+    ) -> HypervisorResult<RequestMetadataStats> {
         // Verify total cycle change is not positive and update cycles balance.
         self.validate_cycle_change(system_state.canister_id == CYCLES_MINTING_CANISTER_ID)?;
         self.apply_balance_changes(system_state);
 
         // Verify we don't accept more cycles than are available from each call
-        // context and update each call context balance
+        // context and update each call context balance.
         if !self.call_context_balance_taken.is_empty() {
             let own_canister_id = system_state.canister_id;
             let call_context_manager = system_state
@@ -323,6 +323,16 @@ impl SystemStateChanges {
             }
         }
 
+        // Get a clone of the request metadata of outgoing requests (they are all equivalent)
+        // and their number. This will be used for call tree metrics.
+        let request_stats = RequestMetadataStats {
+            metadata: self
+                .requests
+                .first()
+                .and_then(|request| request.metadata.clone()),
+            count: self.requests.len() as u64,
+        };
+
         // Push outgoing messages.
         let mut callback_changes = BTreeMap::new();
         let nns_subnet_id = network_topology.nns_subnet_id;
@@ -440,7 +450,7 @@ impl SystemStateChanges {
             system_state.global_timer = new_global_timer;
         }
 
-        Ok(())
+        Ok(request_stats)
     }
 
     /// Applies the balance change to the given state.
@@ -546,6 +556,7 @@ pub struct SandboxSafeSystemState {
     global_timer: CanisterTimer,
     canister_version: u64,
     controllers: BTreeSet<PrincipalId>,
+    pub(super) request_metadata: RequestMetadata,
 }
 
 impl SandboxSafeSystemState {
@@ -572,6 +583,7 @@ impl SandboxSafeSystemState {
         global_timer: CanisterTimer,
         canister_version: u64,
         controllers: BTreeSet<PrincipalId>,
+        request_metadata: RequestMetadata,
     ) -> Self {
         Self {
             canister_id,
@@ -595,6 +607,7 @@ impl SandboxSafeSystemState {
             global_timer,
             canister_version,
             controllers,
+            request_metadata,
         }
     }
 
@@ -604,6 +617,7 @@ impl SandboxSafeSystemState {
         network_topology: &NetworkTopology,
         dirty_page_overhead: NumInstructions,
         compute_allocation: ComputeAllocation,
+        request_metadata: RequestMetadata,
     ) -> Self {
         let call_context_balances = match system_state.call_context_manager() {
             Some(call_context_manager) => call_context_manager
@@ -659,6 +673,7 @@ impl SandboxSafeSystemState {
             system_state.global_timer,
             system_state.canister_version,
             system_state.controllers.clone(),
+            request_metadata,
         )
     }
 
@@ -1158,6 +1173,16 @@ impl SandboxSafeSystemState {
     }
 }
 
+/// Holds the metadata and the number of downstream requests. Requests created during the same
+/// execution have the same metadata. This fact is reflected by the use of a counter, rather than
+/// a list of identical metadata.
+///      
+/// This is used for call tree metrics.
+pub struct RequestMetadataStats {
+    pub metadata: Option<RequestMetadata>,
+    pub count: u64,
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;
diff --git a/rs/system_api/tests/common/mod.rs b/rs/system_api/tests/common/mod.rs
index c30a909ee61..2c47385a068 100644
--- a/rs/system_api/tests/common/mod.rs
+++ b/rs/system_api/tests/common/mod.rs
@@ -22,7 +22,7 @@ use ic_test_utilities::{
     types::ids::{call_context_test_id, canister_test_id, subnet_test_id, user_test_id},
 };
 use ic_types::{
-    messages::{CallContextId, CallbackId, RejectContext},
+    messages::{CallContextId, CallbackId, RejectContext, RequestMetadata},
     methods::SystemMethod,
     ComputeAllocation, Cycles, MemoryAllocation, NumInstructions, PrincipalId, Time,
 };
@@ -134,6 +134,7 @@ pub fn get_system_api(
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         execution_parameters().compute_allocation,
+        RequestMetadata::new(0, mock_time()),
     );
     SystemApiImpl::new(
         api_type,
@@ -165,6 +166,7 @@ pub fn get_system_state() -> SystemState {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::new(50),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
     system_state
 }
diff --git a/rs/system_api/tests/sandbox_safe_system_state.rs b/rs/system_api/tests/sandbox_safe_system_state.rs
index 770741d3eb3..45c70190b78 100644
--- a/rs/system_api/tests/sandbox_safe_system_state.rs
+++ b/rs/system_api/tests/sandbox_safe_system_state.rs
@@ -22,7 +22,7 @@ use ic_test_utilities::{
 };
 use ic_types::nominal_cycles::NominalCycles;
 use ic_types::{
-    messages::{CanisterMessage, MAX_INTER_CANISTER_PAYLOAD_IN_BYTES},
+    messages::{CanisterMessage, RequestMetadata, MAX_INTER_CANISTER_PAYLOAD_IN_BYTES},
     ComputeAllocation, Cycles, NumInstructions,
 };
 use prometheus::IntCounter;
@@ -64,6 +64,7 @@ fn push_output_request_fails_not_enough_cycles_for_request() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     assert_eq!(
@@ -115,6 +116,7 @@ fn push_output_request_fails_not_enough_cycles_for_response() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     assert_eq!(
@@ -148,6 +150,7 @@ fn push_output_request_succeeds_with_enough_cycles() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     let prepayment_for_response_execution =
@@ -191,6 +194,7 @@ fn correct_charging_source_canister_for_a_request() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     let request = RequestBuilder::default()
@@ -339,6 +343,7 @@ fn is_controller_test() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     // Users IDs 1 and 2 are controllers, hence is_controller should return true,
@@ -423,6 +428,7 @@ fn test_inter_canister_call(
         topo,
         SchedulerConfig::application_subnet().dirty_page_overhead,
         ComputeAllocation::default(),
+        RequestMetadata::new(0, mock_time()),
     );
 
     let request = RequestBuilder::default()
diff --git a/rs/system_api/tests/system_api.rs b/rs/system_api/tests/system_api.rs
index 596ed8106cd..64dcdeab089 100644
--- a/rs/system_api/tests/system_api.rs
+++ b/rs/system_api/tests/system_api.rs
@@ -27,7 +27,9 @@ use ic_test_utilities::{
     },
 };
 use ic_types::{
-    messages::{CallContextId, CallbackId, RejectContext, MAX_RESPONSE_COUNT_BYTES},
+    messages::{
+        CallContextId, CallbackId, RejectContext, RequestMetadata, MAX_RESPONSE_COUNT_BYTES,
+    },
     methods::{Callback, WasmClosure},
     time, CanisterTimer, CountBytes, Cycles, NumInstructions, PrincipalId, Time,
 };
@@ -1165,6 +1167,7 @@ fn test_canister_balance() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::new(50),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
 
     let mut api = get_system_api(
@@ -1193,6 +1196,7 @@ fn test_canister_cycle_balance() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::new(50),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
 
     let mut api = get_system_api(
@@ -1227,6 +1231,7 @@ fn test_msg_cycles_available_traps() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             available_cycles,
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
 
     let api = get_system_api(
@@ -1388,6 +1393,7 @@ fn msg_cycles_accept_all_cycles_in_call_context() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::from(amount),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
     let mut api = get_system_api(
         ApiTypeBuilder::build_update_api(),
@@ -1411,6 +1417,7 @@ fn msg_cycles_accept_all_cycles_in_call_context_when_more_asked() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::new(40),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
     let mut api = get_system_api(
         ApiTypeBuilder::build_update_api(),
@@ -1443,6 +1450,7 @@ fn call_perform_not_enough_cycles_does_not_trap() {
             CallOrigin::CanisterUpdate(canister_test_id(33), CallbackId::from(5)),
             Cycles::new(40),
             Time::from_nanos_since_unix_epoch(0),
+            RequestMetadata::new(0, mock_time()),
         );
     let mut api = get_system_api(
         ApiTypeBuilder::build_update_api(),
@@ -1493,6 +1501,7 @@ fn update_available_memory_updates_subnet_available_memory() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         execution_parameters().compute_allocation,
+        RequestMetadata::new(0, mock_time()),
     );
     let mut api = SystemApiImpl::new(
         ApiTypeBuilder::build_update_api(),
@@ -1547,6 +1556,7 @@ fn push_output_request_respects_memory_limits() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         execution_parameters().compute_allocation,
+        RequestMetadata::new(0, mock_time()),
     );
     let own_canister_id = system_state.canister_id;
     let callback_id = sandbox_safe_system_state
@@ -1652,6 +1662,7 @@ fn push_output_request_oversized_request_memory_limits() {
         &NetworkTopology::default(),
         SchedulerConfig::application_subnet().dirty_page_overhead,
         execution_parameters().compute_allocation,
+        RequestMetadata::new(0, mock_time()),
     );
     let own_canister_id = system_state.canister_id;
     let callback_id = sandbox_safe_system_state
diff --git a/rs/test_utilities/src/state.rs b/rs/test_utilities/src/state.rs
index 905b9e16713..23bd7add2ce 100644
--- a/rs/test_utilities/src/state.rs
+++ b/rs/test_utilities/src/state.rs
@@ -33,7 +33,7 @@ use ic_types::methods::{Callback, WasmClosure};
 use ic_types::time::UNIX_EPOCH;
 use ic_types::{batch::RawQueryStats, messages::CallbackId};
 use ic_types::{
-    messages::{Ingress, Request, RequestOrResponse},
+    messages::{Ingress, Request, RequestMetadata, RequestOrResponse},
     nominal_cycles::NominalCycles,
     xnet::{StreamHeader, StreamIndex, StreamIndexedQueue},
     CanisterId, ComputeAllocation, Cycles, ExecutionRound, MemoryAllocation, NumBytes, PrincipalId,
@@ -307,6 +307,7 @@ impl CanisterStateBuilder {
                 call_context.call_origin().clone(),
                 call_context.available_cycles(),
                 call_context.time(),
+                call_context.metadata().clone(),
             );
 
             let call_context_in_call_context_manager = call_context_manager
@@ -468,6 +469,7 @@ impl CallContextBuilder {
             false,
             Cycles::zero(),
             self.time,
+            RequestMetadata::new(0, mock_time()),
         )
     }
 }
@@ -741,6 +743,7 @@ pub fn register_callback(
         CallOrigin::CanisterUpdate(originator, callback_id),
         Cycles::zero(),
         Time::from_nanos_since_unix_epoch(0),
+        RequestMetadata::new(0, mock_time()),
     );
 
     call_context_manager.register_callback(Callback::new(
diff --git a/rs/test_utilities/src/wasmtime_instance.rs b/rs/test_utilities/src/wasmtime_instance.rs
index 661b75136ad..96e24768135 100644
--- a/rs/test_utilities/src/wasmtime_instance.rs
+++ b/rs/test_utilities/src/wasmtime_instance.rs
@@ -14,7 +14,7 @@ use ic_system_api::{
     sandbox_safe_system_state::SandboxSafeSystemState, ExecutionParameters, InstructionLimits,
     ModificationTracking, SystemApiImpl,
 };
-use ic_types::{ComputeAllocation, MemoryAllocation, NumInstructions};
+use ic_types::{messages::RequestMetadata, ComputeAllocation, MemoryAllocation, NumInstructions};
 use ic_wasm_types::BinaryEncodedWasm;
 
 use crate::{
@@ -133,6 +133,7 @@ impl WasmtimeInstanceBuilder {
             &self.network_topology,
             dirty_page_overhead,
             ComputeAllocation::default(),
+            RequestMetadata::new(0, mock_time()),
         );
 
         let subnet_memory_capacity = i64::MAX / 2;
diff --git a/rs/types/types/src/messages.rs b/rs/types/types/src/messages.rs
index f80b18c3cc6..1daaef3cb02 100644
--- a/rs/types/types/src/messages.rs
+++ b/rs/types/types/src/messages.rs
@@ -684,10 +684,10 @@ mod tests {
     fn serialize_request_via_bincode() {
         for metadata in [
             None,
-            Some(RequestMetadata {
-                call_tree_depth: 13,
-                call_tree_start_time: Time::from_nanos_since_unix_epoch(17),
-            }),
+            Some(RequestMetadata::new(
+                13,
+                Time::from_nanos_since_unix_epoch(17),
+            )),
         ] {
             let request = Request {
                 receiver: CanisterId::from(13),
diff --git a/rs/types/types/src/messages/inter_canister.rs b/rs/types/types/src/messages/inter_canister.rs
index 646ede6d7ab..66c3bd9bd98 100644
--- a/rs/types/types/src/messages/inter_canister.rs
+++ b/rs/types/types/src/messages/inter_canister.rs
@@ -40,11 +40,38 @@ pub type CallContextId = Id<CallContextIdTag, u64>;
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct RequestMetadata {
     /// Indicates how many steps down the call tree a request is, starting at 0.
-    pub call_tree_depth: u64,
+    call_tree_depth: u64,
     /// The block time (on the respective subnet) at the start of the call at the
-    /// root of the call tree that this request is part of. This is used for metrics
-    /// only.
-    pub call_tree_start_time: Time,
+    /// root of the call tree that this request is part of.
+    call_tree_start_time: Time,
+}
+
+impl RequestMetadata {
+    pub fn new(call_tree_depth: u64, call_tree_start_time: Time) -> Self {
+        Self {
+            call_tree_depth,
+            call_tree_start_time,
+        }
+    }
+
+    /// Creates `RequestMetadata` for a new call tree, i.e. with a given start time and depth 0.
+    pub fn for_new_call_tree(time: Time) -> Self {
+        Self::new(0, time)
+    }
+
+    /// Creates `RequestMetadata` for a downstream call from another metadata, i.e. with depth
+    /// increased by 1 and the same `call_tree_start_time`.
+    pub fn for_downstream_call(&self) -> Self {
+        Self::new(self.call_tree_depth + 1, self.call_tree_start_time)
+    }
+
+    pub fn call_tree_depth(&self) -> &u64 {
+        &self.call_tree_depth
+    }
+
+    pub fn call_tree_start_time(&self) -> &Time {
+        &self.call_tree_start_time
+    }
 }
 
 /// Canister-to-canister request message.
diff --git a/rs/types/types_test_utils/src/arbitrary.rs b/rs/types/types_test_utils/src/arbitrary.rs
index c00ef6f4838..37dcfd4bb44 100644
--- a/rs/types/types_test_utils/src/arbitrary.rs
+++ b/rs/types/types_test_utils/src/arbitrary.rs
@@ -105,10 +105,10 @@ prop_compose! {
         call_tree_depth in any::<u64>(),
         call_tree_start_time_nanos in any::<u64>(),
     ) -> RequestMetadata {
-        RequestMetadata {
+        RequestMetadata::new(
             call_tree_depth,
-            call_tree_start_time: Time::from_nanos_since_unix_epoch(call_tree_start_time_nanos),
-        }
+            Time::from_nanos_since_unix_epoch(call_tree_start_time_nanos),
+        )
     }
 }