# 1.41.0 is Debian stable
1.41.0,
# 1.45.2 is MSRV for lightning-net-tokio, lightning-block-sync, and coverage generation
- 1.45.2]
+ 1.45.2,
+ # 1.49.0 is MSRV for no_std builds using hashbrown
+ 1.49.0]
include:
- toolchain: stable
build-net-tokio: true
+ build-no-std: true
- toolchain: stable
platform: macos-latest
build-net-tokio: true
+ build-no-std: true
- toolchain: stable
platform: windows-latest
build-net-tokio: true
+ build-no-std: true
- toolchain: beta
build-net-tokio: true
+ build-no-std: true
+ - toolchain: 1.36.0
+ build-no-std: false
+ - toolchain: 1.41.0
+ build-no-std: false
- toolchain: 1.45.2
build-net-tokio: true
+ build-no-std: false
coverage: true
+ - toolchain: 1.49.0
+ build-no-std: true
runs-on: ${{ matrix.platform }}
steps:
- name: Checkout source code
run: RUSTFLAGS="-C link-dead-code" cargo build --verbose --color always
- name: Build on Rust ${{ matrix.toolchain }}
if: "! matrix.build-net-tokio"
- run: cargo build --verbose --color always -p lightning && cargo build --verbose --color always -p lightning-invoice && cargo build --verbose --color always -p lightning-persister
+ run: |
+ cargo build --verbose --color always -p lightning
+ cargo build --verbose --color always -p lightning-invoice
+ cargo build --verbose --color always -p lightning-persister
- name: Build Block Sync Clients on Rust ${{ matrix.toolchain }} with features
if: "matrix.build-net-tokio && !matrix.coverage"
run: |
cargo build --verbose --color always --features rpc-client
cargo build --verbose --color always --features rpc-client,rest-client
cargo build --verbose --color always --features rpc-client,rest-client,tokio
- cd ..
- name: Build Block Sync Clients on Rust ${{ matrix.toolchain }} with features and full code-linking for coverage generation
if: matrix.coverage
run: |
RUSTFLAGS="-C link-dead-code" cargo build --verbose --color always --features rpc-client
RUSTFLAGS="-C link-dead-code" cargo build --verbose --color always --features rpc-client,rest-client
RUSTFLAGS="-C link-dead-code" cargo build --verbose --color always --features rpc-client,rest-client,tokio
- cd ..
- name: Test on Rust ${{ matrix.toolchain }} with net-tokio
if: "matrix.build-net-tokio && !matrix.coverage"
run: cargo test --verbose --color always
- name: Test on Rust ${{ matrix.toolchain }} with net-tokio and full code-linking for coverage generation
if: matrix.coverage
run: RUSTFLAGS="-C link-dead-code" cargo test --verbose --color always
+ - name: Test on no_std bullds Rust ${{ matrix.toolchain }}
+ if: "matrix.build-no-std && !matrix.coverage"
+ run: |
+ cd lightning
+ cargo test --verbose --color always --features hashbrown
+ cd ..
+ - name: Test on no_std bullds Rust ${{ matrix.toolchain }} and full code-linking for coverage generation
+ if: "matrix.build-no-std && matrix.coverage"
+ run: |
+ cd lightning
+ RUSTFLAGS="-C link-dead-code" cargo test --verbose --color always --features hashbrown
+ cd ..
- name: Test on Rust ${{ matrix.toolchain }}
if: "! matrix.build-net-tokio"
- run: cargo test --verbose --color always -p lightning && cargo test --verbose --color always -p lightning-invoice && cargo build --verbose --color always -p lightning-persister
+ run: |
+ cargo test --verbose --color always -p lightning
+ cargo test --verbose --color always -p lightning-invoice
+ cargo build --verbose --color always -p lightning-persister
- name: Test Block Sync Clients on Rust ${{ matrix.toolchain }} with features
if: "matrix.build-net-tokio && !matrix.coverage"
run: |
cargo test --verbose --color always --features rpc-client
cargo test --verbose --color always --features rpc-client,rest-client
cargo test --verbose --color always --features rpc-client,rest-client,tokio
- cd ..
- name: Test Block Sync Clients on Rust ${{ matrix.toolchain }} with features and full code-linking for coverage generation
if: matrix.coverage
run: |
RUSTFLAGS="-C link-dead-code" cargo test --verbose --color always --features rpc-client
RUSTFLAGS="-C link-dead-code" cargo test --verbose --color always --features rpc-client,rest-client
RUSTFLAGS="-C link-dead-code" cargo test --verbose --color always --features rpc-client,rest-client,tokio
- cd ..
- name: Install deps for kcov
if: matrix.coverage
run: |
run: |
cd lightning
RUSTFLAGS="--cfg=require_route_graph_test" cargo test
+ RUSTFLAGS="--cfg=require_route_graph_test" cargo test --features hashbrown
cd ..
- name: Run benchmarks on Rust ${{ matrix.toolchain }}
run: |
check_commits:
runs-on: ubuntu-latest
env:
- TOOLCHAIN: stable
+ # rustc 1.53 regressed and panics when building our (perfectly fine) docs.
+ # See https://github.com/rust-lang/rust/issues/84738
+ TOOLCHAIN: 1.52.1
steps:
- name: Checkout source code
uses: actions/checkout@v2
--- /dev/null
+# 0.0.99 - WIP
+
+## Serialization Compatibility
+
+ * Due to a bug discovered in 0.0.98, if a `ChannelManager` is serialized on
+ version 0.0.98 while an `Event::PaymentSent` is pending processing, the
+ `ChannelManager` will fail to deserialize both on version 0.0.98 and later
+ versions. If you have such a `ChannelManager` available, a simple patch will
+ allow it to deserialize, please file an issue if you need assistance.
+
+# 0.0.98 - 2021-06-11
+
+0.0.98 should be considered a release candidate to the first alpha release of
+Rust-Lightning and the broader LDK. It represents several years of work
+designing and fine-tuning a flexible API for integrating lightning into any
+application. LDK should make it easy to build a lightning node or client which
+meets specific requirements that other lightning node software cannot. As
+lightning continues to evolve, and new use-cases for lightning develop, the API
+of LDK will continue to change and expand. However, starting with version 0.1,
+objects serialized with prior versions will be readable with the latest LDK.
+While Rust-Lightning is approaching the 0.1 milestone, language bindings
+components of LDK available at https://github.com/lightningdevkit are still of
+varying quality. Some are also approaching an 0.1 release, while others are
+still much more experimental. Please note that, at 0.0.98, using Rust-Lightning
+on mainnet is *strongly* discouraged.
cargo doc
cargo doc --document-private-items
cd fuzz && cargo check --features=stdin_fuzz
+cd ../lightning && cargo check --no-default-features --features=no_std
honggfuzz = { version = "0.5", optional = true }
libfuzzer-sys = { git = "https://github.com/rust-fuzz/libfuzzer-sys.git", optional = true }
-[patch.crates-io]
-# Rust-Secp256k1 PR 282. This patch should be dropped once that is merged.
-secp256k1 = { git = 'https://github.com/TheBlueMatt/rust-secp256k1', rev = 'c8615128097e0205dcf69e515e56fb57e5c97138' }
-# bitcoin_hashes PR 111 (without the top commit). This patch should be dropped once that is merged.
-bitcoin_hashes = { git = 'https://github.com/TheBlueMatt/bitcoin_hashes', rev = 'c90d26339a3e34fd2f942aa80298f410cc41b743' }
-
[build-dependencies]
cc = "1.0"
impl<Out: Output> Logger for TestLogger<Out> {
fn log(&self, record: &Record) {
write!(LockedWriteAdapter(&self.out),
- "{:<5} {} [{} : {}, {}] {}\n", record.level.to_string(), self.id, record.module_path, record.file, record.line, record.args)
+ "{:<5} {} [{} : {}] {}\n", record.level.to_string(), self.id, record.module_path, record.line, record.args)
.unwrap();
}
}
}
/// The `Validate` trait defines behavior for validating chain data.
-pub(crate) trait Validate {
+///
+/// This trait is sealed and not meant to be implemented outside of this crate.
+pub trait Validate: sealed::Validate {
/// The validated data wrapper which can be dereferenced to obtain the validated data.
type T: std::ops::Deref<Target = Self>;
}
}
+mod sealed {
+ /// Used to prevent implementing [`super::Validate`] outside the crate but still allow its use.
+ pub trait Validate {}
+
+ impl Validate for crate::BlockHeaderData {}
+ impl Validate for bitcoin::blockdata::block::Block {}
+}
+
/// The canonical `Poll` implementation used for a single `BlockSource`.
///
-/// Other `Poll` implementations must be built using `ChainPoller` as it provides the only means of
-/// validating chain data.
-pub struct ChainPoller<B: DerefMut<Target=T> + Sized , T: BlockSource> {
+/// Other `Poll` implementations should be built using `ChainPoller` as it provides the simplest way
+/// of validating chain data and checking consistency.
+pub struct ChainPoller<B: DerefMut<Target=T> + Sized, T: BlockSource> {
block_source: B,
network: Network,
}
-impl<B: DerefMut<Target=T> + Sized , T: BlockSource> ChainPoller<B, T> {
+impl<B: DerefMut<Target=T> + Sized, T: BlockSource> ChainPoller<B, T> {
/// Creates a new poller for the given block source.
///
/// If the `network` parameter is mainnet, then the difficulty between blocks is checked for
[dev-dependencies]
lightning = { version = "0.0.98", path = "../lightning", features = ["_test_utils"] }
-
-[lib]
-crate-type = ["cdylib", "rlib"]
-
/// Default expiry time as defined by [BOLT 11].
///
/// [BOLT 11]: https://github.com/lightningnetwork/lightning-rfc/blob/master/11-payment-encoding.md
-const DEFAULT_EXPIRY_TIME: u64 = 3600;
+pub const DEFAULT_EXPIRY_TIME: u64 = 3600;
/// Default minimum final CLTV expiry as defined by [BOLT 11].
///
+/// Note that this is *not* the same value as rust-lightning's minimum CLTV expiry, which is
+/// provided in [`MIN_FINAL_CLTV_EXPIRY`].
+///
/// [BOLT 11]: https://github.com/lightningnetwork/lightning-rfc/blob/master/11-payment-encoding.md
-const DEFAULT_MIN_FINAL_CLTV_EXPIRY: u64 = 18;
+/// [`MIN_FINAL_CLTV_EXPIRY`]: lightning::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY
+pub const DEFAULT_MIN_FINAL_CLTV_EXPIRY: u64 = 18;
/// This function is used as a static assert for the size of `SystemTime`. If the crate fails to
/// compile due to it this indicates that your system uses unexpected bounds for `SystemTime`. You
use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
use lightning::util::logger::Logger;
-use std::{task, thread};
+use std::task;
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
use std::sync::{Arc, Mutex};
// socket. To wake it up (without otherwise changing its state, we can push a value into this
// Sender.
read_waker: mpsc::Sender<()>,
- // When we are told by rust-lightning to disconnect, we can't return to rust-lightning until we
- // are sure we won't call any more read/write PeerManager functions with the same connection.
- // This is set to true if we're in such a condition (with disconnect checked before with the
- // top-level mutex held) and false when we can return.
- block_disconnect_socket: bool,
read_paused: bool,
rl_requested_disconnect: bool,
id: u64,
} }
}
- macro_rules! prepare_read_write_call {
- () => { {
- let mut us_lock = us.lock().unwrap();
- if us_lock.rl_requested_disconnect {
- shutdown_socket!("disconnect_socket() call from RL", Disconnect::CloseConnection);
- }
- us_lock.block_disconnect_socket = true;
- } }
- }
-
- let read_paused = us.lock().unwrap().read_paused;
+ let read_paused = {
+ let us_lock = us.lock().unwrap();
+ if us_lock.rl_requested_disconnect {
+ shutdown_socket!("disconnect_socket() call from RL", Disconnect::CloseConnection);
+ }
+ us_lock.read_paused
+ };
tokio::select! {
v = write_avail_receiver.recv() => {
assert!(v.is_some()); // We can't have dropped the sending end, its in the us Arc!
- prepare_read_write_call!();
if let Err(e) = peer_manager.write_buffer_space_avail(&mut our_descriptor) {
shutdown_socket!(e, Disconnect::CloseConnection);
}
- us.lock().unwrap().block_disconnect_socket = false;
},
_ = read_wake_receiver.recv() => {},
read = reader.read(&mut buf), if !read_paused => match read {
Ok(0) => shutdown_socket!("Connection closed", Disconnect::PeerDisconnected),
Ok(len) => {
- prepare_read_write_call!();
let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
let mut us_lock = us.lock().unwrap();
match read_res {
},
Err(e) => shutdown_socket!(e, Disconnect::CloseConnection),
}
- us_lock.block_disconnect_socket = false;
},
Err(e) => shutdown_socket!(e, Disconnect::PeerDisconnected),
},
(reader, write_receiver, read_receiver,
Arc::new(Mutex::new(Self {
writer: Some(writer), write_avail, read_waker, read_paused: false,
- block_disconnect_socket: false, rl_requested_disconnect: false,
+ rl_requested_disconnect: false,
id: ID_COUNTER.fetch_add(1, Ordering::AcqRel)
})))
}
}
fn disconnect_socket(&mut self) {
- {
- let mut us = self.conn.lock().unwrap();
- us.rl_requested_disconnect = true;
- us.read_paused = true;
- // Wake up the sending thread, assuming it is still alive
- let _ = us.write_avail.try_send(());
- // Happy-path return:
- if !us.block_disconnect_socket { return; }
- }
- while self.conn.lock().unwrap().block_disconnect_socket {
- thread::yield_now();
- }
+ let mut us = self.conn.lock().unwrap();
+ us.rl_requested_disconnect = true;
+ // Wake up the sending thread, assuming it is still alive
+ let _ = us.write_avail.try_send(());
}
}
impl Clone for SocketDescriptor {
# This is unsafe to use in production because it may result in the counterparty publishing taking our funds.
unsafe_revoked_tx_signing = []
unstable = []
+no_std = ["hashbrown"]
[dependencies]
bitcoin = "0.26"
+hashbrown = { version = "0.11", optional = true }
hex = { version = "0.3", optional = true }
regex = { version = "0.1.80", optional = true }
use util::events::EventHandler;
use prelude::*;
-use std::collections::{HashMap, hash_map};
use std::sync::RwLock;
use core::ops::Deref;
use util::events::Event;
use prelude::*;
-use std::collections::{HashMap, HashSet};
use core::{cmp, mem};
use std::io::Error;
use core::ops::Deref;
for update_step in self.updates.iter() {
update_step.write(w)?;
}
- write_tlv_fields!(w, {}, {});
+ write_tlv_fields!(w, {});
Ok(())
}
}
for _ in 0..len {
updates.push(Readable::read(r)?);
}
- read_tlv_fields!(r, {}, {});
+ read_tlv_fields!(r, {});
Ok(Self { update_id, updates })
}
}
pub(crate) source: HTLCSource
}
impl_writeable_tlv_based!(HTLCUpdate, {
- (0, payment_hash),
- (2, source),
-}, {
- (4, payment_preimage)
-}, {});
+ (0, payment_hash, required),
+ (2, source, required),
+ (4, payment_preimage, option),
+});
/// If an HTLC expires within this many blocks, don't try to claim it in a shared transaction,
/// instead claiming it in its own individual transaction.
htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>,
}
impl_writeable_tlv_based!(HolderSignedTx, {
- (0, txid),
- (2, revocation_key),
- (4, a_htlc_key),
- (6, b_htlc_key),
- (8, delayed_payment_key),
- (10, per_commitment_point),
- (12, feerate_per_kw),
-}, {}, {
- (14, htlc_outputs)
+ (0, txid, required),
+ (2, revocation_key, required),
+ (4, a_htlc_key, required),
+ (6, b_htlc_key, required),
+ (8, delayed_payment_key, required),
+ (10, per_commitment_point, required),
+ (12, feerate_per_kw, required),
+ (14, htlc_outputs, vec_type)
});
/// We use this to track counterparty commitment transactions and htlcs outputs and
}
}
write_tlv_fields!(w, {
- (0, self.counterparty_delayed_payment_base_key),
- (2, self.counterparty_htlc_base_key),
- (4, self.on_counterparty_tx_csv),
- }, {});
+ (0, self.counterparty_delayed_payment_base_key, required),
+ (2, self.counterparty_htlc_base_key, required),
+ (4, self.on_counterparty_tx_csv, required),
+ });
Ok(())
}
}
let mut counterparty_htlc_base_key = OptionDeserWrapper(None);
let mut on_counterparty_tx_csv: u16 = 0;
read_tlv_fields!(r, {
- (0, counterparty_delayed_payment_base_key),
- (2, counterparty_htlc_base_key),
- (4, on_counterparty_tx_csv),
- }, {});
+ (0, counterparty_delayed_payment_base_key, required),
+ (2, counterparty_htlc_base_key, required),
+ (4, on_counterparty_tx_csv, required),
+ });
CounterpartyCommitmentTransaction {
counterparty_delayed_payment_base_key: counterparty_delayed_payment_base_key.0.unwrap(),
counterparty_htlc_base_key: counterparty_htlc_base_key.0.unwrap(),
conf_threshold
}
- fn has_reached_confirmation_threshold(&self, height: u32) -> bool {
- height >= self.confirmation_threshold()
+ fn has_reached_confirmation_threshold(&self, best_block: &BestBlock) -> bool {
+ best_block.height() >= self.confirmation_threshold()
}
}
}
impl_writeable_tlv_based!(OnchainEventEntry, {
- (0, txid),
- (2, height),
- (4, event),
-}, {}, {});
+ (0, txid, required),
+ (2, height, required),
+ (4, event, required),
+});
impl_writeable_tlv_based_enum!(OnchainEvent,
(0, HTLCUpdate) => {
- (0, source),
- (2, payment_hash),
- }, {}, {},
+ (0, source, required),
+ (2, payment_hash, required),
+ },
(1, MaturingOutput) => {
- (0, descriptor),
- }, {}, {},
+ (0, descriptor, required),
+ },
;);
#[cfg_attr(any(test, feature = "fuzztarget", feature = "_test_utils"), derive(PartialEq))]
impl_writeable_tlv_based_enum!(ChannelMonitorUpdateStep,
(0, LatestHolderCommitmentTXInfo) => {
- (0, commitment_tx),
- }, {}, {
- (2, htlc_outputs),
+ (0, commitment_tx, required),
+ (2, htlc_outputs, vec_type),
},
(1, LatestCounterpartyCommitmentTXInfo) => {
- (0, commitment_txid),
- (2, commitment_number),
- (4, their_revocation_point),
- }, {}, {
- (6, htlc_outputs),
+ (0, commitment_txid, required),
+ (2, commitment_number, required),
+ (4, their_revocation_point, required),
+ (6, htlc_outputs, vec_type),
},
(2, PaymentPreimage) => {
- (0, payment_preimage),
- }, {}, {},
+ (0, payment_preimage, required),
+ },
(3, CommitmentSecret) => {
- (0, idx),
- (2, secret),
- }, {}, {},
+ (0, idx, required),
+ (2, secret, required),
+ },
(4, ChannelForceClosed) => {
- (0, should_broadcast),
- }, {}, {},
+ (0, should_broadcast, required),
+ },
;);
/// A ChannelMonitor handles chain events (blocks connected and disconnected) and generates
self.lockdown_from_offchain.write(writer)?;
self.holder_tx_signed.write(writer)?;
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
macro_rules! claim_htlcs {
($commitment_number: expr, $txid: expr) => {
let htlc_claim_reqs = self.get_counterparty_htlc_output_claim_reqs($commitment_number, $txid, None);
- self.onchain_tx_handler.update_claims_view(&Vec::new(), htlc_claim_reqs, self.best_block.height(), broadcaster, fee_estimator, logger);
+ self.onchain_tx_handler.update_claims_view(&Vec::new(), htlc_claim_reqs, self.best_block.height(), self.best_block.height(), broadcaster, fee_estimator, logger);
}
}
if let Some(txid) = self.current_counterparty_commitment_txid {
// *we* sign a holder commitment transaction, not when e.g. a watchtower broadcasts one of our
// holder commitment transactions.
if self.broadcasted_holder_revokable_script.is_some() {
- let (claim_reqs, _) = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, 0);
- self.onchain_tx_handler.update_claims_view(&Vec::new(), claim_reqs, self.best_block.height(), broadcaster, fee_estimator, logger);
+ // Assume that the broadcasted commitment transaction confirmed in the current best
+ // block. Even if not, its a reasonable metric for the bump criteria on the HTLC
+ // transactions.
+ let (claim_reqs, _) = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, self.best_block.height());
+ self.onchain_tx_handler.update_claims_view(&Vec::new(), claim_reqs, self.best_block.height(), self.best_block.height(), broadcaster, fee_estimator, logger);
if let Some(ref tx) = self.prev_holder_signed_commitment_tx {
- let (claim_reqs, _) = self.get_broadcasted_holder_claims(&tx, 0);
- self.onchain_tx_handler.update_claims_view(&Vec::new(), claim_reqs, self.best_block.height(), broadcaster, fee_estimator, logger);
+ let (claim_reqs, _) = self.get_broadcasted_holder_claims(&tx, self.best_block.height());
+ self.onchain_tx_handler.update_claims_view(&Vec::new(), claim_reqs, self.best_block.height(), self.best_block.height(), broadcaster, fee_estimator, logger);
}
}
}
// Last, track onchain revoked commitment transaction and fail backward outgoing HTLCs as payment path is broken
if !claimable_outpoints.is_empty() || per_commitment_option.is_some() { // ie we're confident this is actually ours
// We're definitely a counterparty commitment transaction!
- log_trace!(logger, "Got broadcast of revoked counterparty commitment transaction, going to generate general spend tx with {} inputs", claimable_outpoints.len());
+ log_error!(logger, "Got broadcast of revoked counterparty commitment transaction, going to generate general spend tx with {} inputs", claimable_outpoints.len());
for (idx, outp) in tx.output.iter().enumerate() {
watch_outputs.push((idx as u32, outp.clone()));
}
}
self.counterparty_commitment_txn_on_chain.insert(commitment_txid, commitment_number);
- log_trace!(logger, "Got broadcast of non-revoked counterparty commitment transaction {}", commitment_txid);
+ log_info!(logger, "Got broadcast of non-revoked counterparty commitment transaction {}", commitment_txid);
macro_rules! check_htlc_fails {
($txid: expr, $commitment_tx: expr, $id: tt) => {
let per_commitment_key = ignore_error!(SecretKey::from_slice(&secret));
let per_commitment_point = PublicKey::from_secret_key(&self.secp_ctx, &per_commitment_key);
- log_trace!(logger, "Counterparty HTLC broadcast {}:{}", htlc_txid, 0);
+ log_error!(logger, "Got broadcast of revoked counterparty HTLC transaction, spending {}:{}", htlc_txid, 0);
let revk_outp = RevokedOutput::build(per_commitment_point, self.counterparty_tx_cache.counterparty_delayed_payment_base_key, self.counterparty_tx_cache.counterparty_htlc_base_key, per_commitment_key, tx.output[0].value, self.counterparty_tx_cache.on_counterparty_tx_csv);
let justice_package = PackageTemplate::build_package(htlc_txid, 0, PackageSolvingData::RevokedOutput(revk_outp), height + self.counterparty_tx_cache.on_counterparty_tx_csv as u32, true, height);
let claimable_outpoints = vec!(justice_package);
// Returns (1) `PackageTemplate`s that can be given to the OnChainTxHandler, so that the handler can
// broadcast transactions claiming holder HTLC commitment outputs and (2) a holder revokable
// script so we can detect whether a holder transaction has been seen on-chain.
- fn get_broadcasted_holder_claims(&self, holder_tx: &HolderSignedTx, height: u32) -> (Vec<PackageTemplate>, Option<(Script, PublicKey, PublicKey)>) {
+ fn get_broadcasted_holder_claims(&self, holder_tx: &HolderSignedTx,
conf_height: u32) -> (Vec<PackageTemplate>, Option<(Script, PublicKey, PublicKey)>) {
let mut claim_requests = Vec::with_capacity(holder_tx.htlc_outputs.len());
let redeemscript = chan_utils::get_revokeable_redeemscript(&holder_tx.revocation_key, self.on_holder_tx_csv, &holder_tx.delayed_payment_key);
};
HolderHTLCOutput::build_accepted(payment_preimage, htlc.amount_msat)
};
- let htlc_package = PackageTemplate::build_package(holder_tx.txid, transaction_output_index, PackageSolvingData::HolderHTLCOutput(htlc_output), height, false, height);
+ let htlc_package = PackageTemplate::build_package(holder_tx.txid, transaction_output_index, PackageSolvingData::HolderHTLCOutput(htlc_output), htlc.cltv_expiry, false, conf_height);
claim_requests.push(htlc_package);
}
}
if self.current_holder_commitment_tx.txid == commitment_txid {
is_holder_tx = true;
- log_trace!(logger, "Got latest holder commitment tx broadcast, searching for available HTLCs to claim");
+ log_info!(logger, "Got broadcast of latest holder commitment tx {}, searching for available HTLCs to claim", commitment_txid);
let res = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, height);
let mut to_watch = self.get_broadcasted_holder_watch_outputs(&self.current_holder_commitment_tx, tx);
append_onchain_update!(res, to_watch);
} else if let &Some(ref holder_tx) = &self.prev_holder_signed_commitment_tx {
if holder_tx.txid == commitment_txid {
is_holder_tx = true;
- log_trace!(logger, "Got previous holder commitment tx broadcast, searching for available HTLCs to claim");
+ log_info!(logger, "Got broadcast of previous holder commitment tx {}, searching for available HTLCs to claim", commitment_txid);
let res = self.get_broadcasted_holder_claims(holder_tx, height);
let mut to_watch = self.get_broadcasted_holder_watch_outputs(holder_tx, tx);
append_onchain_update!(res, to_watch);
}
pub fn get_latest_holder_commitment_txn<L: Deref>(&mut self, logger: &L) -> Vec<Transaction> where L::Target: Logger {
- log_trace!(logger, "Getting signed latest holder commitment transaction!");
+ log_debug!(logger, "Getting signed latest holder commitment transaction!");
self.holder_tx_signed = true;
let commitment_tx = self.onchain_tx_handler.get_fully_signed_holder_tx(&self.funding_redeemscript);
let txid = commitment_tx.txid();
} else if htlc.0.cltv_expiry > self.best_block.height() + 1 {
// Don't broadcast HTLC-Timeout transactions immediately as they don't meet the
// current locktime requirements on-chain. We will broadcast them in
- // `block_confirmed` when `would_broadcast_at_height` returns true.
+ // `block_confirmed` when `should_broadcast_holder_commitment_txn` returns true.
// Note that we add + 1 as transactions are broadcastable when they can be
// confirmed in the next block.
continue;
#[cfg(any(test,feature = "unsafe_revoked_tx_signing"))]
/// Note that this includes possibly-locktimed-in-the-future transactions!
fn unsafe_get_latest_holder_commitment_txn<L: Deref>(&mut self, logger: &L) -> Vec<Transaction> where L::Target: Logger {
- log_trace!(logger, "Getting signed copy of latest holder commitment transaction!");
+ log_debug!(logger, "Getting signed copy of latest holder commitment transaction!");
let commitment_tx = self.onchain_tx_handler.get_fully_signed_copy_holder_tx(&self.funding_redeemscript);
let txid = commitment_tx.txid();
let mut holder_transactions = vec![commitment_tx];
if height > self.best_block.height() {
self.best_block = BestBlock::new(block_hash, height);
- self.block_confirmed(height, vec![], vec![], vec![], broadcaster, fee_estimator, logger)
- } else {
+ self.block_confirmed(height, vec![], vec![], vec![], &broadcaster, &fee_estimator, &logger)
+ } else if block_hash != self.best_block.block_hash() {
self.best_block = BestBlock::new(block_hash, height);
self.onchain_events_awaiting_threshold_conf.retain(|ref entry| entry.height <= height);
self.onchain_tx_handler.block_disconnected(height + 1, broadcaster, fee_estimator, logger);
Vec::new()
- }
+ } else { Vec::new() }
}
fn transactions_confirmed<B: Deref, F: Deref, L: Deref>(
self.is_paying_spendable_output(&tx, height, &logger);
}
- self.block_confirmed(height, txn_matched, watch_outputs, claimable_outpoints, broadcaster, fee_estimator, logger)
+ if height > self.best_block.height() {
+ self.best_block = BestBlock::new(block_hash, height);
+ }
+
+ self.block_confirmed(height, txn_matched, watch_outputs, claimable_outpoints, &broadcaster, &fee_estimator, &logger)
}
+ /// Update state for new block(s)/transaction(s) confirmed. Note that the caller must update
+ /// `self.best_block` before calling if a new best blockchain tip is available. More
+ /// concretely, `self.best_block` must never be at a lower height than `conf_height`, avoiding
+ /// complexity especially in `OnchainTx::update_claims_view`.
+ ///
+ /// `conf_height` should be set to the height at which any new transaction(s)/block(s) were
+ /// confirmed at, even if it is not the current best height.
fn block_confirmed<B: Deref, F: Deref, L: Deref>(
&mut self,
- height: u32,
+ conf_height: u32,
txn_matched: Vec<&Transaction>,
mut watch_outputs: Vec<TransactionOutputs>,
mut claimable_outpoints: Vec<PackageTemplate>,
- broadcaster: B,
- fee_estimator: F,
- logger: L,
+ broadcaster: &B,
+ fee_estimator: &F,
+ logger: &L,
) -> Vec<TransactionOutputs>
where
B::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
{
- let should_broadcast = self.would_broadcast_at_height(height, &logger);
+ debug_assert!(self.best_block.height() >= conf_height);
+
+ let should_broadcast = self.should_broadcast_holder_commitment_txn(logger);
if should_broadcast {
let funding_outp = HolderFundingOutput::build(self.funding_redeemscript.clone());
- let commitment_package = PackageTemplate::build_package(self.funding_info.0.txid.clone(), self.funding_info.0.index as u32, PackageSolvingData::HolderFundingOutput(funding_outp), height, false, height);
+ let commitment_package = PackageTemplate::build_package(self.funding_info.0.txid.clone(), self.funding_info.0.index as u32, PackageSolvingData::HolderFundingOutput(funding_outp), self.best_block.height(), false, self.best_block.height());
claimable_outpoints.push(commitment_package);
self.pending_monitor_events.push(MonitorEvent::CommitmentTxBroadcasted(self.funding_info.0));
let commitment_tx = self.onchain_tx_handler.get_fully_signed_holder_tx(&self.funding_redeemscript);
self.holder_tx_signed = true;
- let (mut new_outpoints, _) = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, height);
+ // Because we're broadcasting a commitment transaction, we should construct the package
+ // assuming it gets confirmed in the next block. Sadly, we have code which considers
+ // "not yet confirmed" things as discardable, so we cannot do that here.
+ let (mut new_outpoints, _) = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, self.best_block.height());
let new_outputs = self.get_broadcasted_holder_watch_outputs(&self.current_holder_commitment_tx, &commitment_tx);
if !new_outputs.is_empty() {
watch_outputs.push((self.current_holder_commitment_tx.txid.clone(), new_outputs));
self.onchain_events_awaiting_threshold_conf.drain(..).collect::<Vec<_>>();
let mut onchain_events_reaching_threshold_conf = Vec::new();
for entry in onchain_events_awaiting_threshold_conf {
- if entry.has_reached_confirmation_threshold(height) {
+ if entry.has_reached_confirmation_threshold(&self.best_block) {
onchain_events_reaching_threshold_conf.push(entry);
} else {
self.onchain_events_awaiting_threshold_conf.push(entry);
matured_htlcs.push(source.clone());
}
- log_trace!(logger, "HTLC {} failure update has got enough confirmations to be passed upstream", log_bytes!(payment_hash.0));
+ log_debug!(logger, "HTLC {} failure update has got enough confirmations to be passed upstream", log_bytes!(payment_hash.0));
self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
payment_hash: payment_hash,
payment_preimage: None,
}));
},
OnchainEvent::MaturingOutput { descriptor } => {
- log_trace!(logger, "Descriptor {} has got enough confirmations to be passed upstream", log_spendable!(descriptor));
+ log_debug!(logger, "Descriptor {} has got enough confirmations to be passed upstream", log_spendable!(descriptor));
self.pending_events.push(Event::SpendableOutputs {
outputs: vec![descriptor]
});
}
}
- self.onchain_tx_handler.update_claims_view(&txn_matched, claimable_outpoints, height, &&*broadcaster, &&*fee_estimator, &&*logger);
+ self.onchain_tx_handler.update_claims_view(&txn_matched, claimable_outpoints, conf_height, self.best_block.height(), broadcaster, fee_estimator, logger);
// Determine new outputs to watch by comparing against previously known outputs to watch,
// updating the latter in the process.
false
}
- fn would_broadcast_at_height<L: Deref>(&self, height: u32, logger: &L) -> bool where L::Target: Logger {
+ fn should_broadcast_holder_commitment_txn<L: Deref>(&self, logger: &L) -> bool where L::Target: Logger {
// We need to consider all HTLCs which are:
// * in any unrevoked counterparty commitment transaction, as they could broadcast said
// transactions and we'd end up in a race, or
// to the source, and if we don't fail the channel we will have to ensure that the next
// updates that peer sends us are update_fails, failing the channel if not. It's probably
// easier to just fail the channel as this case should be rare enough anyway.
+ let height = self.best_block.height();
macro_rules! scan_commitment {
($htlcs: expr, $holder_tx: expr) => {
for ref htlc in $htlcs {
height,
event: OnchainEvent::HTLCUpdate { source: source, payment_hash: payment_hash },
};
- log_info!(logger, "Failing HTLC with payment_hash {} timeout by a spend tx, waiting for confirmation (at height{})", log_bytes!(payment_hash.0), entry.confirmation_threshold());
+ log_info!(logger, "Failing HTLC with payment_hash {} timeout by a spend tx, waiting for confirmation (at height {})", log_bytes!(payment_hash.0), entry.confirmation_threshold());
self.onchain_events_awaiting_threshold_conf.push(entry);
}
}
height: height,
event: OnchainEvent::MaturingOutput { descriptor: spendable_output.clone() },
};
- log_trace!(logger, "Maturing {} until {}", log_spendable!(spendable_output), entry.confirmation_threshold());
+ log_info!(logger, "Received spendable output {}, spendable at height {}", log_spendable!(spendable_output), entry.confirmation_threshold());
self.onchain_events_awaiting_threshold_conf.push(entry);
}
}
let lockdown_from_offchain = Readable::read(reader)?;
let holder_tx_signed = Readable::read(reader)?;
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&keys_manager.get_secure_random_bytes());
use ln::msgs::UnsignedChannelAnnouncement;
use prelude::*;
-use std::collections::HashSet;
use core::sync::atomic::{AtomicUsize, Ordering};
use std::io::Error;
use ln::msgs::{DecodeError, MAX_VALUE_MSAT};
}
impl_writeable_tlv_based!(DelayedPaymentOutputDescriptor, {
- (0, outpoint),
- (2, per_commitment_point),
- (4, to_self_delay),
- (6, output),
- (8, revocation_pubkey),
- (10, channel_keys_id),
- (12, channel_value_satoshis),
-}, {}, {});
+ (0, outpoint, required),
+ (2, per_commitment_point, required),
+ (4, to_self_delay, required),
+ (6, output, required),
+ (8, revocation_pubkey, required),
+ (10, channel_keys_id, required),
+ (12, channel_value_satoshis, required),
+});
/// Information about a spendable output to our "payment key". See
/// SpendableOutputDescriptor::StaticPaymentOutput for more details on how to spend this.
pub const MAX_WITNESS_LENGTH: usize = 1 + 73 + 34;
}
impl_writeable_tlv_based!(StaticPaymentOutputDescriptor, {
- (0, outpoint),
- (2, output),
- (4, channel_keys_id),
- (6, channel_value_satoshis),
-}, {}, {});
+ (0, outpoint, required),
+ (2, output, required),
+ (4, channel_keys_id, required),
+ (6, channel_value_satoshis, required),
+});
/// When on-chain outputs are created by rust-lightning (which our counterparty is not able to
/// claim at any point in the future) an event is generated which you must track and be able to
impl_writeable_tlv_based_enum!(SpendableOutputDescriptor,
(0, StaticOutput) => {
- (0, outpoint),
- (2, output),
- }, {}, {},
+ (0, outpoint, required),
+ (2, output, required),
+ },
;
(1, DelayedPaymentOutput),
(2, StaticPaymentOutput),
self.channel_value_satoshis.write(writer)?;
self.channel_keys_id.write(writer)?;
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
&htlc_base_key);
let keys_id = Readable::read(reader)?;
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
Ok(InMemorySigner {
funding_key,
use prelude::*;
use alloc::collections::BTreeMap;
-use std::collections::HashMap;
use core::cmp;
use core::ops::Deref;
use core::mem::replace;
}
impl_writeable_tlv_based!(OnchainEventEntry, {
- (0, txid),
- (2, height),
- (4, event),
-}, {}, {});
+ (0, txid, required),
+ (2, height, required),
+ (4, event, required),
+});
impl_writeable_tlv_based_enum!(OnchainEvent,
(0, Claim) => {
- (0, claim_request),
- }, {}, {},
+ (0, claim_request, required),
+ },
(1, ContentiousOutpoint) => {
- (0, package),
- }, {}, {},
+ (0, package, required),
+ },
;);
impl Readable for Option<Vec<Option<(usize, Signature)>>> {
entry.write(writer)?;
}
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
}
onchain_events_awaiting_threshold_conf.push(Readable::read(reader)?);
}
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&keys_manager.get_secure_random_bytes());
/// (CSV or CLTV following cases). In case of high-fee spikes, claim tx may stuck in the mempool, so you need to bump its feerate quickly using Replace-By-Fee or Child-Pay-For-Parent.
/// Panics if there are signing errors, because signing operations in reaction to on-chain events
/// are not expected to fail, and if they do, we may lose funds.
- fn generate_claim_tx<F: Deref, L: Deref>(&mut self, height: u32, cached_request: &PackageTemplate, fee_estimator: &F, logger: &L) -> Option<(Option<u32>, u64, Transaction)>
+ fn generate_claim_tx<F: Deref, L: Deref>(&mut self, cur_height: u32, cached_request: &PackageTemplate, fee_estimator: &F, logger: &L) -> Option<(Option<u32>, u64, Transaction)>
where F::Target: FeeEstimator,
L::Target: Logger,
{
// Compute new height timer to decide when we need to regenerate a new bumped version of the claim tx (if we
// didn't receive confirmation of it before, or not enough reorg-safe depth on top of it).
- let new_timer = Some(cached_request.get_height_timer(height));
+ let new_timer = Some(cached_request.get_height_timer(cur_height));
if cached_request.is_malleable() {
let predicted_weight = cached_request.package_weight(&self.destination_script);
if let Some((output_value, new_feerate)) = cached_request.compute_package_output(predicted_weight, fee_estimator, logger) {
/// for this channel, provide new relevant on-chain transactions and/or new claim requests.
/// Formerly this was named `block_connected`, but it is now also used for claiming an HTLC output
/// if we receive a preimage after force-close.
- pub(crate) fn update_claims_view<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[&Transaction], requests: Vec<PackageTemplate>, height: u32, broadcaster: &B, fee_estimator: &F, logger: &L)
+ /// `conf_height` represents the height at which the transactions in `txn_matched` were
+ /// confirmed. This does not need to equal the current blockchain tip height, which should be
+ /// provided via `cur_height`, however it must never be higher than `cur_height`.
+ pub(crate) fn update_claims_view<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[&Transaction], requests: Vec<PackageTemplate>, conf_height: u32, cur_height: u32, broadcaster: &B, fee_estimator: &F, logger: &L)
where B::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
{
- log_trace!(logger, "Updating claims view at height {} with {} matched transactions and {} claim requests", height, txn_matched.len(), requests.len());
+ log_debug!(logger, "Updating claims view at height {} with {} matched transactions in block {} and {} claim requests", cur_height, txn_matched.len(), conf_height, requests.len());
let mut preprocessed_requests = Vec::with_capacity(requests.len());
let mut aggregated_request = None;
for req in requests {
// Don't claim a outpoint twice that would be bad for privacy and may uselessly lock a CPFP input for a while
if let Some(_) = self.claimable_outpoints.get(req.outpoints()[0]) {
- log_trace!(logger, "Ignoring second claim for outpoint {}:{}, already registered its claiming request", req.outpoints()[0].txid, req.outpoints()[0].vout);
+ log_info!(logger, "Ignoring second claim for outpoint {}:{}, already registered its claiming request", req.outpoints()[0].txid, req.outpoints()[0].vout);
} else {
let timelocked_equivalent_package = self.locktimed_packages.iter().map(|v| v.1.iter()).flatten()
.find(|locked_package| locked_package.outpoints() == req.outpoints());
if let Some(package) = timelocked_equivalent_package {
- log_trace!(logger, "Ignoring second claim for outpoint {}:{}, we already have one which we're waiting on a timelock at {} for.",
+ log_info!(logger, "Ignoring second claim for outpoint {}:{}, we already have one which we're waiting on a timelock at {} for.",
req.outpoints()[0].txid, req.outpoints()[0].vout, package.package_timelock());
continue;
}
- if req.package_timelock() > height + 1 {
- log_debug!(logger, "Delaying claim of package until its timelock at {} (current height {}), the following outpoints are spent:", req.package_timelock(), height);
+ if req.package_timelock() > cur_height + 1 {
+ log_info!(logger, "Delaying claim of package until its timelock at {} (current height {}), the following outpoints are spent:", req.package_timelock(), cur_height);
for outpoint in req.outpoints() {
- log_debug!(logger, " Outpoint {}", outpoint);
+ log_info!(logger, " Outpoint {}", outpoint);
}
self.locktimed_packages.entry(req.package_timelock()).or_insert(Vec::new()).push(req);
continue;
}
- log_trace!(logger, "Test if outpoint can be aggregated with expiration {} against {}", req.timelock(), height + CLTV_SHARED_CLAIM_BUFFER);
- if req.timelock() <= height + CLTV_SHARED_CLAIM_BUFFER || !req.aggregable() {
+ log_trace!(logger, "Test if outpoint can be aggregated with expiration {} against {}", req.timelock(), cur_height + CLTV_SHARED_CLAIM_BUFFER);
+ if req.timelock() <= cur_height + CLTV_SHARED_CLAIM_BUFFER || !req.aggregable() {
// Don't aggregate if outpoint package timelock is soon or marked as non-aggregable
preprocessed_requests.push(req);
} else if aggregated_request.is_none() {
preprocessed_requests.push(req);
}
- // Claim everything up to and including height + 1
- let remaining_locked_packages = self.locktimed_packages.split_off(&(height + 2));
+ // Claim everything up to and including cur_height + 1
+ let remaining_locked_packages = self.locktimed_packages.split_off(&(cur_height + 2));
for (pop_height, mut entry) in self.locktimed_packages.iter_mut() {
log_trace!(logger, "Restoring delayed claim of package(s) at their timelock at {}.", pop_height);
preprocessed_requests.append(&mut entry);
// Generate claim transactions and track them to bump if necessary at
// height timer expiration (i.e in how many blocks we're going to take action).
for mut req in preprocessed_requests {
- if let Some((new_timer, new_feerate, tx)) = self.generate_claim_tx(height, &req, &*fee_estimator, &*logger) {
+ if let Some((new_timer, new_feerate, tx)) = self.generate_claim_tx(cur_height, &req, &*fee_estimator, &*logger) {
req.set_timer(new_timer);
req.set_feerate(new_feerate);
let txid = tx.txid();
for k in req.outpoints() {
- log_trace!(logger, "Registering claiming request for {}:{}", k.txid, k.vout);
- self.claimable_outpoints.insert(k.clone(), (txid, height));
+ log_info!(logger, "Registering claiming request for {}:{}", k.txid, k.vout);
+ self.claimable_outpoints.insert(k.clone(), (txid, conf_height));
}
self.pending_claim_requests.insert(txid, req);
- log_trace!(logger, "Broadcasting onchain {}", log_tx!(tx));
+ log_info!(logger, "Broadcasting onchain {}", log_tx!(tx));
broadcaster.broadcast_transaction(&tx);
}
}
() => {
let entry = OnchainEventEntry {
txid: tx.txid(),
- height,
+ height: conf_height,
event: OnchainEvent::Claim { claim_request: first_claim_txid_height.0.clone() }
};
if !self.onchain_events_awaiting_threshold_conf.contains(&entry) {
for package in claimed_outputs_material.drain(..) {
let entry = OnchainEventEntry {
txid: tx.txid(),
- height,
+ height: conf_height,
event: OnchainEvent::ContentiousOutpoint { package },
};
if !self.onchain_events_awaiting_threshold_conf.contains(&entry) {
let onchain_events_awaiting_threshold_conf =
self.onchain_events_awaiting_threshold_conf.drain(..).collect::<Vec<_>>();
for entry in onchain_events_awaiting_threshold_conf {
- if entry.has_reached_confirmation_threshold(height) {
+ if entry.has_reached_confirmation_threshold(cur_height) {
match entry.event {
OnchainEvent::Claim { claim_request } => {
// We may remove a whole set of claim outpoints here, as these one may have
// been aggregated in a single tx and claimed so atomically
if let Some(request) = self.pending_claim_requests.remove(&claim_request) {
for outpoint in request.outpoints() {
+ log_debug!(logger, "Removing claim tracking for {} due to maturation of claim tx {}.", outpoint, claim_request);
self.claimable_outpoints.remove(&outpoint);
}
}
},
OnchainEvent::ContentiousOutpoint { package } => {
+ log_debug!(logger, "Removing claim tracking due to maturation of claim tx for outpoints:");
+ log_debug!(logger, " {:?}", package.outpoints());
self.claimable_outpoints.remove(&package.outpoints()[0]);
}
}
// Check if any pending claim request must be rescheduled
for (first_claim_txid, ref request) in self.pending_claim_requests.iter() {
if let Some(h) = request.timer() {
- if height >= h {
+ if cur_height >= h {
bump_candidates.insert(*first_claim_txid, (*request).clone());
}
}
// Build, bump and rebroadcast tx accordingly
log_trace!(logger, "Bumping {} candidates", bump_candidates.len());
for (first_claim_txid, request) in bump_candidates.iter() {
- if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &request, &*fee_estimator, &*logger) {
- log_trace!(logger, "Broadcasting onchain {}", log_tx!(bump_tx));
+ if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(cur_height, &request, &*fee_estimator, &*logger) {
+ log_info!(logger, "Broadcasting RBF-bumped onchain {}", log_tx!(bump_tx));
broadcaster.broadcast_transaction(&bump_tx);
if let Some(request) = self.pending_claim_requests.get_mut(first_claim_txid) {
request.set_timer(new_timer);
}
impl_writeable_tlv_based!(RevokedOutput, {
- (0, per_commitment_point),
- (2, counterparty_delayed_payment_base_key),
- (4, counterparty_htlc_base_key),
- (6, per_commitment_key),
- (8, weight),
- (10, amount),
- (12, on_counterparty_tx_csv),
-}, {}, {});
+ (0, per_commitment_point, required),
+ (2, counterparty_delayed_payment_base_key, required),
+ (4, counterparty_htlc_base_key, required),
+ (6, per_commitment_key, required),
+ (8, weight, required),
+ (10, amount, required),
+ (12, on_counterparty_tx_csv, required),
+});
/// A struct to describe a revoked offered output and corresponding information to generate a
/// solving witness.
}
impl_writeable_tlv_based!(RevokedHTLCOutput, {
- (0, per_commitment_point),
- (2, counterparty_delayed_payment_base_key),
- (4, counterparty_htlc_base_key),
- (6, per_commitment_key),
- (8, weight),
- (10, amount),
- (12, htlc),
-}, {}, {});
+ (0, per_commitment_point, required),
+ (2, counterparty_delayed_payment_base_key, required),
+ (4, counterparty_htlc_base_key, required),
+ (6, per_commitment_key, required),
+ (8, weight, required),
+ (10, amount, required),
+ (12, htlc, required),
+});
/// A struct to describe a HTLC output on a counterparty commitment transaction.
///
}
impl_writeable_tlv_based!(CounterpartyOfferedHTLCOutput, {
- (0, per_commitment_point),
- (2, counterparty_delayed_payment_base_key),
- (4, counterparty_htlc_base_key),
- (6, preimage),
- (8, htlc),
-}, {}, {});
+ (0, per_commitment_point, required),
+ (2, counterparty_delayed_payment_base_key, required),
+ (4, counterparty_htlc_base_key, required),
+ (6, preimage, required),
+ (8, htlc, required),
+});
/// A struct to describe a HTLC output on a counterparty commitment transaction.
///
}
impl_writeable_tlv_based!(CounterpartyReceivedHTLCOutput, {
- (0, per_commitment_point),
- (2, counterparty_delayed_payment_base_key),
- (4, counterparty_htlc_base_key),
- (6, htlc),
-}, {}, {});
+ (0, per_commitment_point, required),
+ (2, counterparty_delayed_payment_base_key, required),
+ (4, counterparty_htlc_base_key, required),
+ (6, htlc, required),
+});
/// A struct to describe a HTLC output on holder commitment transaction.
///
}
impl_writeable_tlv_based!(HolderHTLCOutput, {
- (0, amount),
- (2, cltv_expiry),
-}, {
- (4, preimage),
-}, {});
+ (0, amount, required),
+ (2, cltv_expiry, required),
+ (4, preimage, option)
+});
/// A struct to describe the channel output on the funding transaction.
///
}
impl_writeable_tlv_based!(HolderFundingOutput, {
- (0, funding_redeemscript),
-}, {}, {});
+ (0, funding_redeemscript, required),
+});
/// A wrapper encapsulating all in-protocol differing outputs types.
///
});
}
for (i, (outpoint, out)) in self.inputs.iter().enumerate() {
- log_trace!(logger, "Adding claiming input for outpoint {}:{}", outpoint.txid, outpoint.vout);
+ log_debug!(logger, "Adding claiming input for outpoint {}:{}", outpoint.txid, outpoint.vout);
if !out.finalize_input(&mut bumped_tx, i, onchain_handler) { return None; }
}
- log_trace!(logger, "Finalized transaction {} ready to broadcast", bumped_tx.txid());
+ log_debug!(logger, "Finalized transaction {} ready to broadcast", bumped_tx.txid());
return Some(bumped_tx);
},
PackageMalleability::Untractable => {
debug_assert_eq!(value, 0, "value is ignored for non-malleable packages, should be zero to ensure callsites are correct");
if let Some((outpoint, outp)) = self.inputs.first() {
if let Some(final_tx) = outp.get_finalized_tx(outpoint, onchain_handler) {
- log_trace!(logger, "Adding claiming input for outpoint {}:{}", outpoint.txid, outpoint.vout);
- log_trace!(logger, "Finalized transaction {} ready to broadcast", final_tx.txid());
+ log_debug!(logger, "Adding claiming input for outpoint {}:{}", outpoint.txid, outpoint.vout);
+ log_debug!(logger, "Finalized transaction {} ready to broadcast", final_tx.txid());
return Some(final_tx);
}
return None;
rev_outp.write(writer)?;
}
write_tlv_fields!(writer, {
- (0, self.soonest_conf_deadline),
- (2, self.feerate_previous),
- (4, self.height_original),
- }, { (6, self.height_timer) });
+ (0, self.soonest_conf_deadline, required),
+ (2, self.feerate_previous, required),
+ (4, self.height_original, required),
+ (6, self.height_timer, option)
+ });
Ok(())
}
}
let mut height_timer = None;
let mut height_original = 0;
read_tlv_fields!(reader, {
- (0, soonest_conf_deadline),
- (2, feerate_previous),
- (4, height_original)
- }, { (6, height_timer) });
+ (0, soonest_conf_deadline, required),
+ (2, feerate_previous, required),
+ (4, height_original, required),
+ (6, height_timer, option),
+ });
Ok(PackageTemplate {
inputs,
malleability,
// ...else just increase the previous feerate by 25% (because that's a nice number)
let new_fee = previous_feerate * (predicted_weight as u64) / 750;
if input_amounts <= new_fee {
- log_trace!(logger, "Can't 25% bump new claiming tx, amount {} is too small", input_amounts);
+ log_warn!(logger, "Can't 25% bump new claiming tx, amount {} is too small", input_amounts);
return None;
}
new_fee
}
} else {
- log_trace!(logger, "Can't new-estimation bump new claiming tx, amount {} is too small", input_amounts);
+ log_warn!(logger, "Can't new-estimation bump new claiming tx, amount {} is too small", input_amounts);
return None;
};
pub mod routing;
mod prelude {
- pub use alloc::{vec, vec::Vec, string::String};
-}
\ No newline at end of file
+ #[cfg(feature = "hashbrown")]
+ extern crate hashbrown;
+
+ pub use alloc::{vec, vec::Vec, string::String, collections::VecDeque};
+ #[cfg(not(feature = "hashbrown"))]
+ pub use std::collections::{HashMap, HashSet, hash_map};
+ #[cfg(feature = "hashbrown")]
+ pub use self::hashbrown::{HashMap, HashSet, hash_map};
+}
writer.write_all(secret)?;
writer.write_all(&byte_utils::be64_to_array(*idx))?;
}
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
}
*secret = Readable::read(reader)?;
*idx = Readable::read(reader)?;
}
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
Ok(Self { old_secrets })
}
}
}
impl_writeable_tlv_based!(TxCreationKeys, {
- (0, per_commitment_point),
- (2, revocation_key),
- (4, broadcaster_htlc_key),
- (6, countersignatory_htlc_key),
- (8, broadcaster_delayed_payment_key),
-}, {}, {});
+ (0, per_commitment_point, required),
+ (2, revocation_key, required),
+ (4, broadcaster_htlc_key, required),
+ (6, countersignatory_htlc_key, required),
+ (8, broadcaster_delayed_payment_key, required),
+});
/// One counterparty's public keys which do not change over the life of a channel.
#[derive(Clone, PartialEq)]
}
impl_writeable_tlv_based!(ChannelPublicKeys, {
- (0, funding_pubkey),
- (2, revocation_basepoint),
- (4, payment_point),
- (6, delayed_payment_basepoint),
- (8, htlc_basepoint),
-}, {}, {});
+ (0, funding_pubkey, required),
+ (2, revocation_basepoint, required),
+ (4, payment_point, required),
+ (6, delayed_payment_basepoint, required),
+ (8, htlc_basepoint, required),
+});
impl TxCreationKeys {
/// Create per-state keys from channel base points and the per-commitment point.
}
impl_writeable_tlv_based!(HTLCOutputInCommitment, {
- (0, offered),
- (2, amount_msat),
- (4, cltv_expiry),
- (6, payment_hash),
-}, {
- (8, transaction_output_index)
-}, {});
+ (0, offered, required),
+ (2, amount_msat, required),
+ (4, cltv_expiry, required),
+ (6, payment_hash, required),
+ (8, transaction_output_index, option),
+});
#[inline]
pub(crate) fn get_htlc_redeemscript_with_explicit_keys(htlc: &HTLCOutputInCommitment, broadcaster_htlc_key: &PublicKey, countersignatory_htlc_key: &PublicKey, revocation_key: &PublicKey) -> Script {
}
impl_writeable_tlv_based!(CounterpartyChannelTransactionParameters, {
- (0, pubkeys),
- (2, selected_contest_delay),
-}, {}, {});
+ (0, pubkeys, required),
+ (2, selected_contest_delay, required),
+});
impl_writeable_tlv_based!(ChannelTransactionParameters, {
- (0, holder_pubkeys),
- (2, holder_selected_contest_delay),
- (4, is_outbound_from_holder),
-}, {
- (6, counterparty_parameters),
- (8, funding_outpoint),
-}, {});
+ (0, holder_pubkeys, required),
+ (2, holder_selected_contest_delay, required),
+ (4, is_outbound_from_holder, required),
+ (6, counterparty_parameters, option),
+ (8, funding_outpoint, option),
+});
/// Static channel fields used to build transactions given per-commitment fields, organized by
/// broadcaster/countersignatory.
}
impl_writeable_tlv_based!(HolderCommitmentTransaction, {
- (0, inner),
- (2, counterparty_sig),
- (4, holder_sig_first),
-}, {}, {
- (6, counterparty_htlc_sigs),
+ (0, inner, required),
+ (2, counterparty_sig, required),
+ (4, holder_sig_first, required),
+ (6, counterparty_htlc_sigs, vec_type),
});
impl HolderCommitmentTransaction {
}
impl_writeable_tlv_based!(BuiltCommitmentTransaction, {
- (0, transaction),
- (2, txid)
-}, {}, {});
+ (0, transaction, required),
+ (2, txid, required),
+});
impl BuiltCommitmentTransaction {
/// Get the SIGHASH_ALL sighash value of the transaction.
}
impl_writeable_tlv_based!(CommitmentTransaction, {
- (0, commitment_number),
- (2, to_broadcaster_value_sat),
- (4, to_countersignatory_value_sat),
- (6, feerate_per_kw),
- (8, keys),
- (10, built),
-}, {}, {
- (12, htlcs),
+ (0, commitment_number, required),
+ (2, to_broadcaster_value_sat, required),
+ (4, to_countersignatory_value_sat, required),
+ (6, feerate_per_kw, required),
+ (8, keys, required),
+ (10, built, required),
+ (12, htlcs, vec_type),
});
impl CommitmentTransaction {
use util::test_utils;
use prelude::*;
-use std::collections::HashMap;
use std::sync::{Arc, Mutex};
// If persister_fail is true, we have the persister return a PermanentFailure
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
check_added_monitors!(nodes[1], 1);
- // Attempt to forward a third payment but fail due to the second channel being unavailable
- // for forwarding.
+ // Forward a third payment which will also be added to the holding cell, despite the channel
+ // being paused waiting a monitor update.
let (_, payment_hash_3, payment_secret_3) = get_payment_preimage_hash!(nodes[2]);
{
let net_graph_msg_handler = &nodes[0].net_graph_msg_handler;
commitment_signed_dance!(nodes[1], nodes[0], send_event.commitment_msg, false, true);
check_added_monitors!(nodes[1], 0);
- let mut events_2 = nodes[1].node.get_and_clear_pending_msg_events();
- assert_eq!(events_2.len(), 1);
- match events_2.remove(0) {
- MessageSendEvent::UpdateHTLCs { node_id, updates } => {
- assert_eq!(node_id, nodes[0].node.get_our_node_id());
- assert!(updates.update_fulfill_htlcs.is_empty());
- assert_eq!(updates.update_fail_htlcs.len(), 1);
- assert!(updates.update_fail_malformed_htlcs.is_empty());
- assert!(updates.update_add_htlcs.is_empty());
- assert!(updates.update_fee.is_none());
-
- nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fail_htlcs[0]);
- commitment_signed_dance!(nodes[0], nodes[1], updates.commitment_signed, false, true);
-
- let msg_events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(msg_events.len(), 1);
- match msg_events[0] {
- MessageSendEvent::PaymentFailureNetworkUpdate { update: msgs::HTLCFailChannelUpdate::ChannelUpdateMessage { ref msg }} => {
- assert_eq!(msg.contents.short_channel_id, chan_2.0.contents.short_channel_id);
- assert_eq!(msg.contents.flags & 2, 2); // temp disabled
- },
- _ => panic!("Unexpected event"),
- }
-
- let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- if let Event::PaymentFailed { payment_hash, rejected_by_dest, .. } = events[0] {
- assert_eq!(payment_hash, payment_hash_3);
- assert!(!rejected_by_dest);
- } else { panic!("Unexpected event!"); }
- },
- _ => panic!("Unexpected event type!"),
- };
+ // Call forward_pending_htlcs and check that the new HTLC was simply added to the holding cell
+ // and not forwarded.
+ expect_pending_htlcs_forwardable!(nodes[1]);
+ check_added_monitors!(nodes[1], 0);
+ assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
let (payment_preimage_4, payment_hash_4) = if test_ignore_second_cs {
// Try to route another payment backwards from 2 to make sure 1 holds off on responding
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Previous monitor update failure prevented generation of RAA".to_string(), 1);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
- assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
(Some(payment_preimage_4), Some(payment_hash_4))
} else { (None, None) };
nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &messages_a.0);
commitment_signed_dance!(nodes[0], nodes[1], messages_a.1, false);
- let events_4 = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events_4.len(), 1);
- if let Event::PaymentFailed { payment_hash, rejected_by_dest, .. } = events_4[0] {
- assert_eq!(payment_hash, payment_hash_1);
- assert!(rejected_by_dest);
- } else { panic!("Unexpected event!"); }
+ expect_payment_failed!(nodes[0], payment_hash_1, true);
nodes[2].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &send_event_b.msgs[0]);
+ let as_cs;
if test_ignore_second_cs {
nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &send_event_b.commitment_msg);
check_added_monitors!(nodes[2], 1);
nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &bs_revoke_and_ack);
check_added_monitors!(nodes[1], 1);
- let as_cs = get_htlc_update_msgs!(nodes[1], nodes[2].node.get_our_node_id());
- assert!(as_cs.update_add_htlcs.is_empty());
- assert!(as_cs.update_fail_htlcs.is_empty());
- assert!(as_cs.update_fail_malformed_htlcs.is_empty());
- assert!(as_cs.update_fulfill_htlcs.is_empty());
- assert!(as_cs.update_fee.is_none());
+ as_cs = get_htlc_update_msgs!(nodes[1], nodes[2].node.get_our_node_id());
nodes[1].node.handle_commitment_signed(&nodes[2].node.get_our_node_id(), &bs_cs.commitment_signed);
check_added_monitors!(nodes[1], 1);
- let as_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[2].node.get_our_node_id());
-
- nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &as_cs.commitment_signed);
+ } else {
+ nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &send_event_b.commitment_msg);
check_added_monitors!(nodes[2], 1);
- let bs_second_raa = get_event_msg!(nodes[2], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id());
- nodes[2].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &as_raa);
- check_added_monitors!(nodes[2], 1);
- assert!(nodes[2].node.get_and_clear_pending_msg_events().is_empty());
+ let bs_revoke_and_commit = nodes[2].node.get_and_clear_pending_msg_events();
+ assert_eq!(bs_revoke_and_commit.len(), 2);
+ match bs_revoke_and_commit[0] {
+ MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
+ assert_eq!(*node_id, nodes[1].node.get_our_node_id());
+ nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &msg);
+ check_added_monitors!(nodes[1], 1);
+ },
+ _ => panic!("Unexpected event"),
+ }
- nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &bs_second_raa);
- check_added_monitors!(nodes[1], 1);
- assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
- } else {
- commitment_signed_dance!(nodes[2], nodes[1], send_event_b.commitment_msg, false);
+ as_cs = get_htlc_update_msgs!(nodes[1], nodes[2].node.get_our_node_id());
+
+ match bs_revoke_and_commit[1] {
+ MessageSendEvent::UpdateHTLCs { ref node_id, ref updates } => {
+ assert_eq!(*node_id, nodes[1].node.get_our_node_id());
+ assert!(updates.update_add_htlcs.is_empty());
+ assert!(updates.update_fail_htlcs.is_empty());
+ assert!(updates.update_fail_malformed_htlcs.is_empty());
+ assert!(updates.update_fulfill_htlcs.is_empty());
+ assert!(updates.update_fee.is_none());
+ nodes[1].node.handle_commitment_signed(&nodes[2].node.get_our_node_id(), &updates.commitment_signed);
+ check_added_monitors!(nodes[1], 1);
+ },
+ _ => panic!("Unexpected event"),
+ }
}
+ assert_eq!(as_cs.update_add_htlcs.len(), 1);
+ assert!(as_cs.update_fail_htlcs.is_empty());
+ assert!(as_cs.update_fail_malformed_htlcs.is_empty());
+ assert!(as_cs.update_fulfill_htlcs.is_empty());
+ assert!(as_cs.update_fee.is_none());
+ let as_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[2].node.get_our_node_id());
+
+
+ nodes[2].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &as_cs.update_add_htlcs[0]);
+ nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &as_cs.commitment_signed);
+ check_added_monitors!(nodes[2], 1);
+ let bs_second_raa = get_event_msg!(nodes[2], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id());
+
+ nodes[2].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &as_raa);
+ check_added_monitors!(nodes[2], 1);
+ let bs_second_cs = get_htlc_update_msgs!(nodes[2], nodes[1].node.get_our_node_id());
+
+ nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &bs_second_raa);
+ check_added_monitors!(nodes[1], 1);
+ assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+ nodes[1].node.handle_commitment_signed(&nodes[2].node.get_our_node_id(), &bs_second_cs.commitment_signed);
+ check_added_monitors!(nodes[1], 1);
+ let as_second_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[2].node.get_our_node_id());
+
+ nodes[2].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &as_second_raa);
+ check_added_monitors!(nodes[2], 1);
+ assert!(nodes[2].node.get_and_clear_pending_msg_events().is_empty());
+
expect_pending_htlcs_forwardable!(nodes[2]);
let events_6 = nodes[2].node.get_and_clear_pending_events();
- assert_eq!(events_6.len(), 1);
+ assert_eq!(events_6.len(), 2);
match events_6[0] {
Event::PaymentReceived { payment_hash, .. } => { assert_eq!(payment_hash, payment_hash_2); },
_ => panic!("Unexpected event"),
};
+ match events_6[1] {
+ Event::PaymentReceived { payment_hash, .. } => { assert_eq!(payment_hash, payment_hash_3); },
+ _ => panic!("Unexpected event"),
+ };
if test_ignore_second_cs {
expect_pending_htlcs_forwardable!(nodes[1]);
fn test_monitor_update_fail_claim() {
// Basic test for monitor update failures when processing claim_funds calls.
// We set up a simple 3-node network, sending a payment from A to B and failing B's monitor
- // update to claim the payment. We then send a payment C->B->A, making the forward of this
- // payment from B to A fail due to the paused channel. Finally, we restore the channel monitor
- // updating and claim the payment on B.
+ // update to claim the payment. We then send two payments C->B->A, which are held at B.
+ // Finally, we restore the channel monitor updating and claim the payment on B, forwarding
+ // the payments from C onwards to A.
let chanmon_cfgs = create_chanmon_cfgs(3);
let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
*nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
assert!(nodes[1].node.claim_funds(payment_preimage_1));
+ nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Temporary failure claiming HTLC, treating as success: Failed to update ChannelMonitor".to_string(), 1);
check_added_monitors!(nodes[1], 1);
+ // Note that at this point there is a pending commitment transaction update for A being held by
+ // B. Even when we go to send the payment from C through B to A, B will not update this
+ // already-signed commitment transaction and will instead wait for it to resolve before
+ // forwarding the payment onwards.
+
let (_, payment_hash_2, payment_secret_2) = get_payment_preimage_hash!(nodes[0]);
+ let route;
{
let net_graph_msg_handler = &nodes[2].net_graph_msg_handler;
- let route = get_route(&nodes[2].node.get_our_node_id(), &net_graph_msg_handler.network_graph.read().unwrap(), &nodes[0].node.get_our_node_id(), Some(InvoiceFeatures::known()), None, &Vec::new(), 1000000, TEST_FINAL_CLTV, &logger).unwrap();
+ route = get_route(&nodes[2].node.get_our_node_id(), &net_graph_msg_handler.network_graph.read().unwrap(), &nodes[0].node.get_our_node_id(), Some(InvoiceFeatures::known()), None, &Vec::new(), 1_000_000, TEST_FINAL_CLTV, &logger).unwrap();
nodes[2].node.send_payment(&route, payment_hash_2, &Some(payment_secret_2)).unwrap();
check_added_monitors!(nodes[2], 1);
}
nodes[1].node.handle_update_add_htlc(&nodes[2].node.get_our_node_id(), &payment_event.msgs[0]);
let events = nodes[1].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 0);
- nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Temporary failure claiming HTLC, treating as success: Failed to update ChannelMonitor".to_string(), 1);
commitment_signed_dance!(nodes[1], nodes[2], payment_event.commitment_msg, false, true);
- let bs_fail_update = get_htlc_update_msgs!(nodes[1], nodes[2].node.get_our_node_id());
- nodes[2].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &bs_fail_update.update_fail_htlcs[0]);
- commitment_signed_dance!(nodes[2], nodes[1], bs_fail_update.commitment_signed, false, true);
-
- let msg_events = nodes[2].node.get_and_clear_pending_msg_events();
- assert_eq!(msg_events.len(), 1);
- match msg_events[0] {
- MessageSendEvent::PaymentFailureNetworkUpdate { update: msgs::HTLCFailChannelUpdate::ChannelUpdateMessage { ref msg }} => {
- assert_eq!(msg.contents.short_channel_id, chan_1.0.contents.short_channel_id);
- assert_eq!(msg.contents.flags & 2, 2); // temp disabled
- },
- _ => panic!("Unexpected event"),
- }
+ let (_, payment_hash_3, payment_secret_3) = get_payment_preimage_hash!(nodes[0]);
+ nodes[2].node.send_payment(&route, payment_hash_3, &Some(payment_secret_3)).unwrap();
+ check_added_monitors!(nodes[2], 1);
- let events = nodes[2].node.get_and_clear_pending_events();
+ let mut events = nodes[2].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 1);
- if let Event::PaymentFailed { payment_hash, rejected_by_dest, .. } = events[0] {
- assert_eq!(payment_hash, payment_hash_2);
- assert!(!rejected_by_dest);
- } else { panic!("Unexpected event!"); }
+ let payment_event = SendEvent::from_event(events.pop().unwrap());
+ nodes[1].node.handle_update_add_htlc(&nodes[2].node.get_our_node_id(), &payment_event.msgs[0]);
+ let events = nodes[1].node.get_and_clear_pending_msg_events();
+ assert_eq!(events.len(), 0);
+ commitment_signed_dance!(nodes[1], nodes[2], payment_event.commitment_msg, false, true);
// Now restore monitor updating on the 0<->1 channel and claim the funds on B.
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_1.2).unwrap().clone();
let bs_fulfill_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_fulfill_update.update_fulfill_htlcs[0]);
commitment_signed_dance!(nodes[0], nodes[1], bs_fulfill_update.commitment_signed, false);
+ expect_payment_sent!(nodes[0], payment_preimage_1);
+
+ // Get the payment forwards, note that they were batched into one commitment update.
+ expect_pending_htlcs_forwardable!(nodes[1]);
+ check_added_monitors!(nodes[1], 1);
+ let bs_forward_update = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+ nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &bs_forward_update.update_add_htlcs[0]);
+ nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &bs_forward_update.update_add_htlcs[1]);
+ commitment_signed_dance!(nodes[0], nodes[1], bs_forward_update.commitment_signed, false);
+ expect_pending_htlcs_forwardable!(nodes[0]);
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- if let Event::PaymentSent { payment_preimage, .. } = events[0] {
- assert_eq!(payment_preimage, payment_preimage_1);
- } else { panic!("Unexpected event!"); }
+ assert_eq!(events.len(), 2);
+ match events[0] {
+ Event::PaymentReceived { ref payment_hash, ref payment_preimage, ref payment_secret, amt, user_payment_id: _ } => {
+ assert_eq!(payment_hash_2, *payment_hash);
+ assert!(payment_preimage.is_none());
+ assert_eq!(payment_secret_2, *payment_secret);
+ assert_eq!(1_000_000, amt);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[1] {
+ Event::PaymentReceived { ref payment_hash, ref payment_preimage, ref payment_secret, amt, user_payment_id: _ } => {
+ assert_eq!(payment_hash_3, *payment_hash);
+ assert!(payment_preimage.is_none());
+ assert_eq!(payment_secret_3, *payment_secret);
+ assert_eq!(1_000_000, amt);
+ },
+ _ => panic!("Unexpected event"),
+ }
}
#[test]
/// commitment transaction without it as otherwise we'll have to force-close the channel to
/// claim it before the timeout (obviously doesn't apply to revoked HTLCs that we can't claim
/// anyway). That said, ChannelMonitor does this for us (see
- /// ChannelMonitor::would_broadcast_at_height) so we actually remove the HTLC from our own
- /// local state before then, once we're sure that the next commitment_signed and
+ /// ChannelMonitor::should_broadcast_holder_commitment_txn) so we actually remove the HTLC from
+ /// our own local state before then, once we're sure that the next commitment_signed and
/// ChannelMonitor::provide_latest_local_commitment_tx will not include this HTLC.
LocalRemoved(InboundHTLCRemovalReason),
}
}
/// Information needed for constructing an invoice route hint for this channel.
-#[derive(Clone)]
+#[derive(Clone, Debug, PartialEq)]
pub struct CounterpartyForwardingInfo {
/// Base routing fee in millisatoshis.
pub fee_base_msat: u32,
next_local_commitment_tx_fee_info_cached: Mutex<Option<CommitmentTxInfoCached>>,
#[cfg(any(test, feature = "fuzztarget"))]
next_remote_commitment_tx_fee_info_cached: Mutex<Option<CommitmentTxInfoCached>>,
+
+ /// lnd has a long-standing bug where, upon reconnection, if the channel is not yet confirmed
+ /// they will not send a channel_reestablish until the channel locks in. Then, they will send a
+ /// funding_locked *before* sending the channel_reestablish (which is clearly a violation of
+ /// the BOLT specs). We copy c-lightning's workaround here and simply store the funding_locked
+ /// message until we receive a channel_reestablish.
+ ///
+ /// See-also <https://github.com/lightningnetwork/lnd/issues/4006>
+ pub workaround_lnd_bug_4006: Option<msgs::FundingLocked>,
}
#[cfg(any(test, feature = "fuzztarget"))]
next_local_commitment_tx_fee_info_cached: Mutex::new(None),
#[cfg(any(test, feature = "fuzztarget"))]
next_remote_commitment_tx_fee_info_cached: Mutex::new(None),
+
+ workaround_lnd_bug_4006: None,
})
}
next_local_commitment_tx_fee_info_cached: Mutex::new(None),
#[cfg(any(test, feature = "fuzztarget"))]
next_remote_commitment_tx_fee_info_cached: Mutex::new(None),
+
+ workaround_lnd_bug_4006: None,
};
Ok(chan)
let mut local_htlc_total_msat = 0;
let mut value_to_self_msat_offset = 0;
- log_trace!(logger, "Building commitment transaction number {} (really {} xor {}) for {}, generated by {} with fee {}...", commitment_number, (INITIAL_COMMITMENT_NUMBER - commitment_number), get_commitment_transaction_number_obscure_factor(&self.get_holder_pubkeys().payment_point, &self.get_counterparty_pubkeys().payment_point, self.is_outbound()), if local { "us" } else { "remote" }, if generated_by_local { "us" } else { "remote" }, feerate_per_kw);
+ log_trace!(logger, "Building commitment transaction number {} (really {} xor {}) for channel {} for {}, generated by {} with fee {}...",
+ commitment_number, (INITIAL_COMMITMENT_NUMBER - commitment_number),
+ get_commitment_transaction_number_obscure_factor(&self.get_holder_pubkeys().payment_point, &self.get_counterparty_pubkeys().payment_point, self.is_outbound()),
+ log_bytes!(self.channel_id), if local { "us" } else { "remote" }, if generated_by_local { "us" } else { "remote" }, feerate_per_kw);
macro_rules! get_htlc_in_commitment {
($htlc: expr, $offered: expr) => {
_ => {}
}
}
- log_trace!(logger, "Adding HTLC claim to holding_cell! Current state: {}", self.channel_state);
+ log_trace!(logger, "Adding HTLC claim to holding_cell in channel {}! Current state: {}", log_bytes!(self.channel_id()), self.channel_state);
self.holding_cell_htlc_updates.push(HTLCUpdateAwaitingACK::ClaimHTLC {
payment_preimage: payment_preimage_arg, htlc_id: htlc_id_arg,
});
debug_assert!(false, "Have an inbound HTLC we tried to claim before it was fully committed to");
return Ok((None, Some(monitor_update)));
}
- log_trace!(logger, "Upgrading HTLC {} to LocalRemoved with a Fulfill!", log_bytes!(htlc.payment_hash.0));
+ log_trace!(logger, "Upgrading HTLC {} to LocalRemoved with a Fulfill in channel {}!", log_bytes!(htlc.payment_hash.0), log_bytes!(self.channel_id));
htlc.state = InboundHTLCState::LocalRemoved(InboundHTLCRemovalReason::Fulfill(payment_preimage_arg.clone()));
}
_ => {}
}
}
- log_trace!(logger, "Placing failure for HTLC ID {} in holding cell", htlc_id_arg);
+ log_trace!(logger, "Placing failure for HTLC ID {} in holding cell in channel {}.", htlc_id_arg, log_bytes!(self.channel_id()));
self.holding_cell_htlc_updates.push(HTLCUpdateAwaitingACK::FailHTLC {
htlc_id: htlc_id_arg,
err_packet,
return Ok(None);
}
- log_trace!(logger, "Failing HTLC ID {} back with a update_fail_htlc message", htlc_id_arg);
+ log_trace!(logger, "Failing HTLC ID {} back with a update_fail_htlc message in channel {}.", htlc_id_arg, log_bytes!(self.channel_id()));
{
let htlc = &mut self.pending_inbound_htlcs[pending_idx];
htlc.state = InboundHTLCState::LocalRemoved(InboundHTLCRemovalReason::FailRelay(err_packet.clone()));
let initial_commitment_bitcoin_tx = trusted_tx.built_transaction();
let sighash = initial_commitment_bitcoin_tx.get_sighash_all(&funding_script, self.channel_value_satoshis);
// They sign the holder commitment transaction...
- log_trace!(logger, "Checking funding_created tx signature {} by key {} against tx {} (sighash {}) with redeemscript {}", log_bytes!(sig.serialize_compact()[..]), log_bytes!(self.counterparty_funding_pubkey().serialize()), encode::serialize_hex(&initial_commitment_bitcoin_tx.transaction), log_bytes!(sighash[..]), encode::serialize_hex(&funding_script));
+ log_trace!(logger, "Checking funding_created tx signature {} by key {} against tx {} (sighash {}) with redeemscript {} for channel {}.",
+ log_bytes!(sig.serialize_compact()[..]), log_bytes!(self.counterparty_funding_pubkey().serialize()),
+ encode::serialize_hex(&initial_commitment_bitcoin_tx.transaction), log_bytes!(sighash[..]),
+ encode::serialize_hex(&funding_script), log_bytes!(self.channel_id()));
secp_check!(self.secp_ctx.verify(&sighash, &sig, self.counterparty_funding_pubkey()), "Invalid funding_created signature from peer".to_owned());
}
let counterparty_trusted_tx = counterparty_initial_commitment_tx.trust();
let counterparty_initial_bitcoin_tx = counterparty_trusted_tx.built_transaction();
- log_trace!(logger, "Initial counterparty ID {} tx {}", counterparty_initial_bitcoin_tx.txid, encode::serialize_hex(&counterparty_initial_bitcoin_tx.transaction));
+ log_trace!(logger, "Initial counterparty tx for channel {} is: txid {} tx {}",
+ log_bytes!(self.channel_id()), counterparty_initial_bitcoin_tx.txid, encode::serialize_hex(&counterparty_initial_bitcoin_tx.transaction));
let counterparty_signature = self.holder_signer.sign_counterparty_commitment(&counterparty_initial_commitment_tx, &self.secp_ctx)
.map_err(|_| ChannelError::Close("Failed to get signatures for new commitment_signed".to_owned()))?.0;
self.cur_counterparty_commitment_transaction_number -= 1;
self.cur_holder_commitment_transaction_number -= 1;
+ log_info!(logger, "Generated funding_signed for peer for channel {}", log_bytes!(self.channel_id()));
+
Ok((msgs::FundingSigned {
channel_id: self.channel_id,
signature
let counterparty_trusted_tx = counterparty_initial_commitment_tx.trust();
let counterparty_initial_bitcoin_tx = counterparty_trusted_tx.built_transaction();
- log_trace!(logger, "Initial counterparty ID {} tx {}", counterparty_initial_bitcoin_tx.txid, encode::serialize_hex(&counterparty_initial_bitcoin_tx.transaction));
+ log_trace!(logger, "Initial counterparty tx for channel {} is: txid {} tx {}",
+ log_bytes!(self.channel_id()), counterparty_initial_bitcoin_tx.txid, encode::serialize_hex(&counterparty_initial_bitcoin_tx.transaction));
let holder_signer = self.build_holder_transaction_keys(self.cur_holder_commitment_transaction_number)?;
let initial_commitment_tx = self.build_commitment_transaction(self.cur_holder_commitment_transaction_number, &holder_signer, true, false, self.feerate_per_kw, logger).0;
self.cur_holder_commitment_transaction_number -= 1;
self.cur_counterparty_commitment_transaction_number -= 1;
+ log_info!(logger, "Received funding_signed from peer for channel {}", log_bytes!(self.channel_id()));
+
Ok((channel_monitor, self.funding_transaction.as_ref().cloned().unwrap()))
}
- pub fn funding_locked(&mut self, msg: &msgs::FundingLocked) -> Result<(), ChannelError> {
+ pub fn funding_locked<L: Deref>(&mut self, msg: &msgs::FundingLocked, logger: &L) -> Result<(), ChannelError> where L::Target: Logger {
if self.channel_state & (ChannelState::PeerDisconnected as u32) == ChannelState::PeerDisconnected as u32 {
- return Err(ChannelError::Close("Peer sent funding_locked when we needed a channel_reestablish".to_owned()));
+ self.workaround_lnd_bug_4006 = Some(msg.clone());
+ return Err(ChannelError::Ignore("Peer sent funding_locked when we needed a channel_reestablish. The peer is likely lnd, see https://github.com/lightningnetwork/lnd/issues/4006".to_owned()));
}
let non_shutdown_state = self.channel_state & (!MULTI_STATE_FLAGS);
self.counterparty_prev_commitment_point = self.counterparty_cur_commitment_point;
self.counterparty_cur_commitment_point = Some(msg.next_per_commitment_point);
+
+ log_info!(logger, "Received funding_locked from peer for channel {}", log_bytes!(self.channel_id()));
+
Ok(())
}
if pending_remote_value_msat - msg.amount_msat - chan_reserve_msat < remote_fee_cost_incl_stuck_buffer_msat {
// Note that if the pending_forward_status is not updated here, then it's because we're already failing
// the HTLC, i.e. its status is already set to failing.
- log_info!(logger, "Attempting to fail HTLC due to fee spike buffer violation");
+ log_info!(logger, "Attempting to fail HTLC due to fee spike buffer violation in channel {}. Rebalancing is required.", log_bytes!(self.channel_id()));
pending_forward_status = create_pending_htlc_status(self, pending_forward_status, 0x1000|7);
}
} else {
let bitcoin_tx = trusted_tx.built_transaction();
let sighash = bitcoin_tx.get_sighash_all(&funding_script, self.channel_value_satoshis);
- log_trace!(logger, "Checking commitment tx signature {} by key {} against tx {} (sighash {}) with redeemscript {}", log_bytes!(msg.signature.serialize_compact()[..]), log_bytes!(self.counterparty_funding_pubkey().serialize()), encode::serialize_hex(&bitcoin_tx.transaction), log_bytes!(sighash[..]), encode::serialize_hex(&funding_script));
+ log_trace!(logger, "Checking commitment tx signature {} by key {} against tx {} (sighash {}) with redeemscript {} in channel {}",
+ log_bytes!(msg.signature.serialize_compact()[..]),
+ log_bytes!(self.counterparty_funding_pubkey().serialize()), encode::serialize_hex(&bitcoin_tx.transaction),
+ log_bytes!(sighash[..]), encode::serialize_hex(&funding_script), log_bytes!(self.channel_id()));
if let Err(_) = self.secp_ctx.verify(&sighash, &msg.signature, &self.counterparty_funding_pubkey()) {
return Err((None, ChannelError::Close("Invalid commitment tx signature from peer".to_owned())));
}
let htlc_tx = self.build_htlc_transaction(&commitment_txid, &htlc, true, &keys, feerate_per_kw);
let htlc_redeemscript = chan_utils::get_htlc_redeemscript(&htlc, &keys);
let htlc_sighash = hash_to_message!(&bip143::SigHashCache::new(&htlc_tx).signature_hash(0, &htlc_redeemscript, htlc.amount_msat / 1000, SigHashType::All)[..]);
- log_trace!(logger, "Checking HTLC tx signature {} by key {} against tx {} (sighash {}) with redeemscript {}", log_bytes!(msg.htlc_signatures[idx].serialize_compact()[..]), log_bytes!(keys.countersignatory_htlc_key.serialize()), encode::serialize_hex(&htlc_tx), log_bytes!(htlc_sighash[..]), encode::serialize_hex(&htlc_redeemscript));
+ log_trace!(logger, "Checking HTLC tx signature {} by key {} against tx {} (sighash {}) with redeemscript {} in channel {}.",
+ log_bytes!(msg.htlc_signatures[idx].serialize_compact()[..]), log_bytes!(keys.countersignatory_htlc_key.serialize()),
+ encode::serialize_hex(&htlc_tx), log_bytes!(htlc_sighash[..]), encode::serialize_hex(&htlc_redeemscript), log_bytes!(self.channel_id()));
if let Err(_) = self.secp_ctx.verify(&htlc_sighash, &msg.htlc_signatures[idx], &keys.countersignatory_htlc_key) {
return Err((None, ChannelError::Close("Invalid HTLC tx signature from peer".to_owned())));
}
Some(forward_info.clone())
} else { None };
if let Some(forward_info) = new_forward {
+ log_trace!(logger, "Updating HTLC {} to AwaitingRemoteRevokeToAnnounce due to commitment_signed in channel {}.",
+ log_bytes!(htlc.payment_hash.0), log_bytes!(self.channel_id));
htlc.state = InboundHTLCState::AwaitingRemoteRevokeToAnnounce(forward_info);
need_commitment = true;
}
if let Some(fail_reason) = if let &mut OutboundHTLCState::RemoteRemoved(ref mut fail_reason) = &mut htlc.state {
Some(fail_reason.take())
} else { None } {
+ log_trace!(logger, "Updating HTLC {} to AwaitingRemoteRevokeToRemove due to commitment_signed in channel {}.",
+ log_bytes!(htlc.payment_hash.0), log_bytes!(self.channel_id));
htlc.state = OutboundHTLCState::AwaitingRemoteRevokeToRemove(fail_reason);
need_commitment = true;
}
self.latest_monitor_update_id = monitor_update.update_id;
monitor_update.updates.append(&mut additional_update.updates);
}
+ log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updated HTLC state but awaiting a monitor update resolution to reply.",
+ log_bytes!(self.channel_id));
// TODO: Call maybe_propose_first_closing_signed on restoration (or call it here and
// re-send the message on restoration)
return Err((Some(monitor_update), ChannelError::Ignore("Previous monitor update failure prevented generation of RAA".to_owned())));
(None, self.maybe_propose_first_closing_signed(fee_estimator))
} else { (None, None) };
+ log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updating HTLC state and responding with{} a revoke_and_ack.",
+ log_bytes!(self.channel_id()), if commitment_signed.is_some() { " our own commitment_signed and" } else { "" });
+
Ok((msgs::RevokeAndACK {
channel_id: self.channel_id,
per_commitment_secret,
fn free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> Result<(Option<(msgs::CommitmentUpdate, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>), ChannelError> where L::Target: Logger {
assert_eq!(self.channel_state & ChannelState::MonitorUpdateFailed as u32, 0);
if self.holding_cell_htlc_updates.len() != 0 || self.holding_cell_update_fee.is_some() {
- log_trace!(logger, "Freeing holding cell with {} HTLC updates{}", self.holding_cell_htlc_updates.len(), if self.holding_cell_update_fee.is_some() { " and a fee update" } else { "" });
+ log_trace!(logger, "Freeing holding cell with {} HTLC updates{} in channel {}", self.holding_cell_htlc_updates.len(),
+ if self.holding_cell_update_fee.is_some() { " and a fee update" } else { "" }, log_bytes!(self.channel_id()));
let mut monitor_update = ChannelMonitorUpdate {
update_id: self.latest_monitor_update_id + 1, // We don't increment this yet!
Err(e) => {
match e {
ChannelError::Ignore(ref msg) => {
- log_info!(logger, "Failed to send HTLC with payment_hash {} due to {}", log_bytes!(payment_hash.0), msg);
+ log_info!(logger, "Failed to send HTLC with payment_hash {} due to {} in channel {}",
+ log_bytes!(payment_hash.0), msg, log_bytes!(self.channel_id()));
// If we fail to send here, then this HTLC should
// be failed backwards. Failing to send here
// indicates that this HTLC may keep being put back
self.latest_monitor_update_id = monitor_update.update_id;
monitor_update.updates.append(&mut additional_update.updates);
+ log_debug!(logger, "Freeing holding cell in channel {} resulted in {}{} HTLCs added, {} HTLCs fulfilled, and {} HTLCs failed.",
+ log_bytes!(self.channel_id()), if update_fee.is_some() { "a fee update, " } else { "" },
+ update_add_htlcs.len(), update_fulfill_htlcs.len(), update_fail_htlcs.len());
+
Ok((Some((msgs::CommitmentUpdate {
update_add_htlcs,
update_fulfill_htlcs,
self.counterparty_cur_commitment_point = Some(msg.next_per_commitment_point);
self.cur_counterparty_commitment_transaction_number -= 1;
- log_trace!(logger, "Updating HTLCs on receipt of RAA...");
+ log_trace!(logger, "Updating HTLCs on receipt of RAA in channel {}...", log_bytes!(self.channel_id()));
let mut to_forward_infos = Vec::new();
let mut revoked_htlcs = Vec::new();
let mut update_fail_htlcs = Vec::new();
});
for htlc in pending_inbound_htlcs.iter_mut() {
let swap = if let &InboundHTLCState::AwaitingRemoteRevokeToAnnounce(_) = &htlc.state {
- log_trace!(logger, " ...promoting inbound AwaitingRemoteRevokeToAnnounce {} to Committed", log_bytes!(htlc.payment_hash.0));
true
} else if let &InboundHTLCState::AwaitingAnnouncedRemoteRevoke(_) = &htlc.state {
- log_trace!(logger, " ...promoting inbound AwaitingAnnouncedRemoteRevoke {} to Committed", log_bytes!(htlc.payment_hash.0));
true
} else { false };
if swap {
mem::swap(&mut state, &mut htlc.state);
if let InboundHTLCState::AwaitingRemoteRevokeToAnnounce(forward_info) = state {
+ log_trace!(logger, " ...promoting inbound AwaitingRemoteRevokeToAnnounce {} to AwaitingAnnouncedRemoteRevoke", log_bytes!(htlc.payment_hash.0));
htlc.state = InboundHTLCState::AwaitingAnnouncedRemoteRevoke(forward_info);
require_commitment = true;
} else if let InboundHTLCState::AwaitingAnnouncedRemoteRevoke(forward_info) = state {
match forward_info {
PendingHTLCStatus::Fail(fail_msg) => {
+ log_trace!(logger, " ...promoting inbound AwaitingAnnouncedRemoteRevoke {} to LocalRemoved due to PendingHTLCStatus indicating failure", log_bytes!(htlc.payment_hash.0));
require_commitment = true;
match fail_msg {
HTLCFailureMsg::Relay(msg) => {
}
},
PendingHTLCStatus::Forward(forward_info) => {
+ log_trace!(logger, " ...promoting inbound AwaitingAnnouncedRemoteRevoke {} to Committed", log_bytes!(htlc.payment_hash.0));
to_forward_infos.push((forward_info, htlc.htlc_id));
htlc.state = InboundHTLCState::Committed;
}
}
self.monitor_pending_forwards.append(&mut to_forward_infos);
self.monitor_pending_failures.append(&mut revoked_htlcs);
+ log_debug!(logger, "Received a valid revoke_and_ack for channel {} but awaiting a monitor update resolution to reply.", log_bytes!(self.channel_id()));
return Ok((None, Vec::new(), Vec::new(), None, monitor_update, Vec::new()))
}
self.latest_monitor_update_id = monitor_update.update_id;
monitor_update.updates.append(&mut additional_update.updates);
+ log_debug!(logger, "Received a valid revoke_and_ack for channel {}. Responding with a commitment update with {} HTLCs failed.",
+ log_bytes!(self.channel_id()), update_fail_htlcs.len() + update_fail_malformed_htlcs.len());
Ok((Some(msgs::CommitmentUpdate {
update_add_htlcs: Vec::new(),
update_fulfill_htlcs: Vec::new(),
commitment_signed
}), to_forward_infos, revoked_htlcs, None, monitor_update, htlcs_to_fail))
} else {
+ log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id()));
Ok((None, to_forward_infos, revoked_htlcs, self.maybe_propose_first_closing_signed(fee_estimator), monitor_update, htlcs_to_fail))
}
}
}
self.channel_state |= ChannelState::PeerDisconnected as u32;
- log_debug!(logger, "Peer disconnection resulted in {} remote-announced HTLC drops on channel {}", inbound_drop_count, log_bytes!(self.channel_id()));
+ log_trace!(logger, "Peer disconnection resulted in {} remote-announced HTLC drops on channel {}", inbound_drop_count, log_bytes!(self.channel_id()));
}
/// Indicates that a ChannelMonitor update failed to be stored by the client and further
self.monitor_pending_revoke_and_ack = false;
self.monitor_pending_commitment_signed = false;
let order = self.resend_order.clone();
- log_trace!(logger, "Restored monitor updating resulting in {}{} commitment update and {} RAA, with {} first",
- if funding_broadcastable.is_some() { "a funding broadcastable, " } else { "" },
- if commitment_update.is_some() { "a" } else { "no" },
- if raa.is_some() { "an" } else { "no" },
+ log_debug!(logger, "Restored monitor updating in channel {} resulting in {}{} commitment update and {} RAA, with {} first",
+ log_bytes!(self.channel_id()), if funding_broadcastable.is_some() { "a funding broadcastable, " } else { "" },
+ if commitment_update.is_some() { "a" } else { "no" }, if raa.is_some() { "an" } else { "no" },
match order { RAACommitmentOrder::CommitmentFirst => "commitment", RAACommitmentOrder::RevokeAndACKFirst => "RAA"});
(raa, commitment_update, order, forwards, failures, funding_broadcastable, funding_locked)
}
}
}
- log_trace!(logger, "Regenerated latest commitment update with {} update_adds, {} update_fulfills, {} update_fails, and {} update_fail_malformeds",
- update_add_htlcs.len(), update_fulfill_htlcs.len(), update_fail_htlcs.len(), update_fail_malformed_htlcs.len());
+ log_trace!(logger, "Regenerated latest commitment update in channel {} with {} update_adds, {} update_fulfills, {} update_fails, and {} update_fail_malformeds",
+ log_bytes!(self.channel_id()), update_add_htlcs.len(), update_fulfill_htlcs.len(), update_fail_htlcs.len(), update_fail_malformed_htlcs.len());
msgs::CommitmentUpdate {
update_add_htlcs, update_fulfill_htlcs, update_fail_htlcs, update_fail_malformed_htlcs,
update_fee: None,
/// is_usable() and considers things like the channel being temporarily disabled.
/// Allowed in any state (including after shutdown)
pub fn is_live(&self) -> bool {
- self.is_usable() && (self.channel_state & (ChannelState::PeerDisconnected as u32 | ChannelState::MonitorUpdateFailed as u32) == 0)
+ self.is_usable() && (self.channel_state & (ChannelState::PeerDisconnected as u32) == 0)
}
/// Returns true if this channel has been marked as awaiting a monitor update to move forward.
// send it immediately instead of waiting for a best_block_updated call (which
// may have already happened for this block).
if let Some(funding_locked) = self.check_get_funding_locked(height) {
+ log_info!(logger, "Sending a funding_locked to our peer for channel {}", log_bytes!(self.channel_id));
return Ok(Some(funding_locked));
}
}
for inp in tx.input.iter() {
if inp.previous_output == funding_txo.into_bitcoin_outpoint() {
- log_trace!(logger, "Detected channel-closing tx {} spending {}:{}, closing channel {}", tx.txid(), inp.previous_output.txid, inp.previous_output.vout, log_bytes!(self.channel_id()));
+ log_info!(logger, "Detected channel-closing tx {} spending {}:{}, closing channel {}", tx.txid(), inp.previous_output.txid, inp.previous_output.vout, log_bytes!(self.channel_id()));
return Err(msgs::ErrorMessage {
channel_id: self.channel_id(),
data: "Commitment or closing transaction was confirmed on chain.".to_owned()
///
/// May return some HTLCs (and their payment_hash) which have timed out and should be failed
/// back.
- pub fn best_block_updated(&mut self, height: u32, highest_header_time: u32) -> Result<(Option<msgs::FundingLocked>, Vec<(HTLCSource, PaymentHash)>), msgs::ErrorMessage> {
+ pub fn best_block_updated<L: Deref>(&mut self, height: u32, highest_header_time: u32, logger: &L)
+ -> Result<(Option<msgs::FundingLocked>, Vec<(HTLCSource, PaymentHash)>), msgs::ErrorMessage> where L::Target: Logger {
let mut timed_out_htlcs = Vec::new();
let unforwarded_htlc_cltv_limit = height + HTLC_FAIL_BACK_BUFFER;
self.holding_cell_htlc_updates.retain(|htlc_update| {
self.update_time_counter = cmp::max(self.update_time_counter, highest_header_time);
if let Some(funding_locked) = self.check_get_funding_locked(height) {
+ log_info!(logger, "Sending a funding_locked to our peer for channel {}", log_bytes!(self.channel_id));
return Ok((Some(funding_locked), timed_out_htlcs));
}
/// Indicates the funding transaction is no longer confirmed in the main chain. This may
/// force-close the channel, but may also indicate a harmless reorganization of a block or two
/// before the channel has reached funding_locked and we can just wait for more blocks.
- pub fn funding_transaction_unconfirmed(&mut self) -> Result<(), msgs::ErrorMessage> {
+ pub fn funding_transaction_unconfirmed<L: Deref>(&mut self, logger: &L) -> Result<(), msgs::ErrorMessage> where L::Target: Logger {
if self.funding_tx_confirmation_height != 0 {
// We handle the funding disconnection by calling best_block_updated with a height one
// below where our funding was connected, implying a reorg back to conf_height - 1.
// larger. If we don't know that time has moved forward, we can just set it to the last
// time we saw and it will be ignored.
let best_time = self.update_time_counter;
- match self.best_block_updated(reorg_height, best_time) {
+ match self.best_block_updated(reorg_height, best_time, logger) {
Ok((funding_locked, timed_out_htlcs)) => {
assert!(funding_locked.is_none(), "We can't generate a funding with 0 confirmations?");
assert!(timed_out_htlcs.is_empty(), "We can't have accepted HTLCs with a timeout before our funding confirmation?");
let dummy_pubkey = PublicKey::from_slice(&pk).unwrap();
let data_loss_protect = if self.cur_counterparty_commitment_transaction_number + 1 < INITIAL_COMMITMENT_NUMBER {
let remote_last_secret = self.commitment_secrets.get_secret(self.cur_counterparty_commitment_transaction_number + 2).unwrap();
- log_trace!(logger, "Enough info to generate a Data Loss Protect with per_commitment_secret {}", log_bytes!(remote_last_secret));
+ log_trace!(logger, "Enough info to generate a Data Loss Protect with per_commitment_secret {} for channel {}", log_bytes!(remote_last_secret), log_bytes!(self.channel_id()));
OptionalField::Present(DataLossProtect {
your_last_per_commitment_secret: remote_last_secret,
my_current_per_commitment_point: dummy_pubkey
})
} else {
- log_info!(logger, "Sending a data_loss_protect with no previous remote per_commitment_secret");
+ log_info!(logger, "Sending a data_loss_protect with no previous remote per_commitment_secret for channel {}", log_bytes!(self.channel_id()));
OptionalField::Present(DataLossProtect {
your_last_per_commitment_secret: [0;32],
my_current_per_commitment_point: dummy_pubkey,
/// Adds a pending outbound HTLC to this channel, note that you probably want
/// send_htlc_and_commit instead cause you'll want both messages at once.
- /// This returns an option instead of a pure UpdateAddHTLC as we may be in a state where we are
- /// waiting on the remote peer to send us a revoke_and_ack during which time we cannot add new
- /// HTLCs on the wire or we wouldn't be able to determine what they actually ACK'ed.
- /// You MUST call send_commitment prior to any other calls on this Channel
+ ///
+ /// This returns an optional UpdateAddHTLC as we may be in a state where we cannot add HTLCs on
+ /// the wire:
+ /// * In cases where we're waiting on the remote peer to send us a revoke_and_ack, we
+ /// wouldn't be able to determine what they actually ACK'ed if we have two sets of updates
+ /// awaiting ACK.
+ /// * In cases where we're marked MonitorUpdateFailed, we cannot commit to a new state as we
+ /// may not yet have sent the previous commitment update messages and will need to regenerate
+ /// them.
+ ///
+ /// You MUST call send_commitment prior to calling any other methods on this Channel!
+ ///
/// If an Err is returned, it's a ChannelError::Ignore!
pub fn send_htlc(&mut self, amount_msat: u64, payment_hash: PaymentHash, cltv_expiry: u32, source: HTLCSource, onion_routing_packet: msgs::OnionPacket) -> Result<Option<msgs::UpdateAddHTLC>, ChannelError> {
if (self.channel_state & (ChannelState::ChannelFunded as u32 | BOTH_SIDES_SHUTDOWN_MASK)) != (ChannelState::ChannelFunded as u32) {
return Err(ChannelError::Ignore(format!("Cannot send less than their minimum HTLC value ({})", self.counterparty_htlc_minimum_msat)));
}
- if (self.channel_state & (ChannelState::PeerDisconnected as u32 | ChannelState::MonitorUpdateFailed as u32)) != 0 {
+ if (self.channel_state & (ChannelState::PeerDisconnected as u32)) != 0 {
// Note that this should never really happen, if we're !is_live() on receipt of an
// incoming HTLC for relay will result in us rejecting the HTLC and we won't allow
// the user to send directly into a !is_live() channel. However, if we
// disconnected during the time the previous hop was doing the commitment dance we may
// end up getting here after the forwarding delay. In any case, returning an
// IgnoreError will get ChannelManager to do the right thing and fail backwards now.
- return Err(ChannelError::Ignore("Cannot send an HTLC while disconnected/frozen for channel monitor update".to_owned()));
+ return Err(ChannelError::Ignore("Cannot send an HTLC while disconnected from channel counterparty".to_owned()));
}
let (outbound_htlc_count, htlc_outbound_value_msat) = self.get_outbound_pending_htlc_stats();
}
// Now update local state:
- if (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == (ChannelState::AwaitingRemoteRevoke as u32) {
+ if (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32 | ChannelState::MonitorUpdateFailed as u32)) != 0 {
self.holding_cell_htlc_updates.push(HTLCUpdateAwaitingACK::AddHTLC {
amount_msat,
payment_hash,
signature = res.0;
htlc_signatures = res.1;
- log_trace!(logger, "Signed remote commitment tx {} (txid {}) with redeemscript {} -> {}",
+ log_trace!(logger, "Signed remote commitment tx {} (txid {}) with redeemscript {} -> {} in channel {}",
encode::serialize_hex(&counterparty_commitment_tx.0.trust().built_transaction().transaction),
- &counterparty_commitment_txid,
- encode::serialize_hex(&self.get_funding_redeemscript()),
- log_bytes!(signature.serialize_compact()[..]));
+ &counterparty_commitment_txid, encode::serialize_hex(&self.get_funding_redeemscript()),
+ log_bytes!(signature.serialize_compact()[..]), log_bytes!(self.channel_id()));
for (ref htlc_sig, ref htlc) in htlc_signatures.iter().zip(htlcs) {
- log_trace!(logger, "Signed remote HTLC tx {} with redeemscript {} with pubkey {} -> {}",
+ log_trace!(logger, "Signed remote HTLC tx {} with redeemscript {} with pubkey {} -> {} in channel {}",
encode::serialize_hex(&chan_utils::build_htlc_transaction(&counterparty_commitment_txid, feerate_per_kw, self.get_holder_selected_contest_delay(), htlc, &counterparty_keys.broadcaster_delayed_payment_key, &counterparty_keys.revocation_key)),
encode::serialize_hex(&chan_utils::get_htlc_redeemscript(&htlc, &counterparty_keys)),
log_bytes!(counterparty_keys.broadcaster_htlc_key.serialize()),
- log_bytes!(htlc_sig.serialize_compact()[..]));
+ log_bytes!(htlc_sig.serialize_compact()[..]), log_bytes!(self.channel_id()));
}
}
self.channel_update_status.write(writer)?;
- write_tlv_fields!(writer, {}, {(0, self.announcement_sigs)});
+ write_tlv_fields!(writer, {(0, self.announcement_sigs, option)});
Ok(())
}
let channel_update_status = Readable::read(reader)?;
let mut announcement_sigs = None;
- read_tlv_fields!(reader, {}, {(0, announcement_sigs)});
+ read_tlv_fields!(reader, {(0, announcement_sigs, option)});
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&keys_source.get_secure_random_bytes());
next_local_commitment_tx_fee_info_cached: Mutex::new(None),
#[cfg(any(test, feature = "fuzztarget"))]
next_remote_commitment_tx_fee_info_cached: Mutex::new(None),
+
+ workaround_lnd_bug_4006: None,
})
}
}
use prelude::*;
use core::{cmp, mem};
use core::cell::RefCell;
-use std::collections::{HashMap, hash_map, HashSet};
use std::io::{Cursor, Read};
use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard};
use core::sync::atomic::{AtomicUsize, Ordering};
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 an HTLC claim. See
-// ChannelMontior::would_broadcast_at_height for a description of why this is needed.
+// ChannelMonitor::should_broadcast_holder_commitment_txn for a description of why this is needed.
#[deny(const_err)]
#[allow(dead_code)]
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)]
+#[derive(Clone, Debug, PartialEq)]
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).
/// point after the funding transaction received enough confirmations).
pub is_funding_locked: bool,
/// True if the channel is (a) confirmed and funding_locked messages have been exchanged, (b)
- /// the peer is connected, (c) no monitor update failure is pending resolution, and (d) the
- /// channel is not currently negotiating a shutdown.
+ /// the peer is connected, and (c) the channel is not currently negotiating a shutdown.
///
/// This is a strict superset of `is_funding_locked`.
pub is_usable: bool,
(false, MsgHandleErrInternal::from_chan_no_close(ChannelError::Ignore(msg), $channel_id.clone()))
},
ChannelError::Close(msg) => {
- log_trace!($self.logger, "Closing channel {} due to close-required error: {}", log_bytes!($channel_id[..]), msg);
+ log_error!($self.logger, "Closing channel {} due to close-required error: {}", log_bytes!($channel_id[..]), msg);
if let Some(short_id) = $channel.get_short_channel_id() {
$short_to_id.remove(&short_id);
}
#[inline]
fn finish_force_close_channel(&self, shutdown_res: ShutdownResult) {
let (monitor_update_option, mut failed_htlcs) = shutdown_res;
- log_trace!(self.logger, "Finishing force-closure of channel {} HTLCs to fail", failed_htlcs.len());
+ log_debug!(self.logger, "Finishing force-closure of channel with {} 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() });
}
return Err(APIError::ChannelUnavailable{err: "No such channel".to_owned()});
}
};
- log_trace!(self.logger, "Force-closing channel {}", log_bytes!(channel_id[..]));
+ log_error!(self.logger, "Force-closing channel {}", log_bytes!(channel_id[..]));
self.finish_force_close_channel(chan.force_shutdown(true));
if let Ok(update) = self.get_channel_update(&chan) {
let mut channel_state = self.channel_state.lock().unwrap();
return Err(APIError::MonitorUpdateFailed);
}
+ log_debug!(self.logger, "Sending payment along path resulted in a commitment_signed for channel {}", log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: path.first().unwrap().pubkey,
updates: msgs::CommitmentUpdate {
onion_packet, ..
}, 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);
+ log_trace!(self.logger, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", prev_short_channel_id, log_bytes!(payment_hash.0), short_chan_id);
let htlc_source = HTLCSource::PreviousHopData(HTLCPreviousHopData {
short_channel_id: prev_short_channel_id,
outpoint: prev_funding_outpoint,
panic!("short_channel_id != 0 should imply any pending_forward entries are of type Forward");
},
HTLCForwardInfo::FailHTLC { htlc_id, err_packet } => {
- log_trace!(self.logger, "Failing HTLC back to channel with short id {} after delay", short_chan_id);
+ log_trace!(self.logger, "Failing HTLC back to channel with short id {} (backward HTLC ID {}) after delay", short_chan_id, htlc_id);
match chan.get_mut().get_update_fail_htlc(htlc_id, err_packet, &self.logger) {
Err(e) => {
if let ChannelError::Ignore(msg) = e {
- log_trace!(self.logger, "Failed to fail backwards to short_id {}: {}", short_chan_id, msg);
+ log_trace!(self.logger, "Failed to fail HTLC with ID {} backwards to short_id {}: {}", htlc_id, short_chan_id, msg);
} else {
panic!("Stated return value requirements in get_update_fail_htlc() were not met");
}
handle_errors.push((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true)));
continue;
}
+ log_debug!(self.logger, "Forwarding HTLCs resulted in a commitment update with {} HTLCs added and {} HTLCs failed for channel {}",
+ add_htlc_msgs.len(), fail_htlc_msgs.len(), log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: chan.get().get_counterparty_node_id(),
updates: msgs::CommitmentUpdate {
}
}
if let Some((msg, commitment_signed)) = msgs {
+ log_debug!(self.logger, "Claiming funds for HTLC with preimage {} resulted in a commitment_signed for channel {}",
+ log_bytes!(payment_preimage.0), log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: chan.get().get_counterparty_node_id(),
updates: msgs::CommitmentUpdate {
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);
+ try_chan_entry!(self, chan.get_mut().funding_locked(&msg, &self.logger), channel_state, chan);
if let Some(announcement_sigs) = self.get_announcement_sigs(chan.get()) {
log_trace!(self.logger, "Sending announcement_signatures for {} in response to funding_locked", log_bytes!(chan.get().channel_id()));
// If we see locking block before receiving remote funding_locked, we broadcast our
Ok(())
}
- fn internal_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) -> Result<(), MsgHandleErrInternal> {
+ /// Returns ShouldPersist if anything changed, otherwise either SkipPersist or an Err.
+ fn internal_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) -> Result<NotifyOption, 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(())
+ return Ok(NotifyOption::SkipPersist)
}
};
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));
+ if chan.get().should_announce() {
+ // If the announcement is about a channel of ours which is public, some
+ // other peer may simply be forwarding all its gossip to us. Don't provide
+ // a scary-looking error message and return Ok instead.
+ return Ok(NotifyOption::SkipPersist);
+ }
+ return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a channel_update for a channel from the wrong node - it shouldn't know about our private channels!".to_owned(), chan_id));
}
try_chan_entry!(self, chan.get_mut().channel_update(&msg), channel_state, chan);
},
hash_map::Entry::Vacant(_) => unreachable!()
}
- Ok(())
+ Ok(NotifyOption::DoPersist)
}
fn internal_channel_reestablish(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelReestablish) -> Result<(), MsgHandleErrInternal> {
- let (htlcs_failed_forward, chan_restoration_res) = {
+ let (htlcs_failed_forward, need_lnd_workaround, chan_restoration_res) = {
let mut channel_state_lock = self.channel_state.lock().unwrap();
let channel_state = &mut *channel_state_lock;
msg,
});
}
- (htlcs_failed_forward, handle_chan_restoration_locked!(self, channel_state_lock, channel_state, chan, revoke_and_ack, commitment_update, order, monitor_update_opt, Vec::new(), None, funding_locked))
+ let need_lnd_workaround = chan.get_mut().workaround_lnd_bug_4006.take();
+ (htlcs_failed_forward, need_lnd_workaround,
+ handle_chan_restoration_locked!(self, channel_state_lock, channel_state, chan, revoke_and_ack, commitment_update, order, monitor_update_opt, Vec::new(), None, funding_locked))
},
hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
}
};
post_handle_chan_restoration!(self, chan_restoration_res);
self.fail_holding_cell_htlcs(htlcs_failed_forward, msg.channel_id);
+
+ if let Some(funding_locked_msg) = need_lnd_workaround {
+ self.internal_funding_locked(counterparty_node_id, &funding_locked_msg)?;
+ }
Ok(())
}
if let Err(_e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
unimplemented!();
}
+ log_debug!(self.logger, "Updating fee resulted in a commitment_signed for channel {}", log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: chan.get().get_counterparty_node_id(),
updates: msgs::CommitmentUpdate {
*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));
+ self.do_chain_event(Some(new_height), |channel| channel.best_block_updated(new_height, header.time, &self.logger));
}
}
*self.best_block.write().unwrap() = BestBlock::new(block_hash, height);
- self.do_chain_event(Some(height), |channel| channel.best_block_updated(height, header.time));
+ self.do_chain_event(Some(height), |channel| channel.best_block_updated(height, header.time, &self.logger));
macro_rules! max_time {
($timestamp: expr) => {
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()))
+ channel.funding_transaction_unconfirmed(&self.logger).map(|_| (None, Vec::new()))
} else { Ok((None, Vec::new())) }
} else { Ok((None, Vec::new())) }
});
}
fn handle_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) {
- let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
- let _ = handle_error!(self, self.internal_channel_update(counterparty_node_id, msg), *counterparty_node_id);
+ PersistenceNotifierGuard::optionally_notify(&self.total_consistency_lock, &self.persistence_notifier, || {
+ if let Ok(persist) = handle_error!(self, self.internal_channel_update(counterparty_node_id, msg), *counterparty_node_id) {
+ persist
+ } else {
+ NotifyOption::SkipPersist
+ }
+ });
}
fn handle_channel_reestablish(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelReestablish) {
impl_writeable_tlv_based_enum!(PendingHTLCRouting,
(0, Forward) => {
- (0, onion_packet),
- (2, short_channel_id),
- }, {}, {},
+ (0, onion_packet, required),
+ (2, short_channel_id, required),
+ },
(1, Receive) => {
- (0, payment_data),
- (2, incoming_cltv_expiry),
- }, {}, {}
+ (0, payment_data, required),
+ (2, incoming_cltv_expiry, required),
+ }
;);
impl_writeable_tlv_based!(PendingHTLCInfo, {
- (0, routing),
- (2, incoming_shared_secret),
- (4, payment_hash),
- (6, amt_to_forward),
- (8, outgoing_cltv_value)
-}, {}, {});
+ (0, routing, required),
+ (2, incoming_shared_secret, required),
+ (4, payment_hash, required),
+ (6, amt_to_forward, required),
+ (8, outgoing_cltv_value, required)
+});
impl_writeable_tlv_based_enum!(HTLCFailureMsg, ;
(0, Relay),
);
impl_writeable_tlv_based!(HTLCPreviousHopData, {
- (0, short_channel_id),
- (2, outpoint),
- (4, htlc_id),
- (6, incoming_packet_shared_secret)
-}, {}, {});
+ (0, short_channel_id, required),
+ (2, outpoint, required),
+ (4, htlc_id, required),
+ (6, incoming_packet_shared_secret, required)
+});
impl_writeable_tlv_based!(ClaimableHTLC, {
- (0, prev_hop),
- (2, value),
- (4, payment_data),
- (6, cltv_expiry),
-}, {}, {});
+ (0, prev_hop, required),
+ (2, value, required),
+ (4, payment_data, required),
+ (6, cltv_expiry, required),
+});
impl_writeable_tlv_based_enum!(HTLCSource,
(0, OutboundRoute) => {
- (0, session_priv),
- (2, first_hop_htlc_msat),
- }, {}, {
- (4, path),
- };
+ (0, session_priv, required),
+ (2, first_hop_htlc_msat, required),
+ (4, path, vec_type),
+ }, ;
(1, PreviousHopData)
);
impl_writeable_tlv_based_enum!(HTLCFailReason,
(0, LightningError) => {
- (0, err),
- }, {}, {},
+ (0, err, required),
+ },
(1, Reason) => {
- (0, failure_code),
- }, {}, {
- (2, data),
+ (0, failure_code, required),
+ (2, data, vec_type),
},
;);
impl_writeable_tlv_based_enum!(HTLCForwardInfo,
(0, AddHTLC) => {
- (0, forward_info),
- (2, prev_short_channel_id),
- (4, prev_htlc_id),
- (6, prev_funding_outpoint),
- }, {}, {},
+ (0, forward_info, required),
+ (2, prev_short_channel_id, required),
+ (4, prev_htlc_id, required),
+ (6, prev_funding_outpoint, required),
+ },
(1, FailHTLC) => {
- (0, htlc_id),
- (2, err_packet),
- }, {}, {},
+ (0, htlc_id, required),
+ (2, err_packet, required),
+ },
;);
impl_writeable_tlv_based!(PendingInboundPayment, {
- (0, payment_secret),
- (2, expiry_time),
- (4, user_payment_id),
- (6, payment_preimage),
- (8, min_value_msat),
-}, {}, {});
+ (0, payment_secret, required),
+ (2, expiry_time, required),
+ (4, user_payment_id, required),
+ (6, payment_preimage, required),
+ (8, min_value_msat, required),
+});
impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable for ChannelManager<Signer, M, T, K, F, L>
where M::Target: chain::Watch<Signer>,
session_priv.write(writer)?;
}
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
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:
+ log_error!(args.logger, "A ChannelMonitor is stale compared to the current ChannelManager! This indicates a potentially-critical violation of the chain::Watch API!");
+ log_error!(args.logger, " The ChannelMonitor for channel {} is at update_id {} but the ChannelManager is at update_id {}.",
+ log_bytes!(channel.channel_id()), monitor.get_latest_update_id(), channel.get_latest_monitor_update_id());
+ log_error!(args.logger, " The chain::Watch API *requires* that monitors are persisted durably before returning,");
+ log_error!(args.logger, " client applications must ensure that ChannelMonitor data is always available and the latest to avoid funds loss!");
+ log_error!(args.logger, " Without the latest ChannelMonitor we cannot continue without risking funds.");
+ log_error!(args.logger, " Please ensure the chain::Watch API requirements are met and file a bug report at https://github.com/rust-bitcoin/rust-lightning");
return Err(DecodeError::InvalidValue);
} 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() ||
by_id.insert(channel.channel_id(), channel);
}
} else {
+ log_error!(args.logger, "Missing ChannelMonitor for channel {} needed by ChannelManager.", log_bytes!(channel.channel_id()));
+ log_error!(args.logger, " The chain::Watch API *requires* that monitors are persisted durably before returning,");
+ log_error!(args.logger, " client applications must ensure that ChannelMonitor data is always available and the latest to avoid funds loss!");
+ log_error!(args.logger, " Without the ChannelMonitor we cannot continue without risking funds.");
+ log_error!(args.logger, " Please ensure the chain::Watch API requirements are met and file a bug report at https://github.com/rust-bitcoin/rust-lightning");
return Err(DecodeError::InvalidValue);
}
}
}
}
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&args.keys_manager.get_secure_random_bytes());
use core::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use core::time::Duration;
+ use ln::functional_test_utils::*;
+ use ln::features::InitFeatures;
+ use ln::msgs::ChannelMessageHandler;
#[test]
fn test_wait_timeout() {
}
}
}
+
+ #[test]
+ fn test_notify_limits() {
+ // Check that a few cases which don't require the persistence of a new ChannelManager,
+ // indeed, do not cause the persistence of a new ChannelManager.
+ let chanmon_cfgs = create_chanmon_cfgs(3);
+ let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+ let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+ let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+ let mut chan = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+
+ // We check that the channel info nodes have doesn't change too early, even though we try
+ // to connect messages with new values
+ chan.0.contents.fee_base_msat *= 2;
+ chan.1.contents.fee_base_msat *= 2;
+ let node_a_chan_info = nodes[0].node.list_channels()[0].clone();
+ let node_b_chan_info = nodes[1].node.list_channels()[0].clone();
+
+ // The first two nodes (which opened a channel) should now require fresh persistence
+ assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ // ... but the last node should not.
+ assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ // After persisting the first two nodes they should no longer need fresh persistence.
+ assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+
+ // Node 3, unrelated to the only channel, shouldn't care if it receives a channel_update
+ // about the channel.
+ nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.0);
+ nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.1);
+ assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1)));
+
+ // The nodes which are a party to the channel should also ignore messages from unrelated
+ // parties.
+ nodes[0].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.0);
+ nodes[0].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1);
+ nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.0);
+ nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1);
+ assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1)));
+ assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1)));
+
+ // At this point the channel info given by peers should still be the same.
+ assert_eq!(nodes[0].node.list_channels()[0], node_a_chan_info);
+ assert_eq!(nodes[1].node.list_channels()[0], node_b_chan_info);
+ }
}
#[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))]
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use core::mem;
-use std::collections::HashMap;
pub const CHAN_CONFIRM_DEPTH: u32 = 10;
pub fn best_block_info(&self) -> (BlockHash, u32) {
self.blocks.lock().unwrap().last().map(|(a, b)| (a.block_hash(), *b)).unwrap()
}
+ pub fn get_block_header(&self, height: u32) -> BlockHeader {
+ self.blocks.lock().unwrap()[height as usize].0
+ }
}
impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
}
}
+#[cfg(test)]
+macro_rules! expect_payment_failure_chan_update {
+ ($node: expr, $scid: expr, $chan_closed: expr) => {
+ let events = $node.node.get_and_clear_pending_msg_events();
+ assert_eq!(events.len(), 1);
+ match events[0] {
+ MessageSendEvent::PaymentFailureNetworkUpdate { ref update } => {
+ match update {
+ &HTLCFailChannelUpdate::ChannelUpdateMessage { ref msg } if !$chan_closed => {
+ assert_eq!(msg.contents.short_channel_id, $scid);
+ assert_eq!(msg.contents.flags & 2, 0);
+ },
+ &HTLCFailChannelUpdate::ChannelClosed { short_channel_id, is_permanent } if $chan_closed => {
+ assert_eq!(short_channel_id, $scid);
+ assert!(is_permanent);
+ },
+ _ => panic!("Unexpected update type"),
+ }
+ },
+ _ => panic!("Unexpected event"),
+ }
+ }
+}
+
#[cfg(test)]
macro_rules! expect_payment_failed {
($node: expr, $expected_payment_hash: expr, $rejected_by_dest: expr $(, $expected_error_code: expr, $expected_error_data: expr)*) => {
//! claim outputs on-chain.
use chain;
-use chain::Listen;
-use chain::Watch;
+use chain::{Confirm, Listen, Watch};
use chain::channelmonitor;
use chain::channelmonitor::{ChannelMonitor, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY};
use chain::transaction::OutPoint;
use prelude::*;
use alloc::collections::BTreeSet;
-use std::collections::{HashMap, HashSet};
use core::default::Default;
use std::sync::{Arc, Mutex};
nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &bs_fail_updates.update_fail_htlcs[0]);
commitment_signed_dance!(nodes[0], nodes[1], bs_fail_updates.commitment_signed, false, true);
- let events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- match events[0] {
- MessageSendEvent::PaymentFailureNetworkUpdate { update: msgs::HTLCFailChannelUpdate::ChannelUpdateMessage { ref msg }} => {
- assert_eq!(msg.contents.short_channel_id, chan_2.0.contents.short_channel_id);
- },
- _ => panic!("Unexpected event"),
- }
-
+ expect_payment_failure_chan_update!(nodes[0], chan_2.0.contents.short_channel_id, false);
expect_payment_failed!(nodes[0], payment_hash_2, false);
// Now forward all the pending HTLCs and claim them back
},
_ => panic!("Unexpected event"),
};
- nodes[1].logger.assert_log("lightning::ln::channel".to_string(), "Attempting to fail HTLC due to fee spike buffer violation".to_string(), 1);
+ nodes[1].logger.assert_log("lightning::ln::channel".to_string(),
+ format!("Attempting to fail HTLC due to fee spike buffer violation in channel {}. Rebalancing is required.", ::hex::encode(raa_msg.channel_id)), 1);
check_added_monitors!(nodes[1], 2);
}
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
{
- // B will rebroadcast its own holder commitment transaction here...just because
+ // B may rebroadcast its own holder commitment transaction here, as a safeguard against
+ // some incredibly unlikely partial-eclipse-attack scenarios. That said, because the
+ // original commitment_tx[0] (also spending chan_2.3) has reached ANTI_REORG_DELAY B really
+ // shouldn't broadcast anything here, and in some connect style scenarios we do not.
let node_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
- assert_eq!(node_txn.len(), 1);
- check_spends!(node_txn[0], chan_2.3);
+ if node_txn.len() == 1 {
+ check_spends!(node_txn[0], chan_2.3);
+ } else {
+ assert_eq!(node_txn.len(), 0);
+ }
}
expect_pending_htlcs_forwardable!(nodes[1]);
nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &update_fail_htlcs[0]);
commitment_signed_dance!(nodes[0], nodes[1], commitment_signed, false, true);
-
- let events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- match events[0] {
- MessageSendEvent::PaymentFailureNetworkUpdate { .. } => {},
- _ => panic!("Unexpected event"),
- }
+ expect_payment_failure_chan_update!(nodes[0], chan_2.0.contents.short_channel_id, true);
expect_payment_failed!(nodes[0], payment_hash, false);
},
_ => panic!("Unexpected event"),
fail_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_hash_6);
}
-fn do_test_drop_messages_peer_disconnect(messages_delivered: u8) {
+fn do_test_drop_messages_peer_disconnect(messages_delivered: u8, simulate_broken_lnd: bool) {
// Test that we can reconnect when in-flight HTLC updates get dropped
let chanmon_cfgs = create_chanmon_cfgs(2);
let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+ let mut as_funding_locked = None;
if messages_delivered == 0 {
- create_chan_between_nodes_with_value_a(&nodes[0], &nodes[1], 100000, 10001, InitFeatures::known(), InitFeatures::known());
+ let (funding_locked, _, _) = create_chan_between_nodes_with_value_a(&nodes[0], &nodes[1], 100000, 10001, InitFeatures::known(), InitFeatures::known());
+ as_funding_locked = Some(funding_locked);
// nodes[1] doesn't receive the funding_locked message (it'll be re-sent on reconnect)
+ // Note that we store it so that if we're running with `simulate_broken_lnd` we can deliver
+ // it before the channel_reestablish message.
} else {
create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
}
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false);
if messages_delivered < 3 {
+ if simulate_broken_lnd {
+ // lnd has a long-standing bug where they send a funding_locked prior to a
+ // channel_reestablish if you reconnect prior to funding_locked time.
+ //
+ // Here we simulate that behavior, delivering a funding_locked immediately on
+ // reconnect. Note that we don't bother skipping the now-duplicate funding_locked sent
+ // in `reconnect_nodes` but we currently don't fail based on that.
+ //
+ // See-also <https://github.com/lightningnetwork/lnd/issues/4006>
+ nodes[1].node.handle_funding_locked(&nodes[0].node.get_our_node_id(), &as_funding_locked.as_ref().unwrap().0);
+ }
// Even if the funding_locked messages get exchanged, as long as nothing further was
// received on either side, both sides will need to resend them.
reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 1), (0, 0), (0, 0), (0, 0), (false, false));
#[test]
fn test_drop_messages_peer_disconnect_a() {
- do_test_drop_messages_peer_disconnect(0);
- do_test_drop_messages_peer_disconnect(1);
- do_test_drop_messages_peer_disconnect(2);
- do_test_drop_messages_peer_disconnect(3);
+ do_test_drop_messages_peer_disconnect(0, true);
+ do_test_drop_messages_peer_disconnect(0, false);
+ do_test_drop_messages_peer_disconnect(1, false);
+ do_test_drop_messages_peer_disconnect(2, false);
}
#[test]
fn test_drop_messages_peer_disconnect_b() {
- do_test_drop_messages_peer_disconnect(4);
- do_test_drop_messages_peer_disconnect(5);
- do_test_drop_messages_peer_disconnect(6);
+ do_test_drop_messages_peer_disconnect(3, false);
+ do_test_drop_messages_peer_disconnect(4, false);
+ do_test_drop_messages_peer_disconnect(5, false);
+ do_test_drop_messages_peer_disconnect(6, false);
}
#[test]
let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
- create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known());
+ let chan_2 = create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known());
// Make sure all nodes are at the same starting height
connect_blocks(&nodes[0], 2*CHAN_CONFIRM_DEPTH + 1 - nodes[0].best_block_info().1);
_ => unreachable!(),
}
expect_payment_failed!(nodes[0], second_payment_hash, false);
- if let &MessageSendEvent::PaymentFailureNetworkUpdate { ref update } = &nodes[0].node.get_and_clear_pending_msg_events()[0] {
- match update {
- &HTLCFailChannelUpdate::ChannelUpdateMessage { .. } => {},
- _ => panic!("Unexpected event"),
- }
- } else {
- panic!("Unexpected event");
- }
+ expect_payment_failure_chan_update!(nodes[0], chan_2.0.contents.short_channel_id, false);
} else {
expect_payment_failed!(nodes[1], second_payment_hash, true);
}
}
macro_rules! check_spendable_outputs {
- ($node: expr, $der_idx: expr, $keysinterface: expr, $chan_value: expr) => {
+ ($node: expr, $keysinterface: expr) => {
{
let mut events = $node.chain_monitor.chain_monitor.get_and_clear_pending_events();
let mut txn = Vec::new();
mine_transaction(&nodes[1], &node_txn[0]);
connect_blocks(&nodes[1], BREAKDOWN_TIMEOUT as u32 - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
assert_eq!(spend_txn[0].input.len(), 1);
check_spends!(spend_txn[0], node_txn[0]);
check_added_monitors!(nodes[1], 1);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], node_txn[0]);
}
mine_transaction(&nodes[1], &node_txn[0]);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 3);
check_spends!(spend_txn[0], revoked_local_txn[0]); // to_remote output on revoked remote commitment_tx
check_spends!(spend_txn[1], node_txn[0]);
mine_transaction(&nodes[1], &node_txn[0]);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], node_txn[0]);
}
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
expect_payment_failed!(nodes[1], our_payment_hash, true);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 3); // SpendableOutput: remote_commitment_tx.to_remote, timeout_tx.output
check_spends!(spend_txn[0], commitment_tx[0]);
check_spends!(spend_txn[1], node_txn[1]);
mine_transaction(&nodes[1], &node_txn[0]);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], node_txn[0]);
}
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
// Check B's ChannelMonitor was able to generate the right spendable output descriptor
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
assert_eq!(spend_txn[0].input.len(), 1);
check_spends!(spend_txn[0], node_txn[1]);
// didn't try to generate any new transactions.
// Check A's ChannelMonitor was able to generate the right spendable output descriptor
- let spend_txn = check_spendable_outputs!(nodes[0], 1, node_cfgs[0].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[0], node_cfgs[0].keys_manager);
assert_eq!(spend_txn.len(), 3);
assert_eq!(spend_txn[0].input.len(), 1);
check_spends!(spend_txn[0], revoked_local_txn[0]); // spending to_remote output from revoked local tx
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
{
commitment_signed_dance!(nodes[0], nodes[1], &htlc_updates.commitment_signed, false, true);
- let events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- match events[0] {
- MessageSendEvent::PaymentFailureNetworkUpdate { update: msgs::HTLCFailChannelUpdate::ChannelClosed { .. } } => {
- },
- _ => { panic!("Unexpected event"); }
- }
+ expect_payment_failure_chan_update!(nodes[0], chan_2.0.contents.short_channel_id, true);
}
expect_payment_failed!(nodes[0], duplicate_payment_hash, false);
connect_blocks(&nodes[1], BREAKDOWN_TIMEOUT as u32 - 1);
// Verify that B is able to spend its own HTLC-Success tx thanks to spendable output event given back by its ChannelMonitor
- let spend_txn = check_spendable_outputs!(nodes[1], 1, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
assert_eq!(spend_txn[0].input.len(), 1);
check_spends!(spend_txn[0], node_tx);
expect_payment_failed!(nodes[0], our_payment_hash, true);
// Verify that A is able to spend its own HTLC-Timeout tx thanks to spendable output event given back by its ChannelMonitor
- let spend_txn = check_spendable_outputs!(nodes[0], 1, node_cfgs[0].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[0], node_cfgs[0].keys_manager);
assert_eq!(spend_txn.len(), 3);
check_spends!(spend_txn[0], local_txn[0]);
assert_eq!(spend_txn[1].input.len(), 1);
// Verify that A is able to spend its own HTLC-Timeout tx thanks to spendable output event given back by its ChannelMonitor
let new_keys_manager = test_utils::TestKeysInterface::new(&seed, Network::Testnet);
- let spend_txn = check_spendable_outputs!(nodes[0], 1, new_keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[0], new_keys_manager);
assert_eq!(spend_txn.len(), 3);
check_spends!(spend_txn[0], local_txn_1[0]);
assert_eq!(spend_txn[1].input.len(), 1);
mine_transaction(&nodes[0], &closing_tx);
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[0], 2, node_cfgs[0].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[0], node_cfgs[0].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], closing_tx);
mine_transaction(&nodes[1], &closing_tx);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[1], 2, node_cfgs[1].keys_manager, 100000);
+ let spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], closing_tx);
}
// us to surface its failure to the user.
chan_stat = get_channel_value_stat!(nodes[0], chan.2);
assert_eq!(chan_stat.holding_cell_outbound_amount_msat, 0);
- nodes[0].logger.assert_log("lightning::ln::channel".to_string(), "Freeing holding cell with 1 HTLC updates".to_string(), 1);
- let failure_log = format!("Failed to send HTLC with payment_hash {} due to Cannot send value that would put our balance under counterparty-announced channel reserve value ({})", log_bytes!(our_payment_hash.0), chan_stat.channel_reserve_msat);
+ nodes[0].logger.assert_log("lightning::ln::channel".to_string(), format!("Freeing holding cell with 1 HTLC updates in channel {}", hex::encode(chan.2)), 1);
+ let failure_log = format!("Failed to send HTLC with payment_hash {} due to Cannot send value that would put our balance under counterparty-announced channel reserve value ({}) in channel {}",
+ hex::encode(our_payment_hash.0), chan_stat.channel_reserve_msat, hex::encode(chan.2));
nodes[0].logger.assert_log("lightning::ln::channel".to_string(), failure_log.to_string(), 1);
// Check that the payment failed to be sent out.
// to surface its failure to the user. The first payment should succeed.
chan_stat = get_channel_value_stat!(nodes[0], chan.2);
assert_eq!(chan_stat.holding_cell_outbound_amount_msat, 0);
- nodes[0].logger.assert_log("lightning::ln::channel".to_string(), "Freeing holding cell with 2 HTLC updates".to_string(), 1);
- let failure_log = format!("Failed to send HTLC with payment_hash {} due to Cannot send value that would put our balance under counterparty-announced channel reserve value ({})", log_bytes!(payment_hash_2.0), chan_stat.channel_reserve_msat);
+ nodes[0].logger.assert_log("lightning::ln::channel".to_string(), format!("Freeing holding cell with 2 HTLC updates in channel {}", hex::encode(chan.2)), 1);
+ let failure_log = format!("Failed to send HTLC with payment_hash {} due to Cannot send value that would put our balance under counterparty-announced channel reserve value ({}) in channel {}",
+ hex::encode(payment_hash_2.0), chan_stat.channel_reserve_msat, hex::encode(chan.2));
nodes[0].logger.assert_log("lightning::ln::channel".to_string(), failure_log.to_string(), 1);
// Check that the second payment failed to be sent out.
_ => panic!("Unexpected event"),
};
nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &raa);
- let fail_msg_event = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(fail_msg_event.len(), 1);
- match &fail_msg_event[0] {
- &MessageSendEvent::PaymentFailureNetworkUpdate { .. } => {},
- _ => panic!("Unexpected event"),
- }
- let failure_event = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(failure_event.len(), 1);
- match &failure_event[0] {
- &Event::PaymentFailed { rejected_by_dest, .. } => {
- assert!(!rejected_by_dest);
- },
- _ => panic!("Unexpected event"),
- }
+ expect_payment_failure_chan_update!(nodes[0], chan_1_2.0.contents.short_channel_id, false);
+ expect_payment_failed!(nodes[0], our_payment_hash, false);
check_added_monitors!(nodes[0], 1);
}
assert_eq!(node_txn[0].output.len(), 2);
mine_transaction(&nodes[0], &node_txn[0]);
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
- let spend_txn = check_spendable_outputs!(nodes[0], 1, node_cfgs[0].keys_manager, 1000000);
+ let spend_txn = check_spendable_outputs!(nodes[0], node_cfgs[0].keys_manager);
assert_eq!(spend_txn.len(), 1);
check_spends!(spend_txn[0], node_txn[0]);
}
} else { panic!(); }
assert_eq!(nodes[1].node.list_channels().len(), 0);
}
+
+fn do_test_tx_confirmed_skipping_blocks_immediate_broadcast(test_height_before_timelock: bool) {
+ // In the first version of the chain::Confirm interface, after a refactor was made to not
+ // broadcast CSV-locked transactions until their CSV lock is up, we wouldn't reliably broadcast
+ // transactions after a `transactions_confirmed` call. Specifically, if the chain, provided via
+ // `best_block_updated` is at height N, and a transaction output which we wish to spend at
+ // height N-1 (due to a CSV to height N-1) is provided at height N, we will not broadcast the
+ // spending transaction until height N+1 (or greater). This was due to the way
+ // `ChannelMonitor::transactions_confirmed` worked, only checking if we should broadcast a
+ // spending transaction at the height the input transaction was confirmed at, not whether we
+ // should broadcast a spending transaction at the current height.
+ // A second, similar, issue involved failing HTLCs backwards - because we only provided the
+ // height at which transactions were confirmed to `OnchainTx::update_claims_view`, it wasn't
+ // aware that the anti-reorg-delay had, in fact, already expired, waiting to fail-backwards
+ // until we learned about an additional block.
+ //
+ // As an additional check, if `test_height_before_timelock` is set, we instead test that we
+ // aren't broadcasting transactions too early (ie not broadcasting them at all).
+ let chanmon_cfgs = create_chanmon_cfgs(3);
+ let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+ let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+ let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+ *nodes[0].connect_style.borrow_mut() = ConnectStyle::BestBlockFirstSkippingBlocks;
+
+ create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+ let (chan_announce, _, channel_id, _) = create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known());
+ let (_, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000);
+ nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false);
+ nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false);
+
+ nodes[1].node.force_close_channel(&channel_id).unwrap();
+ check_closed_broadcast!(nodes[1], true);
+ check_added_monitors!(nodes[1], 1);
+ let node_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
+ assert_eq!(node_txn.len(), 1);
+
+ let conf_height = nodes[1].best_block_info().1;
+ if !test_height_before_timelock {
+ connect_blocks(&nodes[1], 24 * 6);
+ }
+ nodes[1].chain_monitor.chain_monitor.transactions_confirmed(
+ &nodes[1].get_block_header(conf_height), &[(0, &node_txn[0])], conf_height);
+ if test_height_before_timelock {
+ // If we confirmed the close transaction, but timelocks have not yet expired, we should not
+ // generate any events or broadcast any transactions
+ assert!(nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().is_empty());
+ assert!(nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
+ } else {
+ // We should broadcast an HTLC transaction spending our funding transaction first
+ let spending_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
+ assert_eq!(spending_txn.len(), 2);
+ assert_eq!(spending_txn[0], node_txn[0]);
+ check_spends!(spending_txn[1], node_txn[0]);
+ // We should also generate a SpendableOutputs event with the to_self output (as its
+ // timelock is up).
+ let descriptor_spend_txn = check_spendable_outputs!(nodes[1], node_cfgs[1].keys_manager);
+ assert_eq!(descriptor_spend_txn.len(), 1);
+
+ // If we also discover that the HTLC-Timeout transaction was confirmed some time ago, we
+ // should immediately fail-backwards the HTLC to the previous hop, without waiting for an
+ // additional block built on top of the current chain.
+ nodes[1].chain_monitor.chain_monitor.transactions_confirmed(
+ &nodes[1].get_block_header(conf_height + 1), &[(0, &spending_txn[1])], conf_height + 1);
+ expect_pending_htlcs_forwardable!(nodes[1]);
+ check_added_monitors!(nodes[1], 1);
+
+ let updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+ assert!(updates.update_add_htlcs.is_empty());
+ assert!(updates.update_fulfill_htlcs.is_empty());
+ assert_eq!(updates.update_fail_htlcs.len(), 1);
+ assert!(updates.update_fail_malformed_htlcs.is_empty());
+ assert!(updates.update_fee.is_none());
+ nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fail_htlcs[0]);
+ commitment_signed_dance!(nodes[0], nodes[1], updates.commitment_signed, true, true);
+ expect_payment_failed!(nodes[0], payment_hash, false);
+ expect_payment_failure_chan_update!(nodes[0], chan_announce.contents.short_channel_id, true);
+ }
+}
+#[test]
+fn test_tx_confirmed_skipping_blocks_immediate_broadcast() {
+ do_test_tx_confirmed_skipping_blocks_immediate_broadcast(false);
+ do_test_tx_confirmed_skipping_blocks_immediate_broadcast(true);
+}
use std::io::Read;
use util::events::MessageSendEventsProvider;
+use util::logger;
use util::ser::{Readable, Writeable, Writer, FixedLengthReader, HighZeroBytesDroppedVarInt};
use ln::{PaymentPreimage, PaymentHash, PaymentSecret};
msg: Option<ErrorMessage>
},
/// The peer did something harmless that we weren't able to process, just log and ignore
+ // New code should *not* use this. New code must use IgnoreAndLog, below!
IgnoreError,
+ /// The peer did something harmless that we weren't able to meaningfully process.
+ /// If the error is logged, log it at the given level.
+ IgnoreAndLog(logger::Level),
/// The peer did something incorrect. Tell them.
SendErrorMessage {
/// The message to send.
},
OnionHopDataFormat::NonFinalNode { short_channel_id } => {
encode_varint_length_prefixed_tlv!(w, {
- (2, HighZeroBytesDroppedVarInt(self.amt_to_forward)),
- (4, HighZeroBytesDroppedVarInt(self.outgoing_cltv_value)),
- (6, short_channel_id)
- }, { });
+ (2, HighZeroBytesDroppedVarInt(self.amt_to_forward), required),
+ (4, HighZeroBytesDroppedVarInt(self.outgoing_cltv_value), required),
+ (6, short_channel_id, required)
+ });
},
OnionHopDataFormat::FinalNode { ref payment_data } => {
if let Some(final_data) = payment_data {
if final_data.total_msat > MAX_VALUE_MSAT { panic!("We should never be sending infinite/overflow onion payments"); }
}
encode_varint_length_prefixed_tlv!(w, {
- (2, HighZeroBytesDroppedVarInt(self.amt_to_forward)),
- (4, HighZeroBytesDroppedVarInt(self.outgoing_cltv_value))
- }, {
- (8, payment_data)
+ (2, HighZeroBytesDroppedVarInt(self.amt_to_forward), required),
+ (4, HighZeroBytesDroppedVarInt(self.outgoing_cltv_value), required),
+ (8, payment_data, option)
});
},
}
let mut cltv_value = HighZeroBytesDroppedVarInt(0u32);
let mut short_id: Option<u64> = None;
let mut payment_data: Option<FinalOnionHopData> = None;
- decode_tlv!(&mut rd, {
- (2, amt),
- (4, cltv_value)
- }, {
- (6, short_id),
- (8, payment_data)
+ decode_tlv_stream!(&mut rd, {
+ (2, amt, required),
+ (4, cltv_value, required),
+ (6, short_id, option),
+ (8, payment_data, option),
});
rd.eat_remaining().map_err(|_| DecodeError::ShortRead)?;
let format = if let Some(short_channel_id) = short_id {
use prelude::*;
use alloc::collections::LinkedList;
-use std::collections::{HashMap,hash_map,HashSet};
+use alloc::fmt::Debug;
use std::sync::{Arc, Mutex};
use core::sync::atomic::{AtomicUsize, Ordering};
use core::{cmp, hash, fmt, mem};
CM::Target: ChannelMessageHandler,
RM::Target: RoutingMessageHandler {
/// A message handler which handles messages specific to channels. Usually this is just a
- /// ChannelManager object or a ErroringMessageHandler.
+ /// [`ChannelManager`] object or an [`ErroringMessageHandler`].
+ ///
+ /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
pub chan_handler: CM,
/// A message handler which handles messages updating our knowledge of the network channel
- /// graph. Usually this is just a NetGraphMsgHandlerMonitor object or an IgnoringMessageHandler.
+ /// graph. Usually this is just a [`NetGraphMsgHandler`] object or an
+ /// [`IgnoringMessageHandler`].
+ ///
+ /// [`NetGraphMsgHandler`]: crate::routing::network_graph::NetGraphMsgHandler
pub route_handler: RM,
}
///
/// For efficiency, Clone should be relatively cheap for this type.
///
-/// You probably want to just extend an int and put a file descriptor in a struct and implement
-/// send_data. Note that if you are using a higher-level net library that may call close() itself,
-/// be careful to ensure you don't have races whereby you might register a new connection with an
-/// fd which is the same as a previous one which has yet to be removed via
-/// PeerManager::socket_disconnected().
+/// Two descriptors may compare equal (by [`cmp::Eq`] and [`hash::Hash`]) as long as the original
+/// has been disconnected, the [`PeerManager`] has been informed of the disconnection (either by it
+/// having triggered the disconnection or a call to [`PeerManager::socket_disconnected`]), and no
+/// further calls to the [`PeerManager`] related to the original socket occur. This allows you to
+/// use a file descriptor for your SocketDescriptor directly, however for simplicity you may wish
+/// to simply use another value which is guaranteed to be globally unique instead.
pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
/// Attempts to send some data from the given slice to the peer.
///
/// Returns the amount of data which was sent, possibly 0 if the socket has since disconnected.
- /// Note that in the disconnected case, socket_disconnected must still fire and further write
- /// attempts may occur until that time.
+ /// Note that in the disconnected case, [`PeerManager::socket_disconnected`] must still be
+ /// called and further write attempts may occur until that time.
///
- /// If the returned size is smaller than data.len(), a write_available event must
- /// trigger the next time more data can be written. Additionally, until the a send_data event
- /// completes fully, no further read_events should trigger on the same peer!
+ /// If the returned size is smaller than `data.len()`, a
+ /// [`PeerManager::write_buffer_space_avail`] call must be made the next time more data can be
+ /// written. Additionally, until a `send_data` event completes fully, no further
+ /// [`PeerManager::read_event`] calls should be made for the same peer! Because this is to
+ /// prevent denial-of-service issues, you should not read or buffer any data from the socket
+ /// until then.
///
- /// If a read_event on this descriptor had previously returned true (indicating that read
- /// events should be paused to prevent DoS in the send buffer), resume_read may be set
- /// indicating that read events on this descriptor should resume. A resume_read of false does
- /// *not* imply that further read events should be paused.
+ /// If a [`PeerManager::read_event`] call on this descriptor had previously returned true
+ /// (indicating that read events should be paused to prevent DoS in the send buffer),
+ /// `resume_read` may be set indicating that read events on this descriptor should resume. A
+ /// `resume_read` of false carries no meaning, and should not cause any action.
fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize;
- /// Disconnect the socket pointed to by this SocketDescriptor. Once this function returns, no
- /// more calls to write_buffer_space_avail, read_event or socket_disconnected may be made with
- /// this descriptor. No socket_disconnected call should be generated as a result of this call,
- /// though races may occur whereby disconnect_socket is called after a call to
- /// socket_disconnected but prior to socket_disconnected returning.
+ /// Disconnect the socket pointed to by this SocketDescriptor.
+ ///
+ /// You do *not* need to call [`PeerManager::socket_disconnected`] with this socket after this
+ /// call (doing so is a noop).
fn disconnect_socket(&mut self);
}
NodesSyncing(PublicKey),
}
+/// When the outbound buffer has this many messages, we'll stop reading bytes from the peer until
+/// we have fewer than this many messages in the outbound buffer again.
+/// We also use this as the target number of outbound gossip messages to keep in the write buffer,
+/// refilled as we send bytes.
+const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 10;
+/// When the outbound buffer has this many messages, we'll simply skip relaying gossip messages to
+/// the peer.
+const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = 20;
+
struct Peer {
channel_encryptor: PeerChannelEncryptor,
their_node_id: Option<PublicKey>,
struct PeerHolder<Descriptor: SocketDescriptor> {
peers: HashMap<Descriptor, Peer>,
- /// Added to by do_read_event for cases where we pushed a message onto the send buffer but
- /// didn't call do_attempt_write_data to avoid reentrancy. Cleared in process_events()
- peers_needing_send: HashSet<Descriptor>,
/// Only add to this set when noise completes:
node_id_to_descriptor: HashMap<PublicKey, Descriptor>,
}
/// helps with issues such as long function definitions.
pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'f L>;
-/// A PeerManager manages a set of peers, described by their SocketDescriptor and marshalls socket
-/// events into messages which it passes on to its MessageHandlers.
+/// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls
+/// socket events into messages which it passes on to its [`MessageHandler`].
+///
+/// Locks are taken internally, so you must never assume that reentrancy from a
+/// [`SocketDescriptor`] call back into [`PeerManager`] methods will not deadlock.
+///
+/// Calls to [`read_event`] will decode relevant messages and pass them to the
+/// [`ChannelMessageHandler`], likely doing message processing in-line. Thus, the primary form of
+/// parallelism in Rust-Lightning is in calls to [`read_event`]. Note, however, that calls to any
+/// [`PeerManager`] functions related to the same connection must occur only in serial, making new
+/// calls only after previous ones have returned.
///
/// Rather than using a plain PeerManager, it is preferable to use either a SimpleArcPeerManager
/// a SimpleRefPeerManager, for conciseness. See their documentation for more details, but
/// essentially you should default to using a SimpleRefPeerManager, and use a
/// SimpleArcPeerManager when you require a PeerManager with a static lifetime, such as when
/// you're using lightning-net-tokio.
+///
+/// [`read_event`]: PeerManager::read_event
pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> where
CM::Target: ChannelMessageHandler,
RM::Target: RoutingMessageHandler,
}
}
-/// Manages and reacts to connection events. You probably want to use file descriptors as PeerIds.
-/// PeerIds may repeat, but only after socket_disconnected() has been called.
impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<Descriptor, CM, RM, L> where
CM::Target: ChannelMessageHandler,
RM::Target: RoutingMessageHandler,
message_handler,
peers: Mutex::new(PeerHolder {
peers: HashMap::new(),
- peers_needing_send: HashSet::new(),
node_id_to_descriptor: HashMap::new()
}),
our_node_secret,
///
/// Returns a small number of bytes to send to the remote node (currently always 50).
///
- /// Panics if descriptor is duplicative with some other descriptor which has not yet had a
- /// socket_disconnected().
+ /// Panics if descriptor is duplicative with some other descriptor which has not yet been
+ /// [`socket_disconnected()`].
+ ///
+ /// [`socket_disconnected()`]: PeerManager::socket_disconnected
pub fn new_outbound_connection(&self, their_node_id: PublicKey, descriptor: Descriptor) -> Result<Vec<u8>, PeerHandleError> {
let mut peer_encryptor = PeerChannelEncryptor::new_outbound(their_node_id.clone(), self.get_ephemeral_key());
let res = peer_encryptor.get_act_one().to_vec();
/// call socket_disconnected for the new descriptor but must disconnect the connection
/// immediately.
///
- /// Panics if descriptor is duplicative with some other descriptor which has not yet had
- /// socket_disconnected called.
+ /// Panics if descriptor is duplicative with some other descriptor which has not yet been
+ /// [`socket_disconnected()`].
+ ///
+ /// [`socket_disconnected()`]: PeerManager::socket_disconnected
pub fn new_inbound_connection(&self, descriptor: Descriptor) -> Result<(), PeerHandleError> {
let peer_encryptor = PeerChannelEncryptor::new_inbound(&self.our_node_secret);
let pending_read_buffer = [0; 50].to_vec(); // Noise act one is 50 bytes
}
fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
- macro_rules! encode_and_send_msg {
- ($msg: expr) => {
- {
- log_trace!(self.logger, "Encoding and sending sync update message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg)[..]));
- }
- }
- }
- const MSG_BUFF_SIZE: usize = 10;
while !peer.awaiting_write_event {
- if peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE {
+ if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE {
match peer.sync_status {
InitSyncTracker::NoSyncRequested => {},
InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
- let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
+ let steps = ((OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
let all_messages = self.message_handler.route_handler.get_next_channel_announcements(c, steps);
for &(ref announce, ref update_a_option, ref update_b_option) in all_messages.iter() {
- encode_and_send_msg!(announce);
+ self.enqueue_message(peer, announce);
if let &Some(ref update_a) = update_a_option {
- encode_and_send_msg!(update_a);
+ self.enqueue_message(peer, update_a);
}
if let &Some(ref update_b) = update_b_option {
- encode_and_send_msg!(update_b);
+ self.enqueue_message(peer, update_b);
}
peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
}
}
},
InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
- let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
+ let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len()) as u8;
let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
for msg in all_messages.iter() {
- encode_and_send_msg!(msg);
+ self.enqueue_message(peer, msg);
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
}
if all_messages.is_empty() || all_messages.len() != steps as usize {
},
InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
InitSyncTracker::NodesSyncing(key) => {
- let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
+ let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len()) as u8;
let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
for msg in all_messages.iter() {
- encode_and_send_msg!(msg);
+ self.enqueue_message(peer, msg);
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
}
if all_messages.is_empty() || all_messages.len() != steps as usize {
Some(buff) => buff,
};
- let should_be_reading = peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE;
+ let should_be_reading = peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE;
let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..];
let data_sent = descriptor.send_data(pending, should_be_reading);
peer.pending_outbound_buffer_first_msg_offset += data_sent;
///
/// May return an Err to indicate that the connection should be closed.
///
- /// Will most likely call send_data on the descriptor passed in (or the descriptor handed into
- /// new_*\_connection) before returning. Thus, be very careful with reentrancy issues! The
- /// invariants around calling write_buffer_space_avail in case a write did not fully complete
- /// must still hold - be ready to call write_buffer_space_avail again if a write call generated
- /// here isn't sufficient! Panics if the descriptor was not previously registered in a
- /// new_\*_connection event.
+ /// May call [`send_data`] on the descriptor passed in (or an equal descriptor) before
+ /// returning. Thus, be very careful with reentrancy issues! The invariants around calling
+ /// [`write_buffer_space_avail`] in case a write did not fully complete must still hold - be
+ /// ready to call `[write_buffer_space_avail`] again if a write call generated here isn't
+ /// sufficient!
+ ///
+ /// [`send_data`]: SocketDescriptor::send_data
+ /// [`write_buffer_space_avail`]: PeerManager::write_buffer_space_avail
pub fn write_buffer_space_avail(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> {
let mut peers = self.peers.lock().unwrap();
match peers.peers.get_mut(descriptor) {
- None => panic!("Descriptor for write_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user found that the socket
+ // was writeable, then we told the user to `disconnect_socket()`, then they called
+ // this method. Return an error to make sure we get disconnected.
+ return Err(PeerHandleError { no_connection_possible: false });
+ },
Some(peer) => {
peer.awaiting_write_event = false;
self.do_attempt_write_data(descriptor, peer);
///
/// May return an Err to indicate that the connection should be closed.
///
- /// Will *not* call back into send_data on any descriptors to avoid reentrancy complexity.
- /// Thus, however, you almost certainly want to call process_events() after any read_event to
- /// generate send_data calls to handle responses.
+ /// Will *not* call back into [`send_data`] on any descriptors to avoid reentrancy complexity.
+ /// Thus, however, you should call [`process_events`] after any `read_event` to generate
+ /// [`send_data`] calls to handle responses.
///
- /// If Ok(true) is returned, further read_events should not be triggered until a send_data call
- /// on this file descriptor has resume_read set (preventing DoS issues in the send buffer).
+ /// If `Ok(true)` is returned, further read_events should not be triggered until a
+ /// [`send_data`] call on this descriptor has `resume_read` set (preventing DoS issues in the
+ /// send buffer).
///
- /// Panics if the descriptor was not previously registered in a new_*_connection event.
+ /// [`send_data`]: SocketDescriptor::send_data
+ /// [`process_events`]: PeerManager::process_events
pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
match self.do_read_event(peer_descriptor, data) {
Ok(res) => Ok(res),
}
/// Append a message to a peer's pending outbound/write buffer, and update the map of peers needing sends accordingly.
- fn enqueue_message<M: Encode + Writeable>(&self, peers_needing_send: &mut HashSet<Descriptor>, peer: &mut Peer, descriptor: Descriptor, message: &M) {
+ fn enqueue_message<M: Encode + Writeable + Debug>(&self, peer: &mut Peer, message: &M) {
let mut buffer = VecWriter(Vec::new());
wire::write(message, &mut buffer).unwrap(); // crash if the write failed
let encoded_message = buffer.0;
- log_trace!(self.logger, "Enqueueing message of type {} to {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+ log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap()));
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..]));
- peers_needing_send.insert(descriptor);
}
fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
let pause_read = {
let mut peers_lock = self.peers.lock().unwrap();
let peers = &mut *peers_lock;
+ let mut msgs_to_forward = Vec::new();
+ let mut peer_node_id = None;
let pause_read = match peers.peers.get_mut(peer_descriptor) {
- None => panic!("Descriptor for read_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user read some bytes
+ // from the socket, then we told the user to `disconnect_socket()`, then they
+ // called this method. Return an error to make sure we get disconnected.
+ return Err(PeerHandleError { no_connection_possible: false });
+ },
Some(peer) => {
assert!(peer.pending_read_buffer.len() > 0);
assert!(peer.pending_read_buffer.len() > peer.pending_read_buffer_pos);
match e.action {
msgs::ErrorAction::DisconnectPeer { msg: _ } => {
//TODO: Try to push msg
- log_trace!(self.logger, "Got Err handling message, disconnecting peer because {}", e.err);
+ log_debug!(self.logger, "Error handling message; disconnecting peer with: {}", e.err);
return Err(PeerHandleError{ no_connection_possible: false });
},
+ msgs::ErrorAction::IgnoreAndLog(level) => {
+ log_given_level!(self.logger, level, "Error handling message; ignoring: {}", e.err);
+ continue
+ },
msgs::ErrorAction::IgnoreError => {
- log_trace!(self.logger, "Got Err handling message, ignoring because {}", e.err);
+ log_debug!(self.logger, "Error handling message; ignoring: {}", e.err);
continue;
},
msgs::ErrorAction::SendErrorMessage { msg } => {
- log_trace!(self.logger, "Got Err handling message, sending Error message because {}", e.err);
- self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &msg);
+ log_debug!(self.logger, "Error handling message; sending error message with: {}", e.err);
+ self.enqueue_message(peer, &msg);
continue;
},
}
return Err(PeerHandleError{ no_connection_possible: false })
},
hash_map::Entry::Vacant(entry) => {
- log_trace!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap()));
+ log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap()));
entry.insert(peer_descriptor.clone())
},
};
insert_node_id!();
let features = InitFeatures::known();
let resp = msgs::Init { features };
- self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &resp);
+ self.enqueue_message(peer, &resp);
},
NextNoiseStep::ActThree => {
let their_node_id = try_potential_handleerror!(peer.channel_encryptor.process_act_three(&peer.pending_read_buffer[..]));
insert_node_id!();
let features = InitFeatures::known();
let resp = msgs::Init { features };
- self.enqueue_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), &resp);
+ self.enqueue_message(peer, &resp);
},
NextNoiseStep::NoiseComplete => {
if peer.pending_read_is_header {
match e {
msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
msgs::DecodeError::UnknownRequiredFeature => {
- log_debug!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
+ log_trace!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
continue;
}
msgs::DecodeError::InvalidValue => {
msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
msgs::DecodeError::UnsupportedCompression => {
- log_debug!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
+ log_trace!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
continue;
}
}
}
};
- if let Err(handling_error) = self.handle_message(&mut peers.peers_needing_send, peer, peer_descriptor.clone(), message){
- match handling_error {
+ match self.handle_message(peer, message) {
+ Err(handling_error) => match handling_error {
MessageHandlingError::PeerHandleError(e) => { return Err(e) },
MessageHandlingError::LightningError(e) => {
try_potential_handleerror!(Err(e));
},
- }
+ },
+ Ok(Some(msg)) => {
+ peer_node_id = Some(peer.their_node_id.expect("After noise is complete, their_node_id is always set"));
+ msgs_to_forward.push(msg);
+ },
+ Ok(None) => {},
}
}
}
}
}
- self.do_attempt_write_data(peer_descriptor, peer);
-
- peer.pending_outbound_buffer.len() > 10 // pause_read
+ peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_READ_PAUSE // pause_read
}
};
+ for msg in msgs_to_forward.drain(..) {
+ self.forward_broadcast_msg(peers, &msg, peer_node_id.as_ref());
+ }
+
pause_read
};
}
/// Process an incoming message and return a decision (ok, lightning error, peer handling error) regarding the next action with the peer
- fn handle_message(&self, peers_needing_send: &mut HashSet<Descriptor>, peer: &mut Peer, peer_descriptor: Descriptor, message: wire::Message) -> Result<(), MessageHandlingError> {
- log_trace!(self.logger, "Received message of type {} from {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+ /// Returns the message back if it needs to be broadcasted to all other peers.
+ fn handle_message(&self, peer: &mut Peer, message: wire::Message) -> Result<Option<wire::Message>, MessageHandlingError> {
+ log_trace!(self.logger, "Received message {:?} from {}", message, log_pubkey!(peer.their_node_id.unwrap()));
// Need an Init as first message
if let wire::Message::Init(_) = message {
} else if peer.their_features.is_none() {
- log_trace!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
+ log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
return Err(PeerHandleError{ no_connection_possible: false }.into());
}
+ let mut should_forward = None;
+
match message {
// Setup and Control messages:
wire::Message::Init(msg) => {
if msg.features.requires_unknown_bits() {
- log_info!(self.logger, "Peer features required unknown version bits");
+ log_debug!(self.logger, "Peer features required unknown version bits");
return Err(PeerHandleError{ no_connection_possible: true }.into());
}
if peer.their_features.is_some() {
if msg.features.initial_routing_sync() {
peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
- peers_needing_send.insert(peer_descriptor.clone());
}
if !msg.features.supports_static_remote_key() {
log_debug!(self.logger, "Peer {} does not support static remote key, disconnecting with no_connection_possible", log_pubkey!(peer.their_node_id.unwrap()));
wire::Message::Ping(msg) => {
if msg.ponglen < 65532 {
let resp = msgs::Pong { byteslen: msg.ponglen };
- self.enqueue_message(peers_needing_send, peer, peer_descriptor.clone(), &resp);
+ self.enqueue_message(peer, &resp);
}
},
wire::Message::Pong(_msg) => {
self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg);
},
wire::Message::ChannelAnnouncement(msg) => {
- let should_forward = match self.message_handler.route_handler.handle_channel_announcement(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- };
-
- if should_forward {
- // TODO: forward msg along to all our other peers!
+ if self.message_handler.route_handler.handle_channel_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::ChannelAnnouncement(msg));
}
},
wire::Message::NodeAnnouncement(msg) => {
- let should_forward = match self.message_handler.route_handler.handle_node_announcement(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- };
-
- if should_forward {
- // TODO: forward msg along to all our other peers!
+ if self.message_handler.route_handler.handle_node_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::NodeAnnouncement(msg));
}
},
wire::Message::ChannelUpdate(msg) => {
self.message_handler.chan_handler.handle_channel_update(&peer.their_node_id.unwrap(), &msg);
- let should_forward = match self.message_handler.route_handler.handle_channel_update(&msg) {
- Ok(v) => v,
- Err(e) => { return Err(e.into()); },
- };
-
- if should_forward {
- // TODO: forward msg along to all our other peers!
+ if self.message_handler.route_handler.handle_channel_update(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::ChannelUpdate(msg));
}
},
wire::Message::QueryShortChannelIds(msg) => {
log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", msg_type);
}
};
- Ok(())
+ Ok(should_forward)
+ }
+
+ fn forward_broadcast_msg(&self, peers: &mut PeerHolder<Descriptor>, msg: &wire::Message, except_node: Option<&PublicKey>) {
+ match msg {
+ wire::Message::ChannelAnnouncement(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?} or the announced channel's counterparties: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if peer.their_node_id.as_ref() == Some(&msg.contents.node_id_1) ||
+ peer.their_node_id.as_ref() == Some(&msg.contents.node_id_2) {
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ wire::Message::NodeAnnouncement(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?} or the announced node: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_node_announcement(msg.contents.node_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if peer.their_node_id.as_ref() == Some(&msg.contents.node_id) {
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ wire::Message::ChannelUpdate(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?}: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ _ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),
+ }
}
/// Checks for any events generated by our handlers and processes them. Includes sending most
/// response messages as well as messages generated by calls to handler functions directly (eg
- /// functions like ChannelManager::process_pending_htlc_forward or send_payment).
+ /// functions like [`ChannelManager::process_pending_htlc_forwards`] or [`send_payment`]).
+ ///
+ /// May call [`send_data`] on [`SocketDescriptor`]s. Thus, be very careful with reentrancy
+ /// issues!
+ ///
+ /// [`send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
+ /// [`ChannelManager::process_pending_htlc_forwards`]: crate::ln::channelmanager::ChannelManager::process_pending_htlc_forwards
+ /// [`send_data`]: SocketDescriptor::send_data
pub fn process_events(&self) {
{
// TODO: There are some DoS attacks here where you can flood someone's outbound send
let peers = &mut *peers_lock;
for event in events_generated.drain(..) {
macro_rules! get_peer_for_forwarding {
- ($node_id: expr, $handle_no_such_peer: block) => {
+ ($node_id: expr) => {
{
- let descriptor = match peers.node_id_to_descriptor.get($node_id) {
- Some(descriptor) => descriptor.clone(),
+ match peers.node_id_to_descriptor.get($node_id) {
+ Some(descriptor) => match peers.peers.get_mut(&descriptor) {
+ Some(peer) => {
+ if peer.their_features.is_none() {
+ continue;
+ }
+ peer
+ },
+ None => panic!("Inconsistent peers set state!"),
+ },
None => {
- $handle_no_such_peer;
continue;
},
- };
- match peers.peers.get_mut(&descriptor) {
- Some(peer) => {
- if peer.their_features.is_none() {
- $handle_no_such_peer;
- continue;
- }
- (descriptor, peer)
- },
- None => panic!("Inconsistent peers set state!"),
}
}
}
}
match event {
MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.temporary_channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Drop the pending channel? (or just let it timeout, but that sucks)
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.temporary_channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Drop the pending channel? (or just let it timeout, but that sucks)
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
+ log_debug!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
log_pubkey!(node_id),
log_bytes!(msg.temporary_channel_id),
log_funding_channel_id!(msg.funding_txid, msg.funding_output_index));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: generate a DiscardFunding event indicating to the wallet that
- //they should just throw away this funding transaction
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ // TODO: If the peer is gone we should generate a DiscardFunding event
+ // indicating to the wallet that they should just throw away this funding transaction
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: generate a DiscardFunding event indicating to the wallet that
- //they should just throw away this funding transaction
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
+ log_debug!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: generate a DiscardFunding event indicating to the wallet that
- //they should just throw away this funding transaction
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
- log_trace!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
+ log_debug!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
log_pubkey!(node_id),
update_add_htlcs.len(),
update_fulfill_htlcs.len(),
update_fail_htlcs.len(),
log_bytes!(commitment_signed.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
+ let peer = get_peer_for_forwarding!(node_id);
for msg in update_add_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(peer, msg);
}
for msg in update_fulfill_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(peer, msg);
}
for msg in update_fail_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(peer, msg);
}
for msg in update_fail_malformed_htlcs {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(peer, msg);
}
if let &Some(ref msg) = update_fee {
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(peer, msg);
}
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(peer, commitment_signed);
},
MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
- log_trace!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
+ log_debug!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
log_bytes!(msg.channel_id));
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
- MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
- log_trace!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
- if self.message_handler.route_handler.handle_channel_announcement(msg).is_ok() && self.message_handler.route_handler.handle_channel_update(update_msg).is_ok() {
- let encoded_msg = encode_msg!(msg);
- let encoded_update_msg = encode_msg!(update_msg);
-
- for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
- if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
- !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
- continue
- }
- match peer.their_node_id {
- None => continue,
- Some(their_node_id) => {
- if their_node_id == msg.contents.node_id_1 || their_node_id == msg.contents.node_id_2 {
- continue
- }
- }
- }
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_update_msg[..]));
- self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
- }
+ MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
+ log_debug!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+ if self.message_handler.route_handler.handle_channel_announcement(&msg).is_ok() && self.message_handler.route_handler.handle_channel_update(&update_msg).is_ok() {
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelAnnouncement(msg), None);
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(update_msg), None);
}
},
- MessageSendEvent::BroadcastNodeAnnouncement { ref msg } => {
- log_trace!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler");
- if self.message_handler.route_handler.handle_node_announcement(msg).is_ok() {
- let encoded_msg = encode_msg!(msg);
-
- for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
- if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
- !peer.should_forward_node_announcement(msg.contents.node_id) {
- continue
- }
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
- self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
- }
+ MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
+ log_debug!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler");
+ if self.message_handler.route_handler.handle_node_announcement(&msg).is_ok() {
+ self.forward_broadcast_msg(peers, &wire::Message::NodeAnnouncement(msg), None);
}
},
- MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
- log_trace!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
- if self.message_handler.route_handler.handle_channel_update(msg).is_ok() {
- let encoded_msg = encode_msg!(msg);
-
- for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
- if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
- !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
- continue
- }
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
- self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
- }
+ MessageSendEvent::BroadcastChannelUpdate { msg } => {
+ log_debug!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+ if self.message_handler.route_handler.handle_channel_update(&msg).is_ok() {
+ self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(msg), None);
}
},
MessageSendEvent::PaymentFailureNetworkUpdate { ref update } => {
match *action {
msgs::ErrorAction::DisconnectPeer { ref msg } => {
if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) {
- peers.peers_needing_send.remove(&descriptor);
if let Some(mut peer) = peers.peers.remove(&descriptor) {
if let Some(ref msg) = *msg {
log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
log_pubkey!(node_id),
msg.data);
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+ self.enqueue_message(&mut peer, msg);
// This isn't guaranteed to work, but if there is enough free
// room in the send buffer, put the error message there...
self.do_attempt_write_data(&mut descriptor, &mut peer);
self.message_handler.chan_handler.peer_disconnected(&node_id, false);
}
},
- msgs::ErrorAction::IgnoreError => {},
+ msgs::ErrorAction::IgnoreAndLog(level) => {
+ log_given_level!(self.logger, level, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+ },
+ msgs::ErrorAction::IgnoreError => {
+ log_debug!(self.logger, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+ },
msgs::ErrorAction::SendErrorMessage { ref msg } => {
log_trace!(self.logger, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
log_pubkey!(node_id),
msg.data);
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
- //TODO: Do whatever we're gonna do for handling dropped messages
- });
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
}
},
MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
},
MessageSendEvent::SendShortIdsQuery { ref node_id, ref msg } => {
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
}
MessageSendEvent::SendReplyChannelRange { ref node_id, ref msg } => {
log_trace!(self.logger, "Handling SendReplyChannelRange event in peer_handler for node {} with num_scids={} first_blocknum={} number_of_blocks={}, sync_complete={}",
msg.first_blocknum,
msg.number_of_blocks,
msg.sync_complete);
- let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {});
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
- self.do_attempt_write_data(&mut descriptor, peer);
+ self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
}
}
}
- for mut descriptor in peers.peers_needing_send.drain() {
- match peers.peers.get_mut(&descriptor) {
- Some(peer) => self.do_attempt_write_data(&mut descriptor, peer),
- None => panic!("Inconsistent peers set state!"),
- }
+ for (descriptor, peer) in peers.peers.iter_mut() {
+ self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
}
}
}
/// Indicates that the given socket descriptor's connection is now closed.
- ///
- /// This must only be called if the socket has been disconnected by the peer or your own
- /// decision to disconnect it and must NOT be called in any case where other parts of this
- /// library (eg PeerHandleError, explicit disconnect_socket calls) instruct you to disconnect
- /// the peer.
- ///
- /// Panics if the descriptor was not previously registered in a successful new_*_connection event.
pub fn socket_disconnected(&self, descriptor: &Descriptor) {
self.disconnect_event_internal(descriptor, false);
}
fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) {
let mut peers = self.peers.lock().unwrap();
- peers.peers_needing_send.remove(descriptor);
let peer_option = peers.peers.remove(descriptor);
match peer_option {
- None => panic!("Descriptor for disconnect_event is not already known to PeerManager"),
+ None => {
+ // This is most likely a simple race condition where the user found that the socket
+ // was disconnected, then we told the user to `disconnect_socket()`, then they
+ // called this method. Either way we're disconnected, return.
+ },
Some(peer) => {
match peer.their_node_id {
Some(node_id) => {
/// Disconnect a peer given its node id.
///
- /// Set no_connection_possible to true to prevent any further connection with this peer,
+ /// Set `no_connection_possible` to true to prevent any further connection with this peer,
/// force-closing any channels we have with it.
///
- /// If a peer is connected, this will call `disconnect_socket` on the descriptor for the peer,
- /// so be careful about reentrancy issues.
+ /// If a peer is connected, this will call [`disconnect_socket`] on the descriptor for the
+ /// peer. Thus, be very careful about reentrancy issues.
+ ///
+ /// [`disconnect_socket`]: SocketDescriptor::disconnect_socket
pub fn disconnect_by_node_id(&self, node_id: PublicKey, no_connection_possible: bool) {
let mut peers_lock = self.peers.lock().unwrap();
if let Some(mut descriptor) = peers_lock.node_id_to_descriptor.remove(&node_id) {
log_trace!(self.logger, "Disconnecting peer with id {} due to client request", node_id);
peers_lock.peers.remove(&descriptor);
- peers_lock.peers_needing_send.remove(&descriptor);
self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
descriptor.disconnect_socket();
}
}
/// This function should be called roughly once every 30 seconds.
- /// It will send pings to each peer and disconnect those which did not respond to the last round of pings.
-
- /// Will most likely call send_data on all of the registered descriptors, thus, be very careful with reentrancy issues!
+ /// It will send pings to each peer and disconnect those which did not respond to the last
+ /// round of pings.
+ ///
+ /// May call [`send_data`] on all [`SocketDescriptor`]s. Thus, be very careful with reentrancy
+ /// issues!
+ ///
+ /// [`send_data`]: SocketDescriptor::send_data
pub fn timer_tick_occurred(&self) {
let mut peers_lock = self.peers.lock().unwrap();
{
let peers = &mut *peers_lock;
- let peers_needing_send = &mut peers.peers_needing_send;
let node_id_to_descriptor = &mut peers.node_id_to_descriptor;
let peers = &mut peers.peers;
let mut descriptors_needing_disconnect = Vec::new();
peers.retain(|descriptor, peer| {
if peer.awaiting_pong {
- peers_needing_send.remove(descriptor);
descriptors_needing_disconnect.push(descriptor.clone());
match peer.their_node_id {
Some(node_id) => {
ponglen: 0,
byteslen: 64,
};
- peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(&ping)));
+ self.enqueue_message(peer, &ping);
let mut descriptor_clone = descriptor.clone();
self.do_attempt_write_data(&mut descriptor_clone, peer);
let initial_data = peer_b.new_outbound_connection(a_id, fd_b.clone()).unwrap();
peer_a.new_inbound_connection(fd_a.clone()).unwrap();
assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false);
+ peer_a.process_events();
assert_eq!(peer_b.read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap(), false);
+ peer_b.process_events();
assert_eq!(peer_a.read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap(), false);
(fd_a.clone(), fd_b.clone())
}
// peers[0] awaiting_pong is set to true, but the Peer is still connected
peers[0].timer_tick_occurred();
+ peers[0].process_events();
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
// Since timer_tick_occurred() is called again when awaiting_pong is true, all Peers are disconnected
peers[0].timer_tick_occurred();
+ peers[0].process_events();
assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
}
let (mut fd_a, mut fd_b) = establish_connection(&peers[0], &peers[1]);
// Make each peer to read the messages that the other peer just wrote to them.
+ peers[0].process_events();
peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap();
+ peers[1].process_events();
peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap();
// Check that each peer has received the expected number of channel updates and channel
use bitcoin::hash_types::BlockHash;
use prelude::*;
-use std::collections::HashMap;
use core::mem;
use ln::functional_test_utils::*;
/// A Lightning message returned by [`read()`] when decoding bytes received over the wire. Each
/// variant contains a message from [`msgs`] or otherwise the message type if unknown.
#[allow(missing_docs)]
+#[derive(Debug)]
pub enum Message {
Init(msgs::Init),
Error(msgs::ErrorMessage),
}
/// A number identifying a message to determine how it is encoded on the wire.
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
pub struct MessageType(u16);
impl Message {
+ #[allow(dead_code)] // This method is only used in tests
/// Returns the type that was used to decode the message payload.
pub fn type_id(&self) -> MessageType {
match self {
use ln::msgs::{QueryChannelRange, ReplyChannelRange, QueryShortChannelIds, ReplyShortChannelIdsEnd};
use ln::msgs;
use util::ser::{Writeable, Readable, Writer};
-use util::logger::Logger;
+use util::logger::{Logger, Level};
use util::events::{MessageSendEvent, MessageSendEventsProvider};
use util::scid_utils::{block_from_scid, scid_from_parts, MAX_SCID_BLOCK};
fn handle_htlc_fail_channel_update(&self, update: &msgs::HTLCFailChannelUpdate) {
match update {
&msgs::HTLCFailChannelUpdate::ChannelUpdateMessage { ref msg } => {
+ let chan_enabled = msg.contents.flags & (1 << 1) != (1 << 1);
+ log_debug!(self.logger, "Updating channel with channel_update from a payment failure. Channel {} is {}abled.", msg.contents.short_channel_id, if chan_enabled { "en" } else { "dis" });
let _ = self.network_graph.write().unwrap().update_channel(msg, &self.secp_ctx);
},
&msgs::HTLCFailChannelUpdate::ChannelClosed { short_channel_id, is_permanent } => {
+ log_debug!(self.logger, "{} channel graph entry for {} due to a payment failure.", if is_permanent { "Removing" } else { "Disabling" }, short_channel_id);
self.network_graph.write().unwrap().close_channel_from_update(short_channel_id, is_permanent);
},
&msgs::HTLCFailChannelUpdate::NodeFailure { ref node_id, is_permanent } => {
+ log_debug!(self.logger, "{} node graph entry for {} due to a payment failure.", if is_permanent { "Removing" } else { "Disabling" }, node_id);
self.network_graph.write().unwrap().fail_node(node_id, is_permanent);
},
}
}
impl_writeable_tlv_based!(DirectionalChannelInfo, {
- (0, last_update),
- (2, enabled),
- (4, cltv_expiry_delta),
- (6, htlc_minimum_msat),
- (8, htlc_maximum_msat),
- (10, fees),
- (12, last_update_message),
-}, {}, {});
+ (0, last_update, required),
+ (2, enabled, required),
+ (4, cltv_expiry_delta, required),
+ (6, htlc_minimum_msat, required),
+ (8, htlc_maximum_msat, required),
+ (10, fees, required),
+ (12, last_update_message, required),
+});
#[derive(Clone, Debug, PartialEq)]
/// Details about a channel (both directions).
}
impl_writeable_tlv_based!(ChannelInfo, {
- (0, features),
- (2, node_one),
- (4, one_to_two),
- (6, node_two),
- (8, two_to_one),
- (10, capacity_sats),
- (12, announcement_message),
-}, {}, {});
+ (0, features, required),
+ (2, node_one, required),
+ (4, one_to_two, required),
+ (6, node_two, required),
+ (8, two_to_one, required),
+ (10, capacity_sats, required),
+ (12, announcement_message, required),
+});
/// Fees for routing via a given channel or a node
pub proportional_millionths: u32,
}
-impl_writeable_tlv_based!(RoutingFees, {(0, base_msat), (2, proportional_millionths)}, {}, {});
+impl_writeable_tlv_based!(RoutingFees, {
+ (0, base_msat, required),
+ (2, proportional_millionths, required)
+});
#[derive(Clone, Debug, PartialEq)]
/// Information received in the latest node_announcement from this node.
}
impl_writeable_tlv_based!(NodeAnnouncementInfo, {
- (0, features),
- (2, last_update),
- (4, rgb),
- (6, alias),
-}, {
- (8, announcement_message),
-}, {
- (10, addresses),
+ (0, features, required),
+ (2, last_update, required),
+ (4, rgb, required),
+ (6, alias, required),
+ (8, announcement_message, option),
+ (10, addresses, vec_type),
});
#[derive(Clone, Debug, PartialEq)]
}
}
-impl_writeable_tlv_based!(NodeInfo, {}, {
- (0, lowest_inbound_channel_fees),
- (2, announcement_info),
-}, {
- (4, channels),
+impl_writeable_tlv_based!(NodeInfo, {
+ (0, lowest_inbound_channel_fees, option),
+ (2, announcement_info, option),
+ (4, channels, vec_type),
});
const SERIALIZATION_VERSION: u8 = 1;
node_info.write(writer)?;
}
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
}
let node_info = Readable::read(reader)?;
nodes.insert(node_id, node_info);
}
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
Ok(NetworkGraph {
genesis_hash,
Some(node) => {
if let Some(node_info) = node.announcement_info.as_ref() {
if node_info.last_update >= msg.timestamp {
- return Err(LightningError{err: "Update older than last processed update".to_owned(), action: ErrorAction::IgnoreError});
+ return Err(LightningError{err: "Update older than last processed update".to_owned(), action: ErrorAction::IgnoreAndLog(Level::Trace)});
}
}
Self::remove_channel_in_nodes(&mut self.nodes, &entry.get(), msg.short_channel_id);
*entry.get_mut() = chan_info;
} else {
- return Err(LightningError{err: "Already have knowledge of channel".to_owned(), action: ErrorAction::IgnoreError})
+ return Err(LightningError{err: "Already have knowledge of channel".to_owned(), action: ErrorAction::IgnoreAndLog(Level::Trace)})
}
},
BtreeEntry::Vacant(entry) => {
( $target: expr, $src_node: expr) => {
if let Some(existing_chan_info) = $target.as_ref() {
if existing_chan_info.last_update >= msg.timestamp {
- return Err(LightningError{err: "Update older than last processed update".to_owned(), action: ErrorAction::IgnoreError});
+ return Err(LightningError{err: "Update older than last processed update".to_owned(), action: ErrorAction::IgnoreAndLog(Level::Trace)});
}
chan_was_enabled = existing_chan_info.enabled;
} else {
use prelude::*;
use alloc::collections::BinaryHeap;
use core::cmp;
-use std::collections::HashMap;
use core::ops::Deref;
/// A hop in a route
}
impl_writeable_tlv_based!(RouteHop, {
- (0, pubkey),
- (2, node_features),
- (4, short_channel_id),
- (6, channel_features),
- (8, fee_msat),
- (10, cltv_expiry_delta),
-}, {}, {});
+ (0, pubkey, required),
+ (2, node_features, required),
+ (4, short_channel_id, required),
+ (6, channel_features, required),
+ (8, fee_msat, required),
+ (10, cltv_expiry_delta, required),
+});
/// A route directs a payment from the sender (us) to the recipient. If the recipient supports MPP,
/// it can take multiple paths. Each path is composed of one or more hops through the network.
hop.write(writer)?;
}
}
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
Ok(())
}
}
}
paths.push(hops);
}
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
Ok(Route { paths })
}
}
/// so that we can choose cheaper paths (as per Dijkstra's algorithm).
/// Fee values should be updated only in the context of the whole path, see update_value_and_recompute_fees.
/// These fee values are useful to choose hops as we traverse the graph "payee-to-payer".
-#[derive(Clone)]
+#[derive(Clone, Debug)]
struct PathBuildingHop<'a> {
// The RouteHintHop fields which will eventually be used if this hop is used in a final Route.
// Note that node_features is calculated separately after our initial graph walk.
// - when we want to stop looking for new paths.
let mut already_collected_value_msat = 0;
+ log_trace!(logger, "Building path from {} (payee) to {} (us/payer) for value {} msat.", payee, our_node_id, final_value_msat);
+
macro_rules! add_entry {
// Adds entry which goes from $src_node_id to $dest_node_id
// over the channel with id $chan_id with fees described in
}
}
+ log_trace!(logger, "Starting main path collection loop with {} nodes pre-filled from first/last hops.", targets.len());
+
// At this point, targets are filled with the data from first and
// last hops communicated by the caller, and the payment receiver.
let mut found_new_path = false;
ordered_hops.last_mut().unwrap().0.hop_use_fee_msat = 0;
ordered_hops.last_mut().unwrap().0.cltv_expiry_delta = final_cltv;
+ log_trace!(logger, "Found a path back to us from the target with {} hops contributing up to {} msat: {:?}",
+ ordered_hops.len(), value_contribution_msat, ordered_hops);
+
let mut payment_path = PaymentPath {hops: ordered_hops};
// We could have possibly constructed a slightly inconsistent path: since we reduce
// If we weren't capped by hitting a liquidity limit on a channel in the path,
// we'll probably end up picking the same path again on the next iteration.
// Decrease the available liquidity of a hop in the middle of the path.
- let victim_liquidity = bookkeeped_channels_liquidity_available_msat.get_mut(
- &payment_path.hops[(payment_path.hops.len() - 1) / 2].0.short_channel_id).unwrap();
+ let victim_scid = payment_path.hops[(payment_path.hops.len() - 1) / 2].0.short_channel_id;
+ log_trace!(logger, "Disabling channel {} for future path building iterations to avoid duplicates.", victim_scid);
+ let victim_liquidity = bookkeeped_channels_liquidity_available_msat.get_mut(&victim_scid).unwrap();
*victim_liquidity = 0;
}
// In the latter case, making another path finding attempt won't help,
// because we deterministically terminated the search due to low liquidity.
if already_collected_value_msat >= recommended_value_msat || !found_new_path {
+ log_trace!(logger, "Have now collected {} msat (seeking {} msat) in paths. Last path loop {} a new path.",
+ already_collected_value_msat, recommended_value_msat, if found_new_path { "found" } else { "did not find" });
break 'paths_collection;
} else if found_new_path && already_collected_value_msat == final_value_msat && payment_paths.len() == 1 {
// Further, if this was our first walk of the graph, and we weren't limited by an
// potentially allowing us to pay fees to meet the htlc_minimum on the new path while
// still keeping a lower total fee than this path.
if !hit_minimum_limit {
+ log_trace!(logger, "Collected exactly our payment amount on the first pass, without hitting an htlc_minimum_msat limit, exiting.");
break 'paths_collection;
}
+ log_trace!(logger, "Collected our payment amount on the first pass, but running again to collect extra paths with a potentially higher limit.");
path_value_msat = recommended_value_msat;
}
}
}
let route = Route { paths: selected_paths };
- log_trace!(logger, "Got route: {}", log_route!(route));
+ log_info!(logger, "Got route to {}: {}", payee, log_route!(route));
Ok(route)
}
}
}
+ #[cfg(not(feature = "no_std"))]
pub(super) fn random_init_seed() -> u64 {
// Because the default HashMap in std pulls OS randomness, we can use it as a (bad) RNG.
use core::hash::{BuildHasher, Hasher};
println!("Using seed of {}", seed);
seed
}
+ #[cfg(not(feature = "no_std"))]
use util::ser::Readable;
#[test]
+ #[cfg(not(feature = "no_std"))]
fn generate_routes() {
let mut d = match super::test_utils::get_route_file() {
Ok(f) => f,
}
#[test]
+ #[cfg(not(feature = "no_std"))]
fn generate_routes_mpp() {
let mut d = match super::test_utils::get_route_file() {
Ok(f) => f,
}
}
-#[cfg(test)]
+#[cfg(all(test, not(feature = "no_std")))]
pub(crate) mod test_utils {
use std::fs::File;
/// Tries to open a network graph file, or panics with a URL to fetch it.
}
}
-#[cfg(all(test, feature = "unstable"))]
+#[cfg(all(test, feature = "unstable", not(feature = "no_std")))]
mod benches {
use super::*;
use util::logger::{Logger, Record};
&Event::PaymentReceived { ref payment_hash, ref payment_preimage, ref payment_secret, ref amt, ref user_payment_id } => {
1u8.write(writer)?;
write_tlv_fields!(writer, {
- (0, payment_hash),
- (2, payment_secret),
- (4, amt),
- (6, user_payment_id),
- }, {
- (8, payment_preimage),
+ (0, payment_hash, required),
+ (2, payment_secret, required),
+ (4, amt, required),
+ (6, user_payment_id, required),
+ (8, payment_preimage, option),
});
},
&Event::PaymentSent { ref payment_preimage } => {
2u8.write(writer)?;
write_tlv_fields!(writer, {
- (0, payment_preimage),
- }, {});
- payment_preimage.write(writer)?;
+ (0, payment_preimage, required),
+ });
},
&Event::PaymentFailed { ref payment_hash, ref rejected_by_dest,
#[cfg(test)]
#[cfg(test)]
error_data.write(writer)?;
write_tlv_fields!(writer, {
- (0, payment_hash),
- (2, rejected_by_dest),
- }, {});
+ (0, payment_hash, required),
+ (2, rejected_by_dest, required),
+ });
},
&Event::PendingHTLCsForwardable { time_forwardable: _ } => {
4u8.write(writer)?;
- write_tlv_fields!(writer, {}, {});
+ write_tlv_fields!(writer, {});
// We don't write the time_fordwardable out at all, as we presume when the user
// deserializes us at least that much time has elapsed.
},
&Event::SpendableOutputs { ref outputs } => {
5u8.write(writer)?;
write_tlv_fields!(writer, {
- (0, VecWriteWrapper(outputs)),
- }, {});
+ (0, VecWriteWrapper(outputs), required),
+ });
},
}
Ok(())
let mut amt = 0;
let mut user_payment_id = 0;
read_tlv_fields!(reader, {
- (0, payment_hash),
- (2, payment_secret),
- (4, amt),
- (6, user_payment_id),
- }, {
- (8, payment_preimage),
+ (0, payment_hash, required),
+ (2, payment_secret, required),
+ (4, amt, required),
+ (6, user_payment_id, required),
+ (8, payment_preimage, option),
});
Ok(Some(Event::PaymentReceived {
payment_hash,
let f = || {
let mut payment_preimage = PaymentPreimage([0; 32]);
read_tlv_fields!(reader, {
- (0, payment_preimage),
- }, {});
+ (0, payment_preimage, required),
+ });
Ok(Some(Event::PaymentSent {
payment_preimage,
}))
let mut payment_hash = PaymentHash([0; 32]);
let mut rejected_by_dest = false;
read_tlv_fields!(reader, {
- (0, payment_hash),
- (2, rejected_by_dest),
- }, {});
+ (0, payment_hash, required),
+ (2, rejected_by_dest, required),
+ });
Ok(Some(Event::PaymentFailed {
payment_hash,
rejected_by_dest,
},
4u8 => {
let f = || {
- read_tlv_fields!(reader, {}, {});
+ read_tlv_fields!(reader, {});
Ok(Some(Event::PendingHTLCsForwardable {
time_forwardable: Duration::from_secs(0)
}))
let f = || {
let mut outputs = VecReadWrapper(Vec::new());
read_tlv_fields!(reader, {
- (0, outputs),
- }, {});
+ (0, outputs, required),
+ });
Ok(Some(Event::SpendableOutputs { outputs: outputs.0 }))
};
f()
use core::cmp;
use core::fmt;
-static LOG_LEVEL_NAMES: [&'static str; 6] = ["OFF", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"];
+static LOG_LEVEL_NAMES: [&'static str; 5] = ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"];
/// An enum representing the available verbosity levels of the logger.
#[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
pub enum Level {
- ///Designates logger being silent
- Off,
- /// Designates very serious errors
- Error,
- /// Designates hazardous situations
- Warn,
- /// Designates useful information
- Info,
- /// Designates lower priority information
- Debug,
/// Designates very low priority, often extremely verbose, information
Trace,
+ /// Designates lower priority information
+ Debug,
+ /// Designates useful information
+ Info,
+ /// Designates hazardous situations
+ Warn,
+ /// Designates very serious errors
+ Error,
}
impl PartialOrd for Level {
let wrapper = WrapperLog::new(Arc::clone(&logger));
wrapper.call_macros();
}
+
+ #[test]
+ fn test_log_ordering() {
+ assert!(Level::Error > Level::Warn);
+ assert!(Level::Error >= Level::Warn);
+ assert!(Level::Error >= Level::Error);
+ assert!(Level::Warn > Level::Info);
+ assert!(Level::Warn >= Level::Info);
+ assert!(Level::Warn >= Level::Warn);
+ assert!(Level::Info > Level::Debug);
+ assert!(Level::Info >= Level::Debug);
+ assert!(Level::Info >= Level::Info);
+ assert!(Level::Debug > Level::Trace);
+ assert!(Level::Debug >= Level::Trace);
+ assert!(Level::Debug >= Level::Debug);
+ assert!(Level::Trace >= Level::Trace);
+
+ assert!(Level::Error <= Level::Error);
+ assert!(Level::Warn < Level::Error);
+ assert!(Level::Warn <= Level::Error);
+ assert!(Level::Warn <= Level::Warn);
+ assert!(Level::Info < Level::Warn);
+ assert!(Level::Info <= Level::Warn);
+ assert!(Level::Info <= Level::Info);
+ assert!(Level::Debug < Level::Info);
+ assert!(Level::Debug <= Level::Info);
+ assert!(Level::Debug <= Level::Debug);
+ assert!(Level::Trace < Level::Debug);
+ assert!(Level::Trace <= Level::Debug);
+ assert!(Level::Trace <= Level::Trace);
+ }
}
);
}
+/// Logs an entry at the given level.
+#[macro_export]
+macro_rules! log_given_level {
+ ($logger: expr, $lvl:expr, $($arg:tt)+) => (
+ match $lvl {
+ #[cfg(not(any(feature = "max_level_off")))]
+ $crate::util::logger::Level::Error => log_internal!($logger, $lvl, $($arg)*),
+ #[cfg(not(any(feature = "max_level_off", feature = "max_level_error")))]
+ $crate::util::logger::Level::Warn => log_internal!($logger, $lvl, $($arg)*),
+ #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn")))]
+ $crate::util::logger::Level::Info => log_internal!($logger, $lvl, $($arg)*),
+ #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn", feature = "max_level_info")))]
+ $crate::util::logger::Level::Debug => log_internal!($logger, $lvl, $($arg)*),
+ #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn", feature = "max_level_info", feature = "max_level_debug")))]
+ $crate::util::logger::Level::Trace => log_internal!($logger, $lvl, $($arg)*),
+
+ #[cfg(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn", feature = "max_level_info", feature = "max_level_debug"))]
+ _ => {
+ // The level is disabled at compile-time
+ },
+ }
+ );
+}
+
/// Log an error.
#[macro_export]
macro_rules! log_error {
($logger: expr, $($arg:tt)*) => (
- #[cfg(not(any(feature = "max_level_off")))]
- log_internal!($logger, $crate::util::logger::Level::Error, $($arg)*);
+ log_given_level!($logger, $crate::util::logger::Level::Error, $($arg)*);
)
}
macro_rules! log_warn {
($logger: expr, $($arg:tt)*) => (
- #[cfg(not(any(feature = "max_level_off", feature = "max_level_error")))]
- log_internal!($logger, $crate::util::logger::Level::Warn, $($arg)*);
+ log_given_level!($logger, $crate::util::logger::Level::Warn, $($arg)*);
)
}
macro_rules! log_info {
($logger: expr, $($arg:tt)*) => (
- #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn")))]
- log_internal!($logger, $crate::util::logger::Level::Info, $($arg)*);
+ log_given_level!($logger, $crate::util::logger::Level::Info, $($arg)*);
)
}
macro_rules! log_debug {
($logger: expr, $($arg:tt)*) => (
- #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn", feature = "max_level_info")))]
- log_internal!($logger, $crate::util::logger::Level::Debug, $($arg)*);
+ log_given_level!($logger, $crate::util::logger::Level::Debug, $($arg)*);
)
}
#[macro_export]
macro_rules! log_trace {
($logger: expr, $($arg:tt)*) => (
- #[cfg(not(any(feature = "max_level_off", feature = "max_level_error", feature = "max_level_warn", feature = "max_level_info", feature = "max_level_debug")))]
- log_internal!($logger, $crate::util::logger::Level::Trace, $($arg)*);
+ log_given_level!($logger, $crate::util::logger::Level::Trace, $($arg)*);
)
}
//! Note this is not part of the specs, but follows lnd's signing and verifying protocol, which can is defined as follows:
//!
//! signature = zbase32(SigRec(sha256d(("Lightning Signed Message:" + msg)))
-//! zbase32 from https://philzimmermann.com/docs/human-oriented-base-32-encoding.txt
+//! zbase32 from <https://philzimmermann.com/docs/human-oriented-base-32-encoding.txt>
//! SigRec has first byte 31 + recovery id, followed by 64 byte sig.
//!
//! This implementation is compatible with both lnd's and c-lightning's
//!
-//! https://lightning.readthedocs.io/lightning-signmessage.7.html
-//! https://api.lightning.community/#signmessage
+//! <https://lightning.readthedocs.io/lightning-signmessage.7.html>
+//! <https://api.lightning.community/#signmessage>
use prelude::*;
use crate::util::zbase32;
/// Creates a digital signature of a message given a SecretKey, like the node's secret.
/// A receiver knowing the PublicKey (e.g. the node's id) and the message can be sure that the signature was generated by the caller.
/// Signatures are EC recoverable, meaning that given the message and the signature the PublicKey of the signer can be extracted.
-pub fn sign(msg: &[u8], sk: SecretKey) -> Result<String, Error> {
+pub fn sign(msg: &[u8], sk: &SecretKey) -> Result<String, Error> {
let secp_ctx = Secp256k1::signing_only();
let msg_hash = sha256d::Hash::hash(&[LN_MESSAGE_PREFIX, msg].concat());
- let sig = secp_ctx.sign_recoverable(&Message::from_slice(&msg_hash)?, &sk);
+ let sig = secp_ctx.sign_recoverable(&Message::from_slice(&msg_hash)?, sk);
Ok(zbase32::encode(&sigrec_encode(sig)))
}
/// Verifies a message was signed by a PrivateKey that derives to a given PublicKey, given a message, a signature,
/// and the PublicKey.
-pub fn verify(msg: &[u8], sig: &str, pk: PublicKey) -> bool {
+pub fn verify(msg: &[u8], sig: &str, pk: &PublicKey) -> bool {
match recover_pk(msg, sig) {
- Ok(x) => x == pk,
+ Ok(x) => x == *pk,
Err(_) => false
}
}
#[test]
fn test_sign() {
let message = "test message";
- let zbase32_sig = sign(message.as_bytes(), ONE_KEY);
+ let zbase32_sig = sign(message.as_bytes(), &ONE_KEY);
assert_eq!(zbase32_sig.unwrap(), "d9tibmnic9t5y41hg7hkakdcra94akas9ku3rmmj4ag9mritc8ok4p5qzefs78c9pqfhpuftqqzhydbdwfg7u6w6wdxcqpqn4sj4e73e")
}
#[test]
fn test_verify() {
let message = "another message";
- let sig = sign(message.as_bytes(), ONE_KEY).unwrap();
+ let sig = sign(message.as_bytes(), &ONE_KEY).unwrap();
let pk = PublicKey::from_secret_key(&Secp256k1::signing_only(), &ONE_KEY);
- assert!(verify(message.as_bytes(), &sig, pk))
+ assert!(verify(message.as_bytes(), &sig, &pk))
}
#[test]
];
for c in &corpus {
- assert!(verify(c[1].as_bytes(), c[2], PublicKey::from_str(c[3]).unwrap()))
+ assert!(verify(c[1].as_bytes(), c[2], &PublicKey::from_str(c[3]).unwrap()))
}
}
}
use prelude::*;
use std::io::{Read, Write};
-use std::collections::HashMap;
use core::hash::Hash;
use std::sync::Mutex;
use core::cmp;
// licenses.
macro_rules! encode_tlv {
- ($stream: expr, {$(($type: expr, $field: expr)),*}, {$(($optional_type: expr, $optional_field: expr)),*}) => { {
+ ($stream: expr, $type: expr, $field: expr, required) => {
+ BigSize($type).write($stream)?;
+ BigSize($field.serialized_length() as u64).write($stream)?;
+ $field.write($stream)?;
+ };
+ ($stream: expr, $type: expr, $field: expr, vec_type) => {
+ encode_tlv!($stream, $type, ::util::ser::VecWriteWrapper(&$field), required);
+ };
+ ($stream: expr, $optional_type: expr, $optional_field: expr, option) => {
+ if let Some(ref field) = $optional_field {
+ BigSize($optional_type).write($stream)?;
+ BigSize(field.serialized_length() as u64).write($stream)?;
+ field.write($stream)?;
+ }
+ };
+}
+
+macro_rules! encode_tlv_stream {
+ ($stream: expr, {$(($type: expr, $field: expr, $fieldty: ident)),*}) => { {
#[allow(unused_imports)]
- use util::ser::BigSize;
- // Fields must be serialized in order, so we have to potentially switch between optional
- // fields and normal fields while serializing. Thus, we end up having to loop over the type
- // counts.
- // Sadly, while LLVM does appear smart enough to make `max_field` a constant, it appears to
- // refuse to unroll the loop. If we have enough entries that this is slow we can revisit
- // this design in the future.
- #[allow(unused_mut)]
- let mut max_field: u64 = 0;
- $(
- if $type >= max_field { max_field = $type + 1; }
- )*
+ use {
+ ln::msgs::DecodeError,
+ util::ser,
+ util::ser::BigSize,
+ };
+
$(
- if $optional_type >= max_field { max_field = $optional_type + 1; }
+ encode_tlv!($stream, $type, $field, $fieldty);
)*
- #[allow(unused_variables)]
- for i in 0..max_field {
- $(
- if i == $type {
- BigSize($type).write($stream)?;
- BigSize($field.serialized_length() as u64).write($stream)?;
- $field.write($stream)?;
- }
- )*
+
+ #[allow(unused_mut, unused_variables, unused_assignments)]
+ #[cfg(debug_assertions)]
+ {
+ let mut last_seen: Option<u64> = None;
$(
- if i == $optional_type {
- if let Some(ref field) = $optional_field {
- BigSize($optional_type).write($stream)?;
- BigSize(field.serialized_length() as u64).write($stream)?;
- field.write($stream)?;
- }
+ if let Some(t) = last_seen {
+ debug_assert!(t <= $type);
}
+ last_seen = Some($type);
)*
}
} }
}
macro_rules! get_varint_length_prefixed_tlv_length {
- ({$(($type: expr, $field: expr)),*}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}) => { {
- use util::ser::LengthCalculatingWriter;
- #[allow(unused_mut)]
- let mut len = LengthCalculatingWriter(0);
- {
- $(
- BigSize($type).write(&mut len).expect("No in-memory data may fail to serialize");
- let field_len = $field.serialized_length();
- BigSize(field_len as u64).write(&mut len).expect("No in-memory data may fail to serialize");
- len.0 += field_len;
- )*
- $(
- if let Some(ref field) = $optional_field {
- BigSize($optional_type).write(&mut len).expect("No in-memory data may fail to serialize");
- let field_len = field.serialized_length();
- BigSize(field_len as u64).write(&mut len).expect("No in-memory data may fail to serialize");
- len.0 += field_len;
- }
- )*
+ ($len: expr, $type: expr, $field: expr, required) => {
+ BigSize($type).write(&mut $len).expect("No in-memory data may fail to serialize");
+ let field_len = $field.serialized_length();
+ BigSize(field_len as u64).write(&mut $len).expect("No in-memory data may fail to serialize");
+ $len.0 += field_len;
+ };
+ ($len: expr, $type: expr, $field: expr, vec_type) => {
+ get_varint_length_prefixed_tlv_length!($len, $type, ::util::ser::VecWriteWrapper(&$field), required);
+ };
+ ($len: expr, $optional_type: expr, $optional_field: expr, option) => {
+ if let Some(ref field) = $optional_field {
+ BigSize($optional_type).write(&mut $len).expect("No in-memory data may fail to serialize");
+ let field_len = field.serialized_length();
+ BigSize(field_len as u64).write(&mut $len).expect("No in-memory data may fail to serialize");
+ $len.0 += field_len;
}
- len.0
- } }
+ };
}
macro_rules! encode_varint_length_prefixed_tlv {
- ($stream: expr, {$(($type: expr, $field: expr)),*}, {$(($optional_type: expr, $optional_field: expr)),*}) => { {
+ ($stream: expr, {$(($type: expr, $field: expr, $fieldty: ident)),*}) => { {
use util::ser::BigSize;
- let len = get_varint_length_prefixed_tlv_length!({ $(($type, $field)),* }, { $(($optional_type, $optional_field)),* });
+ let len = {
+ #[allow(unused_mut)]
+ let mut len = ::util::ser::LengthCalculatingWriter(0);
+ $(
+ get_varint_length_prefixed_tlv_length!(len, $type, $field, $fieldty);
+ )*
+ len.0
+ };
BigSize(len as u64).write($stream)?;
- encode_tlv!($stream, { $(($type, $field)),* }, { $(($optional_type, $optional_field)),* });
+ encode_tlv_stream!($stream, { $(($type, $field, $fieldty)),* });
} }
}
+macro_rules! check_tlv_order {
+ ($last_seen_type: expr, $typ: expr, $type: expr, required) => {{
+ #[allow(unused_comparisons)] // Note that $type may be 0 making the second comparison always true
+ let invalid_order = ($last_seen_type.is_none() || $last_seen_type.unwrap() < $type) && $typ.0 > $type;
+ if invalid_order {
+ Err(DecodeError::InvalidValue)?
+ }
+ }};
+ ($last_seen_type: expr, $typ: expr, $type: expr, option) => {{
+ // no-op
+ }};
+ ($last_seen_type: expr, $typ: expr, $type: expr, vec_type) => {{
+ // no-op
+ }};
+}
+
+macro_rules! check_missing_tlv {
+ ($last_seen_type: expr, $type: expr, required) => {{
+ #[allow(unused_comparisons)] // Note that $type may be 0 making the second comparison always true
+ let missing_req_type = $last_seen_type.is_none() || $last_seen_type.unwrap() < $type;
+ if missing_req_type {
+ Err(DecodeError::InvalidValue)?
+ }
+ }};
+ ($last_seen_type: expr, $type: expr, vec_type) => {{
+ // no-op
+ }};
+ ($last_seen_type: expr, $type: expr, option) => {{
+ // no-op
+ }};
+}
+
macro_rules! decode_tlv {
- ($stream: expr, {$(($reqtype: expr, $reqfield: ident)),*}, {$(($type: expr, $field: ident)),*}) => { {
+ ($reader: expr, $field: ident, required) => {{
+ $field = ser::Readable::read(&mut $reader)?;
+ }};
+ ($reader: expr, $field: ident, vec_type) => {{
+ $field = Some(ser::Readable::read(&mut $reader)?);
+ }};
+ ($reader: expr, $field: ident, option) => {{
+ $field = Some(ser::Readable::read(&mut $reader)?);
+ }};
+}
+
+macro_rules! decode_tlv_stream {
+ ($stream: expr, {$(($type: expr, $field: ident, $fieldty: ident)),* $(,)*}) => { {
use ln::msgs::DecodeError;
let mut last_seen_type: Option<u64> = None;
'tlv_read: loop {
}
// As we read types, make sure we hit every required type:
$({
- #[allow(unused_comparisons)] // Note that $reqtype may be 0 making the second comparison always true
- let invalid_order = (last_seen_type.is_none() || last_seen_type.unwrap() < $reqtype) && typ.0 > $reqtype;
- if invalid_order {
- Err(DecodeError::InvalidValue)?
- }
+ check_tlv_order!(last_seen_type, typ, $type, $fieldty);
})*
last_seen_type = Some(typ.0);
let length: ser::BigSize = Readable::read($stream)?;
let mut s = ser::FixedLengthReader::new($stream, length.0);
match typ.0 {
- $($reqtype => {
- $reqfield = ser::Readable::read(&mut s)?;
- if s.bytes_remain() {
- s.eat_remaining()?; // Return ShortRead if there's actually not enough bytes
- Err(DecodeError::InvalidValue)?
- }
- },)*
$($type => {
- $field = Some(ser::Readable::read(&mut s)?);
+ decode_tlv!(s, $field, $fieldty);
if s.bytes_remain() {
s.eat_remaining()?; // Return ShortRead if there's actually not enough bytes
Err(DecodeError::InvalidValue)?
}
// Make sure we got to each required type after we've read every TLV:
$({
- #[allow(unused_comparisons)] // Note that $reqtype may be 0 making the second comparison always true
- let missing_req_type = last_seen_type.is_none() || last_seen_type.unwrap() < $reqtype;
- if missing_req_type {
- Err(DecodeError::InvalidValue)?
- }
+ check_missing_tlv!(last_seen_type, $type, $fieldty);
})*
} }
}
{
// In tests, assert that the hard-coded length matches the actual one
if $len != 0 {
- use util::ser::LengthCalculatingWriter;
- let mut len_calc = LengthCalculatingWriter(0);
+ let mut len_calc = ::util::ser::LengthCalculatingWriter(0);
$( self.$field.write(&mut len_calc).expect("No in-memory data may fail to serialize"); )*
assert_eq!(len_calc.0, $len);
assert_eq!(self.serialized_length(), $len);
#[cfg(any(test, feature = "fuzztarget"))]
{
// In tests, assert that the hard-coded length matches the actual one
- use util::ser::LengthCalculatingWriter;
- let mut len_calc = LengthCalculatingWriter(0);
+ let mut len_calc = ::util::ser::LengthCalculatingWriter(0);
$( self.$field.write(&mut len_calc).expect("No in-memory data may fail to serialize"); )*
assert!(len_calc.0 $cmp len);
assert_eq!(len_calc.0, self.serialized_length());
/// This is the preferred method of adding new fields that old nodes can ignore and still function
/// correctly.
macro_rules! write_tlv_fields {
- ($stream: expr, {$(($type: expr, $field: expr)),* $(,)*}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}) => {
- encode_varint_length_prefixed_tlv!($stream, {$(($type, $field)),*} , {$(($optional_type, $optional_field)),*});
+ ($stream: expr, {$(($type: expr, $field: expr, $fieldty: ident)),* $(,)*}) => {
+ encode_varint_length_prefixed_tlv!($stream, {$(($type, $field, $fieldty)),*});
}
}
/// Reads a suffix added by write_tlv_fields.
macro_rules! read_tlv_fields {
- ($stream: expr, {$(($reqtype: expr, $reqfield: ident)),* $(,)*}, {$(($type: expr, $field: ident)),* $(,)*}) => { {
+ ($stream: expr, {$(($type: expr, $field: ident, $fieldty: ident)),* $(,)*}) => { {
let tlv_len = ::util::ser::BigSize::read($stream)?;
let mut rd = ::util::ser::FixedLengthReader::new($stream, tlv_len.0);
- decode_tlv!(&mut rd, {$(($reqtype, $reqfield)),*}, {$(($type, $field)),*});
+ decode_tlv_stream!(&mut rd, {$(($type, $field, $fieldty)),*});
rd.eat_remaining().map_err(|_| ::ln::msgs::DecodeError::ShortRead)?;
} }
}
-// If we naively create a struct in impl_writeable_tlv_based below, we may end up returning
-// `Self { ,,vecfield: vecfield }` which is obviously incorrect. Instead, we have to match here to
-// detect at least one empty field set and skip the potentially-extra comma.
-macro_rules! _init_tlv_based_struct {
- ($($type: ident)::*, {}, {$($field: ident),*}, {$($vecfield: ident),*}) => {
- Ok($($type)::* {
- $($field),*,
- $($vecfield: $vecfield.unwrap().0),*
- })
+macro_rules! init_tlv_based_struct_field {
+ ($field: ident, option) => {
+ $field
};
- ($($type: ident)::*, {$($reqfield: ident),*}, {}, {$($vecfield: ident),*}) => {
- Ok($($type)::* {
- $($reqfield: $reqfield.0.unwrap()),*,
- $($vecfield: $vecfield.unwrap().0),*
- })
+ ($field: ident, required) => {
+ $field.0.unwrap()
};
- ($($type: ident)::*, {$($reqfield: ident),*}, {$($field: ident),*}, {}) => {
- Ok($($type)::* {
- $($reqfield: $reqfield.0.unwrap()),*,
- $($field),*
- })
+ ($field: ident, vec_type) => {
+ $field.unwrap().0
};
- ($($type: ident)::*, {$($reqfield: ident),*}, {$($field: ident),*}, {$($vecfield: ident),*}) => {
- Ok($($type)::* {
- $($reqfield: $reqfield.0.unwrap()),*,
- $($field),*,
- $($vecfield: $vecfield.unwrap().0),*
- })
- }
}
-// If we don't have any optional types below, but do have some vec types, we end up calling
-// `write_tlv_field!($stream, {..}, {, (vec_ty, vec_val)})`, which is obviously broken.
-// Instead, for write and read we match the missing values and skip the extra comma.
-macro_rules! _write_tlv_fields {
- ($stream: expr, {$(($type: expr, $field: expr)),* $(,)*}, {}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}) => {
- write_tlv_fields!($stream, {$(($type, $field)),*} , {$(($optional_type, $optional_field)),*});
+macro_rules! init_tlv_field_var {
+ ($field: ident, required) => {
+ let mut $field = ::util::ser::OptionDeserWrapper(None);
};
- ($stream: expr, {$(($type: expr, $field: expr)),* $(,)*}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}, {$(($optional_type_2: expr, $optional_field_2: expr)),* $(,)*}) => {
- write_tlv_fields!($stream, {$(($type, $field)),*} , {$(($optional_type, $optional_field)),*, $(($optional_type_2, $optional_field_2)),*});
- }
-}
-macro_rules! _get_tlv_len {
- ({$(($type: expr, $field: expr)),* $(,)*}, {}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}) => {
- get_varint_length_prefixed_tlv_length!({$(($type, $field)),*} , {$(($optional_type, $optional_field)),*})
+ ($field: ident, vec_type) => {
+ let mut $field = Some(::util::ser::VecReadWrapper(Vec::new()));
};
- ({$(($type: expr, $field: expr)),* $(,)*}, {$(($optional_type: expr, $optional_field: expr)),* $(,)*}, {$(($optional_type_2: expr, $optional_field_2: expr)),* $(,)*}) => {
- get_varint_length_prefixed_tlv_length!({$(($type, $field)),*} , {$(($optional_type, $optional_field)),*, $(($optional_type_2, $optional_field_2)),*})
- }
-}
-macro_rules! _read_tlv_fields {
- ($stream: expr, {$(($reqtype: expr, $reqfield: ident)),* $(,)*}, {}, {$(($type: expr, $field: ident)),* $(,)*}) => {
- read_tlv_fields!($stream, {$(($reqtype, $reqfield)),*}, {$(($type, $field)),*});
- };
- ($stream: expr, {$(($reqtype: expr, $reqfield: ident)),* $(,)*}, {$(($type: expr, $field: ident)),* $(,)*}, {$(($type_2: expr, $field_2: ident)),* $(,)*}) => {
- read_tlv_fields!($stream, {$(($reqtype, $reqfield)),*}, {$(($type, $field)),*, $(($type_2, $field_2)),*});
+ ($field: ident, option) => {
+ let mut $field = None;
}
}
/// Implements Readable/Writeable for a struct storing it as a set of TLVs
-/// First block includes all the required fields including a dummy value which is used during
-/// deserialization but which will never be exposed to other code.
-/// The second block includes optional fields.
-/// The third block includes any Vecs which need to have their individual elements serialized.
+/// If $fieldty is `required`, then $field is a required field that is not an Option nor a Vec.
+/// If $fieldty is `option`, then $field is optional field.
+/// if $fieldty is `vec_type`, then $field is a Vec, which needs to have its individual elements
+/// serialized.
macro_rules! impl_writeable_tlv_based {
- ($st: ident, {$(($reqtype: expr, $reqfield: ident)),* $(,)*}, {$(($type: expr, $field: ident)),* $(,)*}, {$(($vectype: expr, $vecfield: ident)),* $(,)*}) => {
+ ($st: ident, {$(($type: expr, $field: ident, $fieldty: ident)),* $(,)*}) => {
impl ::util::ser::Writeable for $st {
fn write<W: ::util::ser::Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
- _write_tlv_fields!(writer, {
- $(($reqtype, self.$reqfield)),*
- }, {
- $(($type, self.$field)),*
- }, {
- $(($vectype, Some(::util::ser::VecWriteWrapper(&self.$vecfield)))),*
+ write_tlv_fields!(writer, {
+ $(($type, self.$field, $fieldty)),*
});
Ok(())
}
#[inline]
fn serialized_length(&self) -> usize {
- let len = _get_tlv_len!({
- $(($reqtype, self.$reqfield)),*
- }, {
- $(($type, self.$field)),*
- }, {
- $(($vectype, Some(::util::ser::VecWriteWrapper(&self.$vecfield)))),*
- });
- use util::ser::{BigSize, LengthCalculatingWriter};
- let mut len_calc = LengthCalculatingWriter(0);
+ use util::ser::BigSize;
+ let len = {
+ #[allow(unused_mut)]
+ let mut len = ::util::ser::LengthCalculatingWriter(0);
+ $(
+ get_varint_length_prefixed_tlv_length!(len, $type, self.$field, $fieldty);
+ )*
+ len.0
+ };
+ let mut len_calc = ::util::ser::LengthCalculatingWriter(0);
BigSize(len as u64).write(&mut len_calc).expect("No in-memory data may fail to serialize");
len + len_calc.0
}
impl ::util::ser::Readable for $st {
fn read<R: ::std::io::Read>(reader: &mut R) -> Result<Self, ::ln::msgs::DecodeError> {
$(
- let mut $reqfield = ::util::ser::OptionDeserWrapper(None);
- )*
- $(
- let mut $field = None;
+ init_tlv_field_var!($field, $fieldty);
)*
- $(
- let mut $vecfield = Some(::util::ser::VecReadWrapper(Vec::new()));
- )*
- _read_tlv_fields!(reader, {
- $(($reqtype, $reqfield)),*
- }, {
- $(($type, $field)),*
- }, {
- $(($vectype, $vecfield)),*
+ read_tlv_fields!(reader, {
+ $(($type, $field, $fieldty)),*
});
- _init_tlv_based_struct!($st, {$($reqfield),*}, {$($field),*}, {$($vecfield),*})
+ Ok(Self {
+ $(
+ $field: init_tlv_based_struct_field!($field, $fieldty)
+ ),*
+ })
}
}
}
/// variants stored directly.
/// The format is, for example
/// impl_writeable_tlv_based_enum!(EnumName,
-/// (0, StructVariantA) => {(0, variant_field)}, {(1, variant_optional_field)}, {},
-/// (1, StructVariantB) => {(0, variant_field_a), (1, variant_field_b)}, {}, {(2, variant_vec_field)};
+/// (0, StructVariantA) => {(0, required_variant_field, required), (1, optional_variant_field, option)},
+/// (1, StructVariantB) => {(0, variant_field_a, required), (1, variant_field_b, required), (2, variant_vec_field, vec_type)};
/// (2, TupleVariantA), (3, TupleVariantB),
/// );
/// The type is written as a single byte, followed by any variant data.
/// Attempts to read an unknown type byte result in DecodeError::UnknownRequiredFeature.
macro_rules! impl_writeable_tlv_based_enum {
($st: ident, $(($variant_id: expr, $variant_name: ident) =>
- {$(($reqtype: expr, $reqfield: ident)),* $(,)*},
- {$(($type: expr, $field: ident)),* $(,)*},
- {$(($vectype: expr, $vecfield: ident)),* $(,)*}
+ {$(($type: expr, $field: ident, $fieldty: ident)),* $(,)*}
),* $(,)*;
$(($tuple_variant_id: expr, $tuple_variant_name: ident)),* $(,)*) => {
impl ::util::ser::Writeable for $st {
fn write<W: ::util::ser::Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
match self {
- $($st::$variant_name { $(ref $reqfield),* $(ref $field),*, $(ref $vecfield),* } => {
+ $($st::$variant_name { $(ref $field),* } => {
let id: u8 = $variant_id;
id.write(writer)?;
- _write_tlv_fields!(writer, {
- $(($reqtype, $reqfield)),*
- }, {
- $(($type, $field)),*
- }, {
- $(($vectype, Some(::util::ser::VecWriteWrapper(&$vecfield)))),*
+ write_tlv_fields!(writer, {
+ $(($type, $field, $fieldty)),*
});
}),*
$($st::$tuple_variant_name (ref field) => {
// in the same function body. Instead, we define a closure and call it.
let f = || {
$(
- let mut $reqfield = ::util::ser::OptionDeserWrapper(None);
- )*
- $(
- let mut $field = None;
- )*
- $(
- let mut $vecfield = Some(::util::ser::VecReadWrapper(Vec::new()));
+ init_tlv_field_var!($field, $fieldty);
)*
- _read_tlv_fields!(reader, {
- $(($reqtype, $reqfield)),*
- }, {
- $(($type, $field)),*
- }, {
- $(($vectype, $vecfield)),*
+ read_tlv_fields!(reader, {
+ $(($type, $field, $fieldty)),*
});
- _init_tlv_based_struct!($st::$variant_name, {$($reqfield),*}, {$($field),*}, {$($vecfield),*})
+ Ok($st::$variant_name {
+ $(
+ $field: init_tlv_based_struct_field!($field, $fieldty)
+ ),*
+ })
};
f()
}),*
let mut a: u64 = 0;
let mut b: u32 = 0;
let mut c: Option<u32> = None;
- decode_tlv!(&mut s, {(2, a), (3, b)}, {(4, c)});
+ decode_tlv_stream!(&mut s, {(2, a, required), (3, b, required), (4, c, option)});
Ok((a, b, c))
}
let mut tlv2: Option<u64> = None;
let mut tlv3: Option<(PublicKey, u64, u64)> = None;
let mut tlv4: Option<u16> = None;
- decode_tlv!(&mut s, {}, {(1, tlv1), (2, tlv2), (3, tlv3), (254, tlv4)});
+ decode_tlv_stream!(&mut s, {(1, tlv1, option), (2, tlv2, option), (3, tlv3, option), (254, tlv4, option)});
Ok((tlv1, tlv2, tlv3, tlv4))
}
let mut stream = VecWriter(Vec::new());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { (1, 1u8) }, { (42, None::<u64>) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(1, 1u8, required), (42, None::<u64>, option)});
assert_eq!(stream.0, ::hex::decode("03010101").unwrap());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { }, { (1, Some(1u8)) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(1, Some(1u8), option)});
assert_eq!(stream.0, ::hex::decode("03010101").unwrap());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { (4, 0xabcdu16) }, { (42, None::<u64>) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(4, 0xabcdu16, required), (42, None::<u64>, option)});
assert_eq!(stream.0, ::hex::decode("040402abcd").unwrap());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { (0xff, 0xabcdu16) }, { (42, None::<u64>) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(42, None::<u64>, option), (0xff, 0xabcdu16, required)});
assert_eq!(stream.0, ::hex::decode("06fd00ff02abcd").unwrap());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { (0, 1u64), (0xff, HighZeroBytesDroppedVarInt(0u64)) }, { (42, None::<u64>) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(0, 1u64, required), (42, None::<u64>, option), (0xff, HighZeroBytesDroppedVarInt(0u64), required)});
assert_eq!(stream.0, ::hex::decode("0e00080000000000000001fd00ff00").unwrap());
stream.0.clear();
- encode_varint_length_prefixed_tlv!(&mut stream, { (0xff, HighZeroBytesDroppedVarInt(0u64)) }, { (0, Some(1u64)) });
+ encode_varint_length_prefixed_tlv!(&mut stream, {(0, Some(1u64), option), (0xff, HighZeroBytesDroppedVarInt(0u64), required)});
assert_eq!(stream.0, ::hex::decode("0e00080000000000000001fd00ff00").unwrap());
Ok(())
use std::sync::{Mutex, Arc};
use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use core::{cmp, mem};
-use std::collections::{HashMap, HashSet, VecDeque};
use chain::keysinterface::InMemorySigner;
pub struct TestVecWriter(pub Vec<u8>);
impl Logger for TestLogger {
fn log(&self, record: &Record) {
*self.lines.lock().unwrap().entry((record.module_path.to_string(), format!("{}", record.args))).or_insert(0) += 1;
- if self.level >= record.level {
+ if record.level >= self.level {
println!("{:<5} {} [{} : {}, {}] {}", record.level.to_string(), self.id, record.module_path, record.file, record.line, record.args);
}
}
projects / rust-lightning / commitdiff