frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat!: Rework sqlite, changesets, persistence and wallet-construction

Browse Source 
Rework sqlite: Instead of only supported one schema (defined in
`bdk_sqlite`), we have a schema per changeset type for more flexiblity.

* rm `bdk_sqlite` crate (as we don't need `bdk_sqlite::Store` anymore).
* add `sqlite` feature on `bdk_chain` which adds methods on each
  changeset type for initializing tables, loading the changeset and
  writing.

Rework changesets: Some callers may want to use `KeychainTxOutIndex`
where `K` may change per descriptor on every run. So we only want to
persist the last revealed indices by `DescriptorId` (which uniquely-ish
identifies the descriptor).

* rm `keychain_added` field from `keychain_txout`'s changeset.
* Add `keychain_added` to `CombinedChangeSet` (which is renamed to
  `WalletChangeSet`).

Rework persistence: add back some safety and convenience when persisting
our types. Working with changeset directly (as we were doing before) can
be cumbersome.

* Intoduce `struct Persisted<T>` which wraps a type `T` which stores
  staged changes to it. This adds safety when creating and or loading
  `T` from db.
* `struct Persisted<T>` methods, `create`, `load` and `persist`, are
  avaliable if `trait PersistWith<Db>` is implemented for `T`. `Db`
  represents the database connection and `PersistWith` should be
  implemented per database-type.
* For async, we have `trait PersistedAsyncWith<Db>`.
* `Wallet` has impls of `PersistedWith<rusqlite::Connection>`,
  `PersistedWith<rusqlite::Transaction>` and
  `PersistedWith<bdk_file_store::Store>` by default.

Rework wallet-construction: Before, we had multiple methods for loading
and creating with different input-counts so it would be unwieldly to add
more parameters in the future. This also makes it difficult to impl
`PersistWith` (which has a single method for `load` that takes in
`PersistWith::LoadParams` and a single method for `create` that takes in
`PersistWith::CreateParams`).

* Introduce a builder pattern when constructing a `Wallet`. For loading
  from persistence or `ChangeSet`, we have `LoadParams`. For creating a
  new wallet, we have `CreateParams`.
This commit is contained in:
志宇
2024-07-11 04:49:01 +00:00
parent d99b3ef4b4
commit 6b43001951
49 changed files with 2217 additions and 2058 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
example-crates/example_bitcoind_rpc_polling/src/main.rs
View File

@@ -38,7 +38,7 @@ const DB_COMMIT_DELAY: Duration = Duration::from_secs(60);
type ChangeSet = (
local_chain::ChangeSet,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet<Keychain>>,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet>,
);
#[derive(Debug)]

4
example-crates/example_cli/src/lib.rs
View File

@@ -30,7 +30,7 @@ use clap::{Parser, Subcommand};
pub type KeychainTxGraph<A> = IndexedTxGraph<A, KeychainTxOutIndex<Keychain>>;
pub type KeychainChangeSet<A> = (
local_chain::ChangeSet,
indexed_tx_graph::ChangeSet<A, keychain_txout::ChangeSet<Keychain>>,
indexed_tx_graph::ChangeSet<A, keychain_txout::ChangeSet>,
);
#[derive(Parser)]
@@ -191,7 +191,7 @@ impl core::fmt::Display for Keychain {
}
pub struct CreateTxChange {
pub index_changeset: keychain_txout::ChangeSet<Keychain>,
pub index_changeset: keychain_txout::ChangeSet,
pub change_keychain: Keychain,
pub index: u32,
}

2
example-crates/example_electrum/src/main.rs
View File

@@ -100,7 +100,7 @@ pub struct ScanOptions {
type ChangeSet = (
local_chain::ChangeSet,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet<Keychain>>,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet>,
);
fn main() -> anyhow::Result<()> {

8
example-crates/example_esplora/src/main.rs
View File

@@ -22,11 +22,11 @@ use example_cli::{
};
const DB_MAGIC: &[u8] = b"bdk_example_esplora";
const DB_PATH: &str = ".bdk_esplora_example.db";
const DB_PATH: &str = "bdk_example_esplora.db";
type ChangeSet = (
local_chain::ChangeSet,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet<Keychain>>,
indexed_tx_graph::ChangeSet<ConfirmationBlockTime, keychain_txout::ChangeSet>,
);
#[derive(Subcommand, Debug, Clone)]
@@ -84,7 +84,7 @@ impl EsploraArgs {
Network::Bitcoin => "https://blockstream.info/api",
Network::Testnet => "https://blockstream.info/testnet/api",
Network::Regtest => "http://localhost:3002",
Network::Signet => "https://mempool.space/signet/api",
Network::Signet => "http://signet.bitcoindevkit.net",
_ => panic!("unsupported network"),
});
@@ -96,7 +96,7 @@ impl EsploraArgs {
#[derive(Parser, Debug, Clone, PartialEq)]
pub struct ScanOptions {
/// Max number of concurrent esplora server requests.
#[clap(long, default_value = "1")]
#[clap(long, default_value = "5")]
pub parallel_requests: usize,
}

2
example-crates/wallet_electrum/Cargo.toml
View File

@@ -4,7 +4,7 @@ version = "0.2.0"
edition = "2021"
[dependencies]
bdk_wallet = { path = "../../crates/wallet" }
bdk_wallet = { path = "../../crates/wallet", feature = ["file_store"] }
bdk_electrum = { path = "../../crates/electrum" }
bdk_file_store = { path = "../../crates/file_store" }
anyhow = "1"

55
example-crates/wallet_electrum/src/main.rs
View File

@@ -1,53 +1,52 @@
const DB_MAGIC: &str = "bdk_wallet_electrum_example";
const SEND_AMOUNT: Amount = Amount::from_sat(5000);
const STOP_GAP: usize = 50;
const BATCH_SIZE: usize = 5;
use anyhow::anyhow;
use bdk_wallet::wallet::CreateParams;
use bdk_wallet::wallet::LoadParams;
use std::io::Write;
use std::str::FromStr;
use bdk_electrum::electrum_client;
use bdk_electrum::BdkElectrumClient;
use bdk_file_store::Store;
use bdk_wallet::bitcoin::Network;
use bdk_wallet::bitcoin::{Address, Amount};
use bdk_wallet::chain::collections::HashSet;
use bdk_wallet::{bitcoin::Network, Wallet};
use bdk_wallet::{KeychainKind, SignOptions};
const DB_MAGIC: &str = "bdk_wallet_electrum_example";
const SEND_AMOUNT: Amount = Amount::from_sat(5000);
const STOP_GAP: usize = 50;
const BATCH_SIZE: usize = 5;
const NETWORK: Network = Network::Testnet;
const EXTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
const INTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
const ELECTRUM_URL: &str = "ssl://electrum.blockstream.info:60002";
fn main() -> Result<(), anyhow::Error> {
let db_path = std::env::temp_dir().join("bdk-electrum-example");
let db_path = "bdk-electrum-example.db";
let mut db =
Store::<bdk_wallet::wallet::ChangeSet>::open_or_create_new(DB_MAGIC.as_bytes(), db_path)?;
let external_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
let internal_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
let changeset = db
.aggregate_changesets()
.map_err(|e| anyhow!("load changes error: {}", e))?;
let mut wallet = Wallet::new_or_load(
external_descriptor,
internal_descriptor,
changeset,
Network::Testnet,
)?;
let load_params = LoadParams::with_descriptors(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let create_params = CreateParams::new(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let mut wallet = match load_params.load_wallet(&mut db)? {
Some(wallet) => wallet,
None => create_params.create_wallet(&mut db)?,
};
let address = wallet.next_unused_address(KeychainKind::External);
if let Some(changeset) = wallet.take_staged() {
db.append_changeset(&changeset)?;
}
wallet.persist(&mut db)?;
println!("Generated Address: {}", address);
let balance = wallet.balance();
println!("Wallet balance before syncing: {} sats", balance.total());
print!("Syncing...");
let client = BdkElectrumClient::new(electrum_client::Client::new(
"ssl://electrum.blockstream.info:60002",
)?);
let client = BdkElectrumClient::new(electrum_client::Client::new(ELECTRUM_URL)?);
// Populate the electrum client's transaction cache so it doesn't redownload transaction we
// already have.
client.populate_tx_cache(&wallet);
client.populate_tx_cache(wallet.tx_graph());
let request = wallet
.start_full_scan()
@@ -71,9 +70,7 @@ fn main() -> Result<(), anyhow::Error> {
println!();
wallet.apply_update(update)?;
if let Some(changeset) = wallet.take_staged() {
db.append_changeset(&changeset)?;
}
wallet.persist(&mut db)?;
let balance = wallet.balance();
println!("Wallet balance after syncing: {} sats", balance.total());

3
example-crates/wallet_esplora_async/Cargo.toml
View File

@@ -6,8 +6,7 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bdk_wallet = { path = "../../crates/wallet" }
bdk_wallet = { path = "../../crates/wallet", features = ["sqlite"] }
bdk_esplora = { path = "../../crates/esplora", features = ["async-https"] }
bdk_sqlite = { path = "../../crates/sqlite" }
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
anyhow = "1"

94
example-crates/wallet_esplora_async/src/main.rs
View File

@@ -1,76 +1,55 @@
use std::{collections::BTreeSet, io::Write, str::FromStr};
use std::{collections::BTreeSet, io::Write};
use anyhow::Ok;
use bdk_esplora::{esplora_client, EsploraAsyncExt};
use bdk_wallet::{
bitcoin::{Address, Amount, Network, Script},
KeychainKind, SignOptions, Wallet,
bitcoin::{Amount, Network},
rusqlite::Connection,
wallet::{CreateParams, LoadParams},
KeychainKind, SignOptions,
};
use bdk_sqlite::{rusqlite::Connection, Store};
const SEND_AMOUNT: Amount = Amount::from_sat(5000);
const STOP_GAP: usize = 50;
const STOP_GAP: usize = 5;
const PARALLEL_REQUESTS: usize = 5;
const DB_PATH: &str = "bdk-example-esplora-async.sqlite";
const NETWORK: Network = Network::Signet;
const EXTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
const INTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
const ESPLORA_URL: &str = "http://signet.bitcoindevkit.net";
#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
let db_path = "bdk-esplora-async-example.sqlite";
let conn = Connection::open(db_path)?;
let mut db = Store::new(conn)?;
let external_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
let internal_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
let changeset = db.read()?;
let mut conn = Connection::open(DB_PATH)?;
let mut wallet = Wallet::new_or_load(
external_descriptor,
internal_descriptor,
changeset,
Network::Signet,
)?;
let load_params = LoadParams::with_descriptors(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let create_params = CreateParams::new(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let mut wallet = match load_params.load_wallet(&mut conn)? {
Some(wallet) => wallet,
None => create_params.create_wallet(&mut conn)?,
};
let address = wallet.next_unused_address(KeychainKind::External);
if let Some(changeset) = wallet.take_staged() {
db.write(&changeset)?;
}
println!("Generated Address: {}", address);
wallet.persist(&mut conn)?;
println!("Next unused address: ({}) {}", address.index, address);
let balance = wallet.balance();
println!("Wallet balance before syncing: {} sats", balance.total());
print!("Syncing...");
let client = esplora_client::Builder::new("http://signet.bitcoindevkit.net").build_async()?;
let client = esplora_client::Builder::new(ESPLORA_URL).build_async()?;
fn generate_inspect(kind: KeychainKind) -> impl FnMut(u32, &Script) + Send + Sync + 'static {
let mut once = Some(());
let mut stdout = std::io::stdout();
move |spk_i, _| {
match once.take() {
Some(_) => print!("\nScanning keychain [{:?}]", kind),
None => print!(" {:<3}", spk_i),
};
stdout.flush().expect("must flush");
}
}
let request = wallet
.start_full_scan()
.inspect_spks_for_all_keychains({
let mut once = BTreeSet::<KeychainKind>::new();
move |keychain, spk_i, _| {
match once.insert(keychain) {
true => print!("\nScanning keychain [{:?}]", keychain),
false => print!(" {:<3}", spk_i),
}
std::io::stdout().flush().expect("must flush")
let request = wallet.start_full_scan().inspect_spks_for_all_keychains({
let mut once = BTreeSet::<KeychainKind>::new();
move |keychain, spk_i, _| {
if once.insert(keychain) {
print!("\nScanning keychain [{:?}] ", keychain);
}
})
.inspect_spks_for_keychain(
KeychainKind::External,
generate_inspect(KeychainKind::External),
)
.inspect_spks_for_keychain(
KeychainKind::Internal,
generate_inspect(KeychainKind::Internal),
);
print!(" {:<3}", spk_i);
std::io::stdout().flush().expect("must flush")
}
});
let mut update = client
.full_scan(request, STOP_GAP, PARALLEL_REQUESTS)
@@ -79,9 +58,7 @@ async fn main() -> Result<(), anyhow::Error> {
let _ = update.graph_update.update_last_seen_unconfirmed(now);
wallet.apply_update(update)?;
if let Some(changeset) = wallet.take_staged() {
db.write(&changeset)?;
}
wallet.persist(&mut conn)?;
println!();
let balance = wallet.balance();
@@ -95,12 +72,9 @@ async fn main() -> Result<(), anyhow::Error> {
std::process::exit(0);
}
let faucet_address = Address::from_str("mkHS9ne12qx9pS9VojpwU5xtRd4T7X7ZUt")?
.require_network(Network::Signet)?;
let mut tx_builder = wallet.build_tx();
tx_builder
.add_recipient(faucet_address.script_pubkey(), SEND_AMOUNT)
.add_recipient(address.script_pubkey(), SEND_AMOUNT)
.enable_rbf();
let mut psbt = tx_builder.finish()?;

71
example-crates/wallet_esplora_blocking/src/main.rs
View File

@@ -1,52 +1,56 @@
const DB_MAGIC: &str = "bdk_wallet_esplora_example";
const SEND_AMOUNT: Amount = Amount::from_sat(1000);
const STOP_GAP: usize = 5;
const PARALLEL_REQUESTS: usize = 1;
use std::{collections::BTreeSet, io::Write, str::FromStr};
use std::{collections::BTreeSet, io::Write};
use bdk_esplora::{esplora_client, EsploraExt};
use bdk_file_store::Store;
use bdk_wallet::{
bitcoin::{Address, Amount, Network},
KeychainKind, SignOptions, Wallet,
bitcoin::{Amount, Network},
wallet::{CreateParams, LoadParams},
KeychainKind, SignOptions,
};
fn main() -> Result<(), anyhow::Error> {
let db_path = std::env::temp_dir().join("bdk-esplora-example");
let mut db =
Store::<bdk_wallet::wallet::ChangeSet>::open_or_create_new(DB_MAGIC.as_bytes(), db_path)?;
let external_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
let internal_descriptor = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
let changeset = db.aggregate_changesets()?;
const DB_MAGIC: &str = "bdk_wallet_esplora_example";
const DB_PATH: &str = "bdk-example-esplora-blocking.db";
const SEND_AMOUNT: Amount = Amount::from_sat(5000);
const STOP_GAP: usize = 5;
const PARALLEL_REQUESTS: usize = 5;
let mut wallet = Wallet::new_or_load(
external_descriptor,
internal_descriptor,
changeset,
Network::Testnet,
)?;
const NETWORK: Network = Network::Signet;
const EXTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)";
const INTERNAL_DESC: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)";
const ESPLORA_URL: &str = "http://signet.bitcoindevkit.net";
fn main() -> Result<(), anyhow::Error> {
let mut db =
Store::<bdk_wallet::wallet::ChangeSet>::open_or_create_new(DB_MAGIC.as_bytes(), DB_PATH)?;
let load_params = LoadParams::with_descriptors(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let create_params = CreateParams::new(EXTERNAL_DESC, INTERNAL_DESC, NETWORK)?;
let mut wallet = match load_params.load_wallet(&mut db)? {
Some(wallet) => wallet,
None => create_params.create_wallet(&mut db)?,
};
let address = wallet.next_unused_address(KeychainKind::External);
if let Some(changeset) = wallet.take_staged() {
db.append_changeset(&changeset)?;
}
println!("Generated Address: {}", address);
wallet.persist(&mut db)?;
println!(
"Next unused address: ({}) {}",
address.index, address.address
);
let balance = wallet.balance();
println!("Wallet balance before syncing: {} sats", balance.total());
print!("Syncing...");
let client =
esplora_client::Builder::new("https://blockstream.info/testnet/api").build_blocking();
let client = esplora_client::Builder::new(ESPLORA_URL).build_blocking();
let request = wallet.start_full_scan().inspect_spks_for_all_keychains({
let mut once = BTreeSet::<KeychainKind>::new();
move |keychain, spk_i, _| {
match once.insert(keychain) {
true => print!("\nScanning keychain [{:?}]", keychain),
false => print!(" {:<3}", spk_i),
};
if once.insert(keychain) {
print!("\nScanning keychain [{:?}] ", keychain);
}
print!(" {:<3}", spk_i);
std::io::stdout().flush().expect("must flush")
}
});
@@ -72,12 +76,9 @@ fn main() -> Result<(), anyhow::Error> {
std::process::exit(0);
}
let faucet_address = Address::from_str("mkHS9ne12qx9pS9VojpwU5xtRd4T7X7ZUt")?
.require_network(Network::Testnet)?;
let mut tx_builder = wallet.build_tx();
tx_builder
.add_recipient(faucet_address.script_pubkey(), SEND_AMOUNT)
.add_recipient(address.script_pubkey(), SEND_AMOUNT)
.enable_rbf();
let mut psbt = tx_builder.finish()?;

2
example-crates/wallet_rpc/Cargo.toml
View File

@@ -6,7 +6,7 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
bdk_wallet = { path = "../../crates/wallet" }
bdk_wallet = { path = "../../crates/wallet", features = ["file_store"] }
bdk_file_store = { path = "../../crates/file_store" }
bdk_bitcoind_rpc = { path = "../../crates/bitcoind_rpc" }

24
example-crates/wallet_rpc/src/main.rs
View File

@@ -5,7 +5,7 @@ use bdk_bitcoind_rpc::{
use bdk_file_store::Store;
use bdk_wallet::{
bitcoin::{Block, Network, Transaction},
wallet::Wallet,
wallet::{CreateParams, LoadParams},
};
use clap::{self, Parser};
use std::{path::PathBuf, sync::mpsc::sync_channel, thread::spawn, time::Instant};
@@ -90,14 +90,14 @@ fn main() -> anyhow::Result<()> {
DB_MAGIC.as_bytes(),
args.db_path,
)?;
let changeset = db.aggregate_changesets()?;
let mut wallet = Wallet::new_or_load(
&args.descriptor,
&args.change_descriptor,
changeset,
args.network,
)?;
let load_params =
LoadParams::with_descriptors(&args.descriptor, &args.change_descriptor, args.network)?;
let create_params = CreateParams::new(&args.descriptor, &args.change_descriptor, args.network)?;
let mut wallet = match load_params.load_wallet(&mut db)? {
Some(wallet) => wallet,
None => create_params.create_wallet(&mut db)?,
};
println!(
"Loaded wallet in {}s",
start_load_wallet.elapsed().as_secs_f32()
@@ -146,9 +146,7 @@ fn main() -> anyhow::Result<()> {
let connected_to = block_emission.connected_to();
let start_apply_block = Instant::now();
wallet.apply_block_connected_to(&block_emission.block, height, connected_to)?;
if let Some(changeset) = wallet.take_staged() {
db.append_changeset(&changeset)?;
}
wallet.persist(&mut db)?;
let elapsed = start_apply_block.elapsed().as_secs_f32();
println!(
"Applied block {} at height {} in {}s",
@@ -158,9 +156,7 @@ fn main() -> anyhow::Result<()> {
Emission::Mempool(mempool_emission) => {
let start_apply_mempool = Instant::now();
wallet.apply_unconfirmed_txs(mempool_emission.iter().map(|(tx, time)| (tx, *time)));
if let Some(changeset) = wallet.take_staged() {
db.append_changeset(&changeset)?;
}
wallet.persist(&mut db)?;
println!(
"Applied unconfirmed transactions in {}s",
start_apply_mempool.elapsed().as_secs_f32()