pub extern crate ark;
pub extern crate bip39;
pub extern crate lightning_invoice;
pub extern crate lnurl as lnurllib;
#[macro_use] extern crate anyhow;
#[macro_use] extern crate async_trait;
#[macro_use] extern crate serde;
pub mod chain;
pub mod exit;
pub mod movement;
pub mod onchain;
pub mod persist;
pub mod round;
pub mod subsystem;
pub mod vtxo;
#[cfg(feature = "pid-lock")]
pub mod pid_lock;
mod arkoor;
mod board;
mod config;
mod daemon;
mod fees;
mod lightning;
mod mailbox;
mod notification;
mod offboard;
#[cfg(feature = "socks5-proxy")]
mod proxy;
mod psbtext;
mod utils;
pub use self::arkoor::ArkoorCreateResult;
pub use self::config::{BarkNetwork, Config};
pub use self::daemon::DaemonHandle;
pub use self::fees::FeeEstimate;
pub use self::notification::{WalletNotification, NotificationStream};
pub use self::vtxo::WalletVtxo;
use std::collections::HashSet;
use std::sync::Arc;
use anyhow::{bail, Context};
use bip39::Mnemonic;
use bitcoin::{Amount, Network, OutPoint};
use bitcoin::bip32::{self, ChildNumber, Fingerprint};
use bitcoin::secp256k1::{self, Keypair, PublicKey};
use log::{trace, info, warn, error};
use tokio::sync::{Mutex, RwLock};
use ark::lightning::PaymentHash;
use ark::{ArkInfo, ProtocolEncoding, Vtxo, VtxoId, VtxoPolicy, VtxoRequest};
use ark::address::VtxoDelivery;
use ark::fees::{validate_and_subtract_fee_min_dust, VtxoFeeInfo};
use ark::vtxo::{Full, PubkeyVtxoPolicy, VtxoRef};
use ark::vtxo::policy::signing::VtxoSigner;
use bitcoin_ext::{BlockHeight, P2TR_DUST, TxStatus};
use server_rpc::{protos, ServerConnection};
use crate::chain::{ChainSource, ChainSourceSpec};
use crate::exit::Exit;
use crate::movement::{Movement, MovementStatus, PaymentMethod};
use crate::movement::manager::MovementManager;
use crate::movement::update::MovementUpdate;
use crate::notification::NotificationDispatch;
use crate::onchain::{ExitUnilaterally, PreparePsbt, SignPsbt, Utxo};
use crate::onchain::DaemonizableOnchainWallet;
use crate::persist::BarkPersister;
use crate::persist::models::{PendingOffboard, RoundStateId, StoredRoundState, Unlocked};
#[cfg(feature = "socks5-proxy")]
use crate::proxy::proxy_for_url;
use crate::round::{RoundParticipation, RoundStateLockIndex, RoundStatus};
use crate::subsystem::{ArkoorMovement, RoundMovement};
use crate::vtxo::{FilterVtxos, RefreshStrategy, VtxoFilter, VtxoState, VtxoStateKind};
#[cfg(all(feature = "wasm-web", feature = "socks5-proxy"))]
compile_error!("features `wasm-web` does not support feature `socks5-proxy");
const BARK_PURPOSE_INDEX: u32 = 350;
const VTXO_KEYS_INDEX: u32 = 0;
const MAILBOX_KEY_INDEX: u32 = 1;
const RECOVERY_MAILBOX_KEY_INDEX: u32 = 2;
lazy_static::lazy_static! {
static ref SECP: secp256k1::Secp256k1<secp256k1::All> = secp256k1::Secp256k1::new();
}
fn log_server_pubkey_changed_error(expected: PublicKey, got: PublicKey) {
error!(
"
Server public key has changed!
The Ark server's public key is different from the one stored when this
wallet was created. This typically happens when:
- The server operator has rotated their keys
- You are connecting to a different server
- The server has been replaced
For safety, this wallet will not connect to the server until you
resolve this. You can recover your funds on-chain by doing an emergency exit.
This will exit your VTXOs to on-chain Bitcoin without needing the server's cooperation.
Expected: {expected}
Got: {got}")
}
#[derive(Debug, Clone)]
pub struct LightningReceiveBalance {
pub total: Amount,
pub claimable: Amount,
}
#[derive(Debug, Clone)]
pub struct Balance {
pub spendable: Amount,
pub pending_lightning_send: Amount,
pub claimable_lightning_receive: Amount,
pub pending_in_round: Amount,
pub pending_exit: Option<Amount>,
pub pending_board: Amount,
}
pub struct UtxoInfo {
pub outpoint: OutPoint,
pub amount: Amount,
pub confirmation_height: Option<u32>,
}
impl From<Utxo> for UtxoInfo {
fn from(value: Utxo) -> Self {
match value {
Utxo::Local(o) => UtxoInfo {
outpoint: o.outpoint,
amount: o.amount,
confirmation_height: o.confirmation_height,
},
Utxo::Exit(e) => UtxoInfo {
outpoint: e.vtxo.point(),
amount: e.vtxo.amount(),
confirmation_height: Some(e.height),
},
}
}
}
pub struct OffchainBalance {
pub available: Amount,
pub pending_in_round: Amount,
pub pending_exit: Amount,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WalletProperties {
pub network: Network,
pub fingerprint: Fingerprint,
pub server_pubkey: Option<PublicKey>,
}
pub struct WalletSeed {
master: bip32::Xpriv,
vtxo: bip32::Xpriv,
}
impl WalletSeed {
fn new(network: Network, seed: &[u8; 64]) -> Self {
let bark_path = [ChildNumber::from_hardened_idx(BARK_PURPOSE_INDEX).unwrap()];
let master = bip32::Xpriv::new_master(network, seed)
.expect("invalid seed")
.derive_priv(&SECP, &bark_path)
.expect("purpose is valid");
let vtxo_path = [ChildNumber::from_hardened_idx(VTXO_KEYS_INDEX).unwrap()];
let vtxo = master.derive_priv(&SECP, &vtxo_path)
.expect("vtxo path is valid");
Self { master, vtxo }
}
fn fingerprint(&self) -> Fingerprint {
self.master.fingerprint(&SECP)
}
fn derive_vtxo_keypair(&self, idx: u32) -> Keypair {
self.vtxo.derive_priv(&SECP, &[idx.into()]).unwrap().to_keypair(&SECP)
}
fn to_mailbox_keypair(&self) -> Keypair {
let mailbox_path = [ChildNumber::from_hardened_idx(MAILBOX_KEY_INDEX).unwrap()];
self.master.derive_priv(&SECP, &mailbox_path).unwrap().to_keypair(&SECP)
}
fn to_recovery_mailbox_keypair(&self) -> Keypair {
let mailbox_path = [ChildNumber::from_hardened_idx(RECOVERY_MAILBOX_KEY_INDEX).unwrap()];
self.master.derive_priv(&SECP, &mailbox_path).unwrap().to_keypair(&SECP)
}
}
pub struct Wallet {
pub chain: Arc<ChainSource>,
pub exit: RwLock<Exit>,
pub movements: Arc<MovementManager>,
notifications: NotificationDispatch,
config: Config,
db: Arc<dyn BarkPersister>,
seed: WalletSeed,
server: parking_lot::RwLock<Option<ServerConnection>>,
inflight_lightning_payments: Mutex<HashSet<PaymentHash>>,
round_state_lock_index: RoundStateLockIndex,
}
impl Wallet {
pub fn chain_source(
config: &Config,
) -> anyhow::Result<ChainSourceSpec> {
if let Some(ref url) = config.esplora_address {
Ok(ChainSourceSpec::Esplora {
url: url.clone(),
})
} else if let Some(ref url) = config.bitcoind_address {
let auth = if let Some(ref c) = config.bitcoind_cookiefile {
bitcoin_ext::rpc::Auth::CookieFile(c.clone())
} else {
bitcoin_ext::rpc::Auth::UserPass(
config.bitcoind_user.clone().context("need bitcoind auth config")?,
config.bitcoind_pass.clone().context("need bitcoind auth config")?,
)
};
Ok(ChainSourceSpec::Bitcoind {
url: url.clone(),
auth,
})
} else {
bail!("Need to either provide esplora or bitcoind info");
}
}
pub fn require_chainsource_version(&self) -> anyhow::Result<()> {
self.chain.require_version()
}
pub async fn network(&self) -> anyhow::Result<Network> {
Ok(self.properties().await?.network)
}
pub async fn peek_next_keypair(&self) -> anyhow::Result<(Keypair, u32)> {
let last_revealed = self.db.get_last_vtxo_key_index().await?;
let index = last_revealed.map(|i| i + 1).unwrap_or(u32::MIN);
let keypair = self.seed.derive_vtxo_keypair(index);
Ok((keypair, index))
}
pub async fn derive_store_next_keypair(&self) -> anyhow::Result<(Keypair, u32)> {
let (keypair, index) = self.peek_next_keypair().await?;
self.db.store_vtxo_key(index, keypair.public_key()).await?;
Ok((keypair, index))
}
#[deprecated(note = "use peek_keypair instead")]
pub async fn peak_keypair(&self, index: u32) -> anyhow::Result<Keypair> {
self.peek_keypair(index).await
}
pub async fn peek_keypair(&self, index: u32) -> anyhow::Result<Keypair> {
let keypair = self.seed.derive_vtxo_keypair(index);
if self.db.get_public_key_idx(&keypair.public_key()).await?.is_some() {
Ok(keypair)
} else {
bail!("VTXO key {} does not exist, please derive it first", index)
}
}
pub async fn pubkey_keypair(&self, public_key: &PublicKey) -> anyhow::Result<Option<(u32, Keypair)>> {
if let Some(index) = self.db.get_public_key_idx(&public_key).await? {
Ok(Some((index, self.seed.derive_vtxo_keypair(index))))
} else {
Ok(None)
}
}
pub async fn get_vtxo_key(&self, vtxo: impl VtxoRef) -> anyhow::Result<Keypair> {
let wallet_vtxo = self.get_vtxo_by_id(vtxo.vtxo_id()).await?;
let pubkey = self.find_signable_clause(&wallet_vtxo.vtxo).await
.context("VTXO is not signable by wallet")?
.pubkey();
let idx = self.db.get_public_key_idx(&pubkey).await?
.context("VTXO key not found")?;
Ok(self.seed.derive_vtxo_keypair(idx))
}
#[deprecated(note = "use peek_address instead")]
pub async fn peak_address(&self, index: u32) -> anyhow::Result<ark::Address> {
self.peek_address(index).await
}
pub async fn peek_address(&self, index: u32) -> anyhow::Result<ark::Address> {
let (_, ark_info) = &self.require_server().await?;
let network = self.properties().await?.network;
let keypair = self.peek_keypair(index).await?;
let mailbox = self.mailbox_identifier();
Ok(ark::Address::builder()
.testnet(network != bitcoin::Network::Bitcoin)
.server_pubkey(ark_info.server_pubkey)
.pubkey_policy(keypair.public_key())
.mailbox(ark_info.mailbox_pubkey, mailbox, &keypair)
.expect("Failed to assign mailbox")
.into_address().unwrap())
}
pub async fn new_address_with_index(&self) -> anyhow::Result<(ark::Address, u32)> {
let (_, index) = self.derive_store_next_keypair().await?;
let addr = self.peek_address(index).await?;
Ok((addr, index))
}
pub async fn new_address(&self) -> anyhow::Result<ark::Address> {
let (addr, _) = self.new_address_with_index().await?;
Ok(addr)
}
pub async fn create(
mnemonic: &Mnemonic,
network: Network,
config: Config,
db: Arc<dyn BarkPersister>,
force: bool,
) -> anyhow::Result<Wallet> {
trace!("Config: {:?}", config);
if let Some(existing) = db.read_properties().await? {
trace!("Existing config: {:?}", existing);
bail!("cannot overwrite already existing config")
}
let server_pubkey = if !force {
match Self::connect_to_server(&config, network).await {
Ok(conn) => {
let ark_info = conn.ark_info().await?;
Some(ark_info.server_pubkey)
}
Err(err) => {
bail!("Failed to connect to provided server (if you are sure use the --force flag): {:#}", err);
}
}
} else {
None
};
let wallet_fingerprint = WalletSeed::new(network, &mnemonic.to_seed("")).fingerprint();
let properties = WalletProperties {
network,
fingerprint: wallet_fingerprint,
server_pubkey,
};
db.init_wallet(&properties).await.context("cannot init wallet in the database")?;
info!("Created wallet with fingerprint: {}", wallet_fingerprint);
if let Some(pk) = server_pubkey {
info!("Stored server pubkey: {}", pk);
}
let wallet = Wallet::open(&mnemonic, db, config).await.context("failed to open wallet")?;
wallet.require_chainsource_version()?;
Ok(wallet)
}
pub async fn create_with_onchain(
mnemonic: &Mnemonic,
network: Network,
config: Config,
db: Arc<dyn BarkPersister>,
onchain: &dyn ExitUnilaterally,
force: bool,
) -> anyhow::Result<Wallet> {
let mut wallet = Wallet::create(mnemonic, network, config, db, force).await?;
wallet.exit.get_mut().load(onchain).await?;
Ok(wallet)
}
pub async fn open(
mnemonic: &Mnemonic,
db: Arc<dyn BarkPersister>,
config: Config,
) -> anyhow::Result<Wallet> {
let properties = db.read_properties().await?.context("Wallet is not initialised")?;
let seed = {
let seed = mnemonic.to_seed("");
WalletSeed::new(properties.network, &seed)
};
if properties.fingerprint != seed.fingerprint() {
bail!("incorrect mnemonic")
}
let chain_source = if let Some(ref url) = config.esplora_address {
ChainSourceSpec::Esplora {
url: url.clone(),
}
} else if let Some(ref url) = config.bitcoind_address {
let auth = if let Some(ref c) = config.bitcoind_cookiefile {
bitcoin_ext::rpc::Auth::CookieFile(c.clone())
} else {
bitcoin_ext::rpc::Auth::UserPass(
config.bitcoind_user.clone().context("need bitcoind auth config")?,
config.bitcoind_pass.clone().context("need bitcoind auth config")?,
)
};
ChainSourceSpec::Bitcoind { url: url.clone(), auth }
} else {
bail!("Need to either provide esplora or bitcoind info");
};
#[cfg(feature = "socks5-proxy")]
let chain_proxy = proxy_for_url(&config.socks5_proxy, chain_source.url())?;
let chain_source_client = ChainSource::new(
chain_source, properties.network, config.fallback_fee_rate,
#[cfg(feature = "socks5-proxy")] chain_proxy.as_deref(),
).await?;
let chain = Arc::new(chain_source_client);
let server = match Self::connect_to_server(&config, properties.network).await {
Ok(s) => Some(s),
Err(e) => {
warn!("Ark server handshake failed: {:#}", e);
None
}
};
let server = parking_lot::RwLock::new(server);
let notifications = NotificationDispatch::new();
let movements = Arc::new(MovementManager::new(db.clone(), notifications.clone()));
let exit = RwLock::new(Exit::new(db.clone(), chain.clone(), movements.clone()).await?);
Ok(Wallet {
config, db, seed, exit, movements, notifications, server, chain,
inflight_lightning_payments: Mutex::new(HashSet::new()),
round_state_lock_index: RoundStateLockIndex::new(),
})
}
pub async fn open_with_onchain(
mnemonic: &Mnemonic,
db: Arc<dyn BarkPersister>,
onchain: &dyn ExitUnilaterally,
cfg: Config,
) -> anyhow::Result<Wallet> {
let mut wallet = Wallet::open(mnemonic, db, cfg).await?;
wallet.exit.get_mut().load(onchain).await?;
Ok(wallet)
}
pub async fn open_with_daemon(
mnemonic: &Mnemonic,
db: Arc<dyn BarkPersister>,
cfg: Config,
onchain: Option<Arc<RwLock<dyn DaemonizableOnchainWallet>>>,
) -> anyhow::Result<(Arc<Wallet>, DaemonHandle)> {
let wallet = Arc::new(Wallet::open(mnemonic, db, cfg).await?);
if let Some(onchain) = onchain.as_ref() {
let mut onchain = onchain.write().await;
wallet.exit.write().await.load(&mut *onchain).await?;
}
let daemon = wallet.clone().run_daemon(onchain)?;
Ok((wallet, daemon))
}
pub fn config(&self) -> &Config {
&self.config
}
pub async fn properties(&self) -> anyhow::Result<WalletProperties> {
let properties = self.db.read_properties().await?.context("Wallet is not initialised")?;
Ok(properties)
}
pub fn fingerprint(&self) -> Fingerprint {
self.seed.fingerprint()
}
async fn connect_to_server(
config: &Config,
network: Network,
) -> anyhow::Result<ServerConnection> {
let mut builder = ServerConnection::builder()
.address(&config.server_address)
.network(network);
#[cfg(feature = "socks5-proxy")]
if let Some(proxy) = proxy_for_url(&config.socks5_proxy, &config.server_address)? {
builder = builder.proxy(&proxy)
}
if let Some(ref token) = config.server_access_token {
builder = builder.access_token(token);
}
builder.connect().await.context("Failed to connect to Ark server")
}
async fn require_server(&self) -> anyhow::Result<(ServerConnection, ArkInfo)> {
let conn = self.server.read().clone()
.context("You should be connected to Ark server to perform this action")?;
let ark_info = conn.ark_info().await?;
if let Some(stored_pubkey) = self.properties().await?.server_pubkey {
if stored_pubkey != ark_info.server_pubkey {
log_server_pubkey_changed_error(stored_pubkey, ark_info.server_pubkey);
bail!("Server public key has changed. You should exit all your VTXOs!");
}
} else {
self.db.set_server_pubkey(ark_info.server_pubkey).await?;
info!("Stored server pubkey for existing wallet: {}", ark_info.server_pubkey);
}
Ok((conn, ark_info))
}
pub async fn refresh_server(&self) -> anyhow::Result<()> {
let server = self.server.read().clone();
let properties = self.properties().await?;
let srv = if let Some(srv) = server {
srv.check_connection().await?;
let ark_info = srv.ark_info().await?;
ark_info.fees.validate().context("invalid fee schedule")?;
if let Some(stored_pubkey) = properties.server_pubkey {
if stored_pubkey != ark_info.server_pubkey {
log_server_pubkey_changed_error(stored_pubkey, ark_info.server_pubkey);
bail!("Server public key has changed. You should exit all your VTXOs!");
}
} else {
self.db.set_server_pubkey(ark_info.server_pubkey).await?;
info!("Stored server pubkey for existing wallet: {}", ark_info.server_pubkey);
}
srv
} else {
let conn = Self::connect_to_server(&self.config, properties.network).await?;
let ark_info = conn.ark_info().await?;
ark_info.fees.validate().context("invalid fee schedule")?;
if let Some(stored_pubkey) = properties.server_pubkey {
if stored_pubkey != ark_info.server_pubkey {
log_server_pubkey_changed_error(stored_pubkey, ark_info.server_pubkey);
bail!("Server public key has changed. You should exit all your VTXOs!");
}
} else {
self.db.set_server_pubkey(ark_info.server_pubkey).await?;
info!("Stored server pubkey for existing wallet: {}", ark_info.server_pubkey);
}
conn
};
let _ = self.server.write().insert(srv);
Ok(())
}
pub async fn ark_info(&self) -> anyhow::Result<Option<ArkInfo>> {
let server = self.server.read().clone();
match server.as_ref() {
Some(srv) => Ok(Some(srv.ark_info().await?)),
_ => Ok(None),
}
}
pub async fn balance(&self) -> anyhow::Result<Balance> {
let vtxos = self.vtxos().await?;
let spendable = {
let mut v = vtxos.iter().collect();
VtxoStateKind::Spendable.filter_vtxos(&mut v).await?;
v.into_iter().map(|v| v.amount()).sum::<Amount>()
};
let pending_lightning_send = self.pending_lightning_send_vtxos().await?.iter()
.map(|v| v.amount())
.sum::<Amount>();
let claimable_lightning_receive = self.claimable_lightning_receive_balance().await?;
let pending_board = self.pending_board_vtxos().await?.iter()
.map(|v| v.amount())
.sum::<Amount>();
let pending_in_round = self.pending_round_balance().await?;
let pending_exit = self.exit.try_read().ok().map(|e| e.pending_total());
Ok(Balance {
spendable,
pending_in_round,
pending_lightning_send,
claimable_lightning_receive,
pending_exit,
pending_board,
})
}
pub async fn validate_vtxo(&self, vtxo: &Vtxo<Full>) -> anyhow::Result<()> {
let tx = self.chain.get_tx(&vtxo.chain_anchor().txid).await
.context("could not fetch chain tx")?;
let tx = tx.with_context(|| {
format!("vtxo chain anchor not found for vtxo: {}", vtxo.chain_anchor().txid)
})?;
vtxo.validate(&tx)?;
Ok(())
}
pub async fn import_vtxo(&self, vtxo: &Vtxo<Full>) -> anyhow::Result<()> {
if self.db.get_wallet_vtxo(vtxo.id()).await?.is_some() {
info!("VTXO {} already exists in wallet, skipping import", vtxo.id());
return Ok(());
}
self.validate_vtxo(vtxo).await.context("VTXO validation failed")?;
if self.find_signable_clause(vtxo).await.is_none() {
bail!("VTXO {} is not owned by this wallet (no signable clause found)", vtxo.id());
}
let current_height = self.chain.tip().await?;
if vtxo.expiry_height() <= current_height {
bail!("Vtxo {} has expired", vtxo.id());
}
self.store_spendable_vtxos([vtxo]).await.context("failed to store imported VTXO")?;
info!("Successfully imported VTXO {}", vtxo.id());
Ok(())
}
pub async fn get_vtxo_by_id(&self, vtxo_id: VtxoId) -> anyhow::Result<WalletVtxo> {
let vtxo = self.db.get_wallet_vtxo(vtxo_id).await
.with_context(|| format!("Error when querying vtxo {} in database", vtxo_id))?
.with_context(|| format!("The VTXO with id {} cannot be found", vtxo_id))?;
Ok(vtxo)
}
#[deprecated(since="0.1.0-beta.5", note = "Use Wallet::history instead")]
pub async fn movements(&self) -> anyhow::Result<Vec<Movement>> {
self.history().await
}
pub async fn history(&self) -> anyhow::Result<Vec<Movement>> {
Ok(self.db.get_all_movements().await?)
}
pub async fn history_by_payment_method(
&self,
payment_method: &PaymentMethod,
) -> anyhow::Result<Vec<Movement>> {
let mut ret = self.db.get_movements_by_payment_method(payment_method).await?;
ret.sort_by_key(|m| m.id);
Ok(ret)
}
pub async fn all_vtxos(&self) -> anyhow::Result<Vec<WalletVtxo>> {
Ok(self.db.get_all_vtxos().await?)
}
pub async fn vtxos(&self) -> anyhow::Result<Vec<WalletVtxo>> {
Ok(self.db.get_vtxos_by_state(&VtxoStateKind::UNSPENT_STATES).await?)
}
pub async fn vtxos_with(&self, filter: &impl FilterVtxos) -> anyhow::Result<Vec<WalletVtxo>> {
let mut vtxos = self.vtxos().await?;
filter.filter_vtxos(&mut vtxos).await.context("error filtering vtxos")?;
Ok(vtxos)
}
pub async fn spendable_vtxos(&self) -> anyhow::Result<Vec<WalletVtxo>> {
Ok(self.vtxos_with(&VtxoStateKind::Spendable).await?)
}
pub async fn spendable_vtxos_with(
&self,
filter: &impl FilterVtxos,
) -> anyhow::Result<Vec<WalletVtxo>> {
let mut vtxos = self.spendable_vtxos().await?;
filter.filter_vtxos(&mut vtxos).await.context("error filtering vtxos")?;
Ok(vtxos)
}
pub async fn get_expiring_vtxos(
&self,
threshold: BlockHeight,
) -> anyhow::Result<Vec<WalletVtxo>> {
let expiry = self.chain.tip().await? + threshold;
let filter = VtxoFilter::new(&self).expires_before(expiry);
Ok(self.spendable_vtxos_with(&filter).await?)
}
pub async fn sync_pending_offboards(&self) -> anyhow::Result<()> {
let pending_offboards: Vec<PendingOffboard> = self.db.get_pending_offboards().await?;
if pending_offboards.is_empty() {
return Ok(());
}
let current_height = self.chain.tip().await?;
let required_confs = self.config.offboard_required_confirmations;
trace!("Checking {} pending offboard transaction(s)", pending_offboards.len());
for pending in pending_offboards {
let status = self.chain.tx_status(pending.offboard_txid).await;
match status {
Ok(TxStatus::Confirmed(block_ref)) => {
let confs = current_height - (block_ref.height - 1);
if confs < required_confs as BlockHeight {
trace!(
"Offboard tx {} has {}/{} confirmations, waiting...",
pending.offboard_txid, confs, required_confs,
);
continue;
}
info!(
"Offboard tx {} confirmed, finalizing movement {}",
pending.offboard_txid, pending.movement_id,
);
for vtxo_id in &pending.vtxo_ids {
if let Err(e) = self.db.update_vtxo_state_checked(
*vtxo_id,
VtxoState::Spent,
&[VtxoStateKind::Locked],
).await {
warn!("Failed to mark vtxo {} as spent: {:#}", vtxo_id, e);
}
}
if let Err(e) = self.movements.finish_movement(
pending.movement_id,
MovementStatus::Successful,
).await {
warn!("Failed to finish movement {}: {:#}", pending.movement_id, e);
}
self.db.remove_pending_offboard(pending.movement_id).await?;
}
Ok(TxStatus::Mempool) => {
if required_confs == 0 {
info!(
"Offboard tx {} in mempool with 0 required confirmations, \
finalizing movement {}",
pending.offboard_txid, pending.movement_id,
);
for vtxo_id in &pending.vtxo_ids {
if let Err(e) = self.db.update_vtxo_state_checked(
*vtxo_id,
VtxoState::Spent,
&[VtxoStateKind::Locked],
).await {
warn!("Failed to mark vtxo {} as spent: {:#}", vtxo_id, e);
}
}
if let Err(e) = self.movements.finish_movement(
pending.movement_id,
MovementStatus::Successful,
).await {
warn!("Failed to finish movement {}: {:#}", pending.movement_id, e);
}
self.db.remove_pending_offboard(pending.movement_id).await?;
} else {
trace!(
"Offboard tx {} still in mempool, waiting...",
pending.offboard_txid,
);
}
}
Ok(TxStatus::NotFound) => {
let age = chrono::Local::now() - pending.created_at;
if age < chrono::Duration::hours(1) {
trace!(
"Offboard tx {} not found, but only {} minutes old — waiting...",
pending.offboard_txid, age.num_minutes(),
);
continue;
}
warn!(
"Offboard tx {} not found after {} minutes, canceling movement {}",
pending.offboard_txid, age.num_minutes(), pending.movement_id,
);
for vtxo_id in &pending.vtxo_ids {
if let Err(e) = self.db.update_vtxo_state_checked(
*vtxo_id,
VtxoState::Spendable,
&[VtxoStateKind::Locked],
).await {
warn!("Failed to restore vtxo {} to spendable: {:#}", vtxo_id, e);
}
}
if let Err(e) = self.movements.finish_movement(
pending.movement_id,
MovementStatus::Failed,
).await {
warn!("Failed to fail movement {}: {:#}", pending.movement_id, e);
}
self.db.remove_pending_offboard(pending.movement_id).await?;
}
Err(e) => {
warn!(
"Failed to check status of offboard tx {}: {:#}",
pending.offboard_txid, e,
);
}
}
}
Ok(())
}
pub async fn maintenance(&self) -> anyhow::Result<()> {
info!("Starting wallet maintenance in interactive mode");
self.sync().await;
let rounds = self.progress_pending_rounds(None).await;
if let Err(e) = rounds.as_ref() {
warn!("Error progressing pending rounds: {:#}", e);
}
let refresh = self.maintenance_refresh().await;
if let Err(e) = refresh.as_ref() {
warn!("Error refreshing VTXOs: {:#}", e);
}
if rounds.is_err() || refresh.is_err() {
bail!("Maintenance encountered errors.\nprogress_rounds: {:#?}\nrefresh: {:#?}", rounds, refresh);
}
Ok(())
}
pub async fn maintenance_delegated(&self) -> anyhow::Result<()> {
info!("Starting wallet maintenance in delegated mode");
self.sync().await;
let rounds = self.progress_pending_rounds(None).await;
if let Err(e) = rounds.as_ref() {
warn!("Error progressing pending rounds: {:#}", e);
}
let refresh = self.maybe_schedule_maintenance_refresh_delegated().await;
if let Err(e) = refresh.as_ref() {
warn!("Error refreshing VTXOs: {:#}", e);
}
if rounds.is_err() || refresh.is_err() {
bail!("Delegated maintenance encountered errors.\nprogress_rounds: {:#?}\nrefresh: {:#?}", rounds, refresh);
}
Ok(())
}
pub async fn maintenance_with_onchain<W: PreparePsbt + SignPsbt + ExitUnilaterally>(
&self,
onchain: &mut W,
) -> anyhow::Result<()> {
info!("Starting wallet maintenance in interactive mode with onchain wallet");
let maintenance = self.maintenance().await;
let exit_sync = self.sync_exits(onchain).await;
if let Err(e) = exit_sync.as_ref() {
warn!("Error syncing exits: {:#}", e);
}
let exit_progress = self.exit.write().await.progress_exits(&self, onchain, None).await;
if let Err(e) = exit_progress.as_ref() {
warn!("Error progressing exits: {:#}", e);
}
if maintenance.is_err() || exit_sync.is_err() || exit_progress.is_err() {
bail!("Maintenance encountered errors.\nmaintenance: {:#?}\nexit_sync: {:#?}\nexit_progress: {:#?}", maintenance, exit_sync, exit_progress);
}
Ok(())
}
pub async fn maintenance_with_onchain_delegated<W: PreparePsbt + SignPsbt + ExitUnilaterally>(
&self,
onchain: &mut W,
) -> anyhow::Result<()> {
info!("Starting wallet maintenance in delegated mode with onchain wallet");
let maintenance = self.maintenance_delegated().await;
let exit_sync = self.sync_exits(onchain).await;
if let Err(e) = exit_sync.as_ref() {
warn!("Error syncing exits: {:#}", e);
}
let exit_progress = self.exit.write().await.progress_exits(&self, onchain, None).await;
if let Err(e) = exit_progress.as_ref() {
warn!("Error progressing exits: {:#}", e);
}
if maintenance.is_err() || exit_sync.is_err() || exit_progress.is_err() {
bail!("Delegated maintenance encountered errors.\nmaintenance: {:#?}\nexit_sync: {:#?}\nexit_progress: {:#?}", maintenance, exit_sync, exit_progress);
}
Ok(())
}
pub async fn maybe_schedule_maintenance_refresh(&self) -> anyhow::Result<Option<RoundStateId>> {
let vtxos = self.get_vtxos_to_refresh().await?;
if vtxos.len() == 0 {
return Ok(None);
}
let participation = match self.build_refresh_participation(vtxos).await? {
Some(participation) => participation,
None => return Ok(None),
};
info!("Scheduling maintenance refresh ({} vtxos)", participation.inputs.len());
let state = self.join_next_round(participation, Some(RoundMovement::Refresh)).await?;
Ok(Some(state.id()))
}
pub async fn maybe_schedule_maintenance_refresh_delegated(
&self,
) -> anyhow::Result<Option<RoundStateId>> {
let vtxos = self.get_vtxos_to_refresh().await?;
if vtxos.len() == 0 {
return Ok(None);
}
let participation = match self.build_refresh_participation(vtxos).await? {
Some(participation) => participation,
None => return Ok(None),
};
info!("Scheduling delegated maintenance refresh ({} vtxos)", participation.inputs.len());
let state = self.join_next_round_delegated(participation, Some(RoundMovement::Refresh)).await?;
Ok(Some(state.id()))
}
pub async fn maintenance_refresh(&self) -> anyhow::Result<Option<RoundStatus>> {
let vtxos = self.get_vtxos_to_refresh().await?;
if vtxos.len() == 0 {
return Ok(None);
}
info!("Performing maintenance refresh");
self.refresh_vtxos(vtxos).await
}
pub async fn sync(&self) {
futures::join!(
async {
if let Err(e) = self.chain.update_fee_rates(self.config.fallback_fee_rate).await {
warn!("Error updating fee rates: {:#}", e);
}
},
async {
if let Err(e) = self.sync_mailbox().await {
warn!("Error in mailbox sync: {:#}", e);
}
},
async {
if let Err(e) = self.sync_pending_rounds().await {
warn!("Error while trying to progress rounds awaiting confirmations: {:#}", e);
}
},
async {
if let Err(e) = self.sync_pending_lightning_send_vtxos().await {
warn!("Error syncing pending lightning payments: {:#}", e);
}
},
async {
if let Err(e) = self.try_claim_all_lightning_receives(false).await {
warn!("Error claiming pending lightning receives: {:#}", e);
}
},
async {
if let Err(e) = self.sync_pending_boards().await {
warn!("Error syncing pending boards: {:#}", e);
}
},
async {
if let Err(e) = self.sync_pending_offboards().await {
warn!("Error syncing pending offboards: {:#}", e);
}
}
);
}
pub async fn sync_exits(
&self,
onchain: &mut dyn ExitUnilaterally,
) -> anyhow::Result<()> {
self.exit.write().await.sync(&self, onchain).await?;
Ok(())
}
pub async fn dangerous_drop_vtxo(&self, vtxo_id: VtxoId) -> anyhow::Result<()> {
warn!("Drop vtxo {} from the database", vtxo_id);
self.db.remove_vtxo(vtxo_id).await?;
Ok(())
}
pub async fn dangerous_drop_all_vtxos(&self) -> anyhow::Result<()> {
warn!("Dropping all vtxos from the db...");
for vtxo in self.vtxos().await? {
self.db.remove_vtxo(vtxo.id()).await?;
}
self.exit.write().await.dangerous_clear_exit().await?;
Ok(())
}
async fn has_counterparty_risk(&self, vtxo: &Vtxo<Full>) -> anyhow::Result<bool> {
for past_pks in vtxo.past_arkoor_pubkeys() {
let mut owns_any = false;
for past_pk in past_pks {
if self.db.get_public_key_idx(&past_pk).await?.is_some() {
owns_any = true;
break;
}
}
if !owns_any {
return Ok(true);
}
}
let my_clause = self.find_signable_clause(vtxo).await;
Ok(!my_clause.is_some())
}
async fn add_should_refresh_vtxos(
&self,
participation: &mut RoundParticipation,
) -> anyhow::Result<()> {
let tip = self.chain.tip().await?;
let mut vtxos_to_refresh = self.spendable_vtxos_with(
&RefreshStrategy::should_refresh(self, tip, self.chain.fee_rates().await.fast),
).await?;
if vtxos_to_refresh.is_empty() {
return Ok(());
}
let excluded_ids = participation.inputs.iter().map(|v| v.vtxo_id())
.collect::<HashSet<_>>();
let mut total_amount = Amount::ZERO;
for i in (0..vtxos_to_refresh.len()).rev() {
let vtxo = &vtxos_to_refresh[i];
if excluded_ids.contains(&vtxo.id()) {
vtxos_to_refresh.swap_remove(i);
continue;
}
total_amount += vtxo.amount();
}
if vtxos_to_refresh.is_empty() {
return Ok(());
}
let (_, ark_info) = self.require_server().await?;
let fee = ark_info.fees.refresh.calculate_no_base_fee(
vtxos_to_refresh.iter().map(|wv| VtxoFeeInfo::from_vtxo_and_tip(&wv.vtxo, tip)),
).context("fee overflowed")?;
let output_amount = match validate_and_subtract_fee_min_dust(total_amount, fee) {
Ok(amount) => amount,
Err(e) => {
trace!("Cannot add should-refresh VTXOs: {}", e);
return Ok(());
},
};
info!(
"Adding {} extra VTXOs to round participation total = {}, fee = {}, output = {}",
vtxos_to_refresh.len(), total_amount, fee, output_amount,
);
let (user_keypair, _) = self.derive_store_next_keypair().await?;
let req = VtxoRequest {
policy: VtxoPolicy::new_pubkey(user_keypair.public_key()),
amount: output_amount,
};
participation.inputs.reserve(vtxos_to_refresh.len());
participation.inputs.extend(vtxos_to_refresh.into_iter().map(|wv| wv.vtxo));
participation.outputs.push(req);
Ok(())
}
pub async fn build_refresh_participation<V: VtxoRef>(
&self,
vtxos: impl IntoIterator<Item = V>,
) -> anyhow::Result<Option<RoundParticipation>> {
let (vtxos, total_amount) = {
let iter = vtxos.into_iter();
let size_hint = iter.size_hint();
let mut vtxos = Vec::<Vtxo<Full>>::with_capacity(size_hint.1.unwrap_or(size_hint.0));
let mut amount = Amount::ZERO;
for vref in iter {
let id = vref.vtxo_id();
if vtxos.iter().any(|v| v.id() == id) {
bail!("duplicate VTXO id: {}", id);
}
let vtxo = if let Some(vtxo) = vref.into_full_vtxo() {
vtxo
} else {
self.get_vtxo_by_id(id).await
.with_context(|| format!("vtxo with id {} not found", id))?.vtxo
};
amount += vtxo.amount();
vtxos.push(vtxo);
}
(vtxos, amount)
};
if vtxos.is_empty() {
info!("Skipping refresh since no VTXOs are provided.");
return Ok(None);
}
ensure!(total_amount >= P2TR_DUST,
"vtxo amount must be at least {} to participate in a round",
P2TR_DUST,
);
let (_, ark_info) = self.require_server().await?;
let current_height = self.chain.tip().await?;
let vtxo_fee_infos = vtxos.iter()
.map(|v| VtxoFeeInfo::from_vtxo_and_tip(v, current_height));
let fee = ark_info.fees.refresh.calculate(vtxo_fee_infos).context("fee overflowed")?;
let output_amount = validate_and_subtract_fee_min_dust(total_amount, fee)?;
info!("Refreshing {} VTXOs (total amount = {}, fee = {}, output = {}).",
vtxos.len(), total_amount, fee, output_amount,
);
let (user_keypair, _) = self.derive_store_next_keypair().await?;
let req = VtxoRequest {
policy: VtxoPolicy::Pubkey(PubkeyVtxoPolicy { user_pubkey: user_keypair.public_key() }),
amount: output_amount,
};
Ok(Some(RoundParticipation {
inputs: vtxos,
outputs: vec![req],
unblinded_mailbox_id: None,
}))
}
pub async fn refresh_vtxos<V: VtxoRef>(
&self,
vtxos: impl IntoIterator<Item = V>,
) -> anyhow::Result<Option<RoundStatus>> {
let mut participation = match self.build_refresh_participation(vtxos).await? {
Some(participation) => participation,
None => return Ok(None),
};
if let Err(e) = self.add_should_refresh_vtxos(&mut participation).await {
warn!("Error trying to add additional VTXOs that should be refreshed: {:#}", e);
}
Ok(Some(self.participate_round(participation, Some(RoundMovement::Refresh)).await?))
}
pub async fn refresh_vtxos_delegated<V: VtxoRef>(
&self,
vtxos: impl IntoIterator<Item = V>,
) -> anyhow::Result<Option<StoredRoundState<Unlocked>>> {
let mut part = match self.build_refresh_participation(vtxos).await? {
Some(participation) => participation,
None => return Ok(None),
};
if let Err(e) = self.add_should_refresh_vtxos(&mut part).await {
warn!("Error trying to add additional VTXOs that should be refreshed: {:#}", e);
}
Ok(Some(self.join_next_round_delegated(part, Some(RoundMovement::Refresh)).await?))
}
pub async fn get_vtxos_to_refresh(&self) -> anyhow::Result<Vec<WalletVtxo>> {
let vtxos = self.spendable_vtxos_with(&RefreshStrategy::should_refresh_if_must(
self,
self.chain.tip().await?,
self.chain.fee_rates().await.fast,
)).await?;
Ok(vtxos)
}
pub async fn get_first_expiring_vtxo_blockheight(
&self,
) -> anyhow::Result<Option<BlockHeight>> {
Ok(self.spendable_vtxos().await?.iter().map(|v| v.expiry_height()).min())
}
pub async fn get_next_required_refresh_blockheight(
&self,
) -> anyhow::Result<Option<BlockHeight>> {
let first_expiry = self.get_first_expiring_vtxo_blockheight().await?;
Ok(first_expiry.map(|h| h.saturating_sub(self.config.vtxo_refresh_expiry_threshold)))
}
async fn select_vtxos_to_cover(
&self,
amount: Amount,
) -> anyhow::Result<Vec<WalletVtxo>> {
let mut vtxos = self.spendable_vtxos().await?;
vtxos.sort_by_key(|v| v.expiry_height());
let mut result = Vec::new();
let mut total_amount = Amount::ZERO;
for input in vtxos {
total_amount += input.amount();
result.push(input);
if total_amount >= amount {
return Ok(result)
}
}
bail!("Insufficient money available. Needed {} but {} is available",
amount, total_amount,
);
}
async fn select_vtxos_to_cover_with_fee<F>(
&self,
amount: Amount,
calc_fee: F,
) -> anyhow::Result<(Vec<WalletVtxo>, Amount)>
where
F: for<'a> Fn(
Amount, std::iter::Copied<std::slice::Iter<'a, VtxoFeeInfo>>,
) -> anyhow::Result<Amount>,
{
let tip = self.chain.tip().await?;
const MAX_ITERATIONS: usize = 100;
let mut fee = Amount::ZERO;
let mut fee_info = Vec::new();
for _ in 0..MAX_ITERATIONS {
let required = amount.checked_add(fee)
.context("Amount + fee overflow")?;
let vtxos = self.select_vtxos_to_cover(required).await
.context("Could not find enough suitable VTXOs to cover payment + fees")?;
fee_info.reserve(vtxos.len());
let mut vtxo_amount = Amount::ZERO;
for vtxo in &vtxos {
vtxo_amount += vtxo.amount();
fee_info.push(VtxoFeeInfo::from_vtxo_and_tip(vtxo, tip));
}
fee = calc_fee(amount, fee_info.iter().copied())?;
if amount + fee <= vtxo_amount {
trace!("Selected vtxos to cover amount + fee: amount = {}, fee = {}, total inputs = {}",
amount, fee, vtxo_amount,
);
return Ok((vtxos, fee));
}
trace!("VTXO sum of {} did not exceed amount {} and fee {}, iterating again",
vtxo_amount, amount, fee,
);
fee_info.clear();
}
bail!("Fee calculation did not converge after maximum iterations")
}
pub fn run_daemon(
self: &Arc<Self>,
onchain: Option<Arc<RwLock<dyn DaemonizableOnchainWallet>>>,
) -> anyhow::Result<DaemonHandle> {
Ok(crate::daemon::start_daemon(self.clone(), onchain))
}
pub async fn register_vtxos_with_server(
&self,
vtxos: &[impl AsRef<Vtxo<Full>>],
) -> anyhow::Result<()> {
if vtxos.is_empty() {
return Ok(());
}
let (mut srv, _) = self.require_server().await?;
srv.client.register_vtxos(protos::RegisterVtxosRequest {
vtxos: vtxos.iter().map(|v| v.as_ref().serialize()).collect(),
}).await.context("failed to register vtxos")?;
Ok(())
}
}