/// Validates the `certified_slice` received from `subnet_id`:
///  * checks its signature against the public key of `subnet_id`;
///  * ensures stream message bounds are valid and slice message bounds are
///    within stream message bounds;
///  * looks for gaps/duplicates in its `messages` w.r.t. `expected`
///    indices;
///  * and ensures signals advance monotonically and only refer to current
///    messages.
///
/// Returns the validation result, including the `CountBytes`-like estimate
/// (deterministic, but not exact) of the slice size in bytes if valid.


Returns the validation result, including the `CountBytes`-like estimate
(deterministic, but not exact) of the slice size in bytes if valid.
Returns the validation result including the ` CountBytes ` - like estimate ( deterministic but not exact ) of the slice size in bytes if valid .
fn validate_slice(
        &self,
        subnet_id: SubnetId,
        certified_slice: &CertifiedStreamSlice,
        expected: &ExpectedIndices,
        validation_context: &ValidationContext,
        state: &ReplicatedState,
    ) -> SliceValidationResult {
        let slice = match self.certified_stream_store.decode_certified_stream_slice(
            subnet_id,
            validation_context.registry_version,
            certified_slice,
        ) {
            Ok(slice) => slice,
            Err(err) => {
                info!(
                    self.log,
                    "Failed to decode stream slice from subnet {}: {}", subnet_id, err
                );
                return SliceValidationResult::Invalid(format!(
                    "Invalid stream from {}: {}",
                    subnet_id, err
                ));
            }
        };
        // Valid stream message bounds.
        if slice.header().begin > slice.header().end {
            warn!(
                self.log,
                "Stream from {}: begin index ({}) after end index ({})",
                subnet_id,
                slice.header().begin,
                slice.header().end
            );
            return SliceValidationResult::Invalid(format!(
                "Invalid stream bounds in stream from {}",
                subnet_id
            ));
        }
        // Expected message index within stream message bounds (always present in the
        // header, even for empty slices).
        if expected.message_index < slice.header().begin
            || slice.header().end < expected.message_index
        {
            warn!(
                self.log,
                "Stream from {}: expecting message {}, outside of stream bounds [{}, {})",
                subnet_id,
                expected.message_index,
                slice.header().begin,
                slice.header().end
            );
            return SliceValidationResult::Invalid(format!(
                "Unexpected messages in stream from {}",
                subnet_id
            ));
        }
        if slice.messages().is_none() && slice.header().signals_end == expected.signal_index {
            // Empty slice: no messages and no additional signals (in addition to what we
            // have in state and any intervening payloads). Not actually invalid, but
            // we don't want it in a payload.
            return SliceValidationResult::Empty;
        }
        if let Some(messages) = slice.messages() {
            // Messages in slice within stream message bounds.
            if messages.begin() < slice.header().begin || messages.end() > slice.header().end {
                warn!(
                    self.log,
                    "Stream from {}: slice bounds [{}, {}) outside of stream bounds [{}, {})",
                    subnet_id,
                    messages.begin(),
                    messages.end(),
                    slice.header().begin,
                    slice.header().end
                );
                return SliceValidationResult::Invalid(format!(
                    "Invalid slice bounds in stream from {}",
                    subnet_id
                ));
            }
            // Messages begin exactly at the expected message index.
            if messages.begin() != expected.message_index {
                warn!(
                    self.log,
                    "Stream from {}: expecting message with index {}, found {}",
                    subnet_id,
                    expected.message_index,
                    messages.begin()
                );
                return SliceValidationResult::Invalid(format!(
                    "Unexpected messages in stream from {}",
                    subnet_id
                ));
            }
            // Ensure the message limit (dictated e.g. by the backlog size) is respected.
            if let Some(msg_limit) = self.get_msg_limit(subnet_id, state) {
                if messages.len() > msg_limit {
                    warn!(
                        self.log,
                        "Stream from {}: slice length ({}) above limit ({})",
                        subnet_id,
                        messages.len(),
                        msg_limit
                    );
                    return SliceValidationResult::Invalid(format!(
                        "Stream from {}: slice length above limit",
                        subnet_id
                    ));
                }
            }
        }
        // `signals_end` must point to a message in the stream (or just past the last
        // message).
        match self.validate_signals(
            subnet_id,
            slice.header().signals_end,
            expected.signal_index,
            state,
        ) {
            SignalsValidationResult::Valid => {
                self.metrics
                    .slice_messages
                    .observe(slice.messages().map(|m| m.len()).unwrap_or(0) as f64);
                self.metrics
                    .slice_payload_size
                    .observe(certified_slice.payload.len() as f64);
                SliceValidationResult::Valid {
                    messages_end: slice
                        .messages()
                        .map(|messages| messages.end())
                        .unwrap_or(expected.message_index),
                    signals_end: slice.header().signals_end,
                    byte_size: (self.count_bytes_fn)(certified_slice)
                        .expect("Failed to unpack CertifiedStresmSlice"),
                }
            }
            SignalsValidationResult::Invalid => SliceValidationResult::Invalid(format!(
                "Unexpected signals in stream from {}",
                subnet_id
            )),
        }
    }
fn validate_slice ( & self , subnet_id : SubnetId , certified_slice : & CertifiedStreamSlice , expected : & ExpectedIndices , validation_context : & ValidationContext , state : & ReplicatedState , ) -> SliceValidationResult { let slice = match self . certified_stream_store . decode_certified_stream_slice ( subnet_id , validation_context . registry_version , certified_slice , ) { Ok ( slice ) => slice , Err ( err ) => { info ! ( self log " " subnet_id err ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id err ) ) ; } } ; // Valid stream message bounds. if slice . header ( ) . begin > slice . header ( ) . end { warn ! ( self log " " subnet_id slice header ( ) begin slice header ( ) end ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) ; } // Expected message index within stream message bounds (always present in the // header, even for empty slices). if expected . message_index < slice . header ( ) . begin || slice . header ( ) . end < expected . message_index { warn ! ( self log " " subnet_id expected message_index slice header ( ) begin slice header ( ) end ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) ; } if slice . messages ( ) . is_none ( ) && slice . header ( ) . signals_end == expected . signal_index { // Empty slice: no messages and no additional signals (in addition to what we // have in state and any intervening payloads). Not actually invalid, but // we don't want it in a payload. return SliceValidationResult :: Empty ; } if let Some ( messages ) = slice . messages ( ) { // Messages in slice within stream message bounds. if messages . begin ( ) < slice . header ( ) . begin || messages . end ( ) > slice . header ( ) . end { warn ! ( self log " " subnet_id messages begin ( ) messages end ( ) slice header ( ) begin slice header ( ) end ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) ; } // Messages begin exactly at the expected message index. if messages . begin ( ) != expected . message_index { warn ! ( self log " " subnet_id expected message_index messages begin ( ) ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) ; } // Ensure the message limit (dictated e.g. by the backlog size) is respected. if let Some ( msg_limit ) = self . get_msg_limit ( subnet_id , state ) { if messages . len ( ) > msg_limit { warn ! ( self log " " subnet_id messages len ( ) msg_limit ) ; return SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) ; } } } // `signals_end` must point to a message in the stream (or just past the last // message). match self . validate_signals ( subnet_id , slice . header ( ) . signals_end , expected . signal_index , state , ) { SignalsValidationResult :: Valid => { self . metrics . slice_messages . observe ( slice . messages ( ) . map ( | m | m . len ( ) ) . unwrap_or ( 0 ) as f64 ) ; self . metrics . slice_payload_size . observe ( certified_slice . payload . len ( ) as f64 ) ; SliceValidationResult :: Valid { messages_end : slice . messages ( ) . map ( | messages | messages . end ( ) ) . unwrap_or ( expected . message_index ) , signals_end : slice . header ( ) . signals_end , byte_size : ( self . count_bytes_fn ) ( certified_slice ) . expect ( " " ) , } } SignalsValidationResult :: Invalid => SliceValidationResult :: Invalid ( format ! ( " " subnet_id ) ) , } }
Validates the `certified_slice` received from `subnet_id`:
 checks its signature against the public key of `subnet_id`;
 ensures stream message bounds are valid and slice message bounds are
within stream message bounds;
 looks for gaps/duplicates in its `messages` w.r.t.
Validates the ` certified_slice ` received from ` subnet_id ` : checks its signature against the public key of ` subnet_id ` ; ensures stream message bounds are valid and slice message bounds are within stream message bounds ; looks for gaps / duplicates in its ` messages ` w . r . t .
&self
subnet_id
SubnetId
certified_slice
&CertifiedStreamSlice
expected
&ExpectedIndices
validation_context
&ValidationContext
state
&ReplicatedState