fn on_state_change(
        &self,
        pool: &dyn ConsensusPool,
        ingress_pool: &dyn IngressPoolSelect,
    ) -> ChangeSet {
        let pool_reader = PoolReader::new(pool);
        trace!(self.log, "on_state_change");

        // Load new transcripts, remove outdated keys.
        self.dkg_key_manager
            .lock()
            .unwrap()
            .on_state_change(&pool_reader);

        // For non-root subnets, we must halt if our registry is outdated
        if let Ok(false) = is_root_subnet(
            self.registry_client.as_ref(),
            self.subnet_id,
            self.registry_client.get_latest_version(),
        ) {
            if let Err(e) = self.check_registry_outdated() {
                info!(
                    every_n_seconds => 5,
                    self.log,
                    "consensus is halted due to outdated registry. {:?}", e
                );
                return ChangeSet::new();
            }
        }

        // Consensus halts if instructed by the registry
        if self.should_halt_by_subnet_record() {
            info!(
                every_n_seconds => 5,
                self.log,
                "consensus is halted by instructions of the subnet record in the registry"
            );
            return ChangeSet::new();
        }

        let finalize = || {
            self.call_with_metrics(ConsensusSubcomponent::Finalizer, || {
                add_all_to_validated(self.finalizer.on_state_change(&pool_reader))
            })
        };
        let make_catch_up_package = || {
            self.call_with_metrics(ConsensusSubcomponent::CatchUpPackageMaker, || {
                add_to_validated(self.catch_up_package_maker.on_state_change(&pool_reader))
            })
        };
        let aggregate = || {
            self.call_with_metrics(ConsensusSubcomponent::Aggregator, || {
                add_all_to_validated(self.aggregator.on_state_change(&pool_reader))
            })
        };
        let notarize = || {
            self.call_with_metrics(ConsensusSubcomponent::Notary, || {
                add_all_to_validated(self.notary.on_state_change(&pool_reader))
            })
        };
        let make_random_beacon = || {
            self.call_with_metrics(ConsensusSubcomponent::RandomBeaconMaker, || {
                add_to_validated(self.random_beacon_maker.on_state_change(&pool_reader))
            })
        };
        let make_random_tape = || {
            self.call_with_metrics(ConsensusSubcomponent::RandomTapeMaker, || {
                add_all_to_validated(self.random_tape_maker.on_state_change(&pool_reader))
            })
        };
        let make_block = || {
            self.call_with_metrics(ConsensusSubcomponent::BlockMaker, || {
                add_to_validated(self.block_maker.on_state_change(&pool_reader, ingress_pool))
            })
        };
        let validate = || {
            self.call_with_metrics(ConsensusSubcomponent::Validator, || {
                self.validator.on_state_change(&pool_reader)
            })
        };
        let purge = || {
            self.call_with_metrics(ConsensusSubcomponent::Purger, || {
                self.purger.on_state_change(&pool_reader)
            })
        };
        let calls: [&'_ dyn Fn() -> ChangeSet; 9] = [
            &finalize,
            &make_catch_up_package,
            &purge,
            &aggregate,
            &notarize,
            &make_random_beacon,
            &make_random_tape,
            &make_block,
            &validate,
        ];

        let changeset = self.schedule.call_next(&calls);

        if let Some(settings) = get_notarization_delay_settings(
            &self.log,
            &*self.registry_client,
            self.subnet_id,
            self.registry_client.get_latest_version(),
        ) {
            let unit_delay = settings.unit_delay;
            let current_time = self.time_source.get_relative_time();
            for (component, last_invoked_time) in self.last_invoked.borrow().iter() {
                let time_since_last_invoked = current_time - *last_invoked_time;
                let component_name = component.as_ref();
                self.metrics
                    .time_since_last_invoked
                    .with_label_values(&[component_name])
                    .set(time_since_last_invoked.as_secs_f64());

                // Log starvation if configured
                if self.config.detect_starvation() && time_since_last_invoked > unit_delay {
                    self.metrics
                        .starvation_counter
                        .with_label_values(&[component_name])
                        .inc();

                    warn!(
                        every_n_seconds => 5,
                        self.log,
                        "starvation detected: {:?} has not been invoked for {:?}",
                        component,
                        time_since_last_invoked
                    );
                }
            }
        }

        #[cfg(feature = "malicious_code")]
        if self.malicious_flags.is_consensus_malicious() {
            crate::consensus::malicious_consensus::maliciously_alter_changeset(
                &pool_reader,
                ingress_pool,
                changeset,
                &self.malicious_flags,
                &self.block_maker,
                &self.finalizer,
                &self.notary,
                &self.log,
            )
        } else {
            changeset
        }

        #[cfg(not(feature = "malicious_code"))]
        changeset
    }
/// Invoke `on_state_change` on each subcomponent in order.
/// Return the first non-empty [ChangeSet] as returned by a subcomponent.
/// Otherwise return an empty [ChangeSet] if all subcomponents return
/// empty.
///
/// There are two decisions that ConsensusImpl makes:
///
/// 1. It must return immediately if one of the subcomponent returns a
/// non-empty [ChangeSet]. It is important that a [ChangeSet] is fully
/// applied to the pool or timer before another subcomponent uses
/// them, because each subcomponent expects to see full state in order to
/// make correct decisions on what to do next.
///
/// 2. The order in which subcomponents are called also matters. At the
/// moment it is important to call finalizer first, because otherwise
/// we'll just keep producing notarized blocks indefintely without
/// finalizing anything, due to the above decision of having to return
/// early. The order of the rest subcomponents decides whom is given
/// a priority, but it should not affect liveness or correctness.
Invoke `on_state_change` on each subcomponent in order.
Return the first non-empty [ChangeSet] as returned by a subcomponent.
Otherwise return an empty [ChangeSet] if all subcomponents return
empty.

There are two decisions that ConsensusImpl makes.

1. It must return immediately if one of the subcomponent returns a
non-empty [ChangeSet]. It is important that a [ChangeSet] is fully
applied to the pool or timer before another subcomponent uses
them, because each subcomponent expects to see full state in order to
make correct decisions on what to do next.

2. The order in which subcomponents are called also matters. At the
moment it is important to call finalizer first, because otherwise
we'll just keep producing notarized blocks indefintely without
finalizing anything, due to the above decision of having to return
early. The order of the rest subcomponents decides whom is given
a priority, but it should not affect liveness or correctness.
Invoke ` on_state_change ` on each subcomponent in order . Return the first non - empty [ ChangeSet ] as returned by a subcomponent . Otherwise return an empty [ ChangeSet ] if all subcomponents return empty . There are two decisions that ConsensusImpl makes . 1 . It must return immediately if one of the subcomponent returns a non - empty [ ChangeSet ] . It is important that a [ ChangeSet ] is fully applied to the pool or timer before another subcomponent uses them because each subcomponent expects to see full state in order to make correct decisions on what to do next . 2 . The order in which subcomponents are called also matters . At the moment it is important to call finalizer first because otherwise we ' ll just keep producing notarized blocks indefintely without finalizing anything due to the above decision of having to return early . The order of the rest subcomponents decides whom is given a priority but it should not affect liveness or correctness .
fn on_state_change(
        &self,
        pool: &dyn ConsensusPool,
        ingress_pool: &dyn IngressPoolSelect,
    ) -> ChangeSet {
        let pool_reader = PoolReader::new(pool);
        trace!(self.log, "on_state_change");
        // Load new transcripts, remove outdated keys.
        self.dkg_key_manager
            .lock()
            .unwrap()
            .on_state_change(&pool_reader);
        // For non-root subnets, we must halt if our registry is outdated
        if let Ok(false) = is_root_subnet(
            self.registry_client.as_ref(),
            self.subnet_id,
            self.registry_client.get_latest_version(),
        ) {
            if let Err(e) = self.check_registry_outdated() {
                info!(
                    every_n_seconds => 5,
                    self.log,
                    "consensus is halted due to outdated registry. {:?}", e
                );
                return ChangeSet::new();
            }
        }
        // Consensus halts if instructed by the registry
        if self.should_halt_by_subnet_record() {
            info!(
                every_n_seconds => 5,
                self.log,
                "consensus is halted by instructions of the subnet record in the registry"
            );
            return ChangeSet::new();
        }
        let finalize = || {
            self.call_with_metrics(ConsensusSubcomponent::Finalizer, || {
                add_all_to_validated(self.finalizer.on_state_change(&pool_reader))
            })
        };
        let make_catch_up_package = || {
            self.call_with_metrics(ConsensusSubcomponent::CatchUpPackageMaker, || {
                add_to_validated(self.catch_up_package_maker.on_state_change(&pool_reader))
            })
        };
        let aggregate = || {
            self.call_with_metrics(ConsensusSubcomponent::Aggregator, || {
                add_all_to_validated(self.aggregator.on_state_change(&pool_reader))
            })
        };
        let notarize = || {
            self.call_with_metrics(ConsensusSubcomponent::Notary, || {
                add_all_to_validated(self.notary.on_state_change(&pool_reader))
            })
        };
        let make_random_beacon = || {
            self.call_with_metrics(ConsensusSubcomponent::RandomBeaconMaker, || {
                add_to_validated(self.random_beacon_maker.on_state_change(&pool_reader))
            })
        };
        let make_random_tape = || {
            self.call_with_metrics(ConsensusSubcomponent::RandomTapeMaker, || {
                add_all_to_validated(self.random_tape_maker.on_state_change(&pool_reader))
            })
        };
        let make_block = || {
            self.call_with_metrics(ConsensusSubcomponent::BlockMaker, || {
                add_to_validated(self.block_maker.on_state_change(&pool_reader, ingress_pool))
            })
        };
        let validate = || {
            self.call_with_metrics(ConsensusSubcomponent::Validator, || {
                self.validator.on_state_change(&pool_reader)
            })
        };
        let purge = || {
            self.call_with_metrics(ConsensusSubcomponent::Purger, || {
                self.purger.on_state_change(&pool_reader)
            })
        };
        let calls: [&'_ dyn Fn() -> ChangeSet; 9] = [
            &finalize,
            &make_catch_up_package,
            &purge,
            &aggregate,
            &notarize,
            &make_random_beacon,
            &make_random_tape,
            &make_block,
            &validate,
        ];
        let changeset = self.schedule.call_next(&calls);
        if let Some(settings) = get_notarization_delay_settings(
            &self.log,
            &*self.registry_client,
            self.subnet_id,
            self.registry_client.get_latest_version(),
        ) {
            let unit_delay = settings.unit_delay;
            let current_time = self.time_source.get_relative_time();
            for (component, last_invoked_time) in self.last_invoked.borrow().iter() {
                let time_since_last_invoked = current_time - *last_invoked_time;
                let component_name = component.as_ref();
                self.metrics
                    .time_since_last_invoked
                    .with_label_values(&[component_name])
                    .set(time_since_last_invoked.as_secs_f64());
                // Log starvation if configured
                if self.config.detect_starvation() && time_since_last_invoked > unit_delay {
                    self.metrics
                        .starvation_counter
                        .with_label_values(&[component_name])
                        .inc();
                    warn!(
                        every_n_seconds => 5,
                        self.log,
                        "starvation detected: {:?} has not been invoked for {:?}",
                        component,
                        time_since_last_invoked
                    );
                }
            }
        }
        #[cfg(feature = "malicious_code")]
        if self.malicious_flags.is_consensus_malicious() {
            crate::consensus::malicious_consensus::maliciously_alter_changeset(
                &pool_reader,
                ingress_pool,
                changeset,
                &self.malicious_flags,
                &self.block_maker,
                &self.finalizer,
                &self.notary,
                &self.log,
            )
        } else {
            changeset
        }
        #[cfg(not(feature = "malicious_code"))]
        changeset
    }
fn on_state_change ( & self , pool : & dyn ConsensusPool , ingress_pool : & dyn IngressPoolSelect , ) -> ChangeSet { let pool_reader = PoolReader :: new ( pool ) ; trace ! ( self log " " ) ; // Load new transcripts, remove outdated keys. self . dkg_key_manager . lock ( ) . unwrap ( ) . on_state_change ( & pool_reader ) ; // For non-root subnets, we must halt if our registry is outdated if let Ok ( false ) = is_root_subnet ( self . registry_client . as_ref ( ) , self . subnet_id , self . registry_client . get_latest_version ( ) , ) { if let Err ( e ) = self . check_registry_outdated ( ) { info ! ( every_n_seconds 5 self log " " e ) ; return ChangeSet :: new ( ) ; } } // Consensus halts if instructed by the registry if self . should_halt_by_subnet_record ( ) { info ! ( every_n_seconds 5 self log " " ) ; return ChangeSet :: new ( ) ; } let finalize = | | { self . call_with_metrics ( ConsensusSubcomponent :: Finalizer , | | { add_all_to_validated ( self . finalizer . on_state_change ( & pool_reader ) ) } ) } ; let make_catch_up_package = | | { self . call_with_metrics ( ConsensusSubcomponent :: CatchUpPackageMaker , | | { add_to_validated ( self . catch_up_package_maker . on_state_change ( & pool_reader ) ) } ) } ; let aggregate = | | { self . call_with_metrics ( ConsensusSubcomponent :: Aggregator , | | { add_all_to_validated ( self . aggregator . on_state_change ( & pool_reader ) ) } ) } ; let notarize = | | { self . call_with_metrics ( ConsensusSubcomponent :: Notary , | | { add_all_to_validated ( self . notary . on_state_change ( & pool_reader ) ) } ) } ; let make_random_beacon = | | { self . call_with_metrics ( ConsensusSubcomponent :: RandomBeaconMaker , | | { add_to_validated ( self . random_beacon_maker . on_state_change ( & pool_reader ) ) } ) } ; let make_random_tape = | | { self . call_with_metrics ( ConsensusSubcomponent :: RandomTapeMaker , | | { add_all_to_validated ( self . random_tape_maker . on_state_change ( & pool_reader ) ) } ) } ; let make_block = | | { self . call_with_metrics ( ConsensusSubcomponent :: BlockMaker , | | { add_to_validated ( self . block_maker . on_state_change ( & pool_reader , ingress_pool ) ) } ) } ; let validate = | | { self . call_with_metrics ( ConsensusSubcomponent :: Validator , | | { self . validator . on_state_change ( & pool_reader ) } ) } ; let purge = | | { self . call_with_metrics ( ConsensusSubcomponent :: Purger , | | { self . purger . on_state_change ( & pool_reader ) } ) } ; let calls : [ & ' _ dyn Fn ( ) -> ChangeSet ; 9 ] = [ & finalize , & make_catch_up_package , & purge , & aggregate , & notarize , & make_random_beacon , & make_random_tape , & make_block , & validate , ] ; let changeset = self . schedule . call_next ( & calls ) ; if let Some ( settings ) = get_notarization_delay_settings ( & self . log , & * self . registry_client , self . subnet_id , self . registry_client . get_latest_version ( ) , ) { let unit_delay = settings . unit_delay ; let current_time = self . time_source . get_relative_time ( ) ; for ( component , last_invoked_time ) in self . last_invoked . borrow ( ) . iter ( ) { let time_since_last_invoked = current_time - * last_invoked_time ; let component_name = component . as_ref ( ) ; self . metrics . time_since_last_invoked . with_label_values ( & [ component_name ] ) . set ( time_since_last_invoked . as_secs_f64 ( ) ) ; // Log starvation if configured if self . config . detect_starvation ( ) && time_since_last_invoked > unit_delay { self . metrics . starvation_counter . with_label_values ( & [ component_name ] ) . inc ( ) ; warn ! ( every_n_seconds 5 self log " " component time_since_last_invoked ) ; } } } # [ cfg ( feature " " ) ] if self . malicious_flags . is_consensus_malicious ( ) { crate :: consensus :: malicious_consensus :: maliciously_alter_changeset ( & pool_reader , ingress_pool , changeset , & self . malicious_flags , & self . block_maker , & self . finalizer , & self . notary , & self . log , ) } else { changeset } # [ cfg ( not ( feature " " ) ) ] changeset }
Invoke `on_state_change` on each subcomponent in order.
Invoke ` on_state_change ` on each subcomponent in order .
&self
pool
&dyn ConsensusPool
ingress_pool
&dyn IngressPoolSelect
&self
pool
&dyn ConsensusPool
ingress_pool
&dyn IngressPoolSelect