use crate::{
block_producer::gas_price::{
ConsensusParametersProvider,
GasPriceProvider as GasPriceProviderConstraint,
},
ports,
ports::BlockProducerDatabase,
Config,
};
use anyhow::{
anyhow,
Context,
};
use fuel_core_storage::transactional::{
AtomicView,
Changes,
};
use fuel_core_types::{
blockchain::{
header::{
ApplicationHeader,
ConsensusHeader,
PartialBlockHeader,
},
primitives::DaBlockHeight,
},
fuel_tx::Transaction,
fuel_types::{
BlockHeight,
Bytes32,
},
services::{
block_producer::Components,
executor::{
TransactionExecutionStatus,
UncommittedResult,
},
},
tai64::Tai64,
};
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::debug;
#[cfg(test)]
mod tests;
pub mod gas_price;
#[derive(Debug, derive_more::Display)]
pub enum Error {
#[display(fmt = "Genesis block is absent")]
NoGenesisBlock,
#[display(
fmt = "The block height {height} should be higher than the previous block height {previous_block}"
)]
BlockHeightShouldBeHigherThanPrevious {
height: BlockHeight,
previous_block: BlockHeight,
},
#[display(fmt = "Previous block height {_0} doesn't exist")]
MissingBlock(BlockHeight),
#[display(
fmt = "Best finalized da_height {best} is behind previous block da_height {previous_block}"
)]
InvalidDaFinalizationState {
best: DaBlockHeight,
previous_block: DaBlockHeight,
},
}
impl From<Error> for anyhow::Error {
fn from(error: Error) -> Self {
anyhow::Error::msg(error)
}
}
pub struct Producer<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider> {
pub config: Config,
pub view_provider: ViewProvider,
pub txpool: TxPool,
pub executor: Arc<Executor>,
pub relayer: Box<dyn ports::Relayer>,
pub lock: Mutex<()>,
pub gas_price_provider: GasPriceProvider,
pub consensus_parameters_provider: ConsensusProvider,
}
impl<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
Producer<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
where
ViewProvider: AtomicView<Height = BlockHeight> + 'static,
ViewProvider::View: BlockProducerDatabase,
GasPriceProvider: GasPriceProviderConstraint,
ConsensusProvider: ConsensusParametersProvider,
{
async fn produce_and_execute<TxSource>(
&self,
height: BlockHeight,
block_time: Tai64,
tx_source: impl FnOnce(BlockHeight) -> TxSource,
) -> anyhow::Result<UncommittedResult<Changes>>
where
Executor: ports::BlockProducer<TxSource> + 'static,
{
let _production_guard = self.lock.lock().await;
let source = tx_source(height);
let header = self.new_header(height, block_time).await?;
let gas_price = self
.gas_price_provider
.gas_price(height.into())
.ok_or(anyhow!("No gas price found for block {height:?}"))?;
let component = Components {
header_to_produce: header,
transactions_source: source,
coinbase_recipient: self.config.coinbase_recipient.unwrap_or_default(),
gas_price,
};
let context_string =
format!("Failed to produce block {height:?} due to execution failure");
let result = self
.executor
.produce_without_commit(component)
.map_err(Into::<anyhow::Error>::into)
.context(context_string)?;
debug!("Produced block with result: {:?}", result.result());
Ok(result)
}
}
impl<ViewProvider, TxPool, Executor, TxSource, GasPriceProvider, ConsensusProvider>
Producer<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
where
ViewProvider: AtomicView<Height = BlockHeight> + 'static,
ViewProvider::View: BlockProducerDatabase,
TxPool: ports::TxPool<TxSource = TxSource> + 'static,
Executor: ports::BlockProducer<TxSource> + 'static,
GasPriceProvider: GasPriceProviderConstraint,
ConsensusProvider: ConsensusParametersProvider,
{
pub async fn produce_and_execute_block_txpool(
&self,
height: BlockHeight,
block_time: Tai64,
) -> anyhow::Result<UncommittedResult<Changes>> {
self.produce_and_execute(height, block_time, |height| {
self.txpool.get_source(height)
})
.await
}
}
impl<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
Producer<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
where
ViewProvider: AtomicView<Height = BlockHeight> + 'static,
ViewProvider::View: BlockProducerDatabase,
Executor: ports::BlockProducer<Vec<Transaction>> + 'static,
GasPriceProvider: GasPriceProviderConstraint,
ConsensusProvider: ConsensusParametersProvider,
{
pub async fn produce_and_execute_block_transactions(
&self,
height: BlockHeight,
block_time: Tai64,
transactions: Vec<Transaction>,
) -> anyhow::Result<UncommittedResult<Changes>> {
self.produce_and_execute(height, block_time, |_| transactions)
.await
}
}
impl<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
Producer<ViewProvider, TxPool, Executor, GasPriceProvider, ConsensusProvider>
where
ViewProvider: AtomicView<Height = BlockHeight> + 'static,
ViewProvider::View: BlockProducerDatabase,
Executor: ports::DryRunner + 'static,
GasPriceProvider: GasPriceProviderConstraint,
ConsensusProvider: ConsensusParametersProvider,
{
pub async fn dry_run(
&self,
transactions: Vec<Transaction>,
height: Option<BlockHeight>,
utxo_validation: Option<bool>,
gas_price: Option<u64>,
) -> anyhow::Result<Vec<TransactionExecutionStatus>> {
let height = height.unwrap_or_else(|| {
self.view_provider
.latest_height()
.unwrap_or_default()
.succ()
.expect("It is impossible to overflow the current block height")
});
let gas_price = gas_price
.or_else(|| self.gas_price_provider.gas_price(height.into()))
.ok_or(anyhow!("No gas price found for height {height:?}"))?;
let header = self._new_header(height, Tai64::now())?;
let component = Components {
header_to_produce: header,
transactions_source: transactions.clone(),
coinbase_recipient: self.config.coinbase_recipient.unwrap_or_default(),
gas_price,
};
let executor = self.executor.clone();
let tx_statuses = tokio_rayon::spawn_fifo(
move || -> anyhow::Result<Vec<TransactionExecutionStatus>> {
Ok(executor.dry_run(component, utxo_validation)?)
},
)
.await?;
if transactions
.iter()
.zip(tx_statuses.iter())
.any(|(transaction, tx_status)| {
transaction.is_script() && tx_status.result.receipts().is_empty()
})
{
Err(anyhow!("Expected at least one set of receipts"))
} else {
Ok(tx_statuses)
}
}
}
pub const NO_NEW_DA_HEIGHT_FOUND: &str = "No new da_height found";
impl<ViewProvider, TxPool, Executor, GP, ConsensusProvider>
Producer<ViewProvider, TxPool, Executor, GP, ConsensusProvider>
where
ViewProvider: AtomicView<Height = BlockHeight> + 'static,
ViewProvider::View: BlockProducerDatabase,
ConsensusProvider: ConsensusParametersProvider,
{
async fn new_header(
&self,
height: BlockHeight,
block_time: Tai64,
) -> anyhow::Result<PartialBlockHeader> {
let mut block_header = self._new_header(height, block_time)?;
let previous_da_height = block_header.da_height;
let gas_limit = self
.consensus_parameters_provider
.consensus_params_at_version(&block_header.consensus_parameters_version)?
.block_gas_limit();
let new_da_height = self
.select_new_da_height(gas_limit, previous_da_height)
.await?;
block_header.application.da_height = new_da_height;
Ok(block_header)
}
async fn select_new_da_height(
&self,
gas_limit: u64,
previous_da_height: DaBlockHeight,
) -> anyhow::Result<DaBlockHeight> {
let mut new_best = previous_da_height;
let mut total_cost: u64 = 0;
let highest = self
.relayer
.wait_for_at_least_height(&previous_da_height)
.await?;
if highest < previous_da_height {
return Err(Error::InvalidDaFinalizationState {
best: highest,
previous_block: previous_da_height,
}
.into());
}
if highest == previous_da_height {
return Ok(highest);
}
let next_da_height = previous_da_height.saturating_add(1);
for height in next_da_height..=highest.0 {
let cost = self
.relayer
.get_cost_for_block(&DaBlockHeight(height))
.await?;
total_cost = total_cost.saturating_add(cost);
if total_cost > gas_limit {
break;
} else {
new_best = DaBlockHeight(height);
}
}
if new_best == previous_da_height {
Err(anyhow!(NO_NEW_DA_HEIGHT_FOUND))
} else {
Ok(new_best)
}
}
fn _new_header(
&self,
height: BlockHeight,
block_time: Tai64,
) -> anyhow::Result<PartialBlockHeader> {
let view = self.view_provider.latest_view();
let previous_block_info = self.previous_block_info(height, &view)?;
let consensus_parameters_version = view.latest_consensus_parameters_version()?;
let state_transition_bytecode_version =
view.latest_state_transition_bytecode_version()?;
Ok(PartialBlockHeader {
application: ApplicationHeader {
da_height: previous_block_info.da_height,
consensus_parameters_version,
state_transition_bytecode_version,
generated: Default::default(),
},
consensus: ConsensusHeader {
prev_root: previous_block_info.prev_root,
height,
time: block_time,
generated: Default::default(),
},
})
}
fn previous_block_info(
&self,
height: BlockHeight,
view: &ViewProvider::View,
) -> anyhow::Result<PreviousBlockInfo> {
let latest_height = self
.view_provider
.latest_height()
.ok_or(Error::NoGenesisBlock)?;
if height <= latest_height {
Err(Error::BlockHeightShouldBeHigherThanPrevious {
height,
previous_block: latest_height,
}
.into())
} else {
let prev_height = height.pred().expect("We checked the height above");
let previous_block = view.get_block(&prev_height)?;
let prev_root = view.block_header_merkle_root(&prev_height)?;
Ok(PreviousBlockInfo {
prev_root,
da_height: previous_block.header().da_height,
})
}
}
}
struct PreviousBlockInfo {
prev_root: Bytes32,
da_height: DaBlockHeight,
}