use std::{
borrow::Cow,
collections::{HashMap, HashSet},
hash::Hash,
};
use zakura_chain::{
block::Height,
ironwood, orchard, sapling, sprout,
transaction::{self, UnminedTx, UnminedTxId, VerifiedUnminedTx},
transparent,
};
use zakura_node_services::mempool::TransactionDependencies;
use crate::components::mempool::pending_outputs::PendingOutputs;
use super::super::SameEffectsTipRejectionError;
#[allow(unused_imports)]
use zakura_chain::transaction::MEMPOOL_TRANSACTION_COST_THRESHOLD;
#[derive(Default)]
pub struct VerifiedSet {
transactions: HashMap<transaction::Hash, VerifiedUnminedTx>,
transaction_dependencies: TransactionDependencies,
created_outputs: HashMap<transparent::OutPoint, transparent::Output>,
transactions_serialized_size: usize,
total_cost: u64,
metrics: MempoolMetrics,
spent_outpoints: HashSet<transparent::OutPoint>,
sprout_nullifiers: HashSet<sprout::Nullifier>,
sapling_nullifiers: HashSet<sapling::Nullifier>,
orchard_nullifiers: HashSet<orchard::Nullifier>,
ironwood_nullifiers: HashSet<ironwood::Nullifier>,
}
impl Drop for VerifiedSet {
fn drop(&mut self) {
self.clear()
}
}
impl VerifiedSet {
pub fn transactions(&self) -> &HashMap<transaction::Hash, VerifiedUnminedTx> {
&self.transactions
}
pub fn transaction_dependencies(&self) -> &TransactionDependencies {
&self.transaction_dependencies
}
pub fn created_output(&self, outpoint: &transparent::OutPoint) -> Option<transparent::Output> {
self.created_outputs.get(outpoint).cloned()
}
pub fn has_spent_outpoint(&self, outpoint: &transparent::OutPoint) -> bool {
self.spent_outpoints.contains(outpoint)
}
pub fn transaction_count(&self) -> usize {
self.transactions.len()
}
pub fn total_cost(&self) -> u64 {
self.total_cost
}
pub fn total_serialized_size(&self) -> usize {
self.transactions_serialized_size
}
pub fn contains(&self, id: &transaction::Hash) -> bool {
self.transactions.contains_key(id)
}
pub fn clear(&mut self) {
self.transactions.clear();
self.transaction_dependencies.clear();
self.spent_outpoints.clear();
self.sprout_nullifiers.clear();
self.sapling_nullifiers.clear();
self.orchard_nullifiers.clear();
self.ironwood_nullifiers.clear();
self.created_outputs.clear();
self.transactions_serialized_size = 0;
self.total_cost = 0;
self.metrics = MempoolMetrics::default();
self.report_metrics();
}
pub fn insert(
&mut self,
mut transaction: VerifiedUnminedTx,
spent_mempool_outpoints: Vec<transparent::OutPoint>,
pending_outputs: &mut PendingOutputs,
height: Option<Height>,
) -> Result<(), SameEffectsTipRejectionError> {
if self.has_spend_conflicts(&transaction.transaction) {
return Err(SameEffectsTipRejectionError::SpendConflict);
}
for outpoint in &spent_mempool_outpoints {
if !self.created_outputs.contains_key(outpoint) {
return Err(SameEffectsTipRejectionError::MissingOutput);
}
}
let tx_id = transaction.transaction.id.mined_id();
self.transaction_dependencies
.add(tx_id, spent_mempool_outpoints);
let tx = &transaction.transaction.transaction;
for (index, output) in tx.outputs().iter().cloned().enumerate() {
let outpoint = transparent::OutPoint::from_usize(tx_id, index);
self.created_outputs.insert(outpoint, output.clone());
pending_outputs.respond(&outpoint, output)
}
self.spent_outpoints.extend(tx.spent_outpoints());
self.sprout_nullifiers.extend(tx.sprout_nullifiers());
self.sapling_nullifiers.extend(tx.sapling_nullifiers());
self.orchard_nullifiers.extend(tx.orchard_nullifiers());
self.ironwood_nullifiers.extend(tx.ironwood_nullifiers());
self.transactions_serialized_size += transaction.transaction.size;
self.total_cost += transaction.cost();
transaction.time = Some(chrono::Utc::now());
transaction.height = height;
self.metrics.add_transaction(&transaction);
self.transactions.insert(tx_id, transaction);
self.report_metrics();
Ok(())
}
#[allow(clippy::unwrap_in_result)]
pub(super) fn evict_one(&mut self) -> Option<Vec<VerifiedUnminedTx>> {
use rand::distributions::{Distribution, WeightedIndex};
use rand::prelude::thread_rng;
let (keys, weights): (Vec<transaction::Hash>, Vec<u64>) = self
.transactions
.iter()
.map(|(&tx_id, tx)| (tx_id, tx.eviction_weight()))
.unzip();
let dist = WeightedIndex::new(weights).expect(
"there is at least one weight, all weights are non-negative, and the total is positive",
);
let key_to_remove = keys
.get(dist.sample(&mut thread_rng()))
.expect("should have a key at every index in the distribution");
Some(self.remove(key_to_remove))
}
pub fn clear_mined_dependencies(&mut self, mined_ids: &HashSet<transaction::Hash>) {
self.transaction_dependencies
.clear_mined_dependencies(mined_ids);
}
pub fn remove_all_that(
&mut self,
predicate: impl Fn(&VerifiedUnminedTx) -> bool,
) -> HashSet<UnminedTxId> {
let keys_to_remove: Vec<_> = self
.transactions
.iter()
.filter_map(|(&tx_id, tx)| predicate(tx).then_some(tx_id))
.collect();
let mut removed_transactions = HashSet::new();
for key_to_remove in keys_to_remove {
if !self.transactions.contains_key(&key_to_remove) {
continue;
}
removed_transactions.extend(
self.remove(&key_to_remove)
.into_iter()
.map(|tx| tx.transaction.id),
);
}
removed_transactions
}
fn remove(&mut self, key_to_remove: &transaction::Hash) -> Vec<VerifiedUnminedTx> {
let removed_transactions: Vec<_> = self
.transaction_dependencies
.remove_all(key_to_remove)
.iter()
.chain(std::iter::once(key_to_remove))
.filter_map(|key_to_remove| {
let Some(removed_tx) = self.transactions.remove(key_to_remove) else {
tracing::warn!(?key_to_remove, "invalid transaction key");
return None;
};
self.transactions_serialized_size -= removed_tx.transaction.size;
self.total_cost -= removed_tx.cost();
self.metrics.remove_transaction(&removed_tx);
self.remove_outputs(&removed_tx.transaction);
Some(removed_tx)
})
.collect();
self.report_metrics();
removed_transactions
}
fn has_spend_conflicts(&self, unmined_tx: &UnminedTx) -> bool {
let tx = &unmined_tx.transaction;
Self::has_conflicts(&self.spent_outpoints, tx.spent_outpoints())
|| Self::has_conflicts(&self.sprout_nullifiers, tx.sprout_nullifiers().copied())
|| Self::has_conflicts(&self.sapling_nullifiers, tx.sapling_nullifiers().copied())
|| Self::has_conflicts(&self.orchard_nullifiers, tx.orchard_nullifiers().copied())
|| Self::has_conflicts(&self.ironwood_nullifiers, tx.ironwood_nullifiers().copied())
}
fn remove_outputs(&mut self, unmined_tx: &UnminedTx) {
let tx = &unmined_tx.transaction;
for index in 0..tx.outputs().len() {
self.created_outputs
.remove(&transparent::OutPoint::from_usize(
unmined_tx.id.mined_id(),
index,
));
}
let spent_outpoints = tx.spent_outpoints().map(Cow::Owned);
let sprout_nullifiers = tx.sprout_nullifiers().map(Cow::Borrowed);
let sapling_nullifiers = tx.sapling_nullifiers().map(Cow::Borrowed);
let orchard_nullifiers = tx.orchard_nullifiers().map(Cow::Borrowed);
let ironwood_nullifiers = tx.ironwood_nullifiers().map(Cow::Borrowed);
Self::remove_from_set(&mut self.spent_outpoints, spent_outpoints);
Self::remove_from_set(&mut self.sprout_nullifiers, sprout_nullifiers);
Self::remove_from_set(&mut self.sapling_nullifiers, sapling_nullifiers);
Self::remove_from_set(&mut self.orchard_nullifiers, orchard_nullifiers);
Self::remove_from_set(&mut self.ironwood_nullifiers, ironwood_nullifiers);
}
fn has_conflicts<T>(set: &HashSet<T>, mut list: impl Iterator<Item = T>) -> bool
where
T: Eq + Hash,
{
list.any(|item| set.contains(&item))
}
fn remove_from_set<'t, T>(set: &mut HashSet<T>, items: impl IntoIterator<Item = Cow<'t, T>>)
where
T: Clone + Eq + Hash + 't,
{
for item in items {
set.remove(&item);
}
}
fn report_metrics(&self) {
metrics::gauge!(
"zcash.mempool.actions.unpaid",
"bk" => "< 0.2",
)
.set(self.metrics.unpaid_actions_with_weight_lt20pct as f64);
metrics::gauge!(
"zcash.mempool.actions.unpaid",
"bk" => "< 0.4",
)
.set(self.metrics.unpaid_actions_with_weight_lt40pct as f64);
metrics::gauge!(
"zcash.mempool.actions.unpaid",
"bk" => "< 0.6",
)
.set(self.metrics.unpaid_actions_with_weight_lt60pct as f64);
metrics::gauge!(
"zcash.mempool.actions.unpaid",
"bk" => "< 0.8",
)
.set(self.metrics.unpaid_actions_with_weight_lt80pct as f64);
metrics::gauge!(
"zcash.mempool.actions.unpaid",
"bk" => "< 1",
)
.set(self.metrics.unpaid_actions_with_weight_lt1 as f64);
metrics::gauge!("zcash.mempool.actions.paid").set(self.metrics.paid_actions as f64);
metrics::gauge!("zcash.mempool.size.transactions",).set(self.transaction_count() as f64);
metrics::gauge!(
"zcash.mempool.size.weighted",
"bk" => "< 1",
)
.set(self.metrics.size_with_weight_lt1 as f64);
metrics::gauge!(
"zcash.mempool.size.weighted",
"bk" => "1",
)
.set(self.metrics.size_with_weight_eq1 as f64);
metrics::gauge!(
"zcash.mempool.size.weighted",
"bk" => "> 1",
)
.set(self.metrics.size_with_weight_gt1 as f64);
metrics::gauge!(
"zcash.mempool.size.weighted",
"bk" => "> 2",
)
.set(self.metrics.size_with_weight_gt2 as f64);
metrics::gauge!(
"zcash.mempool.size.weighted",
"bk" => "> 3",
)
.set(self.metrics.size_with_weight_gt3 as f64);
metrics::gauge!("zcash.mempool.size.bytes",).set(self.transactions_serialized_size as f64);
metrics::gauge!("zcash.mempool.cost.bytes").set(self.total_cost as f64);
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct MempoolMetrics {
unpaid_actions_with_weight_lt20pct: u32,
unpaid_actions_with_weight_lt40pct: u32,
unpaid_actions_with_weight_lt60pct: u32,
unpaid_actions_with_weight_lt80pct: u32,
unpaid_actions_with_weight_lt1: u32,
paid_actions: u32,
size_with_weight_lt1: usize,
size_with_weight_eq1: usize,
size_with_weight_gt1: usize,
size_with_weight_gt2: usize,
size_with_weight_gt3: usize,
}
impl MempoolMetrics {
fn add_transaction(&mut self, transaction: &VerifiedUnminedTx) {
self.add(Self::for_transaction(transaction));
}
fn remove_transaction(&mut self, transaction: &VerifiedUnminedTx) {
self.remove(Self::for_transaction(transaction));
}
fn for_transaction(transaction: &VerifiedUnminedTx) -> Self {
Self::for_values(
transaction.fee_weight_ratio,
transaction.conventional_actions,
transaction.unpaid_actions,
transaction.transaction.size,
)
}
fn for_values(
fee_weight_ratio: f32,
conventional_actions: u32,
unpaid_actions: u32,
size: usize,
) -> Self {
let mut metrics = Self {
paid_actions: conventional_actions - unpaid_actions,
..Default::default()
};
if fee_weight_ratio > 3.0 {
metrics.size_with_weight_gt3 = size;
} else if fee_weight_ratio > 2.0 {
metrics.size_with_weight_gt2 = size;
} else if fee_weight_ratio > 1.0 {
metrics.size_with_weight_gt1 = size;
} else if fee_weight_ratio == 1.0 {
metrics.size_with_weight_eq1 = size;
} else {
metrics.size_with_weight_lt1 = size;
if fee_weight_ratio < 0.2 {
metrics.unpaid_actions_with_weight_lt20pct = unpaid_actions;
} else if fee_weight_ratio < 0.4 {
metrics.unpaid_actions_with_weight_lt40pct = unpaid_actions;
} else if fee_weight_ratio < 0.6 {
metrics.unpaid_actions_with_weight_lt60pct = unpaid_actions;
} else if fee_weight_ratio < 0.8 {
metrics.unpaid_actions_with_weight_lt80pct = unpaid_actions;
} else {
metrics.unpaid_actions_with_weight_lt1 = unpaid_actions;
}
}
metrics
}
fn add(&mut self, other: Self) {
self.unpaid_actions_with_weight_lt20pct += other.unpaid_actions_with_weight_lt20pct;
self.unpaid_actions_with_weight_lt40pct += other.unpaid_actions_with_weight_lt40pct;
self.unpaid_actions_with_weight_lt60pct += other.unpaid_actions_with_weight_lt60pct;
self.unpaid_actions_with_weight_lt80pct += other.unpaid_actions_with_weight_lt80pct;
self.unpaid_actions_with_weight_lt1 += other.unpaid_actions_with_weight_lt1;
self.paid_actions += other.paid_actions;
self.size_with_weight_lt1 += other.size_with_weight_lt1;
self.size_with_weight_eq1 += other.size_with_weight_eq1;
self.size_with_weight_gt1 += other.size_with_weight_gt1;
self.size_with_weight_gt2 += other.size_with_weight_gt2;
self.size_with_weight_gt3 += other.size_with_weight_gt3;
}
fn remove(&mut self, other: Self) {
self.unpaid_actions_with_weight_lt20pct -= other.unpaid_actions_with_weight_lt20pct;
self.unpaid_actions_with_weight_lt40pct -= other.unpaid_actions_with_weight_lt40pct;
self.unpaid_actions_with_weight_lt60pct -= other.unpaid_actions_with_weight_lt60pct;
self.unpaid_actions_with_weight_lt80pct -= other.unpaid_actions_with_weight_lt80pct;
self.unpaid_actions_with_weight_lt1 -= other.unpaid_actions_with_weight_lt1;
self.paid_actions -= other.paid_actions;
self.size_with_weight_lt1 -= other.size_with_weight_lt1;
self.size_with_weight_eq1 -= other.size_with_weight_eq1;
self.size_with_weight_gt1 -= other.size_with_weight_gt1;
self.size_with_weight_gt2 -= other.size_with_weight_gt2;
self.size_with_weight_gt3 -= other.size_with_weight_gt3;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn clear_removes_ironwood_nullifiers() {
let mut verified = VerifiedSet::default();
verified.ironwood_nullifiers.insert(
ironwood::Nullifier::try_from([0; 32]).expect("zero is a valid Pallas base field"),
);
verified.clear();
assert!(verified.ironwood_nullifiers.is_empty());
}
#[test]
fn metrics_are_updated_incrementally() {
let transaction_metrics = [
(0.1, 5, 4, 10),
(0.2, 6, 3, 20),
(0.4, 7, 2, 30),
(0.6, 8, 1, 40),
(0.8, 9, 0, 50),
(1.0, 10, 0, 60),
(1.5, 11, 0, 70),
(2.5, 12, 0, 80),
(3.5, 13, 0, 90),
];
let mut metrics = MempoolMetrics::default();
for &(fee_weight_ratio, conventional_actions, unpaid_actions, size) in &transaction_metrics
{
metrics.add(MempoolMetrics::for_values(
fee_weight_ratio,
conventional_actions,
unpaid_actions,
size,
));
}
assert_eq!(
metrics,
MempoolMetrics {
unpaid_actions_with_weight_lt20pct: 4,
unpaid_actions_with_weight_lt40pct: 3,
unpaid_actions_with_weight_lt60pct: 2,
unpaid_actions_with_weight_lt80pct: 1,
unpaid_actions_with_weight_lt1: 0,
paid_actions: 71,
size_with_weight_lt1: 150,
size_with_weight_eq1: 60,
size_with_weight_gt1: 70,
size_with_weight_gt2: 80,
size_with_weight_gt3: 90,
}
);
for &(fee_weight_ratio, conventional_actions, unpaid_actions, size) in
transaction_metrics.iter().rev()
{
metrics.remove(MempoolMetrics::for_values(
fee_weight_ratio,
conventional_actions,
unpaid_actions,
size,
));
}
assert_eq!(metrics, MempoolMetrics::default());
}
}