Skip to content

Commit

Permalink
Use a cuadratic backup curve for reactivate transactions
Browse files Browse the repository at this point in the history
  • Loading branch information
@viquezclaudio
viquezclaudio committed Jan 13, 2025
1 parent 124258f commit fadf7cb
Show file tree
Hide file tree
Showing 3 changed files with 118 additions and 113 deletions.
5 changes: 4 additions & 1 deletion validator/src/micro.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ use std::{
pin::Pin,
sync::Arc,
task::{Context, Poll},
thread::current,

Check warning on line 6 in validator/src/micro.rs

View workflow job for this annotation

GitHub Actions / Clippy Report

unused import: `thread::current` 

warning: unused import: `thread::current`
 --> validator/src/micro.rs:6:5
  |
6 |     thread::current,
  |     ^^^^^^^^^^^^^^^
  |
  = note: `#[warn(unused_imports)]` on by default
time::{Duration, SystemTime},
};

Expand Down Expand Up @@ -198,7 +199,9 @@ impl<TValidatorNetwork: ValidatorNetwork + 'static> NextProduceMicroBlockEvent<T
continue;
}

self.health_state.write().blk_cnt += 1;
// Each successfull block will decrease the number of inactivations
let current_inactivations = self.health_state.read().inactivations;
self.health_state.write().inactivations = current_inactivations.saturating_sub(1);

let event = result
.map(move |_result| ProduceMicroBlockEvent::MicroBlock)
Expand Down
155 changes: 95 additions & 60 deletions validator/src/validator.rs
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use std::{
cmp,
error::Error,
future::Future,
pin::Pin,
Expand Down Expand Up @@ -51,8 +52,13 @@ use crate::{
r#macro::{MappedReturn, ProduceMacroBlock, ProposalTopic},
};

/// The number of blocks a validator needs to produce in time to improve its health
const VALIDATOR_HEALTH_THRESHOLD: u32 = 5;
/// The number of blocks a validator needs o
const VALIDATOR_YELLOW_HEALTH_INACTIVATIONS: u32 = 2;

const VALIDATOR_RED_HEALTH_INACTIVATIONS: u32 = 4;

// The maximum number of blocks the reactivate transaction can be delayed
const MAX_REACTIVATE_DELAY: u32 = 10_000;

#[derive(PartialEq)]
enum ValidatorStakingState {
Expand All @@ -75,18 +81,22 @@ pub struct ConsensusState {
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum ValidatorHealth {
Green,
Yellow(u32),
Red(u32),
Yellow,
Red,
}

/// Struct that represents the overall Validator Health
pub struct HealthState {
/// The current validator health
pub health: ValidatorHealth,
/// Number of blocks that we have produced in time(without being inactivated)
pub blk_cnt: u32,
/// For testing/debug purposes control wether produced blocks are published by the validator
pub publish: bool,
/// Number of inactivations that have ocurred in the current epoch
pub inactivations: u32,
/// Next block number where the re-activate txns should be sent
pub reactivate_bn: u32,
/// Flag that indicates an ongoing reactivation
pub pending_reactivate: bool,
}

/// Validator inactivity
Expand Down Expand Up @@ -234,7 +244,9 @@ where
let health_state = HealthState {
health: ValidatorHealth::Green,
publish: true,
blk_cnt: 0,
inactivations: 0,
reactivate_bn: 0,
pending_reactivate: false,
};

Self {
Expand Down Expand Up @@ -508,6 +520,10 @@ where
self.on_blockchain_extended(hash);
}
BlockchainEvent::EpochFinalized(ref hash) => {
// Reset the inactivations counter
self.health_state.write().inactivations = 0;
// Reset the validator health every epoch
self.health_state.write().health = ValidatorHealth::Green;
self.init_epoch();
// The on_blockchain_extended is necessary for the order of events to not matter.
self.on_blockchain_extended(hash);
Expand Down Expand Up @@ -537,6 +553,14 @@ where

self.check_reactivate(block.block_number());
self.init_block_producer(Some(hash));

let block_number = block.block_number();
let blockchain = self.blockchain.read();

if block_number == self.health_state.read().reactivate_bn {
let inactivity_state = self.reactivate(&blockchain);
self.validator_state = Some(inactivity_state);
}
}

fn on_blockchain_rebranched(
Expand Down Expand Up @@ -718,6 +742,11 @@ where
)
}

// Computes the next block number where we should send the next reactivate transaction
fn get_reactivate_delay(&self, inactivations: u32) -> u32 {
cmp::min(inactivations.pow(2), MAX_REACTIVATE_DELAY)
}

fn reactivate(&self, blockchain: &Blockchain) -> InactivityState {
let validity_start_height = blockchain.block_number();

Expand All @@ -733,7 +762,7 @@ where

let cn = self.consensus.clone();
spawn(async move {
debug!("Sending reactivate transaction to the network");
info!("Sending reactivate transaction to the network");
if cn
.send_transaction(reactivate_transaction.clone())
.await
Expand Down Expand Up @@ -878,39 +907,34 @@ where
let block_number = blockchain.block_number();
match self.get_staking_state(&blockchain) {
ValidatorStakingState::Active => {
drop(blockchain);
if self.validator_state.is_some() {
drop(blockchain);
self.validator_state = None;
self.health_state.write().pending_reactivate = false;
info!("Automatically reactivated.");
}
let inactivations = self.health_state.read().inactivations;

log::warn!(
address=%self.validator_address.read(),
inactivations,
"Inactivations counter",
);

let validator_health = self.health_state.read().health;

match validator_health {
ValidatorHealth::Green => {}
ValidatorHealth::Yellow(yellow_block_number) => {
debug!(
address = %self.validator_address.read(),
inactivated = yellow_block_number,
good_blocks = %self.health_state.read().blk_cnt,
"Current validator health is yellow",
);
if self.health_state.read().blk_cnt >= VALIDATOR_HEALTH_THRESHOLD {
log::info!("Changing the validator health back to green");
ValidatorHealth::Yellow => {
if inactivations < VALIDATOR_YELLOW_HEALTH_INACTIVATIONS {
log::info!(inactivations, "Changed validator health to green");
self.health_state.write().health = ValidatorHealth::Green;
self.health_state.write().blk_cnt = 0;
}
}
ValidatorHealth::Red(red_block_number) => {
debug!(
address = %self.validator_address.read(),
inactivated = red_block_number,
"Current validator health is red",
);
if self.health_state.read().blk_cnt >= VALIDATOR_HEALTH_THRESHOLD {
log::info!("Changing the validator health back to yellow");
self.health_state.write().health =
ValidatorHealth::Yellow(block_number);
self.health_state.write().blk_cnt = 0;
ValidatorHealth::Red => {
if inactivations < VALIDATOR_RED_HEALTH_INACTIVATIONS {
log::info!(inactivations, "Changed validator health to yellow");
self.health_state.write().health = ValidatorHealth::Yellow;
}
}
}
Expand All @@ -924,37 +948,48 @@ where
.unwrap_or(true)
&& self.automatic_reactivate.load(Ordering::Acquire)
{
let validator_health = self.health_state.read().health;
match validator_health {
ValidatorHealth::Green => {
log::warn!(
address=%self.validator_address.read(),
"The validator was inactivated, changing its health to Yellow",
);
let inactivity_state = self.reactivate(&blockchain);
drop(blockchain);
self.validator_state = Some(inactivity_state);
self.health_state.write().health =
ValidatorHealth::Yellow(block_number);
self.health_state.write().blk_cnt = 0;
}
ValidatorHealth::Yellow(_) => {
log::warn!(
address=%self.validator_address.read(),
"The validator was inactivated again, changing its health to Red",
);
let inactivity_state = self.reactivate(&blockchain);
drop(blockchain);
self.validator_state = Some(inactivity_state);
self.health_state.write().health =
ValidatorHealth::Red(block_number);
self.health_state.write().blk_cnt = 0;
}
ValidatorHealth::Red(_) => {
log::warn!(
"The validator needs human intervention, no automatic reactivate"
);
// Keep track of how many times we have been deactivated in the current epoch.
if !self.health_state.read().pending_reactivate {
let mut health_state = self.health_state.write();
health_state.inactivations += 1;
health_state.reactivate_bn = block_number
+ self.get_reactivate_delay(health_state.inactivations);
health_state.pending_reactivate = true;
drop(health_state);

let inactivations = self.health_state.read().inactivations;

let validator_health = self.health_state.read().health;
match validator_health {
ValidatorHealth::Green => {
if inactivations >= VALIDATOR_YELLOW_HEALTH_INACTIVATIONS {
log::warn!(
inactivations,
"Changed validator health to yellow"
);
self.health_state.write().health = ValidatorHealth::Yellow;
}
}
ValidatorHealth::Yellow => {
if inactivations >= VALIDATOR_RED_HEALTH_INACTIVATIONS {
log::warn!(
inactivations,
"Changed validator health to red"
);
self.health_state.write().health = ValidatorHealth::Red;
}
}
ValidatorHealth::Red => {
log::warn!("Validator health is still red")
}
}

log::warn!(
"Current inactivations counter: {}, next reactivate bn {}, current bn {} ",
inactivations,
self.health_state.read().reactivate_bn,
block_number
);
}
}
}
Expand Down
71 changes: 19 additions & 52 deletions validator/tests/mock.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -202,16 +202,18 @@ async fn validator_can_recover_from_yellow_health() {

validator_proxy.validator_health.write().publish = false;

events.take(10).for_each(|_| future::ready(())).await;
log::info!(
"Validator proxy address {}",
validator_proxy.validator_address.read()
);

events.take(30).for_each(|_| future::ready(())).await;

let current_validator_health = validator_proxy.validator_health.read().health;

match current_validator_health {
ValidatorHealth::Yellow(block_number) => {
log::info!(
"Current validator health is yellow, as expected, inactivated block {}",
block_number
)
ValidatorHealth::Yellow => {
log::info!("Current validator health is yellow, as expecteds",)
}
_ => panic!("Validator Health different than expected"),
};
Expand All @@ -220,7 +222,7 @@ async fn validator_can_recover_from_yellow_health() {
validator_proxy.validator_health.write().publish = true;

let events = blockchain.read().notifier_as_stream();
events.take(40).for_each(|_| future::ready(())).await;
events.take(30).for_each(|_| future::ready(())).await;

assert_eq!(
validator_proxy.validator_health.read().health,
Expand Down Expand Up @@ -252,16 +254,13 @@ async fn validator_health_to_red() {

validator_proxy.validator_health.write().publish = false;

events.take(10).for_each(|_| future::ready(())).await;
events.take(30).for_each(|_| future::ready(())).await;

let current_validator_health = validator_proxy.validator_health.read().health;

match current_validator_health {
ValidatorHealth::Yellow(block_number) => {
log::info!(
"Current validator health is yellow, as expected, inactivated block {}",
block_number
)
ValidatorHealth::Yellow => {
log::info!("Current validator health is yellow, as expected",)
}
_ => panic!("Validator Health different than expected"),
};
Expand All @@ -274,11 +273,8 @@ async fn validator_health_to_red() {
let current_validator_health = validator_proxy.validator_health.read().health;

match current_validator_health {
ValidatorHealth::Red(block_number) => {
log::info!(
"Current validator health is red, as expected, inactivated block {}",
block_number
)
ValidatorHealth::Red => {
log::info!("Current validator health is red, as expected",)
}
_ => panic!("Validator Health different than expected"),
};
Expand All @@ -295,7 +291,6 @@ async fn validator_health_fully_recover() {

// Listen for blockchain events from the new block producer (after a skip block).
let validator = validators.first().unwrap();
let consensus = validator.consensus.clone();
let validator_proxy = validator.proxy();
let validator_address = validator.validator_address();

Expand All @@ -316,16 +311,13 @@ async fn validator_health_fully_recover() {

validator_proxy.validator_health.write().publish = false;

events.take(10).for_each(|_| future::ready(())).await;
events.take(30).for_each(|_| future::ready(())).await;

let current_validator_health = validator_proxy.validator_health.read().health;

match current_validator_health {
ValidatorHealth::Yellow(block_number) => {
log::info!(
"Current validator health is yellow, as expected, inactivated block {}",
block_number
)
ValidatorHealth::Yellow => {
log::info!("Current validator health is yellow, as expected",)
}
_ => panic!("Validator Health different than expected"),
};
Expand All @@ -338,37 +330,12 @@ async fn validator_health_fully_recover() {
let current_validator_health = validator_proxy.validator_health.read().health;

match current_validator_health {
ValidatorHealth::Red(block_number) => {
log::info!(
"Current validator health is red, as expected, inactivated block {}",
block_number
)
ValidatorHealth::Red => {
log::info!("Current validator health is red, as expected")
}
_ => panic!("Validator Health different than expected"),
};

// Since the validator needs manual intervention, we are going to send the reactivate transaction

let reactivate_transaction = TransactionBuilder::new_reactivate_validator(
&validator_proxy.fee_key.read(),
validator_address,
&validator_proxy.signing_key.read(),
Coin::ZERO,
Policy::genesis_block_number(),
NetworkId::UnitAlbatross,
);

spawn(async move {
log::info!("Sending reactivate transaction to the network");
if consensus
.send_transaction(reactivate_transaction.clone())
.await
.is_err()
{
log::error!("Failed to send reactivate transaction");
}
});

validator_proxy.validator_health.write().publish = true;

let events = blockchain.read().notifier_as_stream();
Expand Down

0 comments on commit fadf7cb

Please sign in to comment.