fn complete_candidate_block(
        &self,
        mut candidate_block: &mut CandidateBlock,
    ) -> Result<(), BlockPublisherError> {
        // Check if candidate block is already completed
        if candidate_block.summary.is_some() {
            return Ok(());
        }

        // If only injected batches (or no batches at all) have been scheduled, the block is empty
        // and the caller will just have to wait until new batches are submitted.
        if candidate_block
            .scheduled_batch_ids
            .iter()
            .all(|id| candidate_block.injected_batch_ids.contains(id))
        {
            return Err(BlockCompletionError::BlockEmpty.into());
        }

        candidate_block.scheduler.finalize()?;

        let mut batch_results = HashMap::new();

        // Wait for all injected batches to finish executing
        let mut injected_batch_ids = candidate_block.injected_batch_ids.clone();
        loop {
            match candidate_block.batch_results.recv() {
                Ok(Some(batch_result)) => {
                    injected_batch_ids.remove(batch_result.batch.batch().header_signature());

                    batch_results.insert(
                        batch_result.batch.batch().header_signature().to_string(),
                        batch_result,
                    );

                    if injected_batch_ids.is_empty() {
                        break;
                    }
                }
                Ok(None) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor returned `None` result when more results were expected".into(),
                    ))
                }
                Err(_) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor result sender disconnected when more results were expected"
                            .into(),
                    ))
                }
            }
        }

        // Wait for at least one non-injected batch to finish executing
        if batch_results.len() == candidate_block.injected_batch_ids.len() {
            // Only injected batches have finished executing, so wait for one more batch to execute
            match candidate_block.batch_results.recv() {
                Ok(Some(batch_result)) => {
                    batch_results.insert(
                        batch_result.batch.batch().header_signature().to_string(),
                        batch_result,
                    );
                }
                Ok(None) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor returned `None` result when more results were expected".into(),
                    ))
                }
                Err(_) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor result sender disconnected when more results were expected"
                            .into(),
                    ))
                }
            }
        }

        // Cancel the scheduler
        candidate_block.scheduler.cancel()?;

        // Collect any additional batch execution results that may have completed before the
        // scheduler was cancelled
        while let Ok(Some(batch_result)) = candidate_block.batch_results.recv() {
            batch_results.insert(
                batch_result.batch.batch().header_signature().to_string(),
                batch_result,
            );
        }

        // Parse the batch results into valid batches, invalid batches, and state changes
        let (valid_batches, invalid_batches, state_changes) = candidate_block
            // Get all batch execution results in the order they were originally scheduled in
            .scheduled_batch_ids
            .iter()
            .filter_map(|id| batch_results.remove(id))
            // Split batches and state changes
            .try_fold::<_, _, Result<_, BlockPublisherError>>(
                (vec![], vec![], vec![]),
                |(mut valid_batches, mut invalid_batches, mut state_changes),
                 BatchExecutionResult { batch, receipts }| {
                    let mut batch_is_invalid = false;

                    for receipt in receipts {
                        // If the transaction was invalid notify the observers and mark the batch
                        // as invalid
                        if let TransactionResult::Invalid {
                            error_message,
                            error_data,
                        } = receipt.transaction_result
                        {
                            for observer in &self.invalid_transaction_observers {
                                observer.notify_transaction_invalid(
                                    &receipt.transaction_id,
                                    &error_message,
                                    &error_data,
                                );
                            }
                            batch_is_invalid = true;
                        } else {
                            // Convert the receipt to a list of state changes and add them to the
                            // running list of state changes
                            Vec::try_from(receipt)
                                .map(|mut changes| state_changes.append(&mut changes))
                                .map_err(|err| {
                                    BlockPublisherError::Internal(format!(
                                        "Could not convert valid transaction receipt into state \
                                     changes: {}",
                                        err,
                                    ))
                                })?;
                        }
                    }

                    if batch_is_invalid {
                        debug!(
                            "Batch contains invalid transaction(s), dropping batch: {}",
                            batch.batch().header_signature()
                        );
                        invalid_batches.push(batch.take().0);
                    } else {
                        if batch.batch().trace() {
                            trace!("TRACE {}: executed", batch.batch().header_signature());
                        }
                        valid_batches.push(batch.take().0);
                    }

                    Ok((valid_batches, invalid_batches, state_changes))
                },
            )?;

        // If only injected batches (or no batches at all) were valid, the candidate block needs to
        // be restarted because this one was empty
        let num_non_injected_valid_batches = valid_batches
            .iter()
            .filter(|batch| {
                !candidate_block
                    .injected_batch_ids
                    .contains(batch.header_signature())
            })
            .count();
        if num_non_injected_valid_batches == 0 {
            let mut new_candidate_block =
                self.new_candidate_block(candidate_block.previous_block.clone())?;
            std::mem::swap(candidate_block, &mut new_candidate_block);
            return Err(BlockCompletionError::BlockEmpty.into());
        }

        // Save the batches that were executed and the resulting state root hash of the state
        // changes
        candidate_block.valid_batches = valid_batches;
        candidate_block.invalid_batches = invalid_batches;
        candidate_block.state_root_hash = Some(self.merkle_state.compute_state_id(
            &hex::encode(candidate_block.previous_block.header().state_root_hash()),
            &state_changes,
        )?);

        // Compute and save the summary
        let batch_ids = candidate_block
            .valid_batches
            .iter()
            .map(|batch| batch.header_signature().to_string())
            .collect::<Vec<_>>();
        candidate_block.summary = Some(sha256_digest_strs(&batch_ids));

        Ok(())
    }
/// Completes the in-progress block and saves the results
///
/// This is a helper method for the `summarize_block` and `finalize_block` methods that does the
/// following:
///
/// * Ensures the block is not empty and does not contain only injected batches; if the block
///   does not contain any non-injected batches, a `BlockEmpty` error is returned and the
///   candidate block is left uncompleted.
/// * Deschedules any unexecuted batches
/// * Finalizes the scheduler and collects all execution results
/// * Computes and saves:
///   * All executed batches
///   * The resulting state root hash of the block
///   * The summary of the batches
Completes the in-progress block and saves the results
This is a helper method for the `summarize_block` and `finalize_block` methods that does the
following.

Ensures the block is not empty and does not contain only injected batches; if the block
does not contain any non-injected batches, a `BlockEmpty` error is returned and the
candidate block is left uncompleted.
Deschedules any unexecuted batches
Finalizes the scheduler and collects all execution results
Computes and saves:
All executed batches
The resulting state root hash of the block
The summary of the batches
Completes the in - progress block and saves the results This is a helper method for the ` summarize_block ` and ` finalize_block ` methods that does the following . Ensures the block is not empty and does not contain only injected batches ; if the block does not contain any non - injected batches a ` BlockEmpty ` error is returned and the candidate block is left uncompleted . Deschedules any unexecuted batches Finalizes the scheduler and collects all execution results Computes and saves : All executed batches The resulting state root hash of the block The summary of the batches
fn complete_candidate_block(
        &self,
        mut candidate_block: &mut CandidateBlock,
    ) -> Result<(), BlockPublisherError> {
        // Check if candidate block is already completed
        if candidate_block.summary.is_some() {
            return Ok(());
        }
        // If only injected batches (or no batches at all) have been scheduled, the block is empty
        // and the caller will just have to wait until new batches are submitted.
        if candidate_block
            .scheduled_batch_ids
            .iter()
            .all(|id| candidate_block.injected_batch_ids.contains(id))
        {
            return Err(BlockCompletionError::BlockEmpty.into());
        }
        candidate_block.scheduler.finalize()?;
        let mut batch_results = HashMap::new();
        // Wait for all injected batches to finish executing
        let mut injected_batch_ids = candidate_block.injected_batch_ids.clone();
        loop {
            match candidate_block.batch_results.recv() {
                Ok(Some(batch_result)) => {
                    injected_batch_ids.remove(batch_result.batch.batch().header_signature());
                    batch_results.insert(
                        batch_result.batch.batch().header_signature().to_string(),
                        batch_result,
                    );
                    if injected_batch_ids.is_empty() {
                        break;
                    }
                }
                Ok(None) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor returned `None` result when more results were expected".into(),
                    ))
                }
                Err(_) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor result sender disconnected when more results were expected"
                            .into(),
                    ))
                }
            }
        }
        // Wait for at least one non-injected batch to finish executing
        if batch_results.len() == candidate_block.injected_batch_ids.len() {
            // Only injected batches have finished executing, so wait for one more batch to execute
            match candidate_block.batch_results.recv() {
                Ok(Some(batch_result)) => {
                    batch_results.insert(
                        batch_result.batch.batch().header_signature().to_string(),
                        batch_result,
                    );
                }
                Ok(None) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor returned `None` result when more results were expected".into(),
                    ))
                }
                Err(_) => {
                    return Err(BlockPublisherError::Internal(
                        "Executor result sender disconnected when more results were expected"
                            .into(),
                    ))
                }
            }
        }
        // Cancel the scheduler
        candidate_block.scheduler.cancel()?;
        // Collect any additional batch execution results that may have completed before the
        // scheduler was cancelled
        while let Ok(Some(batch_result)) = candidate_block.batch_results.recv() {
            batch_results.insert(
                batch_result.batch.batch().header_signature().to_string(),
                batch_result,
            );
        }
        // Parse the batch results into valid batches, invalid batches, and state changes
        let (valid_batches, invalid_batches, state_changes) = candidate_block
            // Get all batch execution results in the order they were originally scheduled in
            .scheduled_batch_ids
            .iter()
            .filter_map(|id| batch_results.remove(id))
            // Split batches and state changes
            .try_fold::<_, _, Result<_, BlockPublisherError>>(
                (vec![], vec![], vec![]),
                |(mut valid_batches, mut invalid_batches, mut state_changes),
                 BatchExecutionResult { batch, receipts }| {
                    let mut batch_is_invalid = false;
                    for receipt in receipts {
                        // If the transaction was invalid notify the observers and mark the batch
                        // as invalid
                        if let TransactionResult::Invalid {
                            error_message,
                            error_data,
                        } = receipt.transaction_result
                        {
                            for observer in &self.invalid_transaction_observers {
                                observer.notify_transaction_invalid(
                                    &receipt.transaction_id,
                                    &error_message,
                                    &error_data,
                                );
                            }
                            batch_is_invalid = true;
                        } else {
                            // Convert the receipt to a list of state changes and add them to the
                            // running list of state changes
                            Vec::try_from(receipt)
                                .map(|mut changes| state_changes.append(&mut changes))
                                .map_err(|err| {
                                    BlockPublisherError::Internal(format!(
                                        "Could not convert valid transaction receipt into state \
                                     changes: {}",
                                        err,
                                    ))
                                })?;
                        }
                    }
                    if batch_is_invalid {
                        debug!(
                            "Batch contains invalid transaction(s), dropping batch: {}",
                            batch.batch().header_signature()
                        );
                        invalid_batches.push(batch.take().0);
                    } else {
                        if batch.batch().trace() {
                            trace!("TRACE {}: executed", batch.batch().header_signature());
                        }
                        valid_batches.push(batch.take().0);
                    }
                    Ok((valid_batches, invalid_batches, state_changes))
                },
            )?;
        // If only injected batches (or no batches at all) were valid, the candidate block needs to
        // be restarted because this one was empty
        let num_non_injected_valid_batches = valid_batches
            .iter()
            .filter(|batch| {
                !candidate_block
                    .injected_batch_ids
                    .contains(batch.header_signature())
            })
            .count();
        if num_non_injected_valid_batches == 0 {
            let mut new_candidate_block =
                self.new_candidate_block(candidate_block.previous_block.clone())?;
            std::mem::swap(candidate_block, &mut new_candidate_block);
            return Err(BlockCompletionError::BlockEmpty.into());
        }
        // Save the batches that were executed and the resulting state root hash of the state
        // changes
        candidate_block.valid_batches = valid_batches;
        candidate_block.invalid_batches = invalid_batches;
        candidate_block.state_root_hash = Some(self.merkle_state.compute_state_id(
            &hex::encode(candidate_block.previous_block.header().state_root_hash()),
            &state_changes,
        )?);
        // Compute and save the summary
        let batch_ids = candidate_block
            .valid_batches
            .iter()
            .map(|batch| batch.header_signature().to_string())
            .collect::<Vec<_>>();
        candidate_block.summary = Some(sha256_digest_strs(&batch_ids));
        Ok(())
    }
fn complete_candidate_block ( & self , mut candidate_block : & mut CandidateBlock , ) -> Result < ( ) , BlockPublisherError > { // Check if candidate block is already completed if candidate_block . summary . is_some ( ) { return Ok ( ( ) ) ; } // If only injected batches (or no batches at all) have been scheduled, the block is empty // and the caller will just have to wait until new batches are submitted. if candidate_block . scheduled_batch_ids . iter ( ) . all ( | id | candidate_block . injected_batch_ids . contains ( id ) ) { return Err ( BlockCompletionError :: BlockEmpty . into ( ) ) ; } candidate_block . scheduler . finalize ( ) ? ; let mut batch_results = HashMap :: new ( ) ; // Wait for all injected batches to finish executing let mut injected_batch_ids = candidate_block . injected_batch_ids . clone ( ) ; loop { match candidate_block . batch_results . recv ( ) { Ok ( Some ( batch_result ) ) => { injected_batch_ids . remove ( batch_result . batch . batch ( ) . header_signature ( ) ) ; batch_results . insert ( batch_result . batch . batch ( ) . header_signature ( ) . to_string ( ) , batch_result , ) ; if injected_batch_ids . is_empty ( ) { break ; } } Ok ( None ) => { return Err ( BlockPublisherError :: Internal ( " " . into ( ) , ) ) } Err ( _ ) => { return Err ( BlockPublisherError :: Internal ( " " . into ( ) , ) ) } } } // Wait for at least one non-injected batch to finish executing if batch_results . len ( ) == candidate_block . injected_batch_ids . len ( ) { // Only injected batches have finished executing, so wait for one more batch to execute match candidate_block . batch_results . recv ( ) { Ok ( Some ( batch_result ) ) => { batch_results . insert ( batch_result . batch . batch ( ) . header_signature ( ) . to_string ( ) , batch_result , ) ; } Ok ( None ) => { return Err ( BlockPublisherError :: Internal ( " " . into ( ) , ) ) } Err ( _ ) => { return Err ( BlockPublisherError :: Internal ( " " . into ( ) , ) ) } } } // Cancel the scheduler candidate_block . scheduler . cancel ( ) ? ; // Collect any additional batch execution results that may have completed before the // scheduler was cancelled while let Ok ( Some ( batch_result ) ) = candidate_block . batch_results . recv ( ) { batch_results . insert ( batch_result . batch . batch ( ) . header_signature ( ) . to_string ( ) , batch_result , ) ; } // Parse the batch results into valid batches, invalid batches, and state changes let ( valid_batches , invalid_batches , state_changes ) = candidate_block // Get all batch execution results in the order they were originally scheduled in . scheduled_batch_ids . iter ( ) . filter_map ( | id | batch_results . remove ( id ) ) // Split batches and state changes . try_fold :: < _ , _ , Result < _ , BlockPublisherError > > ( ( vec ! [ ] , vec ! [ ] , vec ! [ ] ) , | ( mut valid_batches , mut invalid_batches , mut state_changes ) , BatchExecutionResult { batch , receipts } | { let mut batch_is_invalid = false ; for receipt in receipts { // If the transaction was invalid notify the observers and mark the batch // as invalid if let TransactionResult :: Invalid { error_message , error_data , } = receipt . transaction_result { for observer in & self . invalid_transaction_observers { observer . notify_transaction_invalid ( & receipt . transaction_id , & error_message , & error_data , ) ; } batch_is_invalid = true ; } else { // Convert the receipt to a list of state changes and add them to the // running list of state changes Vec :: try_from ( receipt ) . map ( | mut changes | state_changes . append ( & mut changes ) ) . map_err ( | err | { BlockPublisherError :: Internal ( format ! ( " \
 " err ) ) } ) ? ; } } if batch_is_invalid { debug ! ( " " batch batch ( ) header_signature ( ) ) ; invalid_batches . push ( batch . take ( ) . 0 ) ; } else { if batch . batch ( ) . trace ( ) { trace ! ( " " batch batch ( ) header_signature ( ) ) ; } valid_batches . push ( batch . take ( ) . 0 ) ; } Ok ( ( valid_batches , invalid_batches , state_changes ) ) } , ) ? ; // If only injected batches (or no batches at all) were valid, the candidate block needs to // be restarted because this one was empty let num_non_injected_valid_batches = valid_batches . iter ( ) . filter ( | batch | { ! candidate_block . injected_batch_ids . contains ( batch . header_signature ( ) ) } ) . count ( ) ; if num_non_injected_valid_batches == 0 { let mut new_candidate_block = self . new_candidate_block ( candidate_block . previous_block . clone ( ) ) ? ; std :: mem :: swap ( candidate_block , & mut new_candidate_block ) ; return Err ( BlockCompletionError :: BlockEmpty . into ( ) ) ; } // Save the batches that were executed and the resulting state root hash of the state // changes candidate_block . valid_batches = valid_batches ; candidate_block . invalid_batches = invalid_batches ; candidate_block . state_root_hash = Some ( self . merkle_state . compute_state_id ( & hex :: encode ( candidate_block . previous_block . header ( ) . state_root_hash ( ) ) , & state_changes , ) ? ) ; // Compute and save the summary let batch_ids = candidate_block . valid_batches . iter ( ) . map ( | batch | batch . header_signature ( ) . to_string ( ) ) . collect :: < Vec < _ > > ( ) ; candidate_block . summary = Some ( sha256_digest_strs ( & batch_ids ) ) ; Ok ( ( ) ) }
Completes the in-progress block and saves the results
This is a helper method for the `summarize_block` and `finalize_block` methods that does the
following:
Completes the in - progress block and saves the results This is a helper method for the ` summarize_block ` and ` finalize_block ` methods that does the following :
&self
candidate_block
&mut CandidateBlock
&self
candidate_block
&mut CandidateBlock