Skip to content

Commit

Permalink
fix: orphan parent bug, improve fault detection and logging
Browse files Browse the repository at this point in the history
  • Loading branch information
@grumbach
grumbach committed Apr 5, 2024
1 parent 52d9673 commit 65924b8
Show file tree
Hide file tree
Showing 5 changed files with 125 additions and 64 deletions.
22 changes: 18 additions & 4 deletions sn_auditor/src/dag_db.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ impl SpendDagDb {

let (spend_type, spends) = match spend {
SpendDagGet::SpendNotFound => ("SpendNotFound", vec![]),
SpendDagGet::SpendIsAnUtxo => ("SpendIsAnUtxo", vec![]),
SpendDagGet::SpendKeyExists => ("SpendKeyExists", vec![]),
SpendDagGet::DoubleSpend(vs) => ("DoubleSpend", vs),
SpendDagGet::Spend(s) => ("Spend", vec![*s]),
};
Expand Down Expand Up @@ -176,7 +176,7 @@ impl SpendDagDb {
let mut w_handle = dag_ref
.write()
.map_err(|e| eyre!("Failed to get write lock: {e}"))?;
w_handle.merge(other);
w_handle.merge(other)?;
Ok(())
}
}
Expand Down Expand Up @@ -218,17 +218,31 @@ fn dag_to_svg(dag: &SpendDag) -> Result<Vec<u8>> {
// - makes spends clickable
// - spend address reveals on hover
// - marks poisoned spends as red
// - marks UTXOs and unknown ancestors as yellow
// - just pray it works on windows
fn quick_edit_svg(svg: Vec<u8>, dag: &SpendDag) -> Result<Vec<u8>> {
let mut str = String::from_utf8(svg).map_err(|err| eyre!("Failed svg conversion: {err}"))?;

let spend_addrs: Vec<_> = dag.all_spends().iter().map(|s| s.address()).collect();
let utxo_addrs = dag.get_utxos();
let unknown_parents = dag.get_unknown_parents();
let all_addrs = spend_addrs
.iter()
.chain(utxo_addrs.iter())
.chain(unknown_parents.iter());

for addr in spend_addrs.iter().chain(utxo_addrs.iter()) {
for addr in all_addrs {
let addr_hex = addr.to_hex().to_string();
let is_fault = !dag.get_spend_faults(addr).is_empty();
let colour = if is_fault { "red" } else { "none" };
let is_known_but_not_gathered = matches!(dag.get_spend(addr), SpendDagGet::SpendKeyExists);
let colour = if is_fault {
"red"
} else if is_known_but_not_gathered {
"yellow"
} else {
"none"
};

let link = format!("<a xlink:href=\"/spend/{addr_hex}\">");
let idxs = dag.get_spend_indexes(addr);
for i in idxs {
Expand Down
25 changes: 14 additions & 11 deletions sn_client/src/audit.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ use super::{

use futures::future::join_all;
use sn_networking::{target_arch::Instant, GetRecordError, NetworkError};
use sn_transfers::{SignedSpend, SpendAddress, WalletError, WalletResult};
use sn_transfers::{Hash, SignedSpend, SpendAddress, WalletError, WalletResult};
use std::{collections::BTreeSet, iter::Iterator};

impl Client {
Expand Down Expand Up @@ -73,17 +73,19 @@ impl Client {
let parent_tx_hash = parent_tx.hash();
let parent_keys = parent_tx.inputs.iter().map(|input| input.unique_pubkey);
let addrs_to_verify = parent_keys.map(|k| SpendAddress::from_unique_pubkey(&k));
debug!("Depth {depth} - Verifying parent Tx : {parent_tx_hash:?}");
debug!("Depth {depth} - Verifying parent Tx : {parent_tx_hash:?} with inputs: {addrs_to_verify:?}");

// get all parent spends in parallel
let tasks: Vec<_> = addrs_to_verify
.into_iter()
.clone()
.map(|a| self.get_spend_from_network(a))
.collect();
let spends = join_all(tasks).await
.into_iter()
.collect::<Result<BTreeSet<_>>>()
.map_err(|err| WalletError::CouldNotVerifyTransfer(format!("at depth {depth} - Failed to get spends from network for parent Tx {parent_tx_hash:?}: {err}")))?;
.zip(addrs_to_verify.into_iter())
.map(|(maybe_spend, a)|
maybe_spend.map_err(|err| WalletError::CouldNotVerifyTransfer(format!("at depth {depth} - Failed to get spend {a:?} from network for parent Tx {parent_tx_hash:?}: {err}"))))
.collect::<WalletResult<BTreeSet<_>>>()?;
debug!(
"Depth {depth} - Got {:?} spends for parent Tx: {parent_tx_hash:?}",
spends.len()
Expand Down Expand Up @@ -184,7 +186,7 @@ impl Client {
.iter()
.map(|output| output.unique_pubkey);
let addrs_to_follow = descendant_keys.map(|k| SpendAddress::from_unique_pubkey(&k));
debug!("Gen {gen} - Following descendant Tx : {descendant_tx_hash:?}");
debug!("Gen {gen} - Following descendant Tx : {descendant_tx_hash:?} with outputs: {addrs_to_follow:?}");

// get all descendant spends in parallel
let tasks: Vec<_> = addrs_to_follow
Expand All @@ -198,8 +200,7 @@ impl Client {
.collect::<Vec<_>>();

// split spends into utxos and spends
let (utxos, spends) = split_utxos_and_spends(spends_res)
.map_err(|err| WalletError::CouldNotVerifyTransfer(format!("at gen {gen} - Failed to get spends from network for descendant Tx {descendant_tx_hash:?}: {err}")))?;
let (utxos, spends) = split_utxos_and_spends(spends_res, gen, descendant_tx_hash)?;
debug!("Gen {gen} - Got {:?} spends and {:?} utxos for descendant Tx: {descendant_tx_hash:?}", spends.len(), utxos.len());
trace!("Spends for {descendant_tx_hash:?} - {spends:?}");
next_gen_utxos.extend(utxos);
Expand Down Expand Up @@ -241,7 +242,9 @@ impl Client {

fn split_utxos_and_spends(
spends_res: Vec<(Result<SignedSpend>, SpendAddress)>,
) -> Result<(Vec<SpendAddress>, Vec<SignedSpend>)> {
gen: usize,
descendant_tx_hash: Hash,
) -> WalletResult<(Vec<SpendAddress>, Vec<SignedSpend>)> {
let mut utxos = Vec::new();
let mut spends = Vec::new();

Expand All @@ -254,8 +257,8 @@ fn split_utxos_and_spends(
utxos.push(addr);
}
Err(err) => {
warn!("Error while following spends: {err}");
return Err(err);
warn!("Error while following spend {addr:?}: {err}");
return Err(WalletError::CouldNotVerifyTransfer(format!("at gen {gen} - Failed to get spend {addr:?} from network for descendant Tx {descendant_tx_hash:?}: {err}")));
}
}
}
Expand Down
6 changes: 6 additions & 0 deletions sn_client/src/audit/dag_error.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,11 @@ pub enum SpendFault {
},
#[error("Invalid transaction for spend at {0:?}: {1}")]
InvalidTransaction(SpendAddress, String),
#[error("Spend at {addr:?} has an unknown ancestor at {ancestor_addr:?}, until this ancestor is added to the DAG, it cannot be verified")]
UnknownAncestor {
addr: SpendAddress,
ancestor_addr: SpendAddress,
},
#[error("Poisoned ancestry for spend at {0:?}: {1}")]
PoisonedAncestry(SpendAddress, String),
#[error("Spend at {addr:?} does not descend from given source: {src:?}")]
Expand All @@ -66,6 +71,7 @@ impl SpendFault {
| SpendFault::MissingAncestry { addr, .. }
| SpendFault::InvalidAncestry { addr, .. }
| SpendFault::InvalidTransaction(addr, _)
| SpendFault::UnknownAncestor { addr, .. }
| SpendFault::PoisonedAncestry(addr, _)
| SpendFault::OrphanSpend { addr, .. } => *addr,
}
Expand Down
112 changes: 71 additions & 41 deletions sn_client/src/audit/spend_dag.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,15 +46,15 @@ type DagIndex = usize;
/// Internal Dag entry type
#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
enum DagEntry {
Utxo(DagIndex),
NotGatheredYet(DagIndex),
DoubleSpend(Vec<(SignedSpend, DagIndex)>),
Spend(Box<SignedSpend>, DagIndex),
}

impl DagEntry {
fn indexes(&self) -> Vec<DagIndex> {
match self {
DagEntry::Utxo(idx) => vec![*idx],
DagEntry::NotGatheredYet(idx) => vec![*idx],
DagEntry::DoubleSpend(spends) => spends.iter().map(|(_, idx)| *idx).collect(),
DagEntry::Spend(_, idx) => vec![*idx],
}
Expand All @@ -64,7 +64,7 @@ impl DagEntry {
match self {
DagEntry::Spend(spend, _) => vec![&**spend],
DagEntry::DoubleSpend(spends) => spends.iter().map(|(s, _)| s).collect(),
DagEntry::Utxo(_) => vec![],
DagEntry::NotGatheredYet(_) => vec![],
}
}
}
Expand All @@ -74,8 +74,8 @@ impl DagEntry {
pub enum SpendDagGet {
/// Spend does not exist in the DAG
SpendNotFound,
/// Spend is an UTXO, meaning it was not spent yet but its ancestors exist
SpendIsAnUtxo,
/// Spend key is refered to by known spends but does not exist in the DAG yet
SpendKeyExists,
/// Spend is a double spend
DoubleSpend(Vec<SignedSpend>),
/// Spend is in the DAG
Expand Down Expand Up @@ -127,7 +127,7 @@ impl SpendDag {
node_idx
}
// or upgrade existing utxo to spend
Some(DagEntry::Utxo(idx)) => {
Some(DagEntry::NotGatheredYet(idx)) => {
self.spends
.insert(spend_addr, DagEntry::Spend(Box::new(spend.clone()), idx));
NodeIndex::new(idx)
Expand Down Expand Up @@ -168,7 +168,7 @@ impl SpendDag {
// add descendant if not already in dag
let spends_at_addr = self.spends.entry(descendant_addr).or_insert_with(|| {
let node_idx = self.dag.add_node(descendant_addr);
DagEntry::Utxo(node_idx.index())
DagEntry::NotGatheredYet(node_idx.index())
});

// link to descendant
Expand All @@ -192,7 +192,7 @@ impl SpendDag {
// add ancestor if not already in dag
let spends_at_addr = self.spends.entry(ancestor_addr).or_insert_with(|| {
let node_idx = self.dag.add_node(ancestor_addr);
DagEntry::Utxo(node_idx.index())
DagEntry::NotGatheredYet(node_idx.index())
});

// link to ancestor
Expand All @@ -206,6 +206,27 @@ impl SpendDag {
true
}

/// Get the unknown parents: all the addresses that are refered to as parents by other spends
/// but don't have parents themselves.
/// Those Spends must exist somewhere on the Network, we just haven't gathered them yet.
pub fn get_unknown_parents(&self) -> BTreeSet<SpendAddress> {
let mut sources = BTreeSet::new();
for node_index in self.dag.node_indices() {
if !self
.dag
.neighbors_directed(node_index, petgraph::Direction::Incoming)
.any(|_| true)
{
let utxo_addr = self.dag[node_index];
sources.insert(utxo_addr);
}
}
sources
}

/// Get the UTXOs: all the addresses that are refered to as children by other spends
/// but that don't have children themselves.
/// Those will eventually exist on the Network as the address is spent by their owners.
pub fn get_utxos(&self) -> BTreeSet<SpendAddress> {
let mut leaves = BTreeSet::new();
for node_index in self.dag.node_indices() {
Expand All @@ -226,18 +247,19 @@ impl SpendDag {
}

/// Merges the given dag into ours
pub fn merge(&mut self, sub_dag: SpendDag) {
pub fn merge(&mut self, sub_dag: SpendDag) -> Result<(), DagError> {
let source = self.source();
info!(
"Merging sub DAG starting at {:?} into our DAG with source {:?}",
sub_dag.source(),
self.source()
source
);
for (addr, spends) in sub_dag.spends {
// only add spends to the dag, ignoring utxos
// only add spends to the dag, ignoring utxos and not yet gathered relatives
// utxos will be added automatically as their ancestors are added
// edges are updated by the insert method
match spends {
DagEntry::Utxo(_) => continue,
DagEntry::NotGatheredYet(_) => continue,
DagEntry::DoubleSpend(spends) => {
for (spend, _) in spends {
self.insert(addr, spend);
Expand All @@ -249,15 +271,15 @@ impl SpendDag {
}
}

// merge errors
self.faults.extend(sub_dag.faults);
// recompute faults
self.record_faults(&source)
}

/// Get the spend at a given address
pub fn get_spend(&self, addr: &SpendAddress) -> SpendDagGet {
match self.spends.get(addr) {
None => SpendDagGet::SpendNotFound,
Some(DagEntry::Utxo(_)) => SpendDagGet::SpendIsAnUtxo,
Some(DagEntry::NotGatheredYet(_)) => SpendDagGet::SpendKeyExists,
Some(DagEntry::DoubleSpend(spends)) => {
SpendDagGet::DoubleSpend(spends.iter().map(|(s, _)| s.clone()).collect())
}
Expand Down Expand Up @@ -314,11 +336,11 @@ impl SpendDag {
Some(DagEntry::Spend(ancestor_spend, _)) => {
ancestors.insert(*ancestor_spend.clone());
}
Some(DagEntry::Utxo(_)) => {
warn!("InvalidAncestry: SpendIsAnUtxo ancestor {ancestor_addr:?} for spend {spend:?}");
return Err(SpendFault::InvalidAncestry {
Some(DagEntry::NotGatheredYet(_)) => {
warn!("UnknownAncestor: ancestor {ancestor_addr:?} was not gathered yet for spend {spend:?}");
return Err(SpendFault::UnknownAncestor {
addr,
invalid_ancestor: ancestor_addr,
ancestor_addr,
});
}
Some(DagEntry::DoubleSpend(_)) => {
Expand All @@ -329,10 +351,11 @@ impl SpendDag {
});
}
None => {
warn!("MissingAncestry: ancestor {ancestor_addr:?} is unknown for spend {spend:?}");
return Err(SpendFault::MissingAncestry {
addr,
invalid_ancestor: ancestor_addr,
})
});
}
}
}
Expand All @@ -345,8 +368,8 @@ impl SpendDag {
let mut to_traverse = BTreeSet::from_iter(vec![addr]);
while let Some(current_addr) = to_traverse.pop_first() {
// get the spend at this address
let indexes = match self.spends.get(current_addr) {
Some(entry) => entry.indexes(),
let dag_entry = match self.spends.get(current_addr) {
Some(entry) => entry,
None => {
warn!("Incoherent DAG, missing descendant spend when expecting one at: {current_addr:?}");
return Err(DagError::IncoherentDag(
Expand All @@ -355,6 +378,14 @@ impl SpendDag {
));
}
};
let (spends, indexes) = (dag_entry.spends(), dag_entry.indexes());

// get descendants via Tx data
let descendants_via_tx: BTreeSet<SpendAddress> = spends
.into_iter()
.flat_map(|s| s.spend.spent_tx.outputs.to_vec())
.map(|o| SpendAddress::from_unique_pubkey(&o.unique_pubkey))
.collect();

// get descendants via DAG
let descendants_via_dag: BTreeSet<&SpendAddress> = indexes
Expand All @@ -366,24 +397,17 @@ impl SpendDag {
})
.collect();

// get descendants via Tx data
let descendants_via_tx: BTreeSet<SpendAddress> = self
.spends
.get(current_addr)
.map(|entry| entry.spends())
.unwrap_or_default()
.into_iter()
.flat_map(|s| s.spend.spent_tx.outputs.to_vec())
.map(|o| SpendAddress::from_unique_pubkey(&o.unique_pubkey))
.collect();

// report inconsistencies
if descendants_via_dag != descendants_via_tx.iter().collect() {
warn!("Incoherent DAG at: {current_addr:?}");
return Err(DagError::IncoherentDag(
*current_addr,
format!("descendants via DAG: {descendants_via_dag:?} do not match descendants via TX: {descendants_via_tx:?}")
));
if matches!(dag_entry, DagEntry::NotGatheredYet(_)) {
debug!("Spend at {current_addr:?} was not gathered yet and has children refering to it, continuing traversal through those children...");
} else {
warn!("Incoherent DAG at: {current_addr:?}");
return Err(DagError::IncoherentDag(
*current_addr,
format!("descendants via DAG: {descendants_via_dag:?} do not match descendants via TX: {descendants_via_tx:?}")
));
}
}

// continue traversal
Expand Down Expand Up @@ -430,6 +454,7 @@ impl SpendDag {
pub fn record_faults(&mut self, source: &SpendAddress) -> Result<(), DagError> {
let faults = self.verify(source)?;

self.faults.clear();
for f in faults {
self.faults.entry(f.spend_address()).or_default().insert(f);
}
Expand Down Expand Up @@ -520,9 +545,14 @@ impl SpendDag {
// get the ancestors of this spend
let ancestor_spends = match self.get_ancestor_spends(spend) {
Ok(a) => a,
Err(e) => {
debug!("Failed to get ancestor spends of: {addr:?} {e}");
recorded_faults.insert(e);
Err(fault) => {
debug!("Failed to get ancestor spends of {addr:?}: {fault}");
recorded_faults.insert(fault.clone());

// if ancestry is invalid, poison all the descendants
let poison = format!("ancestry issue: {fault}");
let descendants_faults = self.poison_all_descendants(spend, poison)?;
recorded_faults.extend(descendants_faults);
return Ok(recorded_faults);
}
};
Expand Down
Loading

0 comments on commit 65924b8

Please sign in to comment.