/// Validate all the batches provided. If the state hash matches the block state hash,
/// commit changes.
Validate all the batches provided. If the state hash matches the block state hash,
commit changes.
Validate all the batches provided . If the state hash matches the block state hash commit changes .
fn execute_batches(
    previous_state_root: &str,
    execution_task_submitter: &ExecutionTaskSubmitter,
    merkle_state: &CborMerkleState,
    batches: Vec<Batch>,
    scheduler_factory: &dyn SchedulerFactory,
    expected_state_root_hash: String,
) -> Result<(), StateVerificationError> {
    let mut scheduler = scheduler_factory.create_scheduler(previous_state_root.to_string())?;
    let (result_tx, result_rx): (Sender<SchedulerEvent>, Receiver<SchedulerEvent>) = channel();
    let error_tx = result_tx.clone();
    // Add callback to convert batch result option to scheduler event
    scheduler.set_result_callback(Box::new(move |batch_result| {
        let scheduler_event = match batch_result {
            Some(result) => SchedulerEvent::Result(result),
            None => SchedulerEvent::Complete,
        };
        if result_tx.send(scheduler_event).is_err() {
            error!("Unable to send batch result; receiver must have dropped");
        }
    }))?;
    // add callback to convert error into scheduler event
    scheduler.set_error_callback(Box::new(move |err| {
        if error_tx.send(SchedulerEvent::Error(err)).is_err() {
            error!("Unable to send scheduler error; receiver must have dropped");
        }
    }))?;
    for batch in batches.into_iter() {
        let batch_pair = batch.into_pair().map_err(|err| {
            StateVerificationError::ExecutionError(format!(
                "Unable to convert batch into BatchPair: {:?}",
                err
            ))
        })?;
        scheduler.add_batch(batch_pair).map_err(|err| {
            StateVerificationError::ExecutionError(format!(
                "While adding a batch to the schedule: {:?}",
                err
            ))
        })?;
    }
    scheduler.finalize().map_err(|err| {
        StateVerificationError::ExecutionError(format!(
            "During call to scheduler.finalize: {:?}",
            err
        ))
    })?;
    execution_task_submitter
        .submit(scheduler.take_task_iterator()?, scheduler.new_notifier()?)
        .map_err(|err| {
            StateVerificationError::ExecutionError(format!(
                "During call to ExecutionTaskSubmitter.submit: {}",
                err
            ))
        })?;
    let mut execution_results = vec![];
    loop {
        match result_rx.recv() {
            Ok(SchedulerEvent::Result(result)) => execution_results.push(result),
            Ok(SchedulerEvent::Complete) => break,
            Ok(SchedulerEvent::Error(err)) => {
                return Err(StateVerificationError::ExecutionError(format!(
                    "During execution: {:?}",
                    err
                )))
            }
            Err(err) => {
                return Err(StateVerificationError::ExecutionError(format!(
                    "Error while trying to receive scheduler event: {:?}",
                    err
                )))
            }
        }
    }
    let mut changes = vec![];
    for batch_result in execution_results {
        if !batch_result.receipts.is_empty() {
            for receipt in batch_result.receipts {
                match receipt.transaction_result.clone() {
                    TransactionResult::Invalid { .. } => {
                        return Err(StateVerificationError::ExecutionError(format!(
                            "Failed validation: batch {} was invalid",
                            batch_result.batch.batch().header_signature(),
                        )));
                    }
                    TransactionResult::Valid { state_changes, .. } => {
                        changes.append(
                            &mut state_changes.into_iter().map(StateChange::from).collect(),
                        );
                    }
                };
            }
        } else {
            return Err(StateVerificationError::ExecutionError(format!(
                "Failed validation: batch {} did not have transaction results",
                &batch_result.batch.batch().header_signature(),
            )));
        };
    }
    let new_root = merkle_state
        .compute_state_id(&previous_state_root.to_string(), &changes)
        .map_err(|err| {
            StateVerificationError::ExecutionError(format!(
                "During ending state hash calculation: {:?}",
                err
            ))
        })?;
    if new_root != expected_state_root_hash {
        return Err(StateVerificationError::InvalidChainError(format!(
            "Computed state root {} does not match state root in block {}",
            new_root, expected_state_root_hash
        )));
    };
    merkle_state
        .commit(&previous_state_root.to_string(), &changes)
        .map_err(|err| {
            StateVerificationError::ExecutionError(format!(
                "During ending state hash calculation: {:?}",
                err
            ))
        })?;
    Ok(())
}
fn execute_batches ( previous_state_root : & str , execution_task_submitter : & ExecutionTaskSubmitter , merkle_state : & CborMerkleState , batches : Vec < Batch > , scheduler_factory : & dyn SchedulerFactory , expected_state_root_hash : String , ) -> Result < ( ) , StateVerificationError > { let mut scheduler = scheduler_factory . create_scheduler ( previous_state_root . to_string ( ) ) ? ; let ( result_tx , result_rx ) : ( Sender < SchedulerEvent > , Receiver < SchedulerEvent > ) = channel ( ) ; let error_tx = result_tx . clone ( ) ; // Add callback to convert batch result option to scheduler event scheduler . set_result_callback ( Box :: new ( move | batch_result | { let scheduler_event = match batch_result { Some ( result ) => SchedulerEvent :: Result ( result ) , None => SchedulerEvent :: Complete , } ; if result_tx . send ( scheduler_event ) . is_err ( ) { error ! ( " " ) ; } } ) ) ? ; // add callback to convert error into scheduler event scheduler . set_error_callback ( Box :: new ( move | err | { if error_tx . send ( SchedulerEvent :: Error ( err ) ) . is_err ( ) { error ! ( " " ) ; } } ) ) ? ; for batch in batches . into_iter ( ) { let batch_pair = batch . into_pair ( ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; scheduler . add_batch ( batch_pair ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; } scheduler . finalize ( ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; execution_task_submitter . submit ( scheduler . take_task_iterator ( ) ? , scheduler . new_notifier ( ) ? ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; let mut execution_results = vec ! [ ] ; loop { match result_rx . recv ( ) { Ok ( SchedulerEvent :: Result ( result ) ) => execution_results . push ( result ) , Ok ( SchedulerEvent :: Complete ) => break , Ok ( SchedulerEvent :: Error ( err ) ) => { return Err ( StateVerificationError :: ExecutionError ( format ! ( " " err ) ) ) } Err ( err ) => { return Err ( StateVerificationError :: ExecutionError ( format ! ( " " err ) ) ) } } } let mut changes = vec ! [ ] ; for batch_result in execution_results { if ! batch_result . receipts . is_empty ( ) { for receipt in batch_result . receipts { match receipt . transaction_result . clone ( ) { TransactionResult :: Invalid { .. } => { return Err ( StateVerificationError :: ExecutionError ( format ! ( " " batch_result batch batch ( ) header_signature ( ) ) ) ) ; } TransactionResult :: Valid { state_changes , .. } => { changes . append ( & mut state_changes . into_iter ( ) . map ( StateChange :: from ) . collect ( ) , ) ; } } ; } } else { return Err ( StateVerificationError :: ExecutionError ( format ! ( " " batch_result batch batch ( ) header_signature ( ) ) ) ) ; } ; } let new_root = merkle_state . compute_state_id ( & previous_state_root . to_string ( ) , & changes ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; if new_root != expected_state_root_hash { return Err ( StateVerificationError :: InvalidChainError ( format ! ( " " new_root expected_state_root_hash ) ) ) ; } ; merkle_state . commit ( & previous_state_root . to_string ( ) , & changes ) . map_err ( | err | { StateVerificationError :: ExecutionError ( format ! ( " " err ) ) } ) ? ; Ok ( ( ) ) }
Validate all the batches provided.
Validate all the batches provided .
previous_state_root
&str
execution_task_submitter
&ExecutionTaskSubmitter
merkle_state
&CborMerkleState
batches
Vec<Batch>
scheduler_factory
&dyn SchedulerFactory
expected_state_root_hash
String