X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fchannelmanager.rs;h=3dcc74f905269a7f8a58978a529f96b6e48a5851;hb=615ef7d6f88464ed8478a8c8cc0e7582ba66d773;hp=ee99012c5846922d7265f2d7336d9d7fec129ed3;hpb=2dd8b3e896374006c898243537b78bb1ecc4fb6d;p=rust-lightning diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index ee99012c..31529343 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -16,8 +16,9 @@ //! It does not manage routing logic (see routing::router::get_route for that) nor does it manage constructing //! on-chain transactions (it only monitors the chain to watch for any force-closes that might //! imply it needs to fail HTLCs/payments/channels it manages). +//! -use bitcoin::blockdata::block::BlockHeader; +use bitcoin::blockdata::block::{Block, BlockHeader}; use bitcoin::blockdata::transaction::Transaction; use bitcoin::blockdata::constants::genesis_block; use bitcoin::network::constants::Network; @@ -27,24 +28,30 @@ use bitcoin::hashes::hmac::{Hmac, HmacEngine}; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::hashes::sha256d::Hash as Sha256dHash; use bitcoin::hashes::cmp::fixed_time_eq; -use bitcoin::hash_types::BlockHash; +use bitcoin::hash_types::{BlockHash, Txid}; use bitcoin::secp256k1::key::{SecretKey,PublicKey}; use bitcoin::secp256k1::Secp256k1; use bitcoin::secp256k1::ecdh::SharedSecret; use bitcoin::secp256k1; -use chain::chaininterface::{BroadcasterInterface,ChainListener,FeeEstimator}; -use chain::transaction::OutPoint; +use chain; +use chain::Confirm; +use chain::Watch; +use chain::chaininterface::{BroadcasterInterface, FeeEstimator}; +use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, ChannelMonitorUpdateErr, HTLC_FAIL_BACK_BUFFER, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY, MonitorEvent, CLOSED_CHANNEL_UPDATE_ID}; +use chain::transaction::{OutPoint, TransactionData}; +// Since this struct is returned in `list_channels` methods, expose it here in case users want to +// construct one themselves. +pub use ln::channel::CounterpartyForwardingInfo; use ln::channel::{Channel, ChannelError}; -use ln::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, ManyChannelMonitor, HTLC_FAIL_BACK_BUFFER, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY, MonitorEvent}; use ln::features::{InitFeatures, NodeFeatures}; use routing::router::{Route, RouteHop}; use ln::msgs; use ln::msgs::NetAddress; use ln::onion_utils; use ln::msgs::{ChannelMessageHandler, DecodeError, LightningError, OptionalField}; -use chain::keysinterface::{ChannelKeys, KeysInterface, KeysManager, InMemoryChannelKeys}; +use chain::keysinterface::{Sign, KeysInterface, KeysManager, InMemorySigner}; use util::config::UserConfig; use util::events::{Event, EventsProvider, MessageSendEvent, MessageSendEventsProvider}; use util::{byte_utils, events}; @@ -56,9 +63,11 @@ use util::errors::APIError; use std::{cmp, mem}; use std::collections::{HashMap, hash_map, HashSet}; use std::io::{Cursor, Read}; -use std::sync::{Arc, Mutex, MutexGuard, RwLock}; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; +#[cfg(any(test, feature = "allow_wallclock_use"))] +use std::time::Instant; use std::marker::{Sync, Send}; use std::ops::Deref; use bitcoin::hashes::hex::ToHex; @@ -87,7 +96,7 @@ enum PendingHTLCRouting { short_channel_id: u64, // This should be NonZero eventually when we bump MSRV }, Receive { - payment_data: Option, + payment_data: msgs::FinalOnionHopData, incoming_cltv_expiry: u32, // Used to track when we should expire pending HTLCs that go unclaimed }, } @@ -116,9 +125,15 @@ pub(super) enum PendingHTLCStatus { pub(super) enum HTLCForwardInfo { AddHTLC { + forward_info: PendingHTLCInfo, + + // These fields are produced in `forward_htlcs()` and consumed in + // `process_pending_htlc_forwards()` for constructing the + // `HTLCSource::PreviousHopData` for failed and forwarded + // HTLCs. prev_short_channel_id: u64, prev_htlc_id: u64, - forward_info: PendingHTLCInfo, + prev_funding_outpoint: OutPoint, }, FailHTLC { htlc_id: u64, @@ -128,26 +143,29 @@ pub(super) enum HTLCForwardInfo { /// Tracks the inbound corresponding to an outbound HTLC #[derive(Clone, PartialEq)] -pub(super) struct HTLCPreviousHopData { +pub(crate) struct HTLCPreviousHopData { short_channel_id: u64, htlc_id: u64, incoming_packet_shared_secret: [u8; 32], + + // This field is consumed by `claim_funds_from_hop()` when updating a force-closed backwards + // channel with a preimage provided by the forward channel. + outpoint: OutPoint, } struct ClaimableHTLC { prev_hop: HTLCPreviousHopData, value: u64, - /// Filled in when the HTLC was received with a payment_secret packet, which contains a - /// total_msat (which may differ from value if this is a Multi-Path Payment) and a + /// Contains a total_msat (which may differ from value if this is a Multi-Path Payment) and a /// payment_secret which prevents path-probing attacks and can associate different HTLCs which /// are part of the same payment. - payment_data: Option, + payment_data: msgs::FinalOnionHopData, cltv_expiry: u32, } /// Tracks the inbound corresponding to an outbound HTLC #[derive(Clone, PartialEq)] -pub(super) enum HTLCSource { +pub(crate) enum HTLCSource { PreviousHopData(HTLCPreviousHopData), OutboundRoute { path: Vec, @@ -180,16 +198,19 @@ pub(super) enum HTLCFailReason { } /// payment_hash type, use to cross-lock hop +/// (C-not exported) as we just use [u8; 32] directly #[derive(Hash, Copy, Clone, PartialEq, Eq, Debug)] pub struct PaymentHash(pub [u8;32]); /// payment_preimage type, use to route payment between hop +/// (C-not exported) as we just use [u8; 32] directly #[derive(Hash, Copy, Clone, PartialEq, Eq, Debug)] pub struct PaymentPreimage(pub [u8;32]); /// payment_secret type, use to authenticate sender to the receiver and tie MPP HTLCs together +/// (C-not exported) as we just use [u8; 32] directly #[derive(Hash, Copy, Clone, PartialEq, Eq, Debug)] pub struct PaymentSecret(pub [u8;32]); -type ShutdownResult = (Option, ChannelMonitorUpdate, Vec<(HTLCSource, PaymentHash)>); +type ShutdownResult = (Option<(OutPoint, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>); /// Error type returned across the channel_state mutex boundary. When an Err is generated for a /// Channel, we generally end up with a ChannelError::Close for which we have to close the channel @@ -297,33 +318,56 @@ pub(super) enum RAACommitmentOrder { } // Note this is only exposed in cfg(test): -pub(super) struct ChannelHolder { - pub(super) by_id: HashMap<[u8; 32], Channel>, +pub(super) struct ChannelHolder { + pub(super) by_id: HashMap<[u8; 32], Channel>, pub(super) short_to_id: HashMap, /// short channel id -> forward infos. Key of 0 means payments received /// Note that while this is held in the same mutex as the channels themselves, no consistency /// guarantees are made about the existence of a channel with the short id here, nor the short /// ids in the PendingHTLCInfo! pub(super) forward_htlcs: HashMap>, - /// (payment_hash, payment_secret) -> Vec for tracking HTLCs that - /// were to us and can be failed/claimed by the user + /// Map from payment hash to any HTLCs which are to us and can be failed/claimed by the user. /// Note that while this is held in the same mutex as the channels themselves, no consistency /// guarantees are made about the channels given here actually existing anymore by the time you /// go to read them! - claimable_htlcs: HashMap<(PaymentHash, Option), Vec>, + claimable_htlcs: HashMap>, /// Messages to send to peers - pushed to in the same lock that they are generated in (except /// for broadcast messages, where ordering isn't as strict). pub(super) pending_msg_events: Vec, } +/// Events which we process internally but cannot be procsesed immediately at the generation site +/// for some reason. They are handled in timer_tick_occurred, so may be processed with +/// quite some time lag. +enum BackgroundEvent { + /// Handle a ChannelMonitorUpdate that closes a channel, broadcasting its current latest holder + /// commitment transaction. + ClosingMonitorUpdate((OutPoint, ChannelMonitorUpdate)), +} + /// State we hold per-peer. In the future we should put channels in here, but for now we only hold /// the latest Init features we heard from the peer. struct PeerState { latest_features: InitFeatures, } -#[cfg(not(any(target_pointer_width = "32", target_pointer_width = "64")))] -const ERR: () = "You need at least 32 bit pointers (well, usize, but we'll assume they're the same) for ChannelManager::latest_block_height"; +/// Stores a PaymentSecret and any other data we may need to validate an inbound payment is +/// actually ours and not some duplicate HTLC sent to us by a node along the route. +/// +/// For users who don't want to bother doing their own payment preimage storage, we also store that +/// here. +struct PendingInboundPayment { + /// The payment secret that the sender must use for us to accept this payment + payment_secret: PaymentSecret, + /// Time at which this HTLC expires - blocks with a header time above this value will result in + /// this payment being removed. + expiry_time: u64, + /// Arbitrary identifier the user specifies (or not) + user_payment_id: u64, + // Other required attributes of the payment, optionally enforced: + payment_preimage: Option, + min_value_msat: Option, +} /// SimpleArcChannelManager is useful when you need a ChannelManager with a static lifetime, e.g. /// when you're using lightning-net-tokio (since tokio::spawn requires parameters with static @@ -332,7 +376,7 @@ const ERR: () = "You need at least 32 bit pointers (well, usize, but we'll assum /// issues such as overly long function definitions. Note that the ChannelManager can take any /// type that implements KeysInterface for its keys manager, but this type alias chooses the /// concrete type of the KeysManager. -pub type SimpleArcChannelManager = Arc, Arc, Arc, Arc, Arc>>; +pub type SimpleArcChannelManager = ChannelManager, Arc, Arc, Arc, Arc>; /// SimpleRefChannelManager is a type alias for a ChannelManager reference, and is the reference /// counterpart to the SimpleArcChannelManager type alias. Use this type by default when you don't @@ -342,7 +386,7 @@ pub type SimpleArcChannelManager = Arc = ChannelManager; +pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L> = ChannelManager; /// Manager which keeps track of a number of channels and sends messages to the appropriate /// channel, also tracking HTLC preimages and forwarding onion packets appropriately. @@ -357,13 +401,13 @@ pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L> = ChannelManage /// /// Note that you can be a bit lazier about writing out ChannelManager than you can be with /// ChannelMonitors. With ChannelMonitors you MUST write each monitor update out to disk before -/// returning from ManyChannelMonitor::add_/update_monitor, with ChannelManagers, writing updates +/// returning from chain::Watch::watch_/update_channel, with ChannelManagers, writing updates /// happens out-of-band (and will prevent any other ChannelManager operations from occurring during /// the serialization process). If the deserialized version is out-of-date compared to the /// ChannelMonitors passed by reference to read(), those channels will be force-closed based on the /// ChannelMonitor state and no funds will be lost (mod on-chain transaction fees). /// -/// Note that the deserializer is only implemented for (Sha256dHash, ChannelManager), which +/// Note that the deserializer is only implemented for (BlockHash, ChannelManager), which /// tells you the last block hash which was block_connect()ed. You MUST rescan any blocks along /// the "reorg path" (ie call block_disconnected() until you get to a common block and then call /// block_connected() to step towards your best block) upon deserialization before using the @@ -373,43 +417,56 @@ pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L> = ChannelManage /// ChannelUpdate messages informing peers that the channel is temporarily disabled. To avoid /// spam due to quick disconnection/reconnection, updates are not sent until the channel has been /// offline for a full minute. In order to track this, you must call -/// timer_chan_freshness_every_min roughly once per minute, though it doesn't have to be perfect. +/// timer_tick_occurred roughly once per minute, though it doesn't have to be perfect. /// /// Rather than using a plain ChannelManager, it is preferable to use either a SimpleArcChannelManager /// a SimpleRefChannelManager, for conciseness. See their documentation for more details, but /// essentially you should default to using a SimpleRefChannelManager, and use a /// SimpleArcChannelManager when you require a ChannelManager with a static lifetime, such as when /// you're using lightning-net-tokio. -pub struct ChannelManager - where M::Target: ManyChannelMonitor, +pub struct ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { default_configuration: UserConfig, genesis_hash: BlockHash, fee_estimator: F, - monitor: M, + chain_monitor: M, tx_broadcaster: T, #[cfg(test)] - pub(super) latest_block_height: AtomicUsize, + pub(super) best_block: RwLock, #[cfg(not(test))] - latest_block_height: AtomicUsize, - last_block_hash: Mutex, + best_block: RwLock, secp_ctx: Secp256k1, - #[cfg(test)] - pub(super) channel_state: Mutex>, - #[cfg(not(test))] - channel_state: Mutex>, + #[cfg(any(test, feature = "_test_utils"))] + pub(super) channel_state: Mutex>, + #[cfg(not(any(test, feature = "_test_utils")))] + channel_state: Mutex>, + + /// Storage for PaymentSecrets and any requirements on future inbound payments before we will + /// expose them to users via a PaymentReceived event. HTLCs which do not meet the requirements + /// here are failed when we process them as pending-forwardable-HTLCs, and entries are removed + /// after we generate a PaymentReceived upon receipt of all MPP parts or when they time out. + /// Locked *after* channel_state. + pending_inbound_payments: Mutex>, + our_network_key: SecretKey, + our_network_pubkey: PublicKey, /// Used to track the last value sent in a node_announcement "timestamp" field. We ensure this /// value increases strictly since we don't assume access to a time source. last_node_announcement_serial: AtomicUsize, + /// The highest block timestamp we've seen, which is usually a good guess at the current time. + /// Assuming most miners are generating blocks with reasonable timestamps, this shouldn't be + /// very far in the past, and can only ever be up to two hours in the future. + highest_seen_timestamp: AtomicUsize, + /// The bulk of our storage will eventually be here (channels and message queues and the like). /// If we are connected to a peer we always at least have an entry here, even if no channels /// are currently open with that peer. @@ -419,30 +476,123 @@ pub struct ChannelManager>>, pending_events: Mutex>, + pending_background_events: Mutex>, /// Used when we have to take a BIG lock to make sure everything is self-consistent. /// Essentially just when we're serializing ourselves out. /// Taken first everywhere where we are making changes before any other locks. + /// When acquiring this lock in read mode, rather than acquiring it directly, call + /// `PersistenceNotifierGuard::new(..)` and pass the lock to it, to ensure the PersistenceNotifier + /// the lock contains sends out a notification when the lock is released. total_consistency_lock: RwLock<()>, + persistence_notifier: PersistenceNotifier, + keys_manager: K, logger: L, } -/// The amount of time we require our counterparty wait to claim their money (ie time between when -/// we, or our watchtower, must check for them having broadcast a theft transaction). -pub(crate) const BREAKDOWN_TIMEOUT: u16 = 6 * 24; -/// The amount of time we're willing to wait to claim money back to us -pub(crate) const MAX_LOCAL_BREAKDOWN_TIMEOUT: u16 = 6 * 24 * 7; +/// Chain-related parameters used to construct a new `ChannelManager`. +/// +/// Typically, the block-specific parameters are derived from the best block hash for the network, +/// as a newly constructed `ChannelManager` will not have created any channels yet. These parameters +/// are not needed when deserializing a previously constructed `ChannelManager`. +pub struct ChainParameters { + /// The network for determining the `chain_hash` in Lightning messages. + pub network: Network, + + /// The hash and height of the latest block successfully connected. + /// + /// Used to track on-chain channel funding outputs and send payments with reliable timelocks. + pub best_block: BestBlock, +} + +/// The best known block as identified by its hash and height. +#[derive(Clone, Copy)] +pub struct BestBlock { + block_hash: BlockHash, + height: u32, +} + +impl BestBlock { + /// Returns the best block from the genesis of the given network. + pub fn from_genesis(network: Network) -> Self { + BestBlock { + block_hash: genesis_block(network).header.block_hash(), + height: 0, + } + } + + /// Returns the best block as identified by the given block hash and height. + pub fn new(block_hash: BlockHash, height: u32) -> Self { + BestBlock { block_hash, height } + } + + /// Returns the best block hash. + pub fn block_hash(&self) -> BlockHash { self.block_hash } + + /// Returns the best block height. + pub fn height(&self) -> u32 { self.height } +} + +/// Whenever we release the `ChannelManager`'s `total_consistency_lock`, from read mode, it is +/// desirable to notify any listeners on `await_persistable_update_timeout`/ +/// `await_persistable_update` that new updates are available for persistence. Therefore, this +/// struct is responsible for locking the total consistency lock and, upon going out of scope, +/// sending the aforementioned notification (since the lock being released indicates that the +/// updates are ready for persistence). +struct PersistenceNotifierGuard<'a> { + persistence_notifier: &'a PersistenceNotifier, + // We hold onto this result so the lock doesn't get released immediately. + _read_guard: RwLockReadGuard<'a, ()>, +} + +impl<'a> PersistenceNotifierGuard<'a> { + fn new(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier) -> Self { + let read_guard = lock.read().unwrap(); + + Self { + persistence_notifier: notifier, + _read_guard: read_guard, + } + } +} + +impl<'a> Drop for PersistenceNotifierGuard<'a> { + fn drop(&mut self) { + self.persistence_notifier.notify(); + } +} + +/// The amount of time in blocks we require our counterparty wait to claim their money (ie time +/// between when we, or our watchtower, must check for them having broadcast a theft transaction). +/// +/// This can be increased (but not decreased) through [`ChannelHandshakeConfig::our_to_self_delay`] +/// +/// [`ChannelHandshakeConfig::our_to_self_delay`]: crate::util::config::ChannelHandshakeConfig::our_to_self_delay +pub const BREAKDOWN_TIMEOUT: u16 = 6 * 24; +/// The amount of time in blocks we're willing to wait to claim money back to us. This matches +/// the maximum required amount in lnd as of March 2021. +pub(crate) const MAX_LOCAL_BREAKDOWN_TIMEOUT: u16 = 2 * 6 * 24 * 7; /// The minimum number of blocks between an inbound HTLC's CLTV and the corresponding outbound -/// HTLC's CLTV. This should always be a few blocks greater than channelmonitor::CLTV_CLAIM_BUFFER, -/// ie the node we forwarded the payment on to should always have enough room to reliably time out -/// the HTLC via a full update_fail_htlc/commitment_signed dance before we hit the -/// CLTV_CLAIM_BUFFER point (we static assert that it's at least 3 blocks more). -const CLTV_EXPIRY_DELTA: u16 = 6 * 12; //TODO? +/// HTLC's CLTV. The current default represents roughly six hours of blocks at six blocks/hour. +/// +/// This can be increased (but not decreased) through [`ChannelConfig::cltv_expiry_delta`] +/// +/// [`ChannelConfig::cltv_expiry_delta`]: crate::util::config::ChannelConfig::cltv_expiry_delta +// This should always be a few blocks greater than channelmonitor::CLTV_CLAIM_BUFFER, +// i.e. the node we forwarded the payment on to should always have enough room to reliably time out +// the HTLC via a full update_fail_htlc/commitment_signed dance before we hit the +// CLTV_CLAIM_BUFFER point (we static assert that it's at least 3 blocks more). +pub const MIN_CLTV_EXPIRY_DELTA: u16 = 6 * 6; pub(super) const CLTV_FAR_FAR_AWAY: u32 = 6 * 24 * 7; //TODO? +/// Minimum CLTV difference between the current block height and received inbound payments. +/// Invoices generated for payment to us must set their `min_final_cltv_expiry` field to at least +/// this value. +pub const MIN_FINAL_CLTV_EXPIRY: u32 = HTLC_FAIL_BACK_BUFFER; + // Check that our CLTV_EXPIRY is at least CLTV_CLAIM_BUFFER + ANTI_REORG_DELAY + LATENCY_GRACE_PERIOD_BLOCKS, // ie that if the next-hop peer fails the HTLC within // LATENCY_GRACE_PERIOD_BLOCKS then we'll still have CLTV_CLAIM_BUFFER left to timeout it onchain, @@ -451,15 +601,16 @@ pub(super) const CLTV_FAR_FAR_AWAY: u32 = 6 * 24 * 7; //TODO? // LATENCY_GRACE_PERIOD_BLOCKS. #[deny(const_err)] #[allow(dead_code)] -const CHECK_CLTV_EXPIRY_SANITY: u32 = CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - CLTV_CLAIM_BUFFER - ANTI_REORG_DELAY - LATENCY_GRACE_PERIOD_BLOCKS; +const CHECK_CLTV_EXPIRY_SANITY: u32 = MIN_CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - CLTV_CLAIM_BUFFER - ANTI_REORG_DELAY - LATENCY_GRACE_PERIOD_BLOCKS; // Check for ability of an attacker to make us fail on-chain by delaying inbound claim. See // ChannelMontior::would_broadcast_at_height for a description of why this is needed. #[deny(const_err)] #[allow(dead_code)] -const CHECK_CLTV_EXPIRY_SANITY_2: u32 = CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - 2*CLTV_CLAIM_BUFFER; +const CHECK_CLTV_EXPIRY_SANITY_2: u32 = MIN_CLTV_EXPIRY_DELTA as u32 - LATENCY_GRACE_PERIOD_BLOCKS - 2*CLTV_CLAIM_BUFFER; /// Details of a channel, as returned by ChannelManager::list_channels and ChannelManager::list_usable_channels +#[derive(Clone)] pub struct ChannelDetails { /// The channel's ID (prior to funding transaction generation, this is a random 32 bytes, /// thereafter this is the txid of the funding transaction xor the funding transaction output). @@ -493,12 +644,16 @@ pub struct ChannelDetails { /// True if the channel is (a) confirmed and funding_locked messages have been exchanged, (b) /// the peer is connected, and (c) no monitor update failure is pending resolution. pub is_live: bool, + + /// Information on the fees and requirements that the counterparty requires when forwarding + /// payments to us through this channel. + pub counterparty_forwarding_info: Option, } /// If a payment fails to send, it can be in one of several states. This enum is returned as the /// Err() type describing which state the payment is in, see the description of individual enum /// states for more. -#[derive(Debug)] +#[derive(Clone, Debug)] pub enum PaymentSendFailure { /// A parameter which was passed to send_payment was invalid, preventing us from attempting to /// send the payment at all. No channel state has been changed or messages sent to peers, and @@ -532,7 +687,7 @@ pub enum PaymentSendFailure { } macro_rules! handle_error { - ($self: ident, $internal: expr, $their_node_id: expr) => { + ($self: ident, $internal: expr, $counterparty_node_id: expr) => { match $internal { Ok(msg) => Ok(msg), Err(MsgHandleErrInternal { err, shutdown_finish }) => { @@ -558,7 +713,7 @@ macro_rules! handle_error { if let msgs::ErrorAction::IgnoreError = err.action { } else { msg_events.push(events::MessageSendEvent::HandleError { - node_id: $their_node_id, + node_id: $counterparty_node_id, action: err.action.clone() }); } @@ -693,10 +848,10 @@ macro_rules! maybe_break_monitor_err { } } -impl ChannelManager - where M::Target: ManyChannelMonitor, +impl ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { @@ -709,28 +864,21 @@ impl /// /// panics if channel_value_satoshis is >= `MAX_FUNDING_SATOSHIS`! /// - /// Users must provide the current blockchain height from which to track onchain channel - /// funding outpoints and send payments with reliable timelocks. - /// /// Users need to notify the new ChannelManager when a new block is connected or - /// disconnected using its `block_connected` and `block_disconnected` methods. - /// However, rather than calling these methods directly, the user should register - /// the ChannelManager as a listener to the BlockNotifier and call the BlockNotifier's - /// `block_(dis)connected` methods, which will notify all registered listeners in one - /// go. - pub fn new(network: Network, fee_est: F, monitor: M, tx_broadcaster: T, logger: L, keys_manager: K, config: UserConfig, current_blockchain_height: usize) -> Self { - let secp_ctx = Secp256k1::new(); + /// disconnected using its `block_connected` and `block_disconnected` methods, starting + /// from after `params.latest_hash`. + pub fn new(fee_est: F, chain_monitor: M, tx_broadcaster: T, logger: L, keys_manager: K, config: UserConfig, params: ChainParameters) -> Self { + let mut secp_ctx = Secp256k1::new(); + secp_ctx.seeded_randomize(&keys_manager.get_secure_random_bytes()); ChannelManager { default_configuration: config.clone(), - genesis_hash: genesis_block(network).header.block_hash(), + genesis_hash: genesis_block(params.network).header.block_hash(), fee_estimator: fee_est, - monitor, + chain_monitor, tx_broadcaster, - latest_block_height: AtomicUsize::new(current_blockchain_height), - last_block_hash: Mutex::new(Default::default()), - secp_ctx, + best_block: RwLock::new(params.best_block), channel_state: Mutex::new(ChannelHolder{ by_id: HashMap::new(), @@ -739,14 +887,21 @@ impl claimable_htlcs: HashMap::new(), pending_msg_events: Vec::new(), }), + pending_inbound_payments: Mutex::new(HashMap::new()), + our_network_key: keys_manager.get_node_secret(), + our_network_pubkey: PublicKey::from_secret_key(&secp_ctx, &keys_manager.get_node_secret()), + secp_ctx, last_node_announcement_serial: AtomicUsize::new(0), + highest_seen_timestamp: AtomicUsize::new(0), per_peer_state: RwLock::new(HashMap::new()), pending_events: Mutex::new(Vec::new()), + pending_background_events: Mutex::new(Vec::new()), total_consistency_lock: RwLock::new(()), + persistence_notifier: PersistenceNotifier::new(), keys_manager, @@ -754,12 +909,18 @@ impl } } + /// Gets the current configuration applied to all new channels, as + pub fn get_current_default_configuration(&self) -> &UserConfig { + &self.default_configuration + } + /// Creates a new outbound channel to the given remote node and with the given value. /// - /// user_id will be provided back as user_channel_id in FundingGenerationReady and - /// FundingBroadcastSafe events to allow tracking of which events correspond with which - /// create_channel call. Note that user_channel_id defaults to 0 for inbound channels, so you - /// may wish to avoid using 0 for user_id here. + /// user_id will be provided back as user_channel_id in FundingGenerationReady events to allow + /// tracking of which events correspond with which create_channel call. Note that the + /// user_channel_id defaults to 0 for inbound channels, so you may wish to avoid using 0 for + /// user_id here. user_id has no meaning inside of LDK, it is simply copied to events and + /// otherwise ignored. /// /// If successful, will generate a SendOpenChannel message event, so you should probably poll /// PeerManager::process_events afterwards. @@ -775,7 +936,10 @@ impl let channel = Channel::new_outbound(&self.fee_estimator, &self.keys_manager, their_network_key, channel_value_satoshis, push_msat, user_id, config)?; let res = channel.get_open_channel(self.genesis_hash.clone()); - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + // We want to make sure the lock is actually acquired by PersistenceNotifierGuard. + debug_assert!(&self.total_consistency_lock.try_write().is_err()); + let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.entry(channel.channel_id()) { hash_map::Entry::Occupied(_) => { @@ -794,7 +958,7 @@ impl Ok(()) } - fn list_channels_with_filter)) -> bool>(&self, f: Fn) -> Vec { + fn list_channels_with_filter)) -> bool>(&self, f: Fn) -> Vec { let mut res = Vec::new(); { let channel_state = self.channel_state.lock().unwrap(); @@ -804,13 +968,14 @@ impl res.push(ChannelDetails { channel_id: (*channel_id).clone(), short_channel_id: channel.get_short_channel_id(), - remote_network_id: channel.get_their_node_id(), + remote_network_id: channel.get_counterparty_node_id(), counterparty_features: InitFeatures::empty(), channel_value_satoshis: channel.get_value_satoshis(), inbound_capacity_msat, outbound_capacity_msat, user_id: channel.get_user_id(), is_live: channel.is_live(), + counterparty_forwarding_info: channel.counterparty_forwarding_info(), }); } } @@ -847,7 +1012,7 @@ impl /// /// May generate a SendShutdown message event on success, which should be relayed. pub fn close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> { - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let (mut failed_htlcs, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); @@ -856,7 +1021,7 @@ impl hash_map::Entry::Occupied(mut chan_entry) => { let (shutdown_msg, failed_htlcs) = chan_entry.get_mut().get_shutdown()?; channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown { - node_id: chan_entry.get().get_their_node_id(), + node_id: chan_entry.get().get_counterparty_node_id(), msg: shutdown_msg }); if chan_entry.get().is_shutdown() { @@ -890,35 +1055,36 @@ impl #[inline] fn finish_force_close_channel(&self, shutdown_res: ShutdownResult) { - let (funding_txo_option, monitor_update, mut failed_htlcs) = shutdown_res; + let (monitor_update_option, mut failed_htlcs) = shutdown_res; log_trace!(self.logger, "Finishing force-closure of channel {} HTLCs to fail", failed_htlcs.len()); for htlc_source in failed_htlcs.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source.0, &htlc_source.1, HTLCFailReason::Reason { failure_code: 0x4000 | 8, data: Vec::new() }); } - if let Some(funding_txo) = funding_txo_option { + if let Some((funding_txo, monitor_update)) = monitor_update_option { // There isn't anything we can do if we get an update failure - we're already // force-closing. The monitor update on the required in-memory copy should broadcast // the latest local state, which is the best we can do anyway. Thus, it is safe to // ignore the result here. - let _ = self.monitor.update_monitor(funding_txo, monitor_update); + let _ = self.chain_monitor.update_channel(funding_txo, monitor_update); } } - /// Force closes a channel, immediately broadcasting the latest local commitment transaction to - /// the chain and rejecting new HTLCs on the given channel. - pub fn force_close_channel(&self, channel_id: &[u8; 32]) { - let _ = self.total_consistency_lock.read().unwrap(); - + fn force_close_channel_with_peer(&self, channel_id: &[u8; 32], peer_node_id: Option<&PublicKey>) -> Result { let mut chan = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; - if let Some(chan) = channel_state.by_id.remove(channel_id) { - if let Some(short_id) = chan.get_short_channel_id() { + if let hash_map::Entry::Occupied(chan) = channel_state.by_id.entry(channel_id.clone()) { + if let Some(node_id) = peer_node_id { + if chan.get().get_counterparty_node_id() != *node_id { + return Err(APIError::ChannelUnavailable{err: "No such channel".to_owned()}); + } + } + if let Some(short_id) = chan.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); } - chan + chan.remove_entry().1 } else { - return; + return Err(APIError::ChannelUnavailable{err: "No such channel".to_owned()}); } }; log_trace!(self.logger, "Force-closing channel {}", log_bytes!(channel_id[..])); @@ -929,17 +1095,39 @@ impl msg: update }); } + + Ok(chan.get_counterparty_node_id()) + } + + /// Force closes a channel, immediately broadcasting the latest local commitment transaction to + /// the chain and rejecting new HTLCs on the given channel. Fails if channel_id is unknown to the manager. + pub fn force_close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + match self.force_close_channel_with_peer(channel_id, None) { + Ok(counterparty_node_id) => { + self.channel_state.lock().unwrap().pending_msg_events.push( + events::MessageSendEvent::HandleError { + node_id: counterparty_node_id, + action: msgs::ErrorAction::SendErrorMessage { + msg: msgs::ErrorMessage { channel_id: *channel_id, data: "Channel force-closed".to_owned() } + }, + } + ); + Ok(()) + }, + Err(e) => Err(e) + } } /// Force close all channels, immediately broadcasting the latest local commitment transaction /// for each to the chain and rejecting new HTLCs on each. pub fn force_close_all_channels(&self) { for chan in self.list_channels() { - self.force_close_channel(&chan.channel_id); + let _ = self.force_close_channel(&chan.channel_id); } } - fn decode_update_add_htlc_onion(&self, msg: &msgs::UpdateAddHTLC) -> (PendingHTLCStatus, MutexGuard>) { + fn decode_update_add_htlc_onion(&self, msg: &msgs::UpdateAddHTLC) -> (PendingHTLCStatus, MutexGuard>) { macro_rules! return_malformed_err { ($msg: expr, $err_code: expr) => { { @@ -1046,7 +1234,7 @@ impl // HTLC_FAIL_BACK_BUFFER blocks to go. // Also, ensure that, in the case of an unknown payment hash, our payment logic has enough time to fail the HTLC backward // before our onchain logic triggers a channel closure (see HTLC_FAIL_BACK_BUFFER rational). - if (msg.cltv_expiry as u64) <= self.latest_block_height.load(Ordering::Acquire) as u64 + HTLC_FAIL_BACK_BUFFER as u64 + 1 { + if (msg.cltv_expiry as u64) <= self.best_block.read().unwrap().height() as u64 + HTLC_FAIL_BACK_BUFFER as u64 + 1 { return_err!("The final CLTV expiry is too soon to handle", 17, &[0;0]); } // final_incorrect_htlc_amount @@ -1064,6 +1252,10 @@ impl msgs::OnionHopDataFormat::FinalNode { payment_data } => payment_data, }; + if payment_data.is_none() { + return_err!("We require payment_secrets", 0x4000|0x2000|3, &[0;0]); + } + // Note that we could obviously respond immediately with an update_fulfill_htlc // message, however that would leak that we are the recipient of this payment, so // instead we stay symmetric with the forwarding case, only responding (after a @@ -1071,7 +1263,7 @@ impl PendingHTLCStatus::Forward(PendingHTLCInfo { routing: PendingHTLCRouting::Receive { - payment_data, + payment_data: payment_data.unwrap(), incoming_cltv_expiry: msg.cltv_expiry, }, payment_hash: msg.payment_hash.clone(), @@ -1126,7 +1318,7 @@ impl PendingHTLCStatus::Forward(PendingHTLCInfo { routing: PendingHTLCRouting::Forward { onion_packet: outgoing_packet, - short_channel_id: short_channel_id, + short_channel_id, }, payment_hash: msg.payment_hash.clone(), incoming_shared_secret: shared_secret, @@ -1159,17 +1351,17 @@ impl if !chan.is_live() { // channel_disabled break Some(("Forwarding channel is not in a ready state.", 0x1000 | 20, Some(self.get_channel_update(chan).unwrap()))); } - if *amt_to_forward < chan.get_their_htlc_minimum_msat() { // amount_below_minimum + if *amt_to_forward < chan.get_counterparty_htlc_minimum_msat() { // amount_below_minimum break Some(("HTLC amount was below the htlc_minimum_msat", 0x1000 | 11, Some(self.get_channel_update(chan).unwrap()))); } - let fee = amt_to_forward.checked_mul(chan.get_fee_proportional_millionths() as u64).and_then(|prop_fee| { (prop_fee / 1000000).checked_add(chan.get_our_fee_base_msat(&self.fee_estimator) as u64) }); + let fee = amt_to_forward.checked_mul(chan.get_fee_proportional_millionths() as u64).and_then(|prop_fee| { (prop_fee / 1000000).checked_add(chan.get_holder_fee_base_msat(&self.fee_estimator) as u64) }); if fee.is_none() || msg.amount_msat < fee.unwrap() || (msg.amount_msat - fee.unwrap()) < *amt_to_forward { // fee_insufficient break Some(("Prior hop has deviated from specified fees parameters or origin node has obsolete ones", 0x1000 | 12, Some(self.get_channel_update(chan).unwrap()))); } - if (msg.cltv_expiry as u64) < (*outgoing_cltv_value) as u64 + CLTV_EXPIRY_DELTA as u64 { // incorrect_cltv_expiry + if (msg.cltv_expiry as u64) < (*outgoing_cltv_value) as u64 + chan.get_cltv_expiry_delta() as u64 { // incorrect_cltv_expiry break Some(("Forwarding node has tampered with the intended HTLC values or origin node has an obsolete cltv_expiry_delta", 0x1000 | 13, Some(self.get_channel_update(chan).unwrap()))); } - let cur_height = self.latest_block_height.load(Ordering::Acquire) as u32 + 1; + let cur_height = self.best_block.read().unwrap().height() + 1; // Theoretically, channel counterparty shouldn't send us a HTLC expiring now, but we want to be robust wrt to counterparty // packet sanitization (see HTLC_FAIL_BACK_BUFFER rational) if msg.cltv_expiry <= cur_height + HTLC_FAIL_BACK_BUFFER as u32 { // expiry_too_soon @@ -1211,23 +1403,23 @@ impl /// only fails if the channel does not yet have an assigned short_id /// May be called with channel_state already locked! - fn get_channel_update(&self, chan: &Channel) -> Result { + fn get_channel_update(&self, chan: &Channel) -> Result { let short_channel_id = match chan.get_short_channel_id() { None => return Err(LightningError{err: "Channel not yet established".to_owned(), action: msgs::ErrorAction::IgnoreError}), Some(id) => id, }; - let were_node_one = PublicKey::from_secret_key(&self.secp_ctx, &self.our_network_key).serialize()[..] < chan.get_their_node_id().serialize()[..]; + let were_node_one = PublicKey::from_secret_key(&self.secp_ctx, &self.our_network_key).serialize()[..] < chan.get_counterparty_node_id().serialize()[..]; let unsigned = msgs::UnsignedChannelUpdate { chain_hash: self.genesis_hash, - short_channel_id: short_channel_id, + short_channel_id, timestamp: chan.get_update_time_counter(), flags: (!were_node_one) as u8 | ((!chan.is_live() as u8) << 1), - cltv_expiry_delta: CLTV_EXPIRY_DELTA, - htlc_minimum_msat: chan.get_our_htlc_minimum_msat(), + cltv_expiry_delta: chan.get_cltv_expiry_delta(), + htlc_minimum_msat: chan.get_counterparty_htlc_minimum_msat(), htlc_maximum_msat: OptionalField::Present(chan.get_announced_htlc_max_msat()), - fee_base_msat: chan.get_our_fee_base_msat(&self.fee_estimator), + fee_base_msat: chan.get_holder_fee_base_msat(&self.fee_estimator), fee_proportional_millionths: chan.get_fee_proportional_millionths(), excess_data: Vec::new(), }; @@ -1255,7 +1447,7 @@ impl } let onion_packet = onion_utils::construct_onion_packet(onion_payloads, onion_keys, prng_seed, payment_hash); - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let err: Result<(), _> = loop { let mut channel_lock = self.channel_state.lock().unwrap(); @@ -1267,7 +1459,7 @@ impl let channel_state = &mut *channel_lock; if let hash_map::Entry::Occupied(mut chan) = channel_state.by_id.entry(id) { match { - if chan.get().get_their_node_id() != path.first().unwrap().pubkey { + if chan.get().get_counterparty_node_id() != path.first().unwrap().pubkey { return Err(APIError::RouteError{err: "Node ID mismatch on first hop!"}); } if !chan.get().is_live() { @@ -1280,7 +1472,7 @@ impl }, onion_packet, &self.logger), channel_state, chan) } { Some((update_add, commitment_signed, monitor_update)) => { - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { maybe_break_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true); // Note that MonitorUpdateFailed here indicates (per function docs) // that we will resend the commitment update once monitor updating @@ -1386,7 +1578,7 @@ impl return Err(PaymentSendFailure::PathParameterError(path_errs)); } - let cur_height = self.latest_block_height.load(Ordering::Acquire) as u32 + 1; + let cur_height = self.best_block.read().unwrap().height() + 1; let mut results = Vec::new(); for path in route.paths.iter() { results.push(self.send_payment_along_path(&path, &payment_hash, payment_secret, total_value, cur_height)); @@ -1413,41 +1605,37 @@ impl } } - /// Call this upon creation of a funding transaction for the given channel. - /// - /// Note that ALL inputs in the transaction pointed to by funding_txo MUST spend SegWit outputs - /// or your counterparty can steal your funds! - /// - /// Panics if a funding transaction has already been provided for this channel. - /// - /// May panic if the funding_txo is duplicative with some other channel (note that this should - /// be trivially prevented by using unique funding transaction keys per-channel). - pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_txo: OutPoint) { - let _ = self.total_consistency_lock.read().unwrap(); - + /// Handles the generation of a funding transaction, optionally (for tests) with a function + /// which checks the correctness of the funding transaction given the associated channel. + fn funding_transaction_generated_intern, &Transaction) -> Result> + (&self, temporary_channel_id: &[u8; 32], funding_transaction: Transaction, find_funding_output: FundingOutput) -> Result<(), APIError> { let (chan, msg) = { let (res, chan) = match self.channel_state.lock().unwrap().by_id.remove(temporary_channel_id) { Some(mut chan) => { - (chan.get_outbound_funding_created(funding_txo, &self.logger) + let funding_txo = find_funding_output(&chan, &funding_transaction)?; + + (chan.get_outbound_funding_created(funding_transaction, funding_txo, &self.logger) .map_err(|e| if let ChannelError::Close(msg) = e { MsgHandleErrInternal::from_finish_shutdown(msg, chan.channel_id(), chan.force_shutdown(true), None) } else { unreachable!(); }) , chan) }, - None => return + None => { return Err(APIError::ChannelUnavailable { err: "No such channel".to_owned() }) }, }; - match handle_error!(self, res, chan.get_their_node_id()) { + match handle_error!(self, res, chan.get_counterparty_node_id()) { Ok(funding_msg) => { (chan, funding_msg) }, - Err(_) => { return; } + Err(_) => { return Err(APIError::ChannelUnavailable { + err: "Error deriving keys or signing initial commitment transactions - either our RNG or our counterparty's RNG is broken or the Signer refused to sign".to_owned() + }) }, } }; let mut channel_state = self.channel_state.lock().unwrap(); channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingCreated { - node_id: chan.get_their_node_id(), - msg: msg, + node_id: chan.get_counterparty_node_id(), + msg, }); match channel_state.by_id.entry(chan.channel_id()) { hash_map::Entry::Occupied(_) => { @@ -1457,9 +1645,72 @@ impl e.insert(chan); } } + Ok(()) + } + + #[cfg(test)] + pub(crate) fn funding_transaction_generated_unchecked(&self, temporary_channel_id: &[u8; 32], funding_transaction: Transaction, output_index: u16) -> Result<(), APIError> { + self.funding_transaction_generated_intern(temporary_channel_id, funding_transaction, |_, tx| { + Ok(OutPoint { txid: tx.txid(), index: output_index }) + }) } - fn get_announcement_sigs(&self, chan: &Channel) -> Option { + /// Call this upon creation of a funding transaction for the given channel. + /// + /// Returns an [`APIError::APIMisuseError`] if the funding_transaction spent non-SegWit outputs + /// or if no output was found which matches the parameters in [`Event::FundingGenerationReady`]. + /// + /// Panics if a funding transaction has already been provided for this channel. + /// + /// May panic if the output found in the funding transaction is duplicative with some other + /// channel (note that this should be trivially prevented by using unique funding transaction + /// keys per-channel). + /// + /// Do NOT broadcast the funding transaction yourself. When we have safely received our + /// counterparty's signature the funding transaction will automatically be broadcast via the + /// [`BroadcasterInterface`] provided when this `ChannelManager` was constructed. + /// + /// Note that this includes RBF or similar transaction replacement strategies - lightning does + /// not currently support replacing a funding transaction on an existing channel. Instead, + /// create a new channel with a conflicting funding transaction. + pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_transaction: Transaction) -> Result<(), APIError> { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + + for inp in funding_transaction.input.iter() { + if inp.witness.is_empty() { + return Err(APIError::APIMisuseError { + err: "Funding transaction must be fully signed and spend Segwit outputs".to_owned() + }); + } + } + self.funding_transaction_generated_intern(temporary_channel_id, funding_transaction, |chan, tx| { + let mut output_index = None; + let expected_spk = chan.get_funding_redeemscript().to_v0_p2wsh(); + for (idx, outp) in tx.output.iter().enumerate() { + if outp.script_pubkey == expected_spk && outp.value == chan.get_value_satoshis() { + if output_index.is_some() { + return Err(APIError::APIMisuseError { + err: "Multiple outputs matched the expected script and value".to_owned() + }); + } + if idx > u16::max_value() as usize { + return Err(APIError::APIMisuseError { + err: "Transaction had more than 2^16 outputs, which is not supported".to_owned() + }); + } + output_index = Some(idx as u16); + } + } + if output_index.is_none() { + return Err(APIError::APIMisuseError { + err: "No output matched the script_pubkey and value in the FundingGenerationReady event".to_owned() + }); + } + Ok(OutPoint { txid: tx.txid(), index: output_index.unwrap() }) + }) + } + + fn get_announcement_sigs(&self, chan: &Channel) -> Option { if !chan.should_announce() { log_trace!(self.logger, "Can't send announcement_signatures for private channel {}", log_bytes!(chan.channel_id())); return None @@ -1506,7 +1757,7 @@ impl /// /// Panics if addresses is absurdly large (more than 500). pub fn broadcast_node_announcement(&self, rgb: [u8; 3], alias: [u8; 32], addresses: Vec) { - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); if addresses.len() > 500 { panic!("More than half the message size was taken up by public addresses!"); @@ -1536,7 +1787,7 @@ impl /// Should only really ever be called in response to a PendingHTLCsForwardable event. /// Will likely generate further events. pub fn process_pending_htlc_forwards(&self) { - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let mut new_events = Vec::new(); let mut failed_forwards = Vec::new(); @@ -1553,9 +1804,11 @@ impl failed_forwards.reserve(pending_forwards.len()); for forward_info in pending_forwards.drain(..) { match forward_info { - HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info } => { + HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info, + prev_funding_outpoint } => { let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id: prev_short_channel_id, + outpoint: prev_funding_outpoint, htlc_id: prev_htlc_id, incoming_packet_shared_secret: forward_info.incoming_shared_secret, }); @@ -1582,10 +1835,12 @@ impl HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info: PendingHTLCInfo { routing: PendingHTLCRouting::Forward { onion_packet, .. - }, incoming_shared_secret, payment_hash, amt_to_forward, outgoing_cltv_value }, } => { + }, incoming_shared_secret, payment_hash, amt_to_forward, outgoing_cltv_value }, + prev_funding_outpoint } => { log_trace!(self.logger, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", log_bytes!(payment_hash.0), prev_short_channel_id, short_chan_id); let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id: prev_short_channel_id, + outpoint: prev_funding_outpoint, htlc_id: prev_htlc_id, incoming_packet_shared_secret: incoming_shared_secret, }); @@ -1657,7 +1912,7 @@ impl Err(e) => { // We surely failed send_commitment due to bad keys, in that case // close channel and then send error message to peer. - let their_node_id = chan.get().get_their_node_id(); + let counterparty_node_id = chan.get().get_counterparty_node_id(); let err: Result<(), _> = match e { ChannelError::Ignore(_) => { panic!("Stated return value requirements in send_commitment() were not met"); @@ -1672,16 +1927,16 @@ impl }, ChannelError::CloseDelayBroadcast(_) => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); } }; - handle_errors.push((their_node_id, err)); + handle_errors.push((counterparty_node_id, err)); continue; } }; - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { - handle_errors.push((chan.get().get_their_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true))); + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { + handle_errors.push((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true))); continue; } channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: chan.get().get_their_node_id(), + node_id: chan.get().get_counterparty_node_id(), updates: msgs::CommitmentUpdate { update_add_htlcs: add_htlc_msgs, update_fulfill_htlcs: Vec::new(), @@ -1700,63 +1955,97 @@ impl match forward_info { HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info: PendingHTLCInfo { routing: PendingHTLCRouting::Receive { payment_data, incoming_cltv_expiry }, - incoming_shared_secret, payment_hash, amt_to_forward, .. }, } => { - let prev_hop = HTLCPreviousHopData { - short_channel_id: prev_short_channel_id, - htlc_id: prev_htlc_id, - incoming_packet_shared_secret: incoming_shared_secret, - }; - - let mut total_value = 0; - let payment_secret_opt = - if let &Some(ref data) = &payment_data { Some(data.payment_secret.clone()) } else { None }; - let htlcs = channel_state.claimable_htlcs.entry((payment_hash, payment_secret_opt)) - .or_insert(Vec::new()); - htlcs.push(ClaimableHTLC { - prev_hop, + incoming_shared_secret, payment_hash, amt_to_forward, .. }, + prev_funding_outpoint } => { + let claimable_htlc = ClaimableHTLC { + prev_hop: HTLCPreviousHopData { + short_channel_id: prev_short_channel_id, + outpoint: prev_funding_outpoint, + htlc_id: prev_htlc_id, + incoming_packet_shared_secret: incoming_shared_secret, + }, value: amt_to_forward, payment_data: payment_data.clone(), cltv_expiry: incoming_cltv_expiry, - }); - if let &Some(ref data) = &payment_data { - for htlc in htlcs.iter() { - total_value += htlc.value; - if htlc.payment_data.as_ref().unwrap().total_msat != data.total_msat { - total_value = msgs::MAX_VALUE_MSAT; - } - if total_value >= msgs::MAX_VALUE_MSAT { break; } - } - if total_value >= msgs::MAX_VALUE_MSAT || total_value > data.total_msat { - for htlc in htlcs.iter() { - let mut htlc_msat_height_data = byte_utils::be64_to_array(htlc.value).to_vec(); - htlc_msat_height_data.extend_from_slice( - &byte_utils::be32_to_array( - self.latest_block_height.load(Ordering::Acquire) - as u32, - ), - ); - failed_forwards.push((HTLCSource::PreviousHopData(HTLCPreviousHopData { - short_channel_id: htlc.prev_hop.short_channel_id, - htlc_id: htlc.prev_hop.htlc_id, - incoming_packet_shared_secret: htlc.prev_hop.incoming_packet_shared_secret, - }), payment_hash, - HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: htlc_msat_height_data } - )); - } - } else if total_value == data.total_msat { - new_events.push(events::Event::PaymentReceived { - payment_hash: payment_hash, - payment_secret: Some(data.payment_secret), - amt: total_value, - }); + }; + + macro_rules! fail_htlc { + ($htlc: expr) => { + let mut htlc_msat_height_data = byte_utils::be64_to_array($htlc.value).to_vec(); + htlc_msat_height_data.extend_from_slice( + &byte_utils::be32_to_array(self.best_block.read().unwrap().height()), + ); + failed_forwards.push((HTLCSource::PreviousHopData(HTLCPreviousHopData { + short_channel_id: $htlc.prev_hop.short_channel_id, + outpoint: prev_funding_outpoint, + htlc_id: $htlc.prev_hop.htlc_id, + incoming_packet_shared_secret: $htlc.prev_hop.incoming_packet_shared_secret, + }), payment_hash, + HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: htlc_msat_height_data } + )); } - } else { - new_events.push(events::Event::PaymentReceived { - payment_hash: payment_hash, - payment_secret: None, - amt: amt_to_forward, - }); } + + // Check that the payment hash and secret are known. Note that we + // MUST take care to handle the "unknown payment hash" and + // "incorrect payment secret" cases here identically or we'd expose + // that we are the ultimate recipient of the given payment hash. + // Further, we must not expose whether we have any other HTLCs + // associated with the same payment_hash pending or not. + let mut payment_secrets = self.pending_inbound_payments.lock().unwrap(); + match payment_secrets.entry(payment_hash) { + hash_map::Entry::Vacant(_) => { + log_trace!(self.logger, "Failing new HTLC with payment_hash {} as we didn't have a corresponding inbound payment.", log_bytes!(payment_hash.0)); + fail_htlc!(claimable_htlc); + }, + hash_map::Entry::Occupied(inbound_payment) => { + if inbound_payment.get().payment_secret != payment_data.payment_secret { + log_trace!(self.logger, "Failing new HTLC with payment_hash {} as it didn't match our expected payment secret.", log_bytes!(payment_hash.0)); + fail_htlc!(claimable_htlc); + } else if inbound_payment.get().min_value_msat.is_some() && payment_data.total_msat < inbound_payment.get().min_value_msat.unwrap() { + log_trace!(self.logger, "Failing new HTLC with payment_hash {} as it didn't match our minimum value (had {}, needed {}).", + log_bytes!(payment_hash.0), payment_data.total_msat, inbound_payment.get().min_value_msat.unwrap()); + fail_htlc!(claimable_htlc); + } else { + let mut total_value = 0; + let htlcs = channel_state.claimable_htlcs.entry(payment_hash) + .or_insert(Vec::new()); + htlcs.push(claimable_htlc); + for htlc in htlcs.iter() { + total_value += htlc.value; + if htlc.payment_data.total_msat != payment_data.total_msat { + log_trace!(self.logger, "Failing HTLCs with payment_hash {} as the HTLCs had inconsistent total values (eg {} and {})", + log_bytes!(payment_hash.0), payment_data.total_msat, htlc.payment_data.total_msat); + total_value = msgs::MAX_VALUE_MSAT; + } + if total_value >= msgs::MAX_VALUE_MSAT { break; } + } + if total_value >= msgs::MAX_VALUE_MSAT || total_value > payment_data.total_msat { + log_trace!(self.logger, "Failing HTLCs with payment_hash {} as the total value {} ran over expected value {} (or HTLCs were inconsistent)", + log_bytes!(payment_hash.0), total_value, payment_data.total_msat); + for htlc in htlcs.iter() { + fail_htlc!(htlc); + } + } else if total_value == payment_data.total_msat { + new_events.push(events::Event::PaymentReceived { + payment_hash, + payment_preimage: inbound_payment.get().payment_preimage, + payment_secret: payment_data.payment_secret, + amt: total_value, + user_payment_id: inbound_payment.get().user_payment_id, + }); + // Only ever generate at most one PaymentReceived + // per registered payment_hash, even if it isn't + // claimed. + inbound_payment.remove_entry(); + } else { + // Nothing to do - we haven't reached the total + // payment value yet, wait until we receive more + // MPP parts. + } + } + }, + }; }, HTLCForwardInfo::AddHTLC { .. } => { panic!("short_channel_id == 0 should imply any pending_forward entries are of type Receive"); @@ -1774,8 +2063,8 @@ impl self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_source, &payment_hash, failure_reason); } - for (their_node_id, err) in handle_errors.drain(..) { - let _ = handle_error!(self, err, their_node_id); + for (counterparty_node_id, err) in handle_errors.drain(..) { + let _ = handle_error!(self, err, counterparty_node_id); } if new_events.is_empty() { return } @@ -1783,13 +2072,42 @@ impl events.append(&mut new_events); } + /// Free the background events, generally called from timer_tick_occurred. + /// + /// Exposed for testing to allow us to process events quickly without generating accidental + /// BroadcastChannelUpdate events in timer_tick_occurred. + /// + /// Expects the caller to have a total_consistency_lock read lock. + fn process_background_events(&self) { + let mut background_events = Vec::new(); + mem::swap(&mut *self.pending_background_events.lock().unwrap(), &mut background_events); + for event in background_events.drain(..) { + match event { + BackgroundEvent::ClosingMonitorUpdate((funding_txo, update)) => { + // The channel has already been closed, so no use bothering to care about the + // monitor updating completing. + let _ = self.chain_monitor.update_channel(funding_txo, update); + }, + } + } + } + + #[cfg(any(test, feature = "_test_utils"))] + pub(crate) fn test_process_background_events(&self) { + self.process_background_events(); + } + /// If a peer is disconnected we mark any channels with that peer as 'disabled'. /// After some time, if channels are still disabled we need to broadcast a ChannelUpdate /// to inform the network about the uselessness of these channels. /// /// This method handles all the details, and must be called roughly once per minute. - pub fn timer_chan_freshness_every_min(&self) { - let _ = self.total_consistency_lock.read().unwrap(); + /// + /// Note that in some rare cases this may generate a `chain::Watch::update_channel` call. + pub fn timer_tick_occurred(&self) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + self.process_background_events(); + let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; for (_, chan) in channel_state.by_id.iter_mut() { @@ -1813,18 +2131,17 @@ impl /// along the path (including in our own channel on which we received it). /// Returns false if no payment was found to fail backwards, true if the process of failing the /// HTLC backwards has been started. - pub fn fail_htlc_backwards(&self, payment_hash: &PaymentHash, payment_secret: &Option) -> bool { - let _ = self.total_consistency_lock.read().unwrap(); + pub fn fail_htlc_backwards(&self, payment_hash: &PaymentHash) -> bool { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let mut channel_state = Some(self.channel_state.lock().unwrap()); - let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&(*payment_hash, *payment_secret)); + let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(payment_hash); if let Some(mut sources) = removed_source { for htlc in sources.drain(..) { if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); } let mut htlc_msat_height_data = byte_utils::be64_to_array(htlc.value).to_vec(); htlc_msat_height_data.extend_from_slice(&byte_utils::be32_to_array( - self.latest_block_height.load(Ordering::Acquire) as u32, - )); + self.best_block.read().unwrap().height())); self.fail_htlc_backwards_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc.prev_hop), payment_hash, HTLCFailReason::Reason { failure_code: 0x4000 | 15, data: htlc_msat_height_data }); @@ -1877,11 +2194,15 @@ impl /// to fail and take the channel_state lock for each iteration (as we take ownership and may /// drop it). In other words, no assumptions are made that entries in claimable_htlcs point to /// still-available channels. - fn fail_htlc_backwards_internal(&self, mut channel_state_lock: MutexGuard>, source: HTLCSource, payment_hash: &PaymentHash, onion_error: HTLCFailReason) { + fn fail_htlc_backwards_internal(&self, mut channel_state_lock: MutexGuard>, source: HTLCSource, payment_hash: &PaymentHash, onion_error: HTLCFailReason) { //TODO: There is a timing attack here where if a node fails an HTLC back to us they can //identify whether we sent it or not based on the (I presume) very different runtime //between the branches here. We should make this async and move it into the forward HTLCs //timer handling. + + // Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called + // from block_connected which may run during initialization prior to the chain_monitor + // being fully configured. See the docs for `ChannelManagerReadArgs` for more. match source { HTLCSource::OutboundRoute { ref path, .. } => { log_trace!(self.logger, "Failing outbound payment HTLC with payment_hash {}", log_bytes!(payment_hash.0)); @@ -1939,7 +2260,7 @@ impl } } }, - HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id, htlc_id, incoming_packet_shared_secret }) => { + HTLCSource::PreviousHopData(HTLCPreviousHopData { short_channel_id, htlc_id, incoming_packet_shared_secret, .. }) => { let err_packet = match onion_error { HTLCFailReason::Reason { failure_code, data } => { log_trace!(self.logger, "Failing HTLC with payment_hash {} backwards from us with code {}", log_bytes!(payment_hash.0), failure_code); @@ -1979,24 +2300,22 @@ impl /// generating message events for the net layer to claim the payment, if possible. Thus, you /// should probably kick the net layer to go send messages if this returns true! /// - /// You must specify the expected amounts for this HTLC, and we will only claim HTLCs - /// available within a few percent of the expected amount. This is critical for several - /// reasons : a) it avoids providing senders with `proof-of-payment` (in the form of the - /// payment_preimage without having provided the full value and b) it avoids certain - /// privacy-breaking recipient-probing attacks which may reveal payment activity to - /// motivated attackers. - /// - /// Note that the privacy concerns in (b) are not relevant in payments with a payment_secret - /// set. Thus, for such payments we will claim any payments which do not under-pay. + /// Note that if you did not set an `amount_msat` when calling [`create_inbound_payment`] or + /// [`create_inbound_payment_for_hash`] you must check that the amount in the `PaymentReceived` + /// event matches your expectation. If you fail to do so and call this method, you may provide + /// the sender "proof-of-payment" when they did not fulfill the full expected payment. /// /// May panic if called except in response to a PaymentReceived event. - pub fn claim_funds(&self, payment_preimage: PaymentPreimage, payment_secret: &Option, expected_amount: u64) -> bool { + /// + /// [`create_inbound_payment`]: Self::create_inbound_payment + /// [`create_inbound_payment_for_hash`]: Self::create_inbound_payment_for_hash + pub fn claim_funds(&self, payment_preimage: PaymentPreimage) -> bool { let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner()); - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let mut channel_state = Some(self.channel_state.lock().unwrap()); - let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&(payment_hash, *payment_secret)); + let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&payment_hash); if let Some(mut sources) = removed_source { assert!(!sources.is_empty()); @@ -2011,31 +2330,22 @@ impl // we got all the HTLCs and then a channel closed while we were waiting for the user to // provide the preimage, so worrying too much about the optimal handling isn't worth // it. - - let (is_mpp, mut valid_mpp) = if let &Some(ref data) = &sources[0].payment_data { - assert!(payment_secret.is_some()); - (true, data.total_msat >= expected_amount) - } else { - assert!(payment_secret.is_none()); - (false, false) - }; - + let mut valid_mpp = true; for htlc in sources.iter() { - if !is_mpp || !valid_mpp { break; } if let None = channel_state.as_ref().unwrap().short_to_id.get(&htlc.prev_hop.short_channel_id) { valid_mpp = false; + break; } } let mut errs = Vec::new(); let mut claimed_any_htlcs = false; for htlc in sources.drain(..) { - if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); } - if (is_mpp && !valid_mpp) || (!is_mpp && (htlc.value < expected_amount || htlc.value > expected_amount * 2)) { + if !valid_mpp { + if channel_state.is_none() { channel_state = Some(self.channel_state.lock().unwrap()); } let mut htlc_msat_height_data = byte_utils::be64_to_array(htlc.value).to_vec(); htlc_msat_height_data.extend_from_slice(&byte_utils::be32_to_array( - self.latest_block_height.load(Ordering::Acquire) as u32, - )); + self.best_block.read().unwrap().height())); self.fail_htlc_backwards_internal(channel_state.take().unwrap(), HTLCSource::PreviousHopData(htlc.prev_hop), &payment_hash, HTLCFailReason::Reason { failure_code: 0x4000|15, data: htlc_msat_height_data }); @@ -2049,10 +2359,7 @@ impl claimed_any_htlcs = true; } else { errs.push(e); } }, - Err(None) if is_mpp => unreachable!("We already checked for channel existence, we can't fail here!"), - Err(None) => { - log_warn!(self.logger, "Channel we expected to claim an HTLC from was closed."); - }, + Err(None) => unreachable!("We already checked for channel existence, we can't fail here!"), Ok(()) => claimed_any_htlcs = true, } } @@ -2062,16 +2369,16 @@ impl // which were generated. channel_state.take(); - for (their_node_id, err) in errs.drain(..) { + for (counterparty_node_id, err) in errs.drain(..) { let res: Result<(), _> = Err(err); - let _ = handle_error!(self, res, their_node_id); + let _ = handle_error!(self, res, counterparty_node_id); } claimed_any_htlcs } else { false } } - fn claim_funds_from_hop(&self, channel_state_lock: &mut MutexGuard>, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage) -> Result<(), Option<(PublicKey, MsgHandleErrInternal)>> { + fn claim_funds_from_hop(&self, channel_state_lock: &mut MutexGuard>, prev_hop: HTLCPreviousHopData, payment_preimage: PaymentPreimage) -> Result<(), Option<(PublicKey, MsgHandleErrInternal)>> { //TODO: Delay the claimed_funds relaying just like we do outbound relay! let channel_state = &mut **channel_state_lock; let chan_id = match channel_state.short_to_id.get(&prev_hop.short_channel_id) { @@ -2086,17 +2393,17 @@ impl match chan.get_mut().get_update_fulfill_htlc_and_commit(prev_hop.htlc_id, payment_preimage, &self.logger) { Ok((msgs, monitor_option)) => { if let Some(monitor_update) = monitor_option { - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { if was_frozen_for_monitor { assert!(msgs.is_none()); } else { - return Err(Some((chan.get().get_their_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err()))); + return Err(Some((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, msgs.is_some()).unwrap_err()))); } } } if let Some((msg, commitment_signed)) = msgs { channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: chan.get().get_their_node_id(), + node_id: chan.get().get_counterparty_node_id(), updates: msgs::CommitmentUpdate { update_add_htlcs: Vec::new(), update_fulfill_htlcs: vec![msg], @@ -2124,7 +2431,7 @@ impl } else { unreachable!(); } } - fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard>, source: HTLCSource, payment_preimage: PaymentPreimage) { + fn claim_funds_internal(&self, mut channel_state_lock: MutexGuard>, source: HTLCSource, payment_preimage: PaymentPreimage) { match source { HTLCSource::OutboundRoute { .. } => { mem::drop(channel_state_lock); @@ -2134,19 +2441,30 @@ impl }); }, HTLCSource::PreviousHopData(hop_data) => { - if let Err((their_node_id, err)) = match self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage) { + let prev_outpoint = hop_data.outpoint; + if let Err((counterparty_node_id, err)) = match self.claim_funds_from_hop(&mut channel_state_lock, hop_data, payment_preimage) { Ok(()) => Ok(()), Err(None) => { - // TODO: There is probably a channel monitor somewhere that needs to - // learn the preimage as the channel already hit the chain and that's - // why it's missing. + let preimage_update = ChannelMonitorUpdate { + update_id: CLOSED_CHANNEL_UPDATE_ID, + updates: vec![ChannelMonitorUpdateStep::PaymentPreimage { + payment_preimage: payment_preimage.clone(), + }], + }; + // We update the ChannelMonitor on the backward link, after + // receiving an offchain preimage event from the forward link (the + // event being update_fulfill_htlc). + if let Err(e) = self.chain_monitor.update_channel(prev_outpoint, preimage_update) { + log_error!(self.logger, "Critical error: failed to update channel monitor with preimage {:?}: {:?}", + payment_preimage, e); + } Ok(()) }, Err(Some(res)) => Err(res), } { mem::drop(channel_state_lock); let res: Result<(), _> = Err(err); - let _ = handle_error!(self, res, their_node_id); + let _ = handle_error!(self, res, counterparty_node_id); } }, } @@ -2154,7 +2472,7 @@ impl /// Gets the node_id held by this ChannelManager pub fn get_our_node_id(&self) -> PublicKey { - PublicKey::from_secret_key(&self.secp_ctx, &self.our_network_key) + self.our_network_pubkey.clone() } /// Restores a single, given channel to normal operation after a @@ -2170,7 +2488,7 @@ impl /// exists largely only to prevent races between this and concurrent update_monitor calls. /// /// Thus, the anticipated use is, at a high level: - /// 1) You register a ManyChannelMonitor with this ChannelManager, + /// 1) You register a chain::Watch with this ChannelManager, /// 2) it stores each update to disk, and begins updating any remote (eg watchtower) copies of /// said ChannelMonitors as it can, returning ChannelMonitorUpdateErr::TemporaryFailures /// any time it cannot do so instantly, @@ -2178,7 +2496,7 @@ impl /// 4) once all remote copies are updated, you call this function with the update_id that /// completed, and once it is the latest the Channel will be re-enabled. pub fn channel_monitor_updated(&self, funding_txo: &OutPoint, highest_applied_update_id: u64) { - let _ = self.total_consistency_lock.read().unwrap(); + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let mut close_results = Vec::new(); let mut htlc_forwards = Vec::new(); @@ -2198,16 +2516,16 @@ impl return; } - let (raa, commitment_update, order, pending_forwards, mut pending_failures, needs_broadcast_safe, funding_locked) = channel.monitor_updating_restored(&self.logger); + let (raa, commitment_update, order, pending_forwards, mut pending_failures, funding_broadcastable, funding_locked) = channel.monitor_updating_restored(&self.logger); if !pending_forwards.is_empty() { - htlc_forwards.push((channel.get_short_channel_id().expect("We can't have pending forwards before funding confirmation"), pending_forwards)); + htlc_forwards.push((channel.get_short_channel_id().expect("We can't have pending forwards before funding confirmation"), funding_txo.clone(), pending_forwards)); } htlc_failures.append(&mut pending_failures); macro_rules! handle_cs { () => { if let Some(update) = commitment_update { pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: channel.get_their_node_id(), + node_id: channel.get_counterparty_node_id(), updates: update, }); } @@ -2215,7 +2533,7 @@ impl macro_rules! handle_raa { () => { if let Some(revoke_and_ack) = raa { pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { - node_id: channel.get_their_node_id(), + node_id: channel.get_counterparty_node_id(), msg: revoke_and_ack, }); } @@ -2230,20 +2548,17 @@ impl handle_cs!(); }, } - if needs_broadcast_safe { - pending_events.push(events::Event::FundingBroadcastSafe { - funding_txo: channel.get_funding_txo().unwrap(), - user_channel_id: channel.get_user_id(), - }); + if let Some(tx) = funding_broadcastable { + self.tx_broadcaster.broadcast_transaction(&tx); } if let Some(msg) = funding_locked { pending_msg_events.push(events::MessageSendEvent::SendFundingLocked { - node_id: channel.get_their_node_id(), + node_id: channel.get_counterparty_node_id(), msg, }); if let Some(announcement_sigs) = self.get_announcement_sigs(channel) { pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures { - node_id: channel.get_their_node_id(), + node_id: channel.get_counterparty_node_id(), msg: announcement_sigs, }); } @@ -2263,12 +2578,12 @@ impl } } - fn internal_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_open_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> { if msg.chain_hash != self.genesis_hash { return Err(MsgHandleErrInternal::send_err_msg_no_close("Unknown genesis block hash".to_owned(), msg.temporary_channel_id.clone())); } - let channel = Channel::new_from_req(&self.fee_estimator, &self.keys_manager, their_node_id.clone(), their_features, msg, 0, &self.default_configuration) + let channel = Channel::new_from_req(&self.fee_estimator, &self.keys_manager, counterparty_node_id.clone(), their_features, msg, 0, &self.default_configuration) .map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.temporary_channel_id))?; let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; @@ -2276,7 +2591,7 @@ impl hash_map::Entry::Occupied(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("temporary_channel_id collision!".to_owned(), msg.temporary_channel_id.clone())), hash_map::Entry::Vacant(entry) => { channel_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg: channel.get_accept_channel(), }); entry.insert(channel); @@ -2285,13 +2600,13 @@ impl Ok(()) } - fn internal_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { + fn internal_accept_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { let (value, output_script, user_id) = { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.temporary_channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.temporary_channel_id)); } try_chan_entry!(self, chan.get_mut().accept_channel(&msg, &self.default_configuration, their_features), channel_state, chan); @@ -2304,36 +2619,42 @@ impl pending_events.push(events::Event::FundingGenerationReady { temporary_channel_id: msg.temporary_channel_id, channel_value_satoshis: value, - output_script: output_script, + output_script, user_channel_id: user_id, }); Ok(()) } - fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> { - let ((funding_msg, monitor_update), mut chan) = { + fn internal_funding_created(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> { + let ((funding_msg, monitor), mut chan) = { + let best_block = *self.best_block.read().unwrap(); let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.temporary_channel_id.clone()) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.temporary_channel_id)); } - (try_chan_entry!(self, chan.get_mut().funding_created(msg, &self.logger), channel_state, chan), chan.remove()) + (try_chan_entry!(self, chan.get_mut().funding_created(msg, best_block, &self.logger), channel_state, chan), chan.remove()) }, hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.temporary_channel_id)) } }; // Because we have exclusive ownership of the channel here we can release the channel_state - // lock before add_monitor - if let Err(e) = self.monitor.add_monitor(monitor_update.get_funding_txo().0, monitor_update) { + // lock before watch_channel + if let Err(e) = self.chain_monitor.watch_channel(monitor.get_funding_txo().0, monitor) { match e { ChannelMonitorUpdateErr::PermanentFailure => { // Note that we reply with the new channel_id in error messages if we gave up on the // channel, not the temporary_channel_id. This is compatible with ourselves, but the // spec is somewhat ambiguous here. Not a huge deal since we'll send error messages for // any messages referencing a previously-closed channel anyway. - return Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure".to_owned(), funding_msg.channel_id, chan.force_shutdown(true), None)); + // We do not do a force-close here as that would generate a monitor update for + // a monitor that we didn't manage to store (and that we don't care about - we + // don't respond with the funding_signed so the channel can never go on chain). + let (_monitor_update, failed_htlcs) = chan.force_shutdown(true); + assert!(failed_htlcs.is_empty()); + return Err(MsgHandleErrInternal::send_err_msg_no_close("ChannelMonitor storage failure".to_owned(), funding_msg.channel_id)); }, ChannelMonitorUpdateErr::TemporaryFailure => { // There's no problem signing a counterparty's funding transaction if our monitor @@ -2352,7 +2673,7 @@ impl }, hash_map::Entry::Vacant(e) => { channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingSigned { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg: funding_msg, }); e.insert(chan); @@ -2361,41 +2682,38 @@ impl Ok(()) } - fn internal_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> { - let (funding_txo, user_id) = { + fn internal_funding_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> { + let funding_tx = { + let best_block = *self.best_block.read().unwrap(); let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } - let monitor = match chan.get_mut().funding_signed(&msg, &self.logger) { + let (monitor, funding_tx) = match chan.get_mut().funding_signed(&msg, best_block, &self.logger) { Ok(update) => update, Err(e) => try_chan_entry!(self, Err(e), channel_state, chan), }; - if let Err(e) = self.monitor.add_monitor(chan.get().get_funding_txo().unwrap(), monitor) { + if let Err(e) = self.chain_monitor.watch_channel(chan.get().get_funding_txo().unwrap(), monitor) { return_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::RevokeAndACKFirst, false, false); } - (chan.get().get_funding_txo().unwrap(), chan.get().get_user_id()) + funding_tx }, hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id)) } }; - let mut pending_events = self.pending_events.lock().unwrap(); - pending_events.push(events::Event::FundingBroadcastSafe { - funding_txo: funding_txo, - user_channel_id: user_id, - }); + self.tx_broadcaster.broadcast_transaction(&funding_tx); Ok(()) } - fn internal_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), MsgHandleErrInternal> { + fn internal_funding_locked(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result<(), MsgHandleErrInternal> { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } try_chan_entry!(self, chan.get_mut().funding_locked(&msg), channel_state, chan); @@ -2411,7 +2729,7 @@ impl // failures is an issue. Note, to achieve its goal, only one of the announcement_sigs needs // to be received, from then sigs are going to be flood to the whole network. channel_state.pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg: announcement_sigs, }); } @@ -2421,26 +2739,26 @@ impl } } - fn internal_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(), MsgHandleErrInternal> { + fn internal_shutdown(&self, counterparty_node_id: &PublicKey, their_features: &InitFeatures, msg: &msgs::Shutdown) -> Result<(), MsgHandleErrInternal> { let (mut dropped_htlcs, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { - if chan_entry.get().get_their_node_id() != *their_node_id { + if chan_entry.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } - let (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&self.fee_estimator, &msg), channel_state, chan_entry); + let (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&self.fee_estimator, &their_features, &msg), channel_state, chan_entry); if let Some(msg) = shutdown { channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } if let Some(msg) = closing_signed { channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } @@ -2468,19 +2786,19 @@ impl Ok(()) } - fn internal_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), MsgHandleErrInternal> { + fn internal_closing_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result<(), MsgHandleErrInternal> { let (tx, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { - if chan_entry.get().get_their_node_id() != *their_node_id { + if chan_entry.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } let (closing_signed, tx) = try_chan_entry!(self, chan_entry.get_mut().closing_signed(&self.fee_estimator, &msg), channel_state, chan_entry); if let Some(msg) = closing_signed { channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } @@ -2514,7 +2832,7 @@ impl Ok(()) } - fn internal_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), MsgHandleErrInternal> { + fn internal_update_add_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), MsgHandleErrInternal> { //TODO: BOLT 4 points out a specific attack where a peer may re-send an onion packet and //determine the state of the payment based on our response/if we forward anything/the time //we take to respond. We should take care to avoid allowing such an attack. @@ -2529,11 +2847,11 @@ impl match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } - let create_pending_htlc_status = |chan: &Channel, pending_forward_info: PendingHTLCStatus, error_code: u16| { + let create_pending_htlc_status = |chan: &Channel, pending_forward_info: PendingHTLCStatus, error_code: u16| { // Ensure error_code has the UPDATE flag set, since by default we send a // channel update along as part of failing the HTLC. assert!((error_code & 0x1000) != 0); @@ -2579,13 +2897,13 @@ impl Ok(()) } - fn internal_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> { + fn internal_update_fulfill_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); let htlc_source = { let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } try_chan_entry!(self, chan.get_mut().update_fulfill_htlc(&msg), channel_state, chan) @@ -2597,12 +2915,12 @@ impl Ok(()) } - fn internal_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), MsgHandleErrInternal> { + fn internal_update_fail_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } try_chan_entry!(self, chan.get_mut().update_fail_htlc(&msg, HTLCFailReason::LightningError { err: msg.reason.clone() }), channel_state, chan); @@ -2612,12 +2930,12 @@ impl Ok(()) } - fn internal_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), MsgHandleErrInternal> { + fn internal_update_fail_malformed_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } if (msg.failure_code & 0x8000) == 0 { @@ -2631,12 +2949,12 @@ impl } } - fn internal_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), MsgHandleErrInternal> { + fn internal_commitment_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(), MsgHandleErrInternal> { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } let (revoke_and_ack, commitment_signed, closing_signed, monitor_update) = @@ -2644,23 +2962,23 @@ impl Err((None, e)) => try_chan_entry!(self, Err(e), channel_state, chan), Err((Some(update), e)) => { assert!(chan.get().is_awaiting_monitor_update()); - let _ = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), update); + let _ = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), update); try_chan_entry!(self, Err(e), channel_state, chan); unreachable!(); }, Ok(res) => res }; - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { return_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::RevokeAndACKFirst, true, commitment_signed.is_some()); //TODO: Rebroadcast closing_signed if present on monitor update restoration } channel_state.pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg: revoke_and_ack, }); if let Some(msg) = commitment_signed { channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), updates: msgs::CommitmentUpdate { update_add_htlcs: Vec::new(), update_fulfill_htlcs: Vec::new(), @@ -2673,7 +2991,7 @@ impl } if let Some(msg) = closing_signed { channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } @@ -2684,8 +3002,8 @@ impl } #[inline] - fn forward_htlcs(&self, per_source_pending_forwards: &mut [(u64, Vec<(PendingHTLCInfo, u64)>)]) { - for &mut (prev_short_channel_id, ref mut pending_forwards) in per_source_pending_forwards { + fn forward_htlcs(&self, per_source_pending_forwards: &mut [(u64, OutPoint, Vec<(PendingHTLCInfo, u64)>)]) { + for &mut (prev_short_channel_id, prev_funding_outpoint, ref mut pending_forwards) in per_source_pending_forwards { let mut forward_event = None; if !pending_forwards.is_empty() { let mut channel_state = self.channel_state.lock().unwrap(); @@ -2698,10 +3016,12 @@ impl PendingHTLCRouting::Receive { .. } => 0, }) { hash_map::Entry::Occupied(mut entry) => { - entry.get_mut().push(HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info }); + entry.get_mut().push(HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_funding_outpoint, + prev_htlc_id, forward_info }); }, hash_map::Entry::Vacant(entry) => { - entry.insert(vec!(HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info })); + entry.insert(vec!(HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_funding_outpoint, + prev_htlc_id, forward_info })); } } } @@ -2718,21 +3038,21 @@ impl } } - fn internal_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), MsgHandleErrInternal> { + fn internal_revoke_and_ack(&self, counterparty_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), MsgHandleErrInternal> { let mut htlcs_to_fail = Vec::new(); let res = loop { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { break Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update(); let (commitment_update, pending_forwards, pending_failures, closing_signed, monitor_update, htlcs_to_fail_in) = break_chan_entry!(self, chan.get_mut().revoke_and_ack(&msg, &self.fee_estimator, &self.logger), channel_state, chan); htlcs_to_fail = htlcs_to_fail_in; - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { if was_frozen_for_monitor { assert!(commitment_update.is_none() && closing_signed.is_none() && pending_forwards.is_empty() && pending_failures.is_empty()); break Err(MsgHandleErrInternal::ignore_no_close("Previous monitor update failure prevented responses to RAA".to_owned())); @@ -2744,40 +3064,40 @@ impl } if let Some(updates) = commitment_update { channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), updates, }); } if let Some(msg) = closing_signed { channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } - break Ok((pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"))) + break Ok((pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"), chan.get().get_funding_txo().unwrap())) }, hash_map::Entry::Vacant(_) => break Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id)) } }; self.fail_holding_cell_htlcs(htlcs_to_fail, msg.channel_id); match res { - Ok((pending_forwards, mut pending_failures, short_channel_id)) => { + Ok((pending_forwards, mut pending_failures, short_channel_id, channel_outpoint)) => { for failure in pending_failures.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2); } - self.forward_htlcs(&mut [(short_channel_id, pending_forwards)]); + self.forward_htlcs(&mut [(short_channel_id, channel_outpoint, pending_forwards)]); Ok(()) }, Err(e) => Err(e) } } - fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> { + fn internal_update_fee(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> { let mut channel_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } try_chan_entry!(self, chan.get_mut().update_fee(&self.fee_estimator, &msg), channel_state, chan); @@ -2787,13 +3107,13 @@ impl Ok(()) } - fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> { + fn internal_announcement_signatures(&self, counterparty_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } if !chan.get().is_usable() { @@ -2841,13 +3161,36 @@ impl Ok(()) } - fn internal_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), MsgHandleErrInternal> { + fn internal_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) -> Result<(), MsgHandleErrInternal> { + let mut channel_state_lock = self.channel_state.lock().unwrap(); + let channel_state = &mut *channel_state_lock; + let chan_id = match channel_state.short_to_id.get(&msg.contents.short_channel_id) { + Some(chan_id) => chan_id.clone(), + None => { + // It's not a local channel + return Ok(()) + } + }; + match channel_state.by_id.entry(chan_id) { + hash_map::Entry::Occupied(mut chan) => { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { + // TODO: see issue #153, need a consistent behavior on obnoxious behavior from random node + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), chan_id)); + } + try_chan_entry!(self, chan.get_mut().channel_update(&msg), channel_state, chan); + }, + hash_map::Entry::Vacant(_) => unreachable!() + } + Ok(()) + } + + fn internal_channel_reestablish(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), MsgHandleErrInternal> { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; match channel_state.by_id.entry(msg.channel_id) { hash_map::Entry::Occupied(mut chan) => { - if chan.get().get_their_node_id() != *their_node_id { + if chan.get().get_counterparty_node_id() != *counterparty_node_id { return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id)); } // Currently, we expect all holding cell update_adds to be dropped on peer @@ -2857,7 +3200,7 @@ impl let (funding_locked, revoke_and_ack, commitment_update, monitor_update_opt, mut order, shutdown) = try_chan_entry!(self, chan.get_mut().channel_reestablish(msg, &self.logger), channel_state, chan); if let Some(monitor_update) = monitor_update_opt { - if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { // channel_reestablish doesn't guarantee the order it returns is sensical // for the messages it returns, but if we're setting what messages to // re-transmit on monitor update success, we need to make sure it is sane. @@ -2873,14 +3216,14 @@ impl } if let Some(msg) = funding_locked { channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingLocked { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg }); } macro_rules! send_raa { () => { if let Some(msg) = revoke_and_ack { channel_state.pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg }); } @@ -2888,7 +3231,7 @@ impl macro_rules! send_cu { () => { if let Some(updates) = commitment_update { channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), updates }); } @@ -2905,7 +3248,7 @@ impl } if let Some(msg) = shutdown { channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown { - node_id: their_node_id.clone(), + node_id: counterparty_node_id.clone(), msg, }); } @@ -2919,10 +3262,11 @@ impl /// If successful, will generate a UpdateHTLCs event, so you should probably poll /// PeerManager::process_events afterwards. /// Note: This API is likely to change! + /// (C-not exported) Cause its doc(hidden) anyway #[doc(hidden)] pub fn update_fee(&self, channel_id: [u8;32], feerate_per_kw: u32) -> Result<(), APIError> { - let _ = self.total_consistency_lock.read().unwrap(); - let their_node_id; + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let counterparty_node_id; let err: Result<(), _> = loop { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = &mut *channel_state_lock; @@ -2939,15 +3283,15 @@ impl if !chan.get().is_live() { return Err(APIError::ChannelUnavailable{err: "Channel is either not yet fully established or peer is currently disconnected".to_owned()}); } - their_node_id = chan.get().get_their_node_id(); + counterparty_node_id = chan.get().get_counterparty_node_id(); if let Some((update_fee, commitment_signed, monitor_update)) = break_chan_entry!(self, chan.get_mut().send_update_fee_and_commit(feerate_per_kw, &self.logger), channel_state, chan) { - if let Err(_e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) { + if let Err(_e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) { unimplemented!(); } channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { - node_id: chan.get().get_their_node_id(), + node_id: chan.get().get_counterparty_node_id(), updates: msgs::CommitmentUpdate { update_add_htlcs: Vec::new(), update_fulfill_htlcs: Vec::new(), @@ -2963,17 +3307,17 @@ impl return Ok(()) }; - match handle_error!(self, err, their_node_id) { + match handle_error!(self, err, counterparty_node_id) { Ok(_) => unreachable!(), Err(e) => { Err(APIError::APIMisuseError { err: e.err })} } } - /// Process pending events from the ManyChannelMonitor. + /// Process pending events from the `chain::Watch`. fn process_pending_monitor_events(&self) { let mut failed_channels = Vec::new(); { - for monitor_event in self.monitor.get_and_clear_pending_monitor_events() { + for monitor_event in self.chain_monitor.release_pending_monitor_events() { match monitor_event { MonitorEvent::HTLCEvent(htlc_update) => { if let Some(preimage) = htlc_update.payment_preimage { @@ -3000,6 +3344,12 @@ impl msg: update }); } + pending_msg_events.push(events::MessageSendEvent::HandleError { + node_id: chan.get_counterparty_node_id(), + action: msgs::ErrorAction::SendErrorMessage { + msg: msgs::ErrorMessage { channel_id: chan.channel_id(), data: "Channel force-closed".to_owned() } + }, + }); } }, } @@ -3010,12 +3360,131 @@ impl self.finish_force_close_channel(failure); } } + + /// Handle a list of channel failures during a block_connected or block_disconnected call, + /// pushing the channel monitor update (if any) to the background events queue and removing the + /// Channel object. + fn handle_init_event_channel_failures(&self, mut failed_channels: Vec) { + for mut failure in failed_channels.drain(..) { + // Either a commitment transactions has been confirmed on-chain or + // Channel::block_disconnected detected that the funding transaction has been + // reorganized out of the main chain. + // We cannot broadcast our latest local state via monitor update (as + // Channel::force_shutdown tries to make us do) as we may still be in initialization, + // so we track the update internally and handle it when the user next calls + // timer_tick_occurred, guaranteeing we're running normally. + if let Some((funding_txo, update)) = failure.0.take() { + assert_eq!(update.updates.len(), 1); + if let ChannelMonitorUpdateStep::ChannelForceClosed { should_broadcast } = update.updates[0] { + assert!(should_broadcast); + } else { unreachable!(); } + self.pending_background_events.lock().unwrap().push(BackgroundEvent::ClosingMonitorUpdate((funding_txo, update))); + } + self.finish_force_close_channel(failure); + } + } + + fn set_payment_hash_secret_map(&self, payment_hash: PaymentHash, payment_preimage: Option, min_value_msat: Option, invoice_expiry_delta_secs: u32, user_payment_id: u64) -> Result { + assert!(invoice_expiry_delta_secs <= 60*60*24*365); // Sadly bitcoin timestamps are u32s, so panic before 2106 + + let payment_secret = PaymentSecret(self.keys_manager.get_secure_random_bytes()); + + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let mut payment_secrets = self.pending_inbound_payments.lock().unwrap(); + match payment_secrets.entry(payment_hash) { + hash_map::Entry::Vacant(e) => { + e.insert(PendingInboundPayment { + payment_secret, min_value_msat, user_payment_id, payment_preimage, + // We assume that highest_seen_timestamp is pretty close to the current time - + // its updated when we receive a new block with the maximum time we've seen in + // a header. It should never be more than two hours in the future. + // Thus, we add two hours here as a buffer to ensure we absolutely + // never fail a payment too early. + // Note that we assume that received blocks have reasonably up-to-date + // timestamps. + expiry_time: self.highest_seen_timestamp.load(Ordering::Acquire) as u64 + invoice_expiry_delta_secs as u64 + 7200, + }); + }, + hash_map::Entry::Occupied(_) => return Err(APIError::APIMisuseError { err: "Duplicate payment hash".to_owned() }), + } + Ok(payment_secret) + } + + /// Gets a payment secret and payment hash for use in an invoice given to a third party wishing + /// to pay us. + /// + /// This differs from [`create_inbound_payment_for_hash`] only in that it generates the + /// [`PaymentHash`] and [`PaymentPreimage`] for you, returning the first and storing the second. + /// + /// The [`PaymentPreimage`] will ultimately be returned to you in the [`PaymentReceived`], which + /// will have the [`PaymentReceived::payment_preimage`] field filled in. That should then be + /// passed directly to [`claim_funds`]. + /// + /// See [`create_inbound_payment_for_hash`] for detailed documentation on behavior and requirements. + /// + /// [`claim_funds`]: Self::claim_funds + /// [`PaymentReceived`]: events::Event::PaymentReceived + /// [`PaymentReceived::payment_preimage`]: events::Event::PaymentReceived::payment_preimage + /// [`create_inbound_payment_for_hash`]: Self::create_inbound_payment_for_hash + pub fn create_inbound_payment(&self, min_value_msat: Option, invoice_expiry_delta_secs: u32, user_payment_id: u64) -> (PaymentHash, PaymentSecret) { + let payment_preimage = PaymentPreimage(self.keys_manager.get_secure_random_bytes()); + let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner()); + + (payment_hash, + self.set_payment_hash_secret_map(payment_hash, Some(payment_preimage), min_value_msat, invoice_expiry_delta_secs, user_payment_id) + .expect("RNG Generated Duplicate PaymentHash")) + } + + /// Gets a [`PaymentSecret`] for a given [`PaymentHash`], for which the payment preimage is + /// stored external to LDK. + /// + /// A [`PaymentReceived`] event will only be generated if the [`PaymentSecret`] matches a + /// payment secret fetched via this method or [`create_inbound_payment`], and which is at least + /// the `min_value_msat` provided here, if one is provided. + /// + /// The [`PaymentHash`] (and corresponding [`PaymentPreimage`]) must be globally unique. This + /// method may return an Err if another payment with the same payment_hash is still pending. + /// + /// `user_payment_id` will be provided back in [`PaymentReceived::user_payment_id`] events to + /// allow tracking of which events correspond with which calls to this and + /// [`create_inbound_payment`]. `user_payment_id` has no meaning inside of LDK, it is simply + /// copied to events and otherwise ignored. It may be used to correlate PaymentReceived events + /// with invoice metadata stored elsewhere. + /// + /// `min_value_msat` should be set if the invoice being generated contains a value. Any payment + /// received for the returned [`PaymentHash`] will be required to be at least `min_value_msat` + /// before a [`PaymentReceived`] event will be generated, ensuring that we do not provide the + /// sender "proof-of-payment" unless they have paid the required amount. + /// + /// `invoice_expiry_delta_secs` describes the number of seconds that the invoice is valid for + /// in excess of the current time. This should roughly match the expiry time set in the invoice. + /// After this many seconds, we will remove the inbound payment, resulting in any attempts to + /// pay the invoice failing. The BOLT spec suggests 7,200 secs as a default validity time for + /// invoices when no timeout is set. + /// + /// Note that we use block header time to time-out pending inbound payments (with some margin + /// to compensate for the inaccuracy of block header timestamps). Thus, in practice we will + /// accept a payment and generate a [`PaymentReceived`] event for some time after the expiry. + /// If you need exact expiry semantics, you should enforce them upon receipt of + /// [`PaymentReceived`]. + /// + /// May panic if `invoice_expiry_delta_secs` is greater than one year. + /// + /// Note that invoices generated for inbound payments should have their `min_final_cltv_expiry` + /// set to at least [`MIN_FINAL_CLTV_EXPIRY`]. + /// + /// [`create_inbound_payment`]: Self::create_inbound_payment + /// [`PaymentReceived`]: events::Event::PaymentReceived + /// [`PaymentReceived::user_payment_id`]: events::Event::PaymentReceived::user_payment_id + pub fn create_inbound_payment_for_hash(&self, payment_hash: PaymentHash, min_value_msat: Option, invoice_expiry_delta_secs: u32, user_payment_id: u64) -> Result { + self.set_payment_hash_secret_map(payment_hash, None, min_value_msat, invoice_expiry_delta_secs, user_payment_id) + } } -impl MessageSendEventsProvider for ChannelManager - where M::Target: ManyChannelMonitor, +impl MessageSendEventsProvider for ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { @@ -3031,10 +3500,10 @@ impl } } -impl EventsProvider for ChannelManager - where M::Target: ManyChannelMonitor, +impl EventsProvider for ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { @@ -3050,18 +3519,142 @@ impl } } -impl - ChainListener for ChannelManager - where M::Target: ManyChannelMonitor, - T::Target: BroadcasterInterface, - K::Target: KeysInterface, - F::Target: FeeEstimator, - L::Target: Logger, +impl chain::Listen for ChannelManager +where + M::Target: chain::Watch, + T::Target: BroadcasterInterface, + K::Target: KeysInterface, + F::Target: FeeEstimator, + L::Target: Logger, +{ + fn block_connected(&self, block: &Block, height: u32) { + { + let best_block = self.best_block.read().unwrap(); + assert_eq!(best_block.block_hash(), block.header.prev_blockhash, + "Blocks must be connected in chain-order - the connected header must build on the last connected header"); + assert_eq!(best_block.height(), height - 1, + "Blocks must be connected in chain-order - the connected block height must be one greater than the previous height"); + } + + let txdata: Vec<_> = block.txdata.iter().enumerate().collect(); + self.transactions_confirmed(&block.header, &txdata, height); + self.best_block_updated(&block.header, height); + } + + fn block_disconnected(&self, header: &BlockHeader, height: u32) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let new_height = height - 1; + { + let mut best_block = self.best_block.write().unwrap(); + assert_eq!(best_block.block_hash(), header.block_hash(), + "Blocks must be disconnected in chain-order - the disconnected header must be the last connected header"); + assert_eq!(best_block.height(), height, + "Blocks must be disconnected in chain-order - the disconnected block must have the correct height"); + *best_block = BestBlock::new(header.prev_blockhash, new_height) + } + + self.do_chain_event(Some(new_height), |channel| channel.best_block_updated(new_height, header.time)); + } +} + +impl chain::Confirm for ChannelManager +where + M::Target: chain::Watch, + T::Target: BroadcasterInterface, + K::Target: KeysInterface, + F::Target: FeeEstimator, + L::Target: Logger, +{ + fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) { + // Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called + // during initialization prior to the chain_monitor being fully configured in some cases. + // See the docs for `ChannelManagerReadArgs` for more. + + let block_hash = header.block_hash(); + log_trace!(self.logger, "{} transactions included in block {} at height {} provided", txdata.len(), block_hash, height); + + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + self.do_chain_event(Some(height), |channel| channel.transactions_confirmed(&block_hash, height, txdata, &self.logger).map(|a| (a, Vec::new()))); + } + + fn best_block_updated(&self, header: &BlockHeader, height: u32) { + // Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called + // during initialization prior to the chain_monitor being fully configured in some cases. + // See the docs for `ChannelManagerReadArgs` for more. + + let block_hash = header.block_hash(); + log_trace!(self.logger, "New best block: {} at height {}", block_hash, height); + + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + + *self.best_block.write().unwrap() = BestBlock::new(block_hash, height); + + self.do_chain_event(Some(height), |channel| channel.best_block_updated(height, header.time)); + + macro_rules! max_time { + ($timestamp: expr) => { + loop { + // Update $timestamp to be the max of its current value and the block + // timestamp. This should keep us close to the current time without relying on + // having an explicit local time source. + // Just in case we end up in a race, we loop until we either successfully + // update $timestamp or decide we don't need to. + let old_serial = $timestamp.load(Ordering::Acquire); + if old_serial >= header.time as usize { break; } + if $timestamp.compare_exchange(old_serial, header.time as usize, Ordering::AcqRel, Ordering::Relaxed).is_ok() { + break; + } + } + } + } + max_time!(self.last_node_announcement_serial); + max_time!(self.highest_seen_timestamp); + let mut payment_secrets = self.pending_inbound_payments.lock().unwrap(); + payment_secrets.retain(|_, inbound_payment| { + inbound_payment.expiry_time > header.time as u64 + }); + } + + fn get_relevant_txids(&self) -> Vec { + let channel_state = self.channel_state.lock().unwrap(); + let mut res = Vec::with_capacity(channel_state.short_to_id.len()); + for chan in channel_state.by_id.values() { + if let Some(funding_txo) = chan.get_funding_txo() { + res.push(funding_txo.txid); + } + } + res + } + + fn transaction_unconfirmed(&self, txid: &Txid) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + self.do_chain_event(None, |channel| { + if let Some(funding_txo) = channel.get_funding_txo() { + if funding_txo.txid == *txid { + channel.funding_transaction_unconfirmed().map(|_| (None, Vec::new())) + } else { Ok((None, Vec::new())) } + } else { Ok((None, Vec::new())) } + }); + } +} + +impl ChannelManager +where + M::Target: chain::Watch, + T::Target: BroadcasterInterface, + K::Target: KeysInterface, + F::Target: FeeEstimator, + L::Target: Logger, { - fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[usize]) { - let header_hash = header.block_hash(); - log_trace!(self.logger, "Block {} at height {} connected with {} txn matched", header_hash, height, txn_matched.len()); - let _ = self.total_consistency_lock.read().unwrap(); + /// Calls a function which handles an on-chain event (blocks dis/connected, transactions + /// un/confirmed, etc) on each channel, handling any resulting errors or messages generated by + /// the function. + fn do_chain_event) -> Result<(Option, Vec<(HTLCSource, PaymentHash)>), msgs::ErrorMessage>> + (&self, height_opt: Option, f: FN) { + // Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called + // during initialization prior to the chain_monitor being fully configured in some cases. + // See the docs for `ChannelManagerReadArgs` for more. + let mut failed_channels = Vec::new(); let mut timed_out_htlcs = Vec::new(); { @@ -3070,7 +3663,7 @@ impl= htlc.cltv_expiry - HTLC_FAIL_BACK_BUFFER { - let mut htlc_msat_height_data = byte_utils::be64_to_array(htlc.value).to_vec(); - htlc_msat_height_data.extend_from_slice(&byte_utils::be32_to_array(height)); - timed_out_htlcs.push((HTLCSource::PreviousHopData(htlc.prev_hop.clone()), payment_hash.clone(), HTLCFailReason::Reason { - failure_code: 0x4000 | 15, - data: htlc_msat_height_data - })); - false - } else { true } + if let Some(height) = height_opt { + channel_state.claimable_htlcs.retain(|payment_hash, htlcs| { + htlcs.retain(|htlc| { + // If height is approaching the number of blocks we think it takes us to get + // our commitment transaction confirmed before the HTLC expires, plus the + // number of blocks we generally consider it to take to do a commitment update, + // just give up on it and fail the HTLC. + if height >= htlc.cltv_expiry - HTLC_FAIL_BACK_BUFFER { + let mut htlc_msat_height_data = byte_utils::be64_to_array(htlc.value).to_vec(); + htlc_msat_height_data.extend_from_slice(&byte_utils::be32_to_array(height)); + timed_out_htlcs.push((HTLCSource::PreviousHopData(htlc.prev_hop.clone()), payment_hash.clone(), HTLCFailReason::Reason { + failure_code: 0x4000 | 15, + data: htlc_msat_height_data + })); + false + } else { true } + }); + !htlcs.is_empty() // Only retain this entry if htlcs has at least one entry. }); - !htlcs.is_empty() // Only retain this entry if htlcs has at least one entry. - }); - } - for failure in failed_channels.drain(..) { - self.finish_force_close_channel(failure); + } } + self.handle_init_event_channel_failures(failed_channels); + for (source, payment_hash, reason) in timed_out_htlcs.drain(..) { self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), source, &payment_hash, reason); } - self.latest_block_height.store(height as usize, Ordering::Release); - *self.last_block_hash.try_lock().expect("block_(dis)connected must not be called in parallel") = header_hash; - loop { - // Update last_node_announcement_serial to be the max of its current value and the - // block timestamp. This should keep us close to the current time without relying on - // having an explicit local time source. - // Just in case we end up in a race, we loop until we either successfully update - // last_node_announcement_serial or decide we don't need to. - let old_serial = self.last_node_announcement_serial.load(Ordering::Acquire); - if old_serial >= header.time as usize { break; } - if self.last_node_announcement_serial.compare_exchange(old_serial, header.time as usize, Ordering::AcqRel, Ordering::Relaxed).is_ok() { - break; - } - } } - /// We force-close the channel without letting our counterparty participate in the shutdown - fn block_disconnected(&self, header: &BlockHeader, _: u32) { - let _ = self.total_consistency_lock.read().unwrap(); - let mut failed_channels = Vec::new(); - { - let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = &mut *channel_lock; - let short_to_id = &mut channel_state.short_to_id; - let pending_msg_events = &mut channel_state.pending_msg_events; - channel_state.by_id.retain(|_, v| { - if v.block_disconnected(header) { - if let Some(short_id) = v.get_short_channel_id() { - short_to_id.remove(&short_id); - } - failed_channels.push(v.force_shutdown(true)); - if let Ok(update) = self.get_channel_update(&v) { - pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate { - msg: update - }); - } - false - } else { - true - } - }); - } - for failure in failed_channels.drain(..) { - self.finish_force_close_channel(failure); - } - self.latest_block_height.fetch_sub(1, Ordering::AcqRel); - *self.last_block_hash.try_lock().expect("block_(dis)connected must not be called in parallel") = header.block_hash(); + /// Blocks until ChannelManager needs to be persisted or a timeout is reached. It returns a bool + /// indicating whether persistence is necessary. Only one listener on + /// `await_persistable_update` or `await_persistable_update_timeout` is guaranteed to be woken + /// up. + /// Note that the feature `allow_wallclock_use` must be enabled to use this function. + #[cfg(any(test, feature = "allow_wallclock_use"))] + pub fn await_persistable_update_timeout(&self, max_wait: Duration) -> bool { + self.persistence_notifier.wait_timeout(max_wait) + } + + /// Blocks until ChannelManager needs to be persisted. Only one listener on + /// `await_persistable_update` or `await_persistable_update_timeout` is guaranteed to be woken + /// up. + pub fn await_persistable_update(&self) { + self.persistence_notifier.wait() + } + + #[cfg(any(test, feature = "_test_utils"))] + pub fn get_persistence_condvar_value(&self) -> bool { + let mutcond = &self.persistence_notifier.persistence_lock; + let &(ref mtx, _) = mutcond; + let guard = mtx.lock().unwrap(); + *guard } } -impl - ChannelMessageHandler for ChannelManager - where M::Target: ManyChannelMonitor, +impl + ChannelMessageHandler for ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { - fn handle_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_open_channel(their_node_id, their_features, msg), *their_node_id); + fn handle_open_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_open_channel(counterparty_node_id, their_features, msg), *counterparty_node_id); } - fn handle_accept_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_accept_channel(their_node_id, their_features, msg), *their_node_id); + fn handle_accept_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_accept_channel(counterparty_node_id, their_features, msg), *counterparty_node_id); } - fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_funding_created(their_node_id, msg), *their_node_id); + fn handle_funding_created(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingCreated) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_funding_created(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_funding_signed(their_node_id, msg), *their_node_id); + fn handle_funding_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingSigned) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_funding_signed(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_funding_locked(their_node_id, msg), *their_node_id); + fn handle_funding_locked(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingLocked) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_funding_locked(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_shutdown(their_node_id, msg), *their_node_id); + fn handle_shutdown(&self, counterparty_node_id: &PublicKey, their_features: &InitFeatures, msg: &msgs::Shutdown) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_shutdown(counterparty_node_id, their_features, msg), *counterparty_node_id); } - fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_closing_signed(their_node_id, msg), *their_node_id); + fn handle_closing_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::ClosingSigned) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_closing_signed(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_update_add_htlc(their_node_id, msg), *their_node_id); + fn handle_update_add_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_update_add_htlc(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg), *their_node_id); + fn handle_update_fulfill_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_update_fulfill_htlc(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg), *their_node_id); + fn handle_update_fail_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_update_fail_htlc(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg), *their_node_id); + fn handle_update_fail_malformed_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_update_fail_malformed_htlc(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_commitment_signed(their_node_id, msg), *their_node_id); + fn handle_commitment_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::CommitmentSigned) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_commitment_signed(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg), *their_node_id); + fn handle_revoke_and_ack(&self, counterparty_node_id: &PublicKey, msg: &msgs::RevokeAndACK) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_revoke_and_ack(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_update_fee(their_node_id, msg), *their_node_id); + fn handle_update_fee(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFee) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_update_fee(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_announcement_signatures(their_node_id, msg), *their_node_id); + fn handle_announcement_signatures(&self, counterparty_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_announcement_signatures(counterparty_node_id, msg), *counterparty_node_id); } - fn handle_channel_reestablish(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) { - let _ = self.total_consistency_lock.read().unwrap(); - let _ = handle_error!(self, self.internal_channel_reestablish(their_node_id, msg), *their_node_id); + fn handle_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_channel_update(counterparty_node_id, msg), *counterparty_node_id); } - fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool) { - let _ = self.total_consistency_lock.read().unwrap(); + fn handle_channel_reestablish(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelReestablish) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); + let _ = handle_error!(self, self.internal_channel_reestablish(counterparty_node_id, msg), *counterparty_node_id); + } + + fn peer_disconnected(&self, counterparty_node_id: &PublicKey, no_connection_possible: bool) { + let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier); let mut failed_channels = Vec::new(); let mut failed_payments = Vec::new(); let mut no_channels_remain = true; @@ -3302,9 +3868,9 @@ impl node_id != their_node_id, - &events::MessageSendEvent::SendOpenChannel { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendFundingCreated { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendFundingSigned { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendFundingLocked { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendAnnouncementSignatures { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::UpdateHTLCs { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendRevokeAndACK { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendClosingSigned { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendShutdown { ref node_id, .. } => node_id != their_node_id, - &events::MessageSendEvent::SendChannelReestablish { ref node_id, .. } => node_id != their_node_id, + &events::MessageSendEvent::SendAcceptChannel { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendOpenChannel { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendFundingCreated { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendFundingSigned { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendFundingLocked { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendAnnouncementSignatures { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::UpdateHTLCs { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendRevokeAndACK { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendClosingSigned { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendShutdown { ref node_id, .. } => node_id != counterparty_node_id, + &events::MessageSendEvent::SendChannelReestablish { ref node_id, .. } => node_id != counterparty_node_id, &events::MessageSendEvent::BroadcastChannelAnnouncement { .. } => true, &events::MessageSendEvent::BroadcastNodeAnnouncement { .. } => true, &events::MessageSendEvent::BroadcastChannelUpdate { .. } => true, - &events::MessageSendEvent::HandleError { ref node_id, .. } => node_id != their_node_id, + &events::MessageSendEvent::HandleError { ref node_id, .. } => node_id != counterparty_node_id, &events::MessageSendEvent::PaymentFailureNetworkUpdate { .. } => true, + &events::MessageSendEvent::SendChannelRangeQuery { .. } => false, + &events::MessageSendEvent::SendShortIdsQuery { .. } => false, + &events::MessageSendEvent::SendReplyChannelRange { .. } => false, } }); } if no_channels_remain { - self.per_peer_state.write().unwrap().remove(their_node_id); + self.per_peer_state.write().unwrap().remove(counterparty_node_id); } for failure in failed_channels.drain(..) { @@ -3380,14 +3949,14 @@ impl { e.insert(Mutex::new(PeerState { latest_features: init_msg.features.clone(), @@ -3403,7 +3972,7 @@ impl, Condvar), +} + +impl PersistenceNotifier { + fn new() -> Self { + Self { + persistence_lock: (Mutex::new(false), Condvar::new()), } } + + fn wait(&self) { + loop { + let &(ref mtx, ref cvar) = &self.persistence_lock; + let mut guard = mtx.lock().unwrap(); + guard = cvar.wait(guard).unwrap(); + let result = *guard; + if result { + *guard = false; + return + } + } + } + + #[cfg(any(test, feature = "allow_wallclock_use"))] + fn wait_timeout(&self, max_wait: Duration) -> bool { + let current_time = Instant::now(); + loop { + let &(ref mtx, ref cvar) = &self.persistence_lock; + let mut guard = mtx.lock().unwrap(); + guard = cvar.wait_timeout(guard, max_wait).unwrap().0; + // Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the + // desired wait time has actually passed, and if not then restart the loop with a reduced wait + // time. Note that this logic can be highly simplified through the use of + // `Condvar::wait_while` and `Condvar::wait_timeout_while`, if and when our MSRV is raised to + // 1.42.0. + let elapsed = current_time.elapsed(); + let result = *guard; + if result || elapsed >= max_wait { + *guard = false; + return result; + } + match max_wait.checked_sub(elapsed) { + None => return result, + Some(_) => continue + } + } + } + + // Signal to the ChannelManager persister that there are updates necessitating persisting to disk. + fn notify(&self) { + let &(ref persist_mtx, ref cnd) = &self.persistence_lock; + let mut persistence_lock = persist_mtx.lock().unwrap(); + *persistence_lock = true; + mem::drop(persistence_lock); + cnd.notify_all(); + } } const SERIALIZATION_VERSION: u8 = 1; @@ -3450,7 +4084,8 @@ impl Writeable for PendingHTLCInfo { }, &PendingHTLCRouting::Receive { ref payment_data, ref incoming_cltv_expiry } => { 1u8.write(writer)?; - payment_data.write(writer)?; + payment_data.payment_secret.write(writer)?; + payment_data.total_msat.write(writer)?; incoming_cltv_expiry.write(writer)?; }, } @@ -3471,7 +4106,10 @@ impl Readable for PendingHTLCInfo { short_channel_id: Readable::read(reader)?, }, 1u8 => PendingHTLCRouting::Receive { - payment_data: Readable::read(reader)?, + payment_data: msgs::FinalOnionHopData { + payment_secret: Readable::read(reader)?, + total_msat: Readable::read(reader)?, + }, incoming_cltv_expiry: Readable::read(reader)?, }, _ => return Err(DecodeError::InvalidValue), @@ -3538,16 +4176,34 @@ impl Readable for PendingHTLCStatus { impl_writeable!(HTLCPreviousHopData, 0, { short_channel_id, + outpoint, htlc_id, incoming_packet_shared_secret }); -impl_writeable!(ClaimableHTLC, 0, { - prev_hop, - value, - payment_data, - cltv_expiry -}); +impl Writeable for ClaimableHTLC { + fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + self.prev_hop.write(writer)?; + self.value.write(writer)?; + self.payment_data.payment_secret.write(writer)?; + self.payment_data.total_msat.write(writer)?; + self.cltv_expiry.write(writer) + } +} + +impl Readable for ClaimableHTLC { + fn read(reader: &mut R) -> Result { + Ok(ClaimableHTLC { + prev_hop: Readable::read(reader)?, + value: Readable::read(reader)?, + payment_data: msgs::FinalOnionHopData { + payment_secret: Readable::read(reader)?, + total_msat: Readable::read(reader)?, + }, + cltv_expiry: Readable::read(reader)?, + }) + } +} impl Writeable for HTLCSource { fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { @@ -3614,9 +4270,10 @@ impl Readable for HTLCFailReason { impl Writeable for HTLCForwardInfo { fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { match self { - &HTLCForwardInfo::AddHTLC { ref prev_short_channel_id, ref prev_htlc_id, ref forward_info } => { + &HTLCForwardInfo::AddHTLC { ref prev_short_channel_id, ref prev_funding_outpoint, ref prev_htlc_id, ref forward_info } => { 0u8.write(writer)?; prev_short_channel_id.write(writer)?; + prev_funding_outpoint.write(writer)?; prev_htlc_id.write(writer)?; forward_info.write(writer)?; }, @@ -3635,6 +4292,7 @@ impl Readable for HTLCForwardInfo { match ::read(reader)? { 0 => Ok(HTLCForwardInfo::AddHTLC { prev_short_channel_id: Readable::read(reader)?, + prev_funding_outpoint: Readable::read(reader)?, prev_htlc_id: Readable::read(reader)?, forward_info: Readable::read(reader)?, }), @@ -3647,22 +4305,33 @@ impl Readable for HTLCForwardInfo { } } -impl Writeable for ChannelManager - where M::Target: ManyChannelMonitor, +impl_writeable!(PendingInboundPayment, 0, { + payment_secret, + expiry_time, + user_payment_id, + payment_preimage, + min_value_msat +}); + +impl Writeable for ChannelManager + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { - let _ = self.total_consistency_lock.write().unwrap(); + let _consistency_lock = self.total_consistency_lock.write().unwrap(); writer.write_all(&[SERIALIZATION_VERSION; 1])?; writer.write_all(&[MIN_SERIALIZATION_VERSION; 1])?; self.genesis_hash.write(writer)?; - (self.latest_block_height.load(Ordering::Acquire) as u32).write(writer)?; - self.last_block_hash.lock().unwrap().write(writer)?; + { + let best_block = self.best_block.read().unwrap(); + best_block.height().write(writer)?; + best_block.block_hash().write(writer)?; + } let channel_state = self.channel_state.lock().unwrap(); let mut unfunded_channels = 0; @@ -3710,7 +4379,27 @@ impl { + 0u8.write(writer)?; + funding_txo.write(writer)?; + monitor_update.write(writer)?; + }, + } + } + (self.last_node_announcement_serial.load(Ordering::Acquire) as u32).write(writer)?; + (self.highest_seen_timestamp.load(Ordering::Acquire) as u32).write(writer)?; + + let pending_inbound_payments = self.pending_inbound_payments.lock().unwrap(); + (pending_inbound_payments.len() as u64).write(writer)?; + for (hash, pending_payment) in pending_inbound_payments.iter() { + hash.write(writer)?; + pending_payment.write(writer)?; + } Ok(()) } @@ -3721,38 +4410,48 @@ impl::read(reader, args). +/// 2) Deserialize the ChannelManager by filling in this struct and calling: +/// <(BlockHash, ChannelManager)>::read(reader, args) /// This may result in closing some Channels if the ChannelMonitor is newer than the stored /// ChannelManager state to ensure no loss of funds. Thus, transactions may be broadcasted. -/// 3) Register all relevant ChannelMonitor outpoints with your chain watch mechanism using -/// ChannelMonitor::get_monitored_outpoints and ChannelMonitor::get_funding_txo(). +/// 3) If you are not fetching full blocks, register all relevant ChannelMonitor outpoints the same +/// way you would handle a `chain::Filter` call using ChannelMonitor::get_outputs_to_watch() and +/// ChannelMonitor::get_funding_txo(). /// 4) Reconnect blocks on your ChannelMonitors. -/// 5) Move the ChannelMonitors into your local ManyChannelMonitor. -/// 6) Disconnect/connect blocks on the ChannelManager. -/// 7) Register the new ChannelManager with your ChainWatchInterface. -pub struct ChannelManagerReadArgs<'a, ChanSigner: 'a + ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> - where M::Target: ManyChannelMonitor, +/// 5) Disconnect/connect blocks on the ChannelManager. +/// 6) Move the ChannelMonitors into your local chain::Watch. +/// +/// Note that the ordering of #4-6 is not of importance, however all three must occur before you +/// call any other methods on the newly-deserialized ChannelManager. +/// +/// Note that because some channels may be closed during deserialization, it is critical that you +/// always deserialize only the latest version of a ChannelManager and ChannelMonitors available to +/// you. If you deserialize an old ChannelManager (during which force-closure transactions may be +/// broadcast), and then later deserialize a newer version of the same ChannelManager (which will +/// not force-close the same channels but consider them live), you may end up revoking a state for +/// which you've already broadcasted the transaction. +pub struct ChannelManagerReadArgs<'a, Signer: 'a + Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { - /// The keys provider which will give us relevant keys. Some keys will be loaded during - /// deserialization. + /// deserialization and KeysInterface::read_chan_signer will be used to read per-Channel + /// signing data. pub keys_manager: K, /// The fee_estimator for use in the ChannelManager in the future. /// /// No calls to the FeeEstimator will be made during deserialization. pub fee_estimator: F, - /// The ManyChannelMonitor for use in the ChannelManager in the future. + /// The chain::Watch for use in the ChannelManager in the future. /// - /// No calls to the ManyChannelMonitor will be made during deserialization. It is assumed that + /// No calls to the chain::Watch will be made during deserialization. It is assumed that /// you have deserialized ChannelMonitors separately and will add them to your - /// ManyChannelMonitor after deserializing this ChannelManager. - pub monitor: M, + /// chain::Watch after deserializing this ChannelManager. + pub chain_monitor: M, /// The BroadcasterInterface which will be used in the ChannelManager in the future and may be /// used to broadcast the latest local commitment transactions of channels which must be @@ -3775,24 +4474,26 @@ pub struct ChannelManagerReadArgs<'a, ChanSigner: 'a + ChannelKeys, M: Deref, T: /// /// In such cases the latest local transactions will be sent to the tx_broadcaster included in /// this struct. - pub channel_monitors: HashMap>, + /// + /// (C-not exported) because we have no HashMap bindings + pub channel_monitors: HashMap>, } -impl<'a, ChanSigner: 'a + ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> - ChannelManagerReadArgs<'a, ChanSigner, M, T, K, F, L> - where M::Target: ManyChannelMonitor, +impl<'a, Signer: 'a + Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> + ChannelManagerReadArgs<'a, Signer, M, T, K, F, L> + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { /// Simple utility function to create a ChannelManagerReadArgs which creates the monitor /// HashMap for you. This is primarily useful for C bindings where it is not practical to /// populate a HashMap directly from C. - pub fn new(keys_manager: K, fee_estimator: F, monitor: M, tx_broadcaster: T, logger: L, default_config: UserConfig, - mut channel_monitors: Vec<&'a mut ChannelMonitor>) -> Self { + pub fn new(keys_manager: K, fee_estimator: F, chain_monitor: M, tx_broadcaster: T, logger: L, default_config: UserConfig, + mut channel_monitors: Vec<&'a mut ChannelMonitor>) -> Self { Self { - keys_manager, fee_estimator, monitor, tx_broadcaster, logger, default_config, + keys_manager, fee_estimator, chain_monitor, tx_broadcaster, logger, default_config, channel_monitors: channel_monitors.drain(..).map(|monitor| { (monitor.get_funding_txo().0, monitor) }).collect() } } @@ -3800,29 +4501,29 @@ impl<'a, ChanSigner: 'a + ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L // Implement ReadableArgs for an Arc'd ChannelManager to make it a bit easier to work with the // SipmleArcChannelManager type: -impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> - ReadableArgs> for (BlockHash, Arc>) - where M::Target: ManyChannelMonitor, +impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> + ReadableArgs> for (BlockHash, Arc>) + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { - fn read(reader: &mut R, args: ChannelManagerReadArgs<'a, ChanSigner, M, T, K, F, L>) -> Result { - let (blockhash, chan_manager) = <(BlockHash, ChannelManager)>::read(reader, args)?; + fn read(reader: &mut R, args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result { + let (blockhash, chan_manager) = <(BlockHash, ChannelManager)>::read(reader, args)?; Ok((blockhash, Arc::new(chan_manager))) } } -impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> - ReadableArgs> for (BlockHash, ChannelManager) - where M::Target: ManyChannelMonitor, +impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> + ReadableArgs> for (BlockHash, ChannelManager) + where M::Target: chain::Watch, T::Target: BroadcasterInterface, - K::Target: KeysInterface, + K::Target: KeysInterface, F::Target: FeeEstimator, L::Target: Logger, { - fn read(reader: &mut R, mut args: ChannelManagerReadArgs<'a, ChanSigner, M, T, K, F, L>) -> Result { + fn read(reader: &mut R, mut args: ChannelManagerReadArgs<'a, Signer, M, T, K, F, L>) -> Result { let _ver: u8 = Readable::read(reader)?; let min_ver: u8 = Readable::read(reader)?; if min_ver > SERIALIZATION_VERSION { @@ -3830,8 +4531,8 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De } let genesis_hash: BlockHash = Readable::read(reader)?; - let latest_block_height: u32 = Readable::read(reader)?; - let last_block_hash: BlockHash = Readable::read(reader)?; + let best_block_height: u32 = Readable::read(reader)?; + let best_block_hash: BlockHash = Readable::read(reader)?; let mut failed_htlcs = Vec::new(); @@ -3840,28 +4541,24 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De let mut by_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128)); let mut short_to_id = HashMap::with_capacity(cmp::min(channel_count as usize, 128)); for _ in 0..channel_count { - let mut channel: Channel = Readable::read(reader)?; - if channel.last_block_connected != Default::default() && channel.last_block_connected != last_block_hash { - return Err(DecodeError::InvalidValue); - } - + let mut channel: Channel = Channel::read(reader, &args.keys_manager)?; let funding_txo = channel.get_funding_txo().ok_or(DecodeError::InvalidValue)?; funding_txo_set.insert(funding_txo.clone()); if let Some(ref mut monitor) = args.channel_monitors.get_mut(&funding_txo) { - if channel.get_cur_local_commitment_transaction_number() < monitor.get_cur_local_commitment_number() || - channel.get_revoked_remote_commitment_transaction_number() < monitor.get_min_seen_secret() || - channel.get_cur_remote_commitment_transaction_number() < monitor.get_cur_remote_commitment_number() || + if channel.get_cur_holder_commitment_transaction_number() < monitor.get_cur_holder_commitment_number() || + channel.get_revoked_counterparty_commitment_transaction_number() < monitor.get_min_seen_secret() || + channel.get_cur_counterparty_commitment_transaction_number() < monitor.get_cur_counterparty_commitment_number() || channel.get_latest_monitor_update_id() > monitor.get_latest_update_id() { // If the channel is ahead of the monitor, return InvalidValue: return Err(DecodeError::InvalidValue); - } else if channel.get_cur_local_commitment_transaction_number() > monitor.get_cur_local_commitment_number() || - channel.get_revoked_remote_commitment_transaction_number() > monitor.get_min_seen_secret() || - channel.get_cur_remote_commitment_transaction_number() > monitor.get_cur_remote_commitment_number() || + } else if channel.get_cur_holder_commitment_transaction_number() > monitor.get_cur_holder_commitment_number() || + channel.get_revoked_counterparty_commitment_transaction_number() > monitor.get_min_seen_secret() || + channel.get_cur_counterparty_commitment_transaction_number() > monitor.get_cur_counterparty_commitment_number() || channel.get_latest_monitor_update_id() < monitor.get_latest_update_id() { // But if the channel is behind of the monitor, close the channel: - let (_, _, mut new_failed_htlcs) = channel.force_shutdown(true); + let (_, mut new_failed_htlcs) = channel.force_shutdown(true); failed_htlcs.append(&mut new_failed_htlcs); - monitor.broadcast_latest_local_commitment_txn(&args.tx_broadcaster, &args.logger); + monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger); } else { if let Some(short_channel_id) = channel.get_short_channel_id() { short_to_id.insert(short_channel_id, channel.channel_id()); @@ -3875,7 +4572,7 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De for (ref funding_txo, ref mut monitor) in args.channel_monitors.iter_mut() { if !funding_txo_set.contains(funding_txo) { - monitor.broadcast_latest_local_commitment_txn(&args.tx_broadcaster, &args.logger); + monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger); } } @@ -3923,17 +4620,36 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De } } + let background_event_count: u64 = Readable::read(reader)?; + let mut pending_background_events_read: Vec = Vec::with_capacity(cmp::min(background_event_count as usize, MAX_ALLOC_SIZE/mem::size_of::())); + for _ in 0..background_event_count { + match ::read(reader)? { + 0 => pending_background_events_read.push(BackgroundEvent::ClosingMonitorUpdate((Readable::read(reader)?, Readable::read(reader)?))), + _ => return Err(DecodeError::InvalidValue), + } + } + let last_node_announcement_serial: u32 = Readable::read(reader)?; + let highest_seen_timestamp: u32 = Readable::read(reader)?; + + let pending_inbound_payment_count: u64 = Readable::read(reader)?; + let mut pending_inbound_payments: HashMap = HashMap::with_capacity(cmp::min(pending_inbound_payment_count as usize, MAX_ALLOC_SIZE/(3*32))); + for _ in 0..pending_inbound_payment_count { + if pending_inbound_payments.insert(Readable::read(reader)?, Readable::read(reader)?).is_some() { + return Err(DecodeError::InvalidValue); + } + } + + let mut secp_ctx = Secp256k1::new(); + secp_ctx.seeded_randomize(&args.keys_manager.get_secure_random_bytes()); let channel_manager = ChannelManager { genesis_hash, fee_estimator: args.fee_estimator, - monitor: args.monitor, + chain_monitor: args.chain_monitor, tx_broadcaster: args.tx_broadcaster, - latest_block_height: AtomicUsize::new(latest_block_height as usize), - last_block_hash: Mutex::new(last_block_hash), - secp_ctx: Secp256k1::new(), + best_block: RwLock::new(BestBlock::new(best_block_hash, best_block_height)), channel_state: Mutex::new(ChannelHolder { by_id, @@ -3942,14 +4658,22 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De claimable_htlcs, pending_msg_events: Vec::new(), }), + pending_inbound_payments: Mutex::new(pending_inbound_payments), + our_network_key: args.keys_manager.get_node_secret(), + our_network_pubkey: PublicKey::from_secret_key(&secp_ctx, &args.keys_manager.get_node_secret()), + secp_ctx, last_node_announcement_serial: AtomicUsize::new(last_node_announcement_serial as usize), + highest_seen_timestamp: AtomicUsize::new(highest_seen_timestamp as usize), per_peer_state: RwLock::new(per_peer_state), pending_events: Mutex::new(pending_events_read), + pending_background_events: Mutex::new(pending_background_events_read), total_consistency_lock: RwLock::new(()), + persistence_notifier: PersistenceNotifier::new(), + keys_manager: args.keys_manager, logger: args.logger, default_configuration: args.default_config, @@ -3962,6 +4686,209 @@ impl<'a, ChanSigner: ChannelKeys + Readable, M: Deref, T: Deref, K: Deref, F: De //TODO: Broadcast channel update for closed channels, but only after we've made a //connection or two. - Ok((last_block_hash.clone(), channel_manager)) + Ok((best_block_hash.clone(), channel_manager)) + } +} + +#[cfg(test)] +mod tests { + use ln::channelmanager::PersistenceNotifier; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::thread; + use std::time::Duration; + + #[test] + fn test_wait_timeout() { + let persistence_notifier = Arc::new(PersistenceNotifier::new()); + let thread_notifier = Arc::clone(&persistence_notifier); + + let exit_thread = Arc::new(AtomicBool::new(false)); + let exit_thread_clone = exit_thread.clone(); + thread::spawn(move || { + loop { + let &(ref persist_mtx, ref cnd) = &thread_notifier.persistence_lock; + let mut persistence_lock = persist_mtx.lock().unwrap(); + *persistence_lock = true; + cnd.notify_all(); + + if exit_thread_clone.load(Ordering::SeqCst) { + break + } + } + }); + + // Check that we can block indefinitely until updates are available. + let _ = persistence_notifier.wait(); + + // Check that the PersistenceNotifier will return after the given duration if updates are + // available. + loop { + if persistence_notifier.wait_timeout(Duration::from_millis(100)) { + break + } + } + + exit_thread.store(true, Ordering::SeqCst); + + // Check that the PersistenceNotifier will return after the given duration even if no updates + // are available. + loop { + if !persistence_notifier.wait_timeout(Duration::from_millis(100)) { + break + } + } + } +} + +#[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))] +pub mod bench { + use chain::Listen; + use chain::chainmonitor::ChainMonitor; + use chain::channelmonitor::Persist; + use chain::keysinterface::{KeysManager, InMemorySigner}; + use ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage}; + use ln::features::{InitFeatures, InvoiceFeatures}; + use ln::functional_test_utils::*; + use ln::msgs::ChannelMessageHandler; + use routing::network_graph::NetworkGraph; + use routing::router::get_route; + use util::test_utils; + use util::config::UserConfig; + use util::events::{Event, EventsProvider, MessageSendEvent, MessageSendEventsProvider}; + + use bitcoin::hashes::Hash; + use bitcoin::hashes::sha256::Hash as Sha256; + use bitcoin::{Block, BlockHeader, Transaction, TxOut}; + + use std::sync::Mutex; + + use test::Bencher; + + struct NodeHolder<'a, P: Persist> { + node: &'a ChannelManager, + &'a test_utils::TestBroadcaster, &'a KeysManager, + &'a test_utils::TestFeeEstimator, &'a test_utils::TestLogger> + } + + #[cfg(test)] + #[bench] + fn bench_sends(bench: &mut Bencher) { + bench_two_sends(bench, test_utils::TestPersister::new(), test_utils::TestPersister::new()); + } + + pub fn bench_two_sends>(bench: &mut Bencher, persister_a: P, persister_b: P) { + // Do a simple benchmark of sending a payment back and forth between two nodes. + // Note that this is unrealistic as each payment send will require at least two fsync + // calls per node. + let network = bitcoin::Network::Testnet; + let genesis_hash = bitcoin::blockdata::constants::genesis_block(network).header.block_hash(); + + let tx_broadcaster = test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())}; + let fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; + + let mut config: UserConfig = Default::default(); + config.own_channel_config.minimum_depth = 1; + + let logger_a = test_utils::TestLogger::with_id("node a".to_owned()); + let chain_monitor_a = ChainMonitor::new(None, &tx_broadcaster, &logger_a, &fee_estimator, &persister_a); + let seed_a = [1u8; 32]; + let keys_manager_a = KeysManager::new(&seed_a, 42, 42); + let node_a = ChannelManager::new(&fee_estimator, &chain_monitor_a, &tx_broadcaster, &logger_a, &keys_manager_a, config.clone(), ChainParameters { + network, + best_block: BestBlock::from_genesis(network), + }); + let node_a_holder = NodeHolder { node: &node_a }; + + let logger_b = test_utils::TestLogger::with_id("node a".to_owned()); + let chain_monitor_b = ChainMonitor::new(None, &tx_broadcaster, &logger_a, &fee_estimator, &persister_b); + let seed_b = [2u8; 32]; + let keys_manager_b = KeysManager::new(&seed_b, 42, 42); + let node_b = ChannelManager::new(&fee_estimator, &chain_monitor_b, &tx_broadcaster, &logger_b, &keys_manager_b, config.clone(), ChainParameters { + network, + best_block: BestBlock::from_genesis(network), + }); + let node_b_holder = NodeHolder { node: &node_b }; + + node_a.create_channel(node_b.get_our_node_id(), 8_000_000, 100_000_000, 42, None).unwrap(); + node_b.handle_open_channel(&node_a.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_a_holder, MessageSendEvent::SendOpenChannel, node_b.get_our_node_id())); + node_a.handle_accept_channel(&node_b.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_b_holder, MessageSendEvent::SendAcceptChannel, node_a.get_our_node_id())); + + let tx; + if let Event::FundingGenerationReady { temporary_channel_id, output_script, .. } = get_event!(node_a_holder, Event::FundingGenerationReady) { + tx = Transaction { version: 2, lock_time: 0, input: Vec::new(), output: vec![TxOut { + value: 8_000_000, script_pubkey: output_script, + }]}; + node_a.funding_transaction_generated(&temporary_channel_id, tx.clone()).unwrap(); + } else { panic!(); } + + node_b.handle_funding_created(&node_a.get_our_node_id(), &get_event_msg!(node_a_holder, MessageSendEvent::SendFundingCreated, node_b.get_our_node_id())); + node_a.handle_funding_signed(&node_b.get_our_node_id(), &get_event_msg!(node_b_holder, MessageSendEvent::SendFundingSigned, node_a.get_our_node_id())); + + assert_eq!(&tx_broadcaster.txn_broadcasted.lock().unwrap()[..], &[tx.clone()]); + + let block = Block { + header: BlockHeader { version: 0x20000000, prev_blockhash: genesis_hash, merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 }, + txdata: vec![tx], + }; + Listen::block_connected(&node_a, &block, 1); + Listen::block_connected(&node_b, &block, 1); + + node_a.handle_funding_locked(&node_b.get_our_node_id(), &get_event_msg!(node_b_holder, MessageSendEvent::SendFundingLocked, node_a.get_our_node_id())); + node_b.handle_funding_locked(&node_a.get_our_node_id(), &get_event_msg!(node_a_holder, MessageSendEvent::SendFundingLocked, node_b.get_our_node_id())); + + let dummy_graph = NetworkGraph::new(genesis_hash); + + let mut payment_count: u64 = 0; + macro_rules! send_payment { + ($node_a: expr, $node_b: expr) => { + let usable_channels = $node_a.list_usable_channels(); + let route = get_route(&$node_a.get_our_node_id(), &dummy_graph, &$node_b.get_our_node_id(), Some(InvoiceFeatures::known()), + Some(&usable_channels.iter().map(|r| r).collect::>()), &[], 10_000, TEST_FINAL_CLTV, &logger_a).unwrap(); + + let mut payment_preimage = PaymentPreimage([0; 32]); + payment_preimage.0[0..8].copy_from_slice(&payment_count.to_le_bytes()); + payment_count += 1; + let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0[..]).into_inner()); + let payment_secret = $node_b.create_inbound_payment_for_hash(payment_hash, None, 7200, 0).unwrap(); + + $node_a.send_payment(&route, payment_hash, &Some(payment_secret)).unwrap(); + let payment_event = SendEvent::from_event($node_a.get_and_clear_pending_msg_events().pop().unwrap()); + $node_b.handle_update_add_htlc(&$node_a.get_our_node_id(), &payment_event.msgs[0]); + $node_b.handle_commitment_signed(&$node_a.get_our_node_id(), &payment_event.commitment_msg); + let (raa, cs) = get_revoke_commit_msgs!(NodeHolder { node: &$node_b }, $node_a.get_our_node_id()); + $node_a.handle_revoke_and_ack(&$node_b.get_our_node_id(), &raa); + $node_a.handle_commitment_signed(&$node_b.get_our_node_id(), &cs); + $node_b.handle_revoke_and_ack(&$node_a.get_our_node_id(), &get_event_msg!(NodeHolder { node: &$node_a }, MessageSendEvent::SendRevokeAndACK, $node_b.get_our_node_id())); + + expect_pending_htlcs_forwardable!(NodeHolder { node: &$node_b }); + expect_payment_received!(NodeHolder { node: &$node_b }, payment_hash, payment_secret, 10_000); + assert!($node_b.claim_funds(payment_preimage)); + + match $node_b.get_and_clear_pending_msg_events().pop().unwrap() { + MessageSendEvent::UpdateHTLCs { node_id, updates } => { + assert_eq!(node_id, $node_a.get_our_node_id()); + $node_a.handle_update_fulfill_htlc(&$node_b.get_our_node_id(), &updates.update_fulfill_htlcs[0]); + $node_a.handle_commitment_signed(&$node_b.get_our_node_id(), &updates.commitment_signed); + }, + _ => panic!("Failed to generate claim event"), + } + + let (raa, cs) = get_revoke_commit_msgs!(NodeHolder { node: &$node_a }, $node_b.get_our_node_id()); + $node_b.handle_revoke_and_ack(&$node_a.get_our_node_id(), &raa); + $node_b.handle_commitment_signed(&$node_a.get_our_node_id(), &cs); + $node_a.handle_revoke_and_ack(&$node_b.get_our_node_id(), &get_event_msg!(NodeHolder { node: &$node_b }, MessageSendEvent::SendRevokeAndACK, $node_a.get_our_node_id())); + + expect_payment_sent!(NodeHolder { node: &$node_a }, payment_preimage); + } + } + + bench.iter(|| { + send_payment!(node_a, node_b); + send_payment!(node_b, node_a); + }); } }