Skip to content

Make full-scan/sync flow easier to reason about. #1838

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
evanlinjin merged 2 commits into from
Feb 28, 2025
Merged

Make full-scan/sync flow easier to reason about. #1838

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ee52745
feat(core)!: Make `TxUpdate` non-exhaustive
evanlinjin Jan 24, 2025
800f358
feat!: Improve spk-based syncing flow
evanlinjin Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 8 additions & 10 deletions crates/chain/benches/canonicalization.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,11 +132,10 @@ pub fn many_conflicting_unconfirmed(c: &mut Criterion) {
}],
..new_tx(i)
};
let update = TxUpdate {
txs: vec![Arc::new(tx)],
..Default::default()
};
let _ = tx_graph.apply_update_at(update, Some(i as u64));
let mut update = TxUpdate::default();
update.seen_ats = [(tx.compute_txid(), i as u64)].into();
update.txs = vec![Arc::new(tx)];
let _ = tx_graph.apply_update(update);
}
}));
c.bench_function("many_conflicting_unconfirmed::list_canonical_txs", {
Expand Down Expand Up @@ -169,11 +168,10 @@ pub fn many_chained_unconfirmed(c: &mut Criterion) {
..new_tx(i)
};
let txid = tx.compute_txid();
let update = TxUpdate {
txs: vec![Arc::new(tx)],
..Default::default()
};
let _ = tx_graph.apply_update_at(update, Some(i as u64));
let mut update = TxUpdate::default();
update.seen_ats = [(txid, i as u64)].into();
update.txs = vec![Arc::new(tx)];
let _ = tx_graph.apply_update(update);
// Store the next prevout.
previous_output = OutPoint::new(txid, 0);
}
Expand Down
25 changes: 0 additions & 25 deletions crates/chain/src/indexed_tx_graph.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,37 +91,12 @@ where
/// Apply an `update` directly.
///
/// `update` is a [`tx_graph::TxUpdate<A>`] and the resultant changes is returned as [`ChangeSet`].
#[cfg(feature = "std")]
#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
pub fn apply_update(&mut self, update: tx_graph::TxUpdate<A>) -> ChangeSet<A, I::ChangeSet> {
let tx_graph = self.graph.apply_update(update);
let indexer = self.index_tx_graph_changeset(&tx_graph);
ChangeSet { tx_graph, indexer }
}

/// Apply the given `update` with an optional `seen_at` timestamp.
///
/// `seen_at` represents when the update is seen (in unix seconds). It is used to determine the
/// `last_seen`s for all transactions in the update which have no corresponding anchor(s). The
/// `last_seen` value is used internally to determine precedence of conflicting unconfirmed
/// transactions (where the transaction with the lower `last_seen` value is omitted from the
/// canonical history).
///
/// Not setting a `seen_at` value means unconfirmed transactions introduced by this update will
/// not be part of the canonical history of transactions.
///
/// Use [`apply_update`](IndexedTxGraph::apply_update) to have the `seen_at` value automatically
/// set to the current time.
pub fn apply_update_at(
&mut self,
update: tx_graph::TxUpdate<A>,
seen_at: Option<u64>,
) -> ChangeSet<A, I::ChangeSet> {
let tx_graph = self.graph.apply_update_at(update, seen_at);
let indexer = self.index_tx_graph_changeset(&tx_graph);
ChangeSet { tx_graph, indexer }
}

/// Insert a floating `txout` of given `outpoint`.
pub fn insert_txout(&mut self, outpoint: OutPoint, txout: TxOut) -> ChangeSet<A, I::ChangeSet> {
let graph = self.graph.insert_txout(outpoint, txout);
Expand Down
62 changes: 15 additions & 47 deletions crates/chain/src/tx_graph.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,26 +110,26 @@ use core::{

impl<A: Ord> From<TxGraph<A>> for TxUpdate<A> {
fn from(graph: TxGraph<A>) -> Self {
Self {
txs: graph.full_txs().map(|tx_node| tx_node.tx).collect(),
txouts: graph
.floating_txouts()
.map(|(op, txo)| (op, txo.clone()))
.collect(),
anchors: graph
.anchors
.into_iter()
.flat_map(|(txid, anchors)| anchors.into_iter().map(move |a| (a, txid)))
.collect(),
seen_ats: graph.last_seen.into_iter().collect(),
}
let mut tx_update = TxUpdate::default();
tx_update.txs = graph.full_txs().map(|tx_node| tx_node.tx).collect();
tx_update.txouts = graph
.floating_txouts()
.map(|(op, txo)| (op, txo.clone()))
.collect();
tx_update.anchors = graph
.anchors
.into_iter()
.flat_map(|(txid, anchors)| anchors.into_iter().map(move |a| (a, txid)))
.collect();
tx_update.seen_ats = graph.last_seen.into_iter().collect();
tx_update
}
}

impl<A: Anchor> From<TxUpdate<A>> for TxGraph<A> {
fn from(update: TxUpdate<A>) -> Self {
let mut graph = TxGraph::<A>::default();
let _ = graph.apply_update_at(update, None);
let _ = graph.apply_update(update);
graph
}
}
Expand Down Expand Up @@ -719,52 +719,20 @@ impl<A: Anchor> TxGraph<A> {
///
/// The returned [`ChangeSet`] is the set difference between `update` and `self` (transactions that
/// exist in `update` but not in `self`).
#[cfg(feature = "std")]
#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
pub fn apply_update(&mut self, update: TxUpdate<A>) -> ChangeSet<A> {
use std::time::*;
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("current time must be greater than epoch anchor");
self.apply_update_at(update, Some(now.as_secs()))
}

/// Extends this graph with the given `update` alongside an optional `seen_at` timestamp.
///
/// `seen_at` represents when the update is seen (in unix seconds). It is used to determine the
/// `last_seen`s for all transactions in the update which have no corresponding anchor(s). The
/// `last_seen` value is used internally to determine precedence of conflicting unconfirmed
/// transactions (where the transaction with the lower `last_seen` value is omitted from the
/// canonical history).
///
/// Not setting a `seen_at` value means unconfirmed transactions introduced by this update will
/// not be part of the canonical history of transactions.
///
/// Use [`apply_update`](TxGraph::apply_update) to have the `seen_at` value automatically set
/// to the current time.
pub fn apply_update_at(&mut self, update: TxUpdate<A>, seen_at: Option<u64>) -> ChangeSet<A> {
let mut changeset = ChangeSet::<A>::default();
let mut unanchored_txs = HashSet::<Txid>::new();
for tx in update.txs {
if unanchored_txs.insert(tx.compute_txid()) {
changeset.merge(self.insert_tx(tx));
}
changeset.merge(self.insert_tx(tx));
}
for (outpoint, txout) in update.txouts {
changeset.merge(self.insert_txout(outpoint, txout));
}
for (anchor, txid) in update.anchors {
unanchored_txs.remove(&txid);
changeset.merge(self.insert_anchor(txid, anchor));
}
for (txid, seen_at) in update.seen_ats {
changeset.merge(self.insert_seen_at(txid, seen_at));
}
if let Some(seen_at) = seen_at {
for txid in unanchored_txs {
changeset.merge(self.insert_seen_at(txid, seen_at));
}
}
changeset
}

Expand Down
121 changes: 56 additions & 65 deletions crates/chain/tests/test_tx_graph.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ fn insert_txouts() {
// Insert partials transactions.
update.txouts.insert(*outpoint, txout.clone());
// Mark them unconfirmed.
update.seen_ats.insert(outpoint.txid, unconf_seen_at);
update.seen_ats.insert((outpoint.txid, unconf_seen_at));
}

// Insert the full transaction.
Expand Down Expand Up @@ -1231,74 +1231,65 @@ fn tx_graph_update_conversion() {

let test_cases: &[TestCase] = &[
("empty_update", TxUpdate::default()),
(
"single_tx",
TxUpdate {
txs: vec![make_tx(0).into()],
..Default::default()
},
),
(
"two_txs",
TxUpdate {
txs: vec![make_tx(0).into(), make_tx(1).into()],
..Default::default()
},
),
(
"with_floating_txouts",
TxUpdate {
txs: vec![make_tx(0).into(), make_tx(1).into()],
txouts: [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("b"), 0), make_txout(2)),
]
.into(),
..Default::default()
},
),
(
"with_anchors",
TxUpdate {
txs: vec![make_tx(0).into(), make_tx(1).into()],
txouts: [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("b"), 0), make_txout(2)),
]
.into(),
anchors: [
(ConfirmationBlockTime::default(), hash!("a")),
(ConfirmationBlockTime::default(), hash!("b")),
]
.into(),
..Default::default()
},
),
(
"with_seen_ats",
TxUpdate {
txs: vec![make_tx(0).into(), make_tx(1).into()],
txouts: [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("d"), 0), make_txout(2)),
]
.into(),
anchors: [
(ConfirmationBlockTime::default(), hash!("a")),
(ConfirmationBlockTime::default(), hash!("b")),
]
.into(),
seen_ats: [(hash!("c"), 12346)].into_iter().collect(),
},
),
("single_tx", {
let mut tx_update = TxUpdate::default();
tx_update.txs = vec![make_tx(0).into()];
tx_update
}),
("two_txs", {
let mut tx_update = TxUpdate::default();
tx_update.txs = vec![make_tx(0).into(), make_tx(1).into()];
tx_update
}),
("with_floating_txouts", {
let mut tx_update = TxUpdate::default();
tx_update.txs = vec![make_tx(0).into(), make_tx(1).into()];
tx_update.txouts = [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("b"), 0), make_txout(2)),
]
.into();
tx_update
}),
("with_anchors", {
let mut tx_update = TxUpdate::default();
tx_update.txs = vec![make_tx(0).into(), make_tx(1).into()];
tx_update.txouts = [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("b"), 0), make_txout(2)),
]
.into();
tx_update.anchors = [
(ConfirmationBlockTime::default(), hash!("a")),
(ConfirmationBlockTime::default(), hash!("b")),
]
.into();
tx_update
}),
("with_seen_ats", {
let mut tx_update = TxUpdate::default();
tx_update.txs = vec![make_tx(0).into(), make_tx(1).into()];
tx_update.txouts = [
(OutPoint::new(hash!("a"), 0), make_txout(0)),
(OutPoint::new(hash!("a"), 1), make_txout(1)),
(OutPoint::new(hash!("d"), 0), make_txout(2)),
]
.into();
tx_update.anchors = [
(ConfirmationBlockTime::default(), hash!("a")),
(ConfirmationBlockTime::default(), hash!("b")),
]
.into();
tx_update.seen_ats = [(hash!("c"), 12346)].into_iter().collect();
tx_update
}),
];

for (test_name, update) in test_cases {
let mut tx_graph = TxGraph::<ConfirmationBlockTime>::default();
let _ = tx_graph.apply_update_at(update.clone(), None);
let _ = tx_graph.apply_update(update.clone());
let update_from_tx_graph: TxUpdate<ConfirmationBlockTime> = tx_graph.into();

assert_eq!(
Expand Down
Loading
Loading