/// Attempts to induct the given message at `stream_index` in the incoming
/// stream from `remote_subnet_id` into `state`, producing a signal onto the
/// provided reverse `stream`. The induction attempt will result in one of
/// the following outcomes (in addition to the signal):
///
///  * enqueuing the message into the corresponding input queue;
///  * a reject response enqueued into the reverse stream: if enqueuing of a
///    request failed (queue full, canister not found, out of memory);
///  * no other action: if the sender canister and source subnet do not
///    match; or enqueuing of a response failed.
///
/// Updates `subnet_available_memory` to reflect any change in memory usage.


enqueuing the message into the corresponding input queue;
a reject response enqueued into the reverse stream: if enqueuing of a
request failed (queue full, canister not found, out of memory);
no other action: if the sender canister and source subnet do not
match; or enqueuing of a response failed.

Updates `subnet_available_memory` to reflect any change in memory usage.
enqueuing the message into the corresponding input queue ; a reject response enqueued into the reverse stream : if enqueuing of a request failed ( queue full canister not found out of memory ) ; no other action : if the sender canister and source subnet do not match ; or enqueuing of a response failed . Updates ` subnet_available_memory ` to reflect any change in memory usage .
fn induct_message(
        &self,
        msg: RequestOrResponse,
        remote_subnet_id: SubnetId,
        stream_index: StreamIndex,
        state: &mut ReplicatedState,
        stream: &mut StreamHandle,
        subnet_available_memory: &mut i64,
    ) {
        let payload_size = match &msg {
            RequestOrResponse::Request(req) => req.payload_size_bytes().get(),
            RequestOrResponse::Response(res) => res.response_payload.size_of().get(),
        };
        let msg_type = match msg {
            RequestOrResponse::Request(_) => LABEL_VALUE_TYPE_REQUEST,
            RequestOrResponse::Response(_) => LABEL_VALUE_TYPE_RESPONSE,
        };
        match state
            .metadata
            .network_topology
            .routing_table
            .route(msg.sender().get())
        {
            Some(host_subnet) => {
                if host_subnet == remote_subnet_id {
                    // Sender is hosted by `remote_subnet_id`, proceed with induction.
                    match state.push_input(
                        QUEUE_INDEX_NONE,
                        msg,
                        self.max_canister_memory_size,
                        subnet_available_memory,
                    ) {
                        // Message successfully inducted, all done.
                        Ok(()) => {
                            self.observe_inducted_message_status(msg_type, LABEL_VALUE_SUCCESS);
                            self.observe_inducted_payload_size(payload_size);
                        }
                        // Message not inducted.
                        Err((err, msg)) => {
                            debug!(self.log, "Induction failed with error '{}', generating reject Response for {:?}", &err, &msg);
                            self.observe_inducted_message_status(msg_type, err.to_label_value());
                            let code = reject_code_for_state_error(&err);
                            self.try_enqueue_reject_response(msg, code, err.to_string(), stream);
                        }
                    }
                } else {
                    // Sender is hosted by a subnet other than `remote_subnet_id`.
                    //
                    // Do not enqueue a reject response as remote subnet is likely malicious and
                    // trying to cause a memory leak by sending bogus messages and never consuming
                    // reject responses.
                    warn!(self.log,
                        "Dropping message from subnet {} claiming to be from sender {} hosted by subnet {}: {:?}",
                        remote_subnet_id,
                        msg.sender(),
                        host_subnet,
                        msg);
                    self.observe_inducted_message_status(
                        msg_type,
                        LABEL_VALUE_SENDER_SUBNET_MISMATCH,
                    );
                }
            }
            // Unknown host subnet for sender.
            //
            // Not enqueuing a reject response, see explanation above.
            None => {
                warn!(self.log,
                    "Dropping message from subnet {} claiming to be from sender {} with unknown host subnet: {:?}",
                    remote_subnet_id,
                    msg.sender(),
                    msg);
                self.observe_inducted_message_status(msg_type, LABEL_VALUE_SENDER_SUBNET_UNKNOWN);
            }
        }
        // Signals use the `StreamIndex` of the incoming message.
        assert_eq!(
            stream.signals_end(),
            stream_index,
            "Expecting signal with stream index {}, got {}",
            stream.signals_end(),
            stream_index
        );
        stream.increment_signals_end();
    }
fn induct_message ( & self , msg : RequestOrResponse , remote_subnet_id : SubnetId , stream_index : StreamIndex , state : & mut ReplicatedState , stream : & mut StreamHandle , subnet_available_memory : & mut i64 , ) { let payload_size = match & msg { RequestOrResponse :: Request ( req ) => req . payload_size_bytes ( ) . get ( ) , RequestOrResponse :: Response ( res ) => res . response_payload . size_of ( ) . get ( ) , } ; let msg_type = match msg { RequestOrResponse :: Request ( _ ) => LABEL_VALUE_TYPE_REQUEST , RequestOrResponse :: Response ( _ ) => LABEL_VALUE_TYPE_RESPONSE , } ; match state . metadata . network_topology . routing_table . route ( msg . sender ( ) . get ( ) ) { Some ( host_subnet ) => { if host_subnet == remote_subnet_id { // Sender is hosted by `remote_subnet_id`, proceed with induction. match state . push_input ( QUEUE_INDEX_NONE , msg , self . max_canister_memory_size , subnet_available_memory , ) { // Message successfully inducted, all done. Ok ( ( ) ) => { self . observe_inducted_message_status ( msg_type , LABEL_VALUE_SUCCESS ) ; self . observe_inducted_payload_size ( payload_size ) ; } // Message not inducted. Err ( ( err , msg ) ) => { debug ! ( self log " " err msg ) ; self . observe_inducted_message_status ( msg_type , err . to_label_value ( ) ) ; let code = reject_code_for_state_error ( & err ) ; self . try_enqueue_reject_response ( msg , code , err . to_string ( ) , stream ) ; } } } else { // Sender is hosted by a subnet other than `remote_subnet_id`. // // Do not enqueue a reject response as remote subnet is likely malicious and // trying to cause a memory leak by sending bogus messages and never consuming // reject responses. warn ! ( self log " " remote_subnet_id msg sender ( ) host_subnet msg ) ; self . observe_inducted_message_status ( msg_type , LABEL_VALUE_SENDER_SUBNET_MISMATCH , ) ; } } // Unknown host subnet for sender. // // Not enqueuing a reject response, see explanation above. None => { warn ! ( self log " " remote_subnet_id msg sender ( ) msg ) ; self . observe_inducted_message_status ( msg_type , LABEL_VALUE_SENDER_SUBNET_UNKNOWN ) ; } } // Signals use the `StreamIndex` of the incoming message. assert_eq ! ( stream signals_end ( ) stream_index " " stream signals_end ( ) stream_index ) ; stream . increment_signals_end ( ) ; }
Attempts to induct the given message at `stream_index` in the incoming
stream from `remote_subnet_id` into `state`, producing a signal onto the
provided reverse `stream`.
Attempts to induct the given message at ` stream_index ` in the incoming stream from ` remote_subnet_id ` into ` state ` producing a signal onto the provided reverse ` stream ` .
&self
msg
RequestOrResponse
remote_subnet_id
SubnetId
stream_index
StreamIndex
state
&mut ReplicatedState
stream
&mut StreamHandle
subnet_available_memory
&mut i64