build-no-std: true
build-futures: true
build-tx-sync: true
+ - toolchain: stable
+ test-custom-message: true
- toolchain: beta
platform: macos-latest
build-net-tokio: true
build-no-std: true
build-futures: true
build-tx-sync: true
+ - toolchain: beta
+ test-custom-message: true
- toolchain: 1.41.1
build-no-std: false
test-log-variants: true
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
+ - name: Test Custom Message Macros on Rust ${{ matrix.toolchain }}
+ if: "matrix.test-custom-message"
+ run: |
+ cd lightning-custom-message
+ cargo test --verbose --color always
- name: Install deps for kcov
if: matrix.coverage
run: |
.idea
lightning/target
lightning/ldk-net_graph-*.bin
+lightning-custom-message/target
no-std-check/target
-
]
exclude = [
+ "lightning-custom-message",
"no-std-check",
]
events::Event::PaymentClaimed { .. } => {},
events::Event::PaymentPathSuccessful { .. } => {},
events::Event::PaymentPathFailed { .. } => {},
+ events::Event::PaymentFailed { .. } => {},
events::Event::ProbeSuccessful { .. } | events::Event::ProbeFailed { .. } => {
// Even though we don't explicitly send probes, because probes are
// detected based on hashing the payment hash+preimage, its rather
use crate::utils::test_logger;
-fn check_eq(btree: &BTreeMap<u8, u8>, indexed: &IndexedMap<u8, u8>) {
+use std::ops::{RangeBounds, Bound};
+
+struct ExclLowerInclUpper(u8, u8);
+impl RangeBounds<u8> for ExclLowerInclUpper {
+ fn start_bound(&self) -> Bound<&u8> { Bound::Excluded(&self.0) }
+ fn end_bound(&self) -> Bound<&u8> { Bound::Included(&self.1) }
+}
+struct ExclLowerExclUpper(u8, u8);
+impl RangeBounds<u8> for ExclLowerExclUpper {
+ fn start_bound(&self) -> Bound<&u8> { Bound::Excluded(&self.0) }
+ fn end_bound(&self) -> Bound<&u8> { Bound::Excluded(&self.1) }
+}
+
+fn check_eq(btree: &BTreeMap<u8, u8>, mut indexed: IndexedMap<u8, u8>) {
assert_eq!(btree.len(), indexed.len());
assert_eq!(btree.is_empty(), indexed.is_empty());
let mut btree_clone = btree.clone();
assert!(btree_clone == *btree);
let mut indexed_clone = indexed.clone();
- assert!(indexed_clone == *indexed);
+ assert!(indexed_clone == indexed);
for k in 0..=255 {
assert_eq!(btree.contains_key(&k), indexed.contains_key(&k));
}
const STRIDE: u8 = 16;
- for k in 0..=255/STRIDE {
- let lower_bound = k * STRIDE;
- let upper_bound = lower_bound + (STRIDE - 1);
- let mut btree_iter = btree.range(lower_bound..=upper_bound);
- let mut indexed_iter = indexed.range(lower_bound..=upper_bound);
- loop {
- let b_v = btree_iter.next();
- let i_v = indexed_iter.next();
- assert_eq!(b_v, i_v);
- if b_v.is_none() { break; }
+ for range_type in 0..4 {
+ for k in 0..=255/STRIDE {
+ let lower_bound = k * STRIDE;
+ let upper_bound = lower_bound + (STRIDE - 1);
+ macro_rules! range { ($map: expr) => {
+ match range_type {
+ 0 => $map.range(lower_bound..upper_bound),
+ 1 => $map.range(lower_bound..=upper_bound),
+ 2 => $map.range(ExclLowerInclUpper(lower_bound, upper_bound)),
+ 3 => $map.range(ExclLowerExclUpper(lower_bound, upper_bound)),
+ _ => unreachable!(),
+ }
+ } }
+ let mut btree_iter = range!(btree);
+ let mut indexed_iter = range!(indexed);
+ loop {
+ let b_v = btree_iter.next();
+ let i_v = indexed_iter.next();
+ assert_eq!(b_v, i_v);
+ if b_v.is_none() { break; }
+ }
}
}
let prev_value_i = indexed.insert(tuple[0], tuple[1]);
assert_eq!(prev_value_b, prev_value_i);
}
- check_eq(&btree, &indexed);
+ check_eq(&btree, indexed.clone());
// Now, modify the maps in all the ways we have to do so, checking that the maps remain
// equivalent as we go.
*v = *k;
*btree.get_mut(k).unwrap() = *k;
}
- check_eq(&btree, &indexed);
+ check_eq(&btree, indexed.clone());
for k in 0..=255 {
match btree.entry(k) {
},
}
}
- check_eq(&btree, &indexed);
+ check_eq(&btree, indexed);
}
pub fn indexedmap_test<Out: test_logger::Output>(data: &[u8], _out: Out) {
0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 8, 153, 192, 0, 2, 27, 0, 0, 25, 0, 0,
0, 1, 0, 0, 0, 125, 255, 2, 68, 226, 0, 6, 11, 0, 1, 5, 0, 0, 0, 0, 29, 129, 25, 192,
];
- nodes[0].rapid_gossip_sync.update_network_graph(&initialization_input[..]).unwrap();
+ nodes[0].rapid_gossip_sync.update_network_graph_no_std(&initialization_input[..], Some(1642291930)).unwrap();
// this should have added two channels
assert_eq!(network_graph.read_only().channels().len(), 3);
--- /dev/null
+[package]
+name = "lightning-custom-message"
+version = "0.0.113"
+authors = ["Jeffrey Czyz"]
+license = "MIT OR Apache-2.0"
+repository = "http://github.com/lightningdevkit/rust-lightning"
+description = """
+Utilities for supporting custom peer-to-peer messages in LDK.
+"""
+edition = "2021"
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[dependencies]
+bitcoin = "0.29.0"
+lightning = { version = "0.0.113", path = "../lightning" }
--- /dev/null
+//! Utilities for supporting custom peer-to-peer messages in LDK.
+//!
+//! [BOLT 1] specifies a custom message type range for use with experimental or application-specific
+//! messages. While a [`CustomMessageHandler`] can be defined to support more than one message type,
+//! defining such a handler requires a significant amount of boilerplate and can be error prone.
+//!
+//! This crate provides the [`composite_custom_message_handler`] macro for easily composing
+//! pre-defined custom message handlers into one handler. The resulting handler can be further
+//! composed with other custom message handlers using the same macro.
+//!
+//! The following example demonstrates defining a `FooBarHandler` to compose separate handlers for
+//! `Foo` and `Bar` messages, and further composing it with a handler for `Baz` messages.
+//!
+//!```
+//! # extern crate bitcoin;
+//! extern crate lightning;
+//! #[macro_use]
+//! extern crate lightning_custom_message;
+//!
+//! # use bitcoin::secp256k1::PublicKey;
+//! # use lightning::io;
+//! # use lightning::ln::msgs::{DecodeError, LightningError};
+//! use lightning::ln::peer_handler::CustomMessageHandler;
+//! use lightning::ln::wire::{CustomMessageReader, self};
+//! use lightning::util::ser::Writeable;
+//! # use lightning::util::ser::Writer;
+//!
+//! // Assume that `FooHandler` and `BarHandler` are defined in one crate and `BazHandler` is
+//! // defined in another crate, handling messages `Foo`, `Bar`, and `Baz`, respectively.
+//!
+//! #[derive(Debug)]
+//! pub struct Foo;
+//!
+//! macro_rules! foo_type_id {
+//! () => { 32768 }
+//! }
+//!
+//! impl wire::Type for Foo {
+//! fn type_id(&self) -> u16 { foo_type_id!() }
+//! }
+//! impl Writeable for Foo {
+//! // ...
+//! # fn write<W: Writer>(&self, _: &mut W) -> Result<(), io::Error> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! pub struct FooHandler;
+//!
+//! impl CustomMessageReader for FooHandler {
+//! // ...
+//! # type CustomMessage = Foo;
+//! # fn read<R: io::Read>(
+//! # &self, _message_type: u16, _buffer: &mut R
+//! # ) -> Result<Option<Self::CustomMessage>, DecodeError> {
+//! # unimplemented!()
+//! # }
+//! }
+//! impl CustomMessageHandler for FooHandler {
+//! // ...
+//! # fn handle_custom_message(
+//! # &self, _msg: Self::CustomMessage, _sender_node_id: &PublicKey
+//! # ) -> Result<(), LightningError> {
+//! # unimplemented!()
+//! # }
+//! # fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! #[derive(Debug)]
+//! pub struct Bar;
+//!
+//! macro_rules! bar_type_id {
+//! () => { 32769 }
+//! }
+//!
+//! impl wire::Type for Bar {
+//! fn type_id(&self) -> u16 { bar_type_id!() }
+//! }
+//! impl Writeable for Bar {
+//! // ...
+//! # fn write<W: Writer>(&self, _: &mut W) -> Result<(), io::Error> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! pub struct BarHandler;
+//!
+//! impl CustomMessageReader for BarHandler {
+//! // ...
+//! # type CustomMessage = Bar;
+//! # fn read<R: io::Read>(
+//! # &self, _message_type: u16, _buffer: &mut R
+//! # ) -> Result<Option<Self::CustomMessage>, DecodeError> {
+//! # unimplemented!()
+//! # }
+//! }
+//! impl CustomMessageHandler for BarHandler {
+//! // ...
+//! # fn handle_custom_message(
+//! # &self, _msg: Self::CustomMessage, _sender_node_id: &PublicKey
+//! # ) -> Result<(), LightningError> {
+//! # unimplemented!()
+//! # }
+//! # fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! #[derive(Debug)]
+//! pub struct Baz;
+//!
+//! macro_rules! baz_type_id {
+//! () => { 32770 }
+//! }
+//!
+//! impl wire::Type for Baz {
+//! fn type_id(&self) -> u16 { baz_type_id!() }
+//! }
+//! impl Writeable for Baz {
+//! // ...
+//! # fn write<W: Writer>(&self, _: &mut W) -> Result<(), io::Error> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! pub struct BazHandler;
+//!
+//! impl CustomMessageReader for BazHandler {
+//! // ...
+//! # type CustomMessage = Baz;
+//! # fn read<R: io::Read>(
+//! # &self, _message_type: u16, _buffer: &mut R
+//! # ) -> Result<Option<Self::CustomMessage>, DecodeError> {
+//! # unimplemented!()
+//! # }
+//! }
+//! impl CustomMessageHandler for BazHandler {
+//! // ...
+//! # fn handle_custom_message(
+//! # &self, _msg: Self::CustomMessage, _sender_node_id: &PublicKey
+//! # ) -> Result<(), LightningError> {
+//! # unimplemented!()
+//! # }
+//! # fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)> {
+//! # unimplemented!()
+//! # }
+//! }
+//!
+//! # fn main() {
+//! // The first crate may define a handler composing `FooHandler` and `BarHandler` and export the
+//! // corresponding message type ids as a macro to use in further composition.
+//!
+//! composite_custom_message_handler!(
+//! pub struct FooBarHandler {
+//! foo: FooHandler,
+//! bar: BarHandler,
+//! }
+//!
+//! pub enum FooBarMessage {
+//! Foo(foo_type_id!()),
+//! Bar(bar_type_id!()),
+//! }
+//! );
+//!
+//! #[macro_export]
+//! macro_rules! foo_bar_type_ids {
+//! () => { foo_type_id!() | bar_type_id!() }
+//! }
+//!
+//! // Another crate can then define a handler further composing `FooBarHandler` with `BazHandler`
+//! // and similarly export the composition of message type ids as a macro.
+//!
+//! composite_custom_message_handler!(
+//! pub struct FooBarBazHandler {
+//! foo_bar: FooBarHandler,
+//! baz: BazHandler,
+//! }
+//!
+//! pub enum FooBarBazMessage {
+//! FooBar(foo_bar_type_ids!()),
+//! Baz(baz_type_id!()),
+//! }
+//! );
+//!
+//! #[macro_export]
+//! macro_rules! foo_bar_baz_type_ids {
+//! () => { foo_bar_type_ids!() | baz_type_id!() }
+//! }
+//! # }
+//!```
+//!
+//! [BOLT 1]: https://github.com/lightning/bolts/blob/master/01-messaging.md
+//! [`CustomMessageHandler`]: crate::lightning::ln::peer_handler::CustomMessageHandler
+
+#![doc(test(no_crate_inject, attr(deny(warnings))))]
+
+pub extern crate bitcoin;
+pub extern crate lightning;
+
+/// Defines a composite type implementing [`CustomMessageHandler`] (and therefore also implementing
+/// [`CustomMessageReader`]), along with a corresponding enumerated custom message [`Type`], from
+/// one or more previously defined custom message handlers.
+///
+/// Useful for parameterizing [`PeerManager`] with custom message handling for one or more sets of
+/// custom messages. Message type ids may be given as a valid `match` pattern, including ranges,
+/// though using OR-ed literal patterns is preferred in order to catch unreachable code for
+/// conflicting handlers.
+///
+/// See [crate documentation] for example usage.
+///
+/// [`CustomMessageHandler`]: crate::lightning::ln::peer_handler::CustomMessageHandler
+/// [`CustomMessageReader`]: crate::lightning::ln::wire::CustomMessageReader
+/// [`Type`]: crate::lightning::ln::wire::Type
+/// [`PeerManager`]: crate::lightning::ln::peer_handler::PeerManager
+/// [crate documentation]: self
+#[macro_export]
+macro_rules! composite_custom_message_handler {
+ (
+ $handler_visibility:vis struct $handler:ident {
+ $($field_visibility:vis $field:ident: $type:ty),* $(,)*
+ }
+
+ $message_visibility:vis enum $message:ident {
+ $($variant:ident($pattern:pat)),* $(,)*
+ }
+ ) => {
+ #[allow(missing_docs)]
+ $handler_visibility struct $handler {
+ $(
+ $field_visibility $field: $type,
+ )*
+ }
+
+ #[allow(missing_docs)]
+ #[derive(Debug)]
+ $message_visibility enum $message {
+ $(
+ $variant(<$type as $crate::lightning::ln::wire::CustomMessageReader>::CustomMessage),
+ )*
+ }
+
+ impl $crate::lightning::ln::peer_handler::CustomMessageHandler for $handler {
+ fn handle_custom_message(
+ &self, msg: Self::CustomMessage, sender_node_id: &$crate::bitcoin::secp256k1::PublicKey
+ ) -> Result<(), $crate::lightning::ln::msgs::LightningError> {
+ match msg {
+ $(
+ $message::$variant(message) => {
+ $crate::lightning::ln::peer_handler::CustomMessageHandler::handle_custom_message(
+ &self.$field, message, sender_node_id
+ )
+ },
+ )*
+ }
+ }
+
+ fn get_and_clear_pending_msg(&self) -> Vec<($crate::bitcoin::secp256k1::PublicKey, Self::CustomMessage)> {
+ vec![].into_iter()
+ $(
+ .chain(
+ self.$field
+ .get_and_clear_pending_msg()
+ .into_iter()
+ .map(|(pubkey, message)| (pubkey, $message::$variant(message)))
+ )
+ )*
+ .collect()
+ }
+ }
+
+ impl $crate::lightning::ln::wire::CustomMessageReader for $handler {
+ type CustomMessage = $message;
+ fn read<R: $crate::lightning::io::Read>(
+ &self, message_type: u16, buffer: &mut R
+ ) -> Result<Option<Self::CustomMessage>, $crate::lightning::ln::msgs::DecodeError> {
+ match message_type {
+ $(
+ $pattern => match <$type>::read(&self.$field, message_type, buffer)? {
+ None => unreachable!(),
+ Some(message) => Ok(Some($message::$variant(message))),
+ },
+ )*
+ _ => Ok(None),
+ }
+ }
+ }
+
+ impl $crate::lightning::ln::wire::Type for $message {
+ fn type_id(&self) -> u16 {
+ match self {
+ $(
+ Self::$variant(message) => message.type_id(),
+ )*
+ }
+ }
+ }
+
+ impl $crate::lightning::util::ser::Writeable for $message {
+ fn write<W: $crate::lightning::util::ser::Writer>(&self, writer: &mut W) -> Result<(), $crate::lightning::io::Error> {
+ match self {
+ $(
+ Self::$variant(message) => message.write(writer),
+ )*
+ }
+ }
+ }
+ }
+}
/// Update network graph from binary data.
/// Returns the last sync timestamp to be used the next time rapid sync data is queried.
///
- /// `network_graph`: network graph to be updated
- ///
/// `update_data`: `&[u8]` binary stream that comprises the update data
pub fn update_network_graph(&self, update_data: &[u8]) -> Result<u32, GraphSyncError> {
let mut read_cursor = io::Cursor::new(update_data);
self.update_network_graph_from_byte_stream(&mut read_cursor)
}
+ /// Update network graph from binary data.
+ /// Returns the last sync timestamp to be used the next time rapid sync data is queried.
+ ///
+ /// `update_data`: `&[u8]` binary stream that comprises the update data
+ /// `current_time_unix`: `Option<u64>` optional current timestamp to verify data age
+ pub fn update_network_graph_no_std(&self, update_data: &[u8], current_time_unix: Option<u64>) -> Result<u32, GraphSyncError> {
+ let mut read_cursor = io::Cursor::new(update_data);
+ self.update_network_graph_from_byte_stream_no_std(&mut read_cursor, current_time_unix)
+ }
+
/// Gets a reference to the underlying [`NetworkGraph`] which was provided in
/// [`RapidGossipSync::new`].
///
use crate::error::GraphSyncError;
use crate::RapidGossipSync;
+#[cfg(all(feature = "std", not(test)))]
+use std::time::{SystemTime, UNIX_EPOCH};
+
#[cfg(not(feature = "std"))]
use alloc::{vec::Vec, borrow::ToOwned};
/// avoid malicious updates being able to trigger excessive memory allocation.
const MAX_INITIAL_NODE_ID_VECTOR_CAPACITY: u32 = 50_000;
+/// We disallow gossip data that's more than two weeks old, per BOLT 7's
+/// suggestion.
+const STALE_RGS_UPDATE_AGE_LIMIT_SECS: u64 = 60 * 60 * 24 * 14;
+
impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L::Target: Logger {
pub(crate) fn update_network_graph_from_byte_stream<R: io::Read>(
+ &self,
+ read_cursor: &mut R,
+ ) -> Result<u32, GraphSyncError> {
+ #[allow(unused_mut)]
+ let mut current_time_unix = None;
+ #[cfg(all(feature = "std", not(test)))]
+ {
+ // Note that many tests rely on being able to set arbitrarily old timestamps, thus we
+ // disable this check during tests!
+ current_time_unix = Some(SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs());
+ }
+ self.update_network_graph_from_byte_stream_no_std(read_cursor, current_time_unix)
+ }
+
+ pub(crate) fn update_network_graph_from_byte_stream_no_std<R: io::Read>(
&self,
mut read_cursor: &mut R,
+ current_time_unix: Option<u64>
) -> Result<u32, GraphSyncError> {
let mut prefix = [0u8; 4];
read_cursor.read_exact(&mut prefix)?;
let chain_hash: BlockHash = Readable::read(read_cursor)?;
let latest_seen_timestamp: u32 = Readable::read(read_cursor)?;
+
+ if let Some(time) = current_time_unix {
+ if (latest_seen_timestamp as u64) < time.saturating_sub(STALE_RGS_UPDATE_AGE_LIMIT_SECS) {
+ return Err(LightningError{err: "Rapid Gossip Sync data is more than two weeks old".to_owned(), action: ErrorAction::IgnoreError}.into());
+ }
+ }
+
// backdate the applied timestamp by a week
let backdated_timestamp = latest_seen_timestamp.saturating_sub(24 * 3600 * 7);
use lightning::util::test_utils::TestLogger;
use crate::error::GraphSyncError;
+ use crate::processing::STALE_RGS_UPDATE_AGE_LIMIT_SECS;
use crate::RapidGossipSync;
+ const VALID_RGS_BINARY: [u8; 300] = [
+ 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
+ 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
+ 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
+ 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
+ 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
+ 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
+ 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
+ 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
+ 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
+ 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
+ 226, 0, 6, 11, 0, 1, 2, 3, 0, 0, 0, 4, 0, 40, 0, 0, 0, 0, 0, 0, 3, 232, 0, 0, 3, 232,
+ 0, 0, 0, 1, 0, 0, 0, 0, 29, 129, 25, 192, 255, 8, 153, 192, 0, 2, 27, 0, 0, 60, 0, 0,
+ 0, 0, 0, 0, 0, 1, 0, 0, 0, 100, 0, 0, 2, 224, 0, 0, 0, 0, 58, 85, 116, 216, 0, 29, 0,
+ 0, 0, 1, 0, 0, 0, 125, 0, 0, 0, 0, 58, 85, 116, 216, 255, 2, 68, 226, 0, 6, 11, 0, 1,
+ 0, 0, 1,
+ ];
+ const VALID_BINARY_TIMESTAMP: u64 = 1642291930;
+
#[test]
fn network_graph_fails_to_update_from_clipped_input() {
let block_hash = genesis_block(Network::Bitcoin).block_hash();
#[test]
fn full_update_succeeds() {
- let valid_input = vec![
- 76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
- 79, 147, 30, 131, 101, 225, 90, 8, 156, 104, 214, 25, 0, 0, 0, 0, 0, 97, 227, 98, 218,
- 0, 0, 0, 4, 2, 22, 7, 207, 206, 25, 164, 197, 231, 230, 231, 56, 102, 61, 250, 251,
- 187, 172, 38, 46, 79, 247, 108, 44, 155, 48, 219, 238, 252, 53, 192, 6, 67, 2, 36, 125,
- 157, 176, 223, 175, 234, 116, 94, 248, 201, 225, 97, 235, 50, 47, 115, 172, 63, 136,
- 88, 216, 115, 11, 111, 217, 114, 84, 116, 124, 231, 107, 2, 158, 1, 242, 121, 152, 106,
- 204, 131, 186, 35, 93, 70, 216, 10, 237, 224, 183, 89, 95, 65, 3, 83, 185, 58, 138,
- 181, 64, 187, 103, 127, 68, 50, 2, 201, 19, 17, 138, 136, 149, 185, 226, 156, 137, 175,
- 110, 32, 237, 0, 217, 90, 31, 100, 228, 149, 46, 219, 175, 168, 77, 4, 143, 38, 128,
- 76, 97, 0, 0, 0, 2, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 0, 1, 0, 0, 255, 2, 68,
- 226, 0, 6, 11, 0, 1, 2, 3, 0, 0, 0, 4, 0, 40, 0, 0, 0, 0, 0, 0, 3, 232, 0, 0, 3, 232,
- 0, 0, 0, 1, 0, 0, 0, 0, 29, 129, 25, 192, 255, 8, 153, 192, 0, 2, 27, 0, 0, 60, 0, 0,
- 0, 0, 0, 0, 0, 1, 0, 0, 0, 100, 0, 0, 2, 224, 0, 0, 0, 0, 58, 85, 116, 216, 0, 29, 0,
- 0, 0, 1, 0, 0, 0, 125, 0, 0, 0, 0, 58, 85, 116, 216, 255, 2, 68, 226, 0, 6, 11, 0, 1,
- 0, 0, 1,
- ];
-
let block_hash = genesis_block(Network::Bitcoin).block_hash();
let logger = TestLogger::new();
let network_graph = NetworkGraph::new(block_hash, &logger);
assert_eq!(network_graph.read_only().channels().len(), 0);
let rapid_sync = RapidGossipSync::new(&network_graph);
- let update_result = rapid_sync.update_network_graph(&valid_input[..]);
+ let update_result = rapid_sync.update_network_graph(&VALID_RGS_BINARY);
if update_result.is_err() {
panic!("Unexpected update result: {:?}", update_result)
}
assert!(after.contains("783241506229452801"));
}
+ #[test]
+ fn full_update_succeeds_at_the_beginning_of_the_unix_era() {
+ let block_hash = genesis_block(Network::Bitcoin).block_hash();
+ let logger = TestLogger::new();
+ let network_graph = NetworkGraph::new(block_hash, &logger);
+
+ assert_eq!(network_graph.read_only().channels().len(), 0);
+
+ let rapid_sync = RapidGossipSync::new(&network_graph);
+ // this is mostly for checking uint underflow issues before the fuzzer does
+ let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(0));
+ assert!(update_result.is_ok());
+ assert_eq!(network_graph.read_only().channels().len(), 2);
+ }
+
+ #[test]
+ fn timestamp_edge_cases_are_handled_correctly() {
+ // this is the timestamp encoded in the binary data of valid_input below
+ let block_hash = genesis_block(Network::Bitcoin).block_hash();
+ let logger = TestLogger::new();
+
+ let latest_succeeding_time = VALID_BINARY_TIMESTAMP + STALE_RGS_UPDATE_AGE_LIMIT_SECS;
+ let earliest_failing_time = latest_succeeding_time + 1;
+
+ {
+ let network_graph = NetworkGraph::new(block_hash, &logger);
+ assert_eq!(network_graph.read_only().channels().len(), 0);
+
+ let rapid_sync = RapidGossipSync::new(&network_graph);
+ let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(latest_succeeding_time));
+ assert!(update_result.is_ok());
+ assert_eq!(network_graph.read_only().channels().len(), 2);
+ }
+
+ {
+ let network_graph = NetworkGraph::new(block_hash, &logger);
+ assert_eq!(network_graph.read_only().channels().len(), 0);
+
+ let rapid_sync = RapidGossipSync::new(&network_graph);
+ let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(earliest_failing_time));
+ assert!(update_result.is_err());
+ if let Err(GraphSyncError::LightningError(lightning_error)) = update_result {
+ assert_eq!(
+ lightning_error.err,
+ "Rapid Gossip Sync data is more than two weeks old"
+ );
+ } else {
+ panic!("Unexpected update result: {:?}", update_result)
+ }
+ }
+ }
+
#[test]
pub fn update_fails_with_unknown_version() {
let unknown_version_input = vec![
commitment_signed_dance!(nodes[0], nodes[1], bs_updates.commitment_signed, false, true);
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 2);
+ assert_eq!(events.len(), 3);
if let Event::PaymentPathFailed { payment_hash, payment_failed_permanently, .. } = events[0] {
assert_eq!(payment_hash, payment_hash_1);
assert!(payment_failed_permanently);
} else { panic!("Unexpected event!"); }
match events[1] {
+ Event::PaymentFailed { payment_hash, .. } => {
+ assert_eq!(payment_hash, payment_hash_1);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[2] {
Event::PendingHTLCsForwardable { .. } => { },
_ => panic!("Unexpected event"),
};
use core::{cmp, mem};
use core::cell::RefCell;
use crate::io::Read;
-use crate::sync::{Arc, Mutex, RwLock, RwLockReadGuard, FairRwLock};
+use crate::sync::{Arc, Mutex, RwLock, RwLockReadGuard, FairRwLock, LockTestExt, LockHeldState};
use core::sync::atomic::{AtomicUsize, Ordering};
use core::time::Duration;
use core::ops::Deref;
/// made before LDK version 0.0.104.
payment_hash: Option<PaymentHash>,
},
- /// After a payment is explicitly abandoned by calling [`ChannelManager::abandon_payment`], it
- /// is marked as abandoned until an [`Event::PaymentFailed`] is generated. A payment could also
- /// be marked as abandoned if pathfinding fails repeatedly or retries have been exhausted.
+ /// After a payment's retries are exhausted per the provided [`Retry`], or it is explicitly
+ /// abandoned via [`ChannelManager::abandon_payment`], it is marked as abandoned until all
+ /// pending HTLCs for this payment resolve and an [`Event::PaymentFailed`] is generated.
Abandoned {
/// Hash of the payment that we have given up trying to send.
payment_hash: PaymentHash,
match $internal {
Ok(msg) => Ok(msg),
Err(MsgHandleErrInternal { err, chan_id, shutdown_finish }) => {
- #[cfg(any(feature = "_test_utils", test))]
- {
- // In testing, ensure there are no deadlocks where the lock is already held upon
- // entering the macro.
- debug_assert!($self.pending_events.try_lock().is_ok());
- debug_assert!($self.per_peer_state.try_write().is_ok());
- }
+ // In testing, ensure there are no deadlocks where the lock is already held upon
+ // entering the macro.
+ debug_assert_ne!($self.pending_events.held_by_thread(), LockHeldState::HeldByThread);
+ debug_assert_ne!($self.per_peer_state.held_by_thread(), LockHeldState::HeldByThread);
let mut msg_events = Vec::with_capacity(2);
///
/// This can be useful for payments that may have been prepared, but ultimately not sent, as a
/// result of a crash. If such a payment exists, is not listed here, and an
- /// [`Event::PaymentSent`] has not been received, you may consider retrying the payment.
+ /// [`Event::PaymentSent`] has not been received, you may consider resending the payment.
///
/// [`Event::PaymentSent`]: events::Event::PaymentSent
pub fn list_recent_payments(&self) -> Vec<RecentPaymentDetails> {
/// If a pending payment is currently in-flight with the same [`PaymentId`] provided, this
/// method will error with an [`APIError::InvalidRoute`]. Note, however, that once a payment
/// is no longer pending (either via [`ChannelManager::abandon_payment`], or handling of an
- /// [`Event::PaymentSent`]) LDK will not stop you from sending a second payment with the same
- /// [`PaymentId`].
+ /// [`Event::PaymentSent`] or [`Event::PaymentFailed`]) LDK will not stop you from sending a
+ /// second payment with the same [`PaymentId`].
///
/// Thus, in order to ensure duplicate payments are not sent, you should implement your own
/// tracking of payments, including state to indicate once a payment has completed. Because you
/// [`Route`], we assume the invoice had the basic_mpp feature set.
///
/// [`Event::PaymentSent`]: events::Event::PaymentSent
+ /// [`Event::PaymentFailed`]: events::Event::PaymentFailed
/// [`PeerManager::process_events`]: crate::ln::peer_handler::PeerManager::process_events
/// [`ChannelMonitorUpdateStatus::InProgress`]: crate::chain::ChannelMonitorUpdateStatus::InProgress
pub fn send_payment(&self, route: &Route, payment_hash: PaymentHash, payment_secret: &Option<PaymentSecret>, payment_id: PaymentId) -> Result<(), PaymentSendFailure> {
}
- /// Retries a payment along the given [`Route`].
- ///
- /// Errors returned are a superset of those returned from [`send_payment`], so see
- /// [`send_payment`] documentation for more details on errors. This method will also error if the
- /// retry amount puts the payment more than 10% over the payment's total amount, if the payment
- /// for the given `payment_id` cannot be found (likely due to timeout or success), or if
- /// further retries have been disabled with [`abandon_payment`].
- ///
- /// [`send_payment`]: [`ChannelManager::send_payment`]
- /// [`abandon_payment`]: [`ChannelManager::abandon_payment`]
- pub fn retry_payment(&self, route: &Route, payment_id: PaymentId) -> Result<(), PaymentSendFailure> {
- let best_block_height = self.best_block.read().unwrap().height();
- self.pending_outbound_payments.retry_payment_with_route(route, payment_id, &self.entropy_source, &self.node_signer, best_block_height,
- |path, payment_params, payment_hash, payment_secret, total_value, cur_height, payment_id, keysend_preimage, session_priv|
- self.send_payment_along_path(path, payment_params, payment_hash, payment_secret, total_value, cur_height, payment_id, keysend_preimage, session_priv))
- }
-
- /// Signals that no further retries for the given payment will occur.
+ /// Signals that no further retries for the given payment should occur. Useful if you have a
+ /// pending outbound payment with retries remaining, but wish to stop retrying the payment before
+ /// retries are exhausted.
///
- /// After this method returns, no future calls to [`retry_payment`] for the given `payment_id`
- /// are allowed. If no [`Event::PaymentFailed`] event had been generated before, one will be
- /// generated as soon as there are no remaining pending HTLCs for this payment.
+ /// If no [`Event::PaymentFailed`] event had been generated before, one will be generated as soon
+ /// as there are no remaining pending HTLCs for this payment.
///
/// Note that calling this method does *not* prevent a payment from succeeding. You must still
/// wait until you receive either a [`Event::PaymentFailed`] or [`Event::PaymentSent`] event to
/// determine the ultimate status of a payment.
///
/// If an [`Event::PaymentFailed`] event is generated and we restart without this
- /// [`ChannelManager`] having been persisted, the payment may still be in the pending state
- /// upon restart. This allows further calls to [`retry_payment`] (and requiring a second call
- /// to [`abandon_payment`] to mark the payment as failed again). Otherwise, future calls to
- /// [`retry_payment`] will fail with [`PaymentSendFailure::ParameterError`].
+ /// [`ChannelManager`] having been persisted, another [`Event::PaymentFailed`] may be generated.
///
- /// [`abandon_payment`]: Self::abandon_payment
- /// [`retry_payment`]: Self::retry_payment
/// [`Event::PaymentFailed`]: events::Event::PaymentFailed
/// [`Event::PaymentSent`]: events::Event::PaymentSent
pub fn abandon_payment(&self, payment_id: PaymentId) {
let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
- if let Some(payment_failed_ev) = self.pending_outbound_payments.abandon_payment(payment_id) {
- self.pending_events.lock().unwrap().push(payment_failed_ev);
- }
+ self.pending_outbound_payments.abandon_payment(payment_id, &self.pending_events);
}
/// Send a spontaneous payment, which is a payment that does not require the recipient to have
let best_block_height = self.best_block.read().unwrap().height();
self.pending_outbound_payments.check_retry_payments(&self.router, || self.list_usable_channels(),
- || self.compute_inflight_htlcs(), &self.entropy_source, &self.node_signer, best_block_height, &self.logger,
+ || self.compute_inflight_htlcs(), &self.entropy_source, &self.node_signer, best_block_height,
+ &self.pending_events, &self.logger,
|path, payment_params, payment_hash, payment_secret, total_value, cur_height, payment_id, keysend_preimage, session_priv|
self.send_payment_along_path(path, payment_params, payment_hash, payment_secret, total_value, cur_height, payment_id, keysend_preimage, session_priv));
/// Fails an HTLC backwards to the sender of it to us.
/// Note that we do not assume that channels corresponding to failed HTLCs are still available.
fn fail_htlc_backwards_internal(&self, source: &HTLCSource, payment_hash: &PaymentHash, onion_error: &HTLCFailReason, destination: HTLCDestination) {
- #[cfg(any(feature = "_test_utils", test))]
- {
- // Ensure that the peer state channel storage lock is not held when calling this
- // function.
- // This ensures that future code doesn't introduce a lock_order requirement for
- // `forward_htlcs` to be locked after the `per_peer_state` peer locks, which calling
- // this function with any `per_peer_state` peer lock aquired would.
- let per_peer_state = self.per_peer_state.read().unwrap();
- for (_, peer) in per_peer_state.iter() {
- debug_assert!(peer.try_lock().is_ok());
- }
+ // Ensure that no peer state channel storage lock is held when calling this function.
+ // This ensures that future code doesn't introduce a lock-order requirement for
+ // `forward_htlcs` to be locked after the `per_peer_state` peer locks, which calling
+ // this function with any `per_peer_state` peer lock acquired would.
+ for (_, peer) in self.per_peer_state.read().unwrap().iter() {
+ debug_assert_ne!(peer.held_by_thread(), LockHeldState::HeldByThread);
}
//TODO: There is a timing attack here where if a node fails an HTLC back to us they can
// being fully configured. See the docs for `ChannelManagerReadArgs` for more.
match source {
HTLCSource::OutboundRoute { ref path, ref session_priv, ref payment_id, ref payment_params, .. } => {
- self.pending_outbound_payments.fail_htlc(source, payment_hash, onion_error, path, session_priv, payment_id, payment_params, self.probing_cookie_secret, &self.secp_ctx, &self.pending_events, &self.logger);
+ if self.pending_outbound_payments.fail_htlc(source, payment_hash, onion_error, path,
+ session_priv, payment_id, payment_params, self.probing_cookie_secret, &self.secp_ctx,
+ &self.pending_events, &self.logger)
+ { self.push_pending_forwards_ev(); }
},
HTLCSource::PreviousHopData(HTLCPreviousHopData { ref short_channel_id, ref htlc_id, ref incoming_packet_shared_secret, ref phantom_shared_secret, ref outpoint }) => {
log_trace!(self.logger, "Failing HTLC with payment_hash {} backwards from us with {:?}", log_bytes!(payment_hash.0), onion_error);
let err_packet = onion_error.get_encrypted_failure_packet(incoming_packet_shared_secret, phantom_shared_secret);
- let mut forward_event = None;
+ let mut push_forward_ev = false;
let mut forward_htlcs = self.forward_htlcs.lock().unwrap();
if forward_htlcs.is_empty() {
- forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS));
+ push_forward_ev = true;
}
match forward_htlcs.entry(*short_channel_id) {
hash_map::Entry::Occupied(mut entry) => {
}
}
mem::drop(forward_htlcs);
+ if push_forward_ev { self.push_pending_forwards_ev(); }
let mut pending_events = self.pending_events.lock().unwrap();
- if let Some(time) = forward_event {
- pending_events.push(events::Event::PendingHTLCsForwardable {
- time_forwardable: time
- });
- }
pending_events.push(events::Event::HTLCHandlingFailed {
prev_channel_id: outpoint.to_channel_id(),
failed_next_destination: destination,
#[inline]
fn forward_htlcs(&self, per_source_pending_forwards: &mut [(u64, OutPoint, u128, Vec<(PendingHTLCInfo, u64)>)]) {
for &mut (prev_short_channel_id, prev_funding_outpoint, prev_user_channel_id, ref mut pending_forwards) in per_source_pending_forwards {
- let mut forward_event = None;
+ let mut push_forward_event = false;
let mut new_intercept_events = Vec::new();
let mut failed_intercept_forwards = Vec::new();
if !pending_forwards.is_empty() {
// We don't want to generate a PendingHTLCsForwardable event if only intercepted
// payments are being processed.
if forward_htlcs_empty {
- forward_event = Some(Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS));
+ push_forward_event = true;
}
entry.insert(vec!(HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
prev_short_channel_id, prev_funding_outpoint, prev_htlc_id, prev_user_channel_id, forward_info })));
let mut events = self.pending_events.lock().unwrap();
events.append(&mut new_intercept_events);
}
+ if push_forward_event { self.push_pending_forwards_ev() }
+ }
+ }
- match forward_event {
- Some(time) => {
- let mut pending_events = self.pending_events.lock().unwrap();
- pending_events.push(events::Event::PendingHTLCsForwardable {
- time_forwardable: time
- });
- }
- None => {},
- }
+ // We only want to push a PendingHTLCsForwardable event if no others are queued.
+ fn push_pending_forwards_ev(&self) {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ let forward_ev_exists = pending_events.iter()
+ .find(|ev| if let events::Event::PendingHTLCsForwardable { .. } = ev { true } else { false })
+ .is_some();
+ if !forward_ev_exists {
+ pending_events.push(events::Event::PendingHTLCsForwardable {
+ time_forwardable:
+ Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS),
+ });
}
}
}
}
- if !forward_htlcs.is_empty() {
+ let pending_outbounds = OutboundPayments { pending_outbound_payments: Mutex::new(pending_outbound_payments.unwrap()), retry_lock: Mutex::new(()) };
+ if !forward_htlcs.is_empty() || pending_outbounds.needs_abandon() {
// If we have pending HTLCs to forward, assume we either dropped a
// `PendingHTLCsForwardable` or the user received it but never processed it as they
// shut down before the timer hit. Either way, set the time_forwardable to a small
inbound_payment_key: expanded_inbound_key,
pending_inbound_payments: Mutex::new(pending_inbound_payments),
- pending_outbound_payments: OutboundPayments { pending_outbound_payments: Mutex::new(pending_outbound_payments.unwrap()) },
+ pending_outbound_payments: pending_outbounds,
pending_intercepted_htlcs: Mutex::new(pending_intercepted_htlcs.unwrap()),
forward_htlcs: Mutex::new(forward_htlcs),
}
}
+/// If we need an unsafe pointer to a `Node` (ie to reference it in a thread
+/// pre-std::thread::scope), this provides that with `Sync`. Note that accessing some of the fields
+/// in the `Node` are not safe to use (i.e. the ones behind an `Rc`), but that's left to the caller
+/// to figure out.
+pub struct NodePtr(pub *const Node<'static, 'static, 'static>);
+impl NodePtr {
+ pub fn from_node<'a, 'b: 'a, 'c: 'b>(node: &Node<'a, 'b, 'c>) -> Self {
+ Self((node as *const Node<'a, 'b, 'c>).cast())
+ }
+}
+unsafe impl Send for NodePtr {}
+unsafe impl Sync for NodePtr {}
+
impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
fn drop(&mut self) {
if !panicking() {
}
pub fn expect_payment_failed_conditions_event<'a, 'b, 'c, 'd, 'e>(
- node: &'a Node<'b, 'c, 'd>, payment_failed_event: Event, expected_payment_hash: PaymentHash,
+ payment_failed_events: Vec<Event>, expected_payment_hash: PaymentHash,
expected_payment_failed_permanently: bool, conditions: PaymentFailedConditions<'e>
) {
- let expected_payment_id = match payment_failed_event {
+ if conditions.expected_mpp_parts_remain { assert_eq!(payment_failed_events.len(), 1); } else { assert_eq!(payment_failed_events.len(), 2); }
+ let expected_payment_id = match &payment_failed_events[0] {
Event::PaymentPathFailed { payment_hash, payment_failed_permanently, path, retry, payment_id, network_update, short_channel_id,
#[cfg(test)]
error_code,
#[cfg(test)]
error_data, .. } => {
- assert_eq!(payment_hash, expected_payment_hash, "unexpected payment_hash");
- assert_eq!(payment_failed_permanently, expected_payment_failed_permanently, "unexpected payment_failed_permanently value");
+ assert_eq!(*payment_hash, expected_payment_hash, "unexpected payment_hash");
+ assert_eq!(*payment_failed_permanently, expected_payment_failed_permanently, "unexpected payment_failed_permanently value");
assert!(retry.is_some(), "expected retry.is_some()");
assert_eq!(retry.as_ref().unwrap().final_value_msat, path.last().unwrap().fee_msat, "Retry amount should match last hop in path");
assert_eq!(retry.as_ref().unwrap().payment_params.payee_pubkey, path.last().unwrap().pubkey, "Retry payee node_id should match last hop in path");
},
Some(NetworkUpdate::ChannelFailure { short_channel_id, is_permanent }) if chan_closed => {
if let Some(scid) = conditions.expected_blamed_scid {
- assert_eq!(short_channel_id, scid);
+ assert_eq!(*short_channel_id, scid);
}
assert!(is_permanent);
},
_ => panic!("Unexpected event"),
};
if !conditions.expected_mpp_parts_remain {
- node.node.abandon_payment(expected_payment_id);
- let events = node.node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- match events[0] {
+ match &payment_failed_events[1] {
Event::PaymentFailed { ref payment_hash, ref payment_id } => {
assert_eq!(*payment_hash, expected_payment_hash, "unexpected second payment_hash");
assert_eq!(*payment_id, expected_payment_id);
node: &'a Node<'b, 'c, 'd>, expected_payment_hash: PaymentHash, expected_payment_failed_permanently: bool,
conditions: PaymentFailedConditions<'e>
) {
- let mut events = node.node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- expect_payment_failed_conditions_event(node, events.pop().unwrap(), expected_payment_hash, expected_payment_failed_permanently, conditions);
+ let events = node.node.get_and_clear_pending_events();
+ expect_payment_failed_conditions_event(events, expected_payment_hash, expected_payment_failed_permanently, conditions);
}
pub fn send_along_route_with_secret<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, route: Route, expected_paths: &[&[&Node<'a, 'b, 'c>]], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: PaymentSecret) -> PaymentId {
}
pub fn pass_failed_payment_back<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_paths_slice: &[&[&Node<'a, 'b, 'c>]], skip_last: bool, our_payment_hash: PaymentHash) {
- let expected_payment_id = pass_failed_payment_back_no_abandon(origin_node, expected_paths_slice, skip_last, our_payment_hash);
- if !skip_last {
- origin_node.node.abandon_payment(expected_payment_id.unwrap());
- let events = origin_node.node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- match events[0] {
- Event::PaymentFailed { ref payment_hash, ref payment_id } => {
- assert_eq!(*payment_hash, our_payment_hash, "unexpected second payment_hash");
- assert_eq!(*payment_id, expected_payment_id.unwrap());
- }
- _ => panic!("Unexpected second event"),
- }
- }
-}
-
-pub fn pass_failed_payment_back_no_abandon<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_paths_slice: &[&[&Node<'a, 'b, 'c>]], skip_last: bool, our_payment_hash: PaymentHash) -> Option<PaymentId> {
let mut expected_paths: Vec<_> = expected_paths_slice.iter().collect();
check_added_monitors!(expected_paths[0].last().unwrap(), expected_paths.len());
per_path_msgs.sort_unstable_by(|(_, node_id_a), (_, node_id_b)| node_id_a.cmp(node_id_b));
expected_paths.sort_unstable_by(|path_a, path_b| path_a[path_a.len() - 2].node.get_our_node_id().cmp(&path_b[path_b.len() - 2].node.get_our_node_id()));
- let mut expected_payment_id = None;
-
for (i, (expected_route, (path_msgs, next_hop))) in expected_paths.iter().zip(per_path_msgs.drain(..)).enumerate() {
let mut next_msgs = Some(path_msgs);
let mut expected_next_node = next_hop;
assert!(origin_node.node.get_and_clear_pending_msg_events().is_empty());
commitment_signed_dance!(origin_node, prev_node, next_msgs.as_ref().unwrap().1, false);
let events = origin_node.node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- expected_payment_id = Some(match events[0] {
+ if i == expected_paths.len() - 1 { assert_eq!(events.len(), 2); } else { assert_eq!(events.len(), 1); }
+
+ let expected_payment_id = match events[0] {
Event::PaymentPathFailed { payment_hash, payment_failed_permanently, all_paths_failed, ref path, ref payment_id, .. } => {
assert_eq!(payment_hash, our_payment_hash);
assert!(payment_failed_permanently);
payment_id.unwrap()
},
_ => panic!("Unexpected event"),
- });
+ };
+ if i == expected_paths.len() - 1 {
+ match events[1] {
+ Event::PaymentFailed { ref payment_hash, ref payment_id } => {
+ assert_eq!(*payment_hash, our_payment_hash, "unexpected second payment_hash");
+ assert_eq!(*payment_id, expected_payment_id);
+ }
+ _ => panic!("Unexpected second event"),
+ }
+ }
}
}
assert!(expected_paths[0].last().unwrap().node.get_and_clear_pending_events().is_empty());
assert!(expected_paths[0].last().unwrap().node.get_and_clear_pending_msg_events().is_empty());
check_added_monitors!(expected_paths[0].last().unwrap(), 0);
-
- expected_payment_id
}
pub fn fail_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_path: &[&Node<'a, 'b, 'c>], our_payment_hash: PaymentHash) {
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
let events = nodes[1].node.get_and_clear_pending_events();
- assert_eq!(events.len(), if deliver_bs_raa { 2 + nodes.len() - 1 } else { 3 + nodes.len() });
+ assert_eq!(events.len(), if deliver_bs_raa { 3 + nodes.len() - 1 } else { 4 + nodes.len() });
match events[0] {
Event::ChannelClosed { reason: ClosureReason::CommitmentTxConfirmed, .. } => { },
_ => panic!("Unexepected event"),
},
_ => panic!("Unexpected event"),
}
- if !deliver_bs_raa {
- match events[2] {
- Event::PendingHTLCsForwardable { .. } => { },
- _ => panic!("Unexpected event"),
- };
- nodes[1].node.abandon_payment(PaymentId(fourth_payment_hash.0));
- let payment_failed_events = nodes[1].node.get_and_clear_pending_events();
- assert_eq!(payment_failed_events.len(), 1);
- match payment_failed_events[0] {
- Event::PaymentFailed { ref payment_hash, .. } => {
- assert_eq!(*payment_hash, fourth_payment_hash);
- },
- _ => panic!("Unexpected event"),
- }
+ match events[2] {
+ Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(*payment_hash, fourth_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
}
nodes[1].node.process_pending_htlc_forwards();
commitment_signed_dance!(nodes[0], nodes[1], commitment_signed, false, true);
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 3);
+ assert_eq!(events.len(), 6);
match events[0] {
Event::PaymentPathFailed { ref payment_hash, ref network_update, .. } => {
assert!(failed_htlcs.insert(payment_hash.0));
_ => panic!("Unexpected event"),
}
match events[1] {
+ Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(*payment_hash, first_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[2] {
Event::PaymentPathFailed { ref payment_hash, ref network_update, .. } => {
assert!(failed_htlcs.insert(payment_hash.0));
assert!(network_update.is_some());
},
_ => panic!("Unexpected event"),
}
- match events[2] {
+ match events[3] {
+ Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(*payment_hash, second_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[4] {
Event::PaymentPathFailed { ref payment_hash, ref network_update, .. } => {
assert!(failed_htlcs.insert(payment_hash.0));
assert!(network_update.is_some());
},
_ => panic!("Unexpected event"),
}
+ match events[5] {
+ Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(*payment_hash, third_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
},
_ => panic!("Unexpected event"),
}
nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &update_add_htlc);
}
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 2);
+ assert_eq!(events.len(), 3);
// Check that Alice fails backward the pending HTLC from the second payment.
match events[0] {
Event::PaymentPathFailed { payment_hash, .. } => {
_ => panic!("Unexpected event"),
}
match events[1] {
+ Event::PaymentFailed { payment_hash, .. } => {
+ assert_eq!(payment_hash, failed_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[2] {
Event::ChannelClosed { reason: ClosureReason::ProcessingError { ref err }, .. } => {
assert_eq!(err, "Remote side tried to send a 0-msat HTLC");
},
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (1, 0), (1, 0), (false, false));
{
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 3);
+ assert_eq!(events.len(), 4);
match events[0] {
Event::PaymentSent { payment_preimage, payment_hash, .. } => {
assert_eq!(payment_preimage, payment_preimage_3);
_ => panic!("Unexpected event"),
}
match events[2] {
+ Event::PaymentFailed { payment_hash, .. } => {
+ assert_eq!(payment_hash, payment_hash_5);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ match events[3] {
Event::PaymentPathSuccessful { .. } => {},
_ => panic!("Unexpected event"),
}
}
let as_events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(as_events.len(), if announce_latest { 5 } else { 3 });
+ assert_eq!(as_events.len(), if announce_latest { 10 } else { 6 });
let mut as_failds = HashSet::new();
let mut as_updates = 0;
for event in as_events.iter() {
if network_update.is_some() {
as_updates += 1;
}
+ } else if let &Event::PaymentFailed { .. } = event {
} else { panic!("Unexpected event"); }
}
assert!(as_failds.contains(&payment_hash_1));
assert!(as_failds.contains(&payment_hash_6));
let bs_events = nodes[1].node.get_and_clear_pending_events();
- assert_eq!(bs_events.len(), if announce_latest { 4 } else { 3 });
+ assert_eq!(bs_events.len(), if announce_latest { 8 } else { 6 });
let mut bs_failds = HashSet::new();
let mut bs_updates = 0;
for event in bs_events.iter() {
if network_update.is_some() {
bs_updates += 1;
}
+ } else if let &Event::PaymentFailed { .. } = event {
} else { panic!("Unexpected event"); }
}
assert!(bs_failds.contains(&payment_hash_1));
// Check that the payment failed to be sent out.
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
+ assert_eq!(events.len(), 2);
match &events[0] {
&Event::PaymentPathFailed { ref payment_id, ref payment_hash, ref payment_failed_permanently, ref network_update, ref all_paths_failed, ref short_channel_id, .. } => {
assert_eq!(PaymentId(our_payment_hash.0), *payment_id.as_ref().unwrap());
},
_ => panic!("Unexpected event"),
}
+ match &events[1] {
+ &Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(our_payment_hash.clone(), *payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
}
// Test that if multiple HTLCs are released from the holding cell and one is
// Check that the second payment failed to be sent out.
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
+ assert_eq!(events.len(), 2);
match &events[0] {
&Event::PaymentPathFailed { ref payment_id, ref payment_hash, ref payment_failed_permanently, ref network_update, ref all_paths_failed, ref short_channel_id, .. } => {
assert_eq!(payment_id_2, *payment_id.as_ref().unwrap());
},
_ => panic!("Unexpected event"),
}
+ match &events[1] {
+ &Event::PaymentFailed { ref payment_hash, .. } => {
+ assert_eq!(payment_hash_2.clone(), *payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
// Complete the first payment and the RAA from the fee update.
let (payment_event, send_raa_event) = {
}
let events_5 = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events_5.len(), 1);
+ assert_eq!(events_5.len(), 2);
// Expect a PaymentPathFailed event with a ChannelFailure network update for the channel between
// the node originating the error to its next hop.
},
_ => panic!("Unexpected event"),
}
+ match events_5[1] {
+ Event::PaymentFailed { payment_hash, .. } => {
+ assert_eq!(payment_hash, our_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
// TODO: Test actual removal of channel from NetworkGraph when it's implemented.
}
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
let events = nodes[0].node.get_and_clear_pending_events();
// Only 2 PaymentPathFailed events should show up, over-dust HTLC has to be failed by timeout tx
- assert_eq!(events.len(), 2);
+ assert_eq!(events.len(), 4);
let mut first_failed = false;
for event in events {
match event {
} else {
assert_eq!(payment_hash, payment_hash_2);
}
- }
+ },
+ Event::PaymentFailed { .. } => {}
_ => panic!("Unexpected event"),
}
}
commitment_signed_dance!(nodes[0], nodes[1], fail_updates_1.commitment_signed, false);
let failure_events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(failure_events.len(), 2);
+ assert_eq!(failure_events.len(), 4);
if let Event::PaymentPathFailed { .. } = failure_events[0] {} else { panic!(); }
- if let Event::PaymentPathFailed { .. } = failure_events[1] {} else { panic!(); }
+ if let Event::PaymentFailed { .. } = failure_events[1] {} else { panic!(); }
+ if let Event::PaymentPathFailed { .. } = failure_events[2] {} else { panic!(); }
+ if let Event::PaymentFailed { .. } = failure_events[3] {} else { panic!(); }
} else {
// Let the second HTLC fail and claim the first
expect_pending_htlcs_forwardable_and_htlc_handling_failed_ignore!(nodes[1], vec![HTLCDestination::FailedPayment { payment_hash: our_payment_hash }]);
nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &fail_updates_1.update_fail_htlcs[0]);
commitment_signed_dance!(nodes[0], nodes[1], fail_updates_1.commitment_signed, false);
- expect_payment_failed_conditions(&nodes[0], our_payment_hash, true, PaymentFailedConditions::new().mpp_parts_remain());
+ expect_payment_failed_conditions(&nodes[0], our_payment_hash, true, PaymentFailedConditions::new());
claim_payment(&nodes[0], &[&nodes[1]], our_payment_preimage);
}
assert_eq!(events.len(), 1);
pass_along_path(&nodes[0], &[&nodes[2], &nodes[3]], 15_000_000, our_payment_hash, Some(our_payment_secret), events.pop().unwrap(), true, None);
- claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[3]], &[&nodes[2], &nodes[3]]], false, our_payment_preimage);
+ do_claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[3]], &[&nodes[2], &nodes[3]]], false, our_payment_preimage);
+ let events = nodes[0].node.get_and_clear_pending_events();
+ assert_eq!(events.len(), 3);
+ match events[0] {
+ Event::PaymentSent { payment_hash, .. } => { // The payment was abandoned earlier, so the fee paid will be None
+ assert_eq!(payment_hash, our_payment_hash);
+ },
+ _ => panic!("Unexpected event")
+ }
+ match events[1] {
+ Event::PaymentPathSuccessful { payment_hash, .. } => {
+ assert_eq!(payment_hash.unwrap(), our_payment_hash);
+ },
+ _ => panic!("Unexpected event")
+ }
+ match events[2] {
+ Event::PaymentPathSuccessful { payment_hash, .. } => {
+ assert_eq!(payment_hash.unwrap(), our_payment_hash);
+ },
+ _ => panic!("Unexpected event")
+ }
}
#[test]
test_spendable_output(&nodes[1], &as_revoked_txn[0]);
let mut payment_failed_events = nodes[1].node.get_and_clear_pending_events();
- expect_payment_failed_conditions_event(&nodes[1], payment_failed_events.pop().unwrap(),
- dust_payment_hash, false, PaymentFailedConditions::new());
- expect_payment_failed_conditions_event(&nodes[1], payment_failed_events.pop().unwrap(),
+ expect_payment_failed_conditions_event(payment_failed_events[..2].to_vec(),
missing_htlc_payment_hash, false, PaymentFailedConditions::new());
- assert!(payment_failed_events.is_empty());
+ expect_payment_failed_conditions_event(payment_failed_events[2..].to_vec(),
+ dust_payment_hash, false, PaymentFailedConditions::new());
connect_blocks(&nodes[1], 1);
test_spendable_output(&nodes[1], &claim_txn[if confirm_htlc_spend_first { 2 } else { 3 }]);
commitment_signed_dance!(nodes[0], nodes[1], update_1_0.commitment_signed, false, true);
let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
+ assert_eq!(events.len(), 2);
if let &Event::PaymentPathFailed { ref payment_failed_permanently, ref network_update, ref all_paths_failed, ref short_channel_id, ref error_code, .. } = &events[0] {
assert_eq!(*payment_failed_permanently, !expected_retryable);
assert_eq!(*all_paths_failed, true);
} else {
panic!("Unexpected event");
}
- nodes[0].node.abandon_payment(payment_id);
- let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- match events[0] {
+ match events[1] {
Event::PaymentFailed { payment_hash: ev_payment_hash, payment_id: ev_payment_id } => {
assert_eq!(*payment_hash, ev_payment_hash);
assert_eq!(payment_id, ev_payment_id);
use crate::chain::keysinterface::{EntropySource, NodeSigner, Recipient};
use crate::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
-use crate::ln::channelmanager::{ChannelDetails, HTLCSource, IDEMPOTENCY_TIMEOUT_TICKS, MIN_HTLC_RELAY_HOLDING_CELL_MILLIS, PaymentId};
+use crate::ln::channelmanager::{ChannelDetails, HTLCSource, IDEMPOTENCY_TIMEOUT_TICKS, PaymentId};
use crate::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA as LDK_DEFAULT_MIN_FINAL_CLTV_EXPIRY_DELTA;
use crate::ln::msgs::DecodeError;
use crate::ln::onion_utils::HTLCFailReason;
use core::cmp;
use core::fmt::{self, Display, Formatter};
use core::ops::Deref;
-use core::time::Duration;
use crate::prelude::*;
use crate::sync::Mutex;
let our_payment_hash;
core::mem::swap(&mut session_privs, match self {
PendingOutboundPayment::Legacy { .. } |
- PendingOutboundPayment::Fulfilled { .. } =>
+ PendingOutboundPayment::Fulfilled { .. } =>
return Err(()),
- PendingOutboundPayment::Retryable { session_privs, payment_hash, .. } |
- PendingOutboundPayment::Abandoned { session_privs, payment_hash, .. } => {
- our_payment_hash = *payment_hash;
- session_privs
- },
+ PendingOutboundPayment::Retryable { session_privs, payment_hash, .. } |
+ PendingOutboundPayment::Abandoned { session_privs, payment_hash, .. } => {
+ our_payment_hash = *payment_hash;
+ session_privs
+ },
});
*self = PendingOutboundPayment::Abandoned { session_privs, payment_hash: our_payment_hash };
Ok(())
///
/// You can freely resend the payment in full (with the parameter error fixed).
///
- /// Because the payment failed outright, no payment tracking is done, you do not need to call
- /// [`ChannelManager::abandon_payment`] and [`ChannelManager::retry_payment`] will *not* work
- /// for this payment.
+ /// Because the payment failed outright, no payment tracking is done and no
+ /// [`Event::PaymentPathFailed`] or [`Event::PaymentFailed`] events will be generated.
///
- /// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
+ /// [`Event::PaymentPathFailed`]: crate::util::events::Event::PaymentPathFailed
+ /// [`Event::PaymentFailed`]: crate::util::events::Event::PaymentFailed
ParameterError(APIError),
/// A parameter in a single path which was passed to send_payment was invalid, preventing us
/// from attempting to send the payment at all.
///
/// You can freely resend the payment in full (with the parameter error fixed).
///
+ /// Because the payment failed outright, no payment tracking is done and no
+ /// [`Event::PaymentPathFailed`] or [`Event::PaymentFailed`] events will be generated.
+ ///
/// The results here are ordered the same as the paths in the route object which was passed to
/// send_payment.
///
- /// Because the payment failed outright, no payment tracking is done, you do not need to call
- /// [`ChannelManager::abandon_payment`] and [`ChannelManager::retry_payment`] will *not* work
- /// for this payment.
- ///
- /// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
+ /// [`Event::PaymentPathFailed`]: crate::util::events::Event::PaymentPathFailed
+ /// [`Event::PaymentFailed`]: crate::util::events::Event::PaymentFailed
PathParameterError(Vec<Result<(), APIError>>),
/// All paths which were attempted failed to send, with no channel state change taking place.
/// You can freely resend the payment in full (though you probably want to do so over different
/// paths than the ones selected).
///
- /// Because the payment failed outright, no payment tracking is done, you do not need to call
- /// [`ChannelManager::abandon_payment`] and [`ChannelManager::retry_payment`] will *not* work
- /// for this payment.
+ /// Because the payment failed outright, no payment tracking is done and no
+ /// [`Event::PaymentPathFailed`] or [`Event::PaymentFailed`] events will be generated.
///
- /// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
+ /// [`Event::PaymentPathFailed`]: crate::util::events::Event::PaymentPathFailed
+ /// [`Event::PaymentFailed`]: crate::util::events::Event::PaymentFailed
AllFailedResendSafe(Vec<APIError>),
/// Indicates that a payment for the provided [`PaymentId`] is already in-flight and has not
- /// yet completed (i.e. generated an [`Event::PaymentSent`]) or been abandoned (via
- /// [`ChannelManager::abandon_payment`]).
+ /// yet completed (i.e. generated an [`Event::PaymentSent`] or [`Event::PaymentFailed`]).
///
/// [`PaymentId`]: crate::ln::channelmanager::PaymentId
/// [`Event::PaymentSent`]: crate::util::events::Event::PaymentSent
- /// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
+ /// [`Event::PaymentFailed`]: crate::util::events::Event::PaymentFailed
DuplicatePayment,
- /// Some paths which were attempted failed to send, though possibly not all. At least some
- /// paths have irrevocably committed to the HTLC and retrying the payment in full would result
- /// in over-/re-payment.
+ /// Some paths that were attempted failed to send, though some paths may have succeeded. At least
+ /// some paths have irrevocably committed to the HTLC.
///
- /// The results here are ordered the same as the paths in the route object which was passed to
- /// send_payment, and any `Err`s which are not [`APIError::MonitorUpdateInProgress`] can be
- /// safely retried via [`ChannelManager::retry_payment`].
+ /// The results here are ordered the same as the paths in the route object that was passed to
+ /// send_payment.
///
- /// Any entries which contain `Err(APIError::MonitorUpdateInprogress)` or `Ok(())` MUST NOT be
- /// retried as they will result in over-/re-payment. These HTLCs all either successfully sent
- /// (in the case of `Ok(())`) or will send once a [`MonitorEvent::Completed`] is provided for
- /// the next-hop channel with the latest update_id.
+ /// Any entries that contain `Err(APIError::MonitorUpdateInprogress)` will send once a
+ /// [`MonitorEvent::Completed`] is provided for the next-hop channel with the latest update_id.
///
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
/// [`MonitorEvent::Completed`]: crate::chain::channelmonitor::MonitorEvent::Completed
PartialFailure {
- /// The errors themselves, in the same order as the route hops.
+ /// The errors themselves, in the same order as the paths from the route.
results: Vec<Result<(), APIError>>,
/// If some paths failed without irrevocably committing to the new HTLC(s), this will
- /// contain a [`RouteParameters`] object which can be used to calculate a new route that
- /// will pay all remaining unpaid balance.
+ /// contain a [`RouteParameters`] object for the failing paths.
failed_paths_retry: Option<RouteParameters>,
/// The payment id for the payment, which is now at least partially pending.
payment_id: PaymentId,
pub(super) struct OutboundPayments {
pub(super) pending_outbound_payments: Mutex<HashMap<PaymentId, PendingOutboundPayment>>,
+ pub(super) retry_lock: Mutex<()>,
}
impl OutboundPayments {
pub(super) fn new() -> Self {
Self {
- pending_outbound_payments: Mutex::new(HashMap::new())
+ pending_outbound_payments: Mutex::new(HashMap::new()),
+ retry_lock: Mutex::new(()),
}
}
pub(super) fn check_retry_payments<R: Deref, ES: Deref, NS: Deref, SP, IH, FH, L: Deref>(
&self, router: &R, first_hops: FH, inflight_htlcs: IH, entropy_source: &ES, node_signer: &NS,
- best_block_height: u32, logger: &L, send_payment_along_path: SP,
+ best_block_height: u32, pending_events: &Mutex<Vec<events::Event>>, logger: &L,
+ send_payment_along_path: SP,
)
where
R::Target: Router,
FH: Fn() -> Vec<ChannelDetails>,
L::Target: Logger,
{
+ let _single_thread = self.retry_lock.lock().unwrap();
loop {
let mut outbounds = self.pending_outbound_payments.lock().unwrap();
let mut retry_id_route_params = None;
}
}
}
+ core::mem::drop(outbounds);
if let Some((payment_id, route_params)) = retry_id_route_params {
- core::mem::drop(outbounds);
if let Err(e) = self.pay_internal(payment_id, None, route_params, router, first_hops(), &inflight_htlcs, entropy_source, node_signer, best_block_height, logger, &send_payment_along_path) {
log_info!(logger, "Errored retrying payment: {:?}", e);
+ // If we error on retry, there is no chance of the payment succeeding and no HTLCs have
+ // been irrevocably committed to, so we can safely abandon.
+ self.abandon_payment(payment_id, pending_events);
}
} else { break }
}
+
+ let mut outbounds = self.pending_outbound_payments.lock().unwrap();
+ outbounds.retain(|pmt_id, pmt| {
+ let mut retain = true;
+ if !pmt.is_auto_retryable_now() && pmt.remaining_parts() == 0 {
+ if pmt.mark_abandoned().is_ok() {
+ pending_events.lock().unwrap().push(events::Event::PaymentFailed {
+ payment_id: *pmt_id,
+ payment_hash: pmt.payment_hash().expect("PendingOutboundPayments::Retryable always has a payment hash set"),
+ });
+ retain = false;
+ }
+ }
+ retain
+ });
+ }
+
+ pub(super) fn needs_abandon(&self) -> bool {
+ let outbounds = self.pending_outbound_payments.lock().unwrap();
+ outbounds.iter().any(|(_, pmt)|
+ !pmt.is_auto_retryable_now() && pmt.remaining_parts() == 0 && !pmt.is_fulfilled())
}
+ /// Will return `Ok(())` iff at least one HTLC is sent for the payment.
fn pay_internal<R: Deref, NS: Deref, ES: Deref, IH, SP, L: Deref>(
&self, payment_id: PaymentId,
initial_send_info: Option<(PaymentHash, &Option<PaymentSecret>, Option<PaymentPreimage>, Retry)>,
.map_err(|e| { self.remove_outbound_if_all_failed(payment_id, &e); e })
}
- // If we failed to send any paths, we should remove the new PaymentId from the
- // `pending_outbound_payments` map, as the user isn't expected to `abandon_payment`.
+ // If we failed to send any paths, remove the new PaymentId from the `pending_outbound_payments`
+ // map as the payment is free to be resent.
fn remove_outbound_if_all_failed(&self, payment_id: PaymentId, err: &PaymentSendFailure) {
if let &PaymentSendFailure::AllFailedResendSafe(_) = err {
let removed = self.pending_outbound_payments.lock().unwrap().remove(&payment_id).is_some();
});
}
+ // Returns a bool indicating whether a PendingHTLCsForwardable event should be generated.
pub(super) fn fail_htlc<L: Deref>(
&self, source: &HTLCSource, payment_hash: &PaymentHash, onion_error: &HTLCFailReason,
path: &Vec<RouteHop>, session_priv: &SecretKey, payment_id: &PaymentId,
payment_params: &Option<PaymentParameters>, probing_cookie_secret: [u8; 32],
secp_ctx: &Secp256k1<secp256k1::All>, pending_events: &Mutex<Vec<events::Event>>, logger: &L
- ) where L::Target: Logger {
+ ) -> bool where L::Target: Logger {
#[cfg(test)]
let (network_update, short_channel_id, payment_retryable, onion_error_code, onion_error_data) = onion_error.decode_onion_failure(secp_ctx, logger, &source);
#[cfg(not(test))]
let (network_update, short_channel_id, payment_retryable, _, _) = onion_error.decode_onion_failure(secp_ctx, logger, &source);
+ let payment_is_probe = payment_is_probe(payment_hash, &payment_id, probing_cookie_secret);
let mut session_priv_bytes = [0; 32];
session_priv_bytes.copy_from_slice(&session_priv[..]);
let mut outbounds = self.pending_outbound_payments.lock().unwrap();
+
+ // If any payments already need retry, there's no need to generate a redundant
+ // `PendingHTLCsForwardable`.
+ let already_awaiting_retry = outbounds.iter().any(|(_, pmt)| {
+ let mut awaiting_retry = false;
+ if pmt.is_auto_retryable_now() {
+ if let PendingOutboundPayment::Retryable { pending_amt_msat, total_msat, .. } = pmt {
+ if pending_amt_msat < total_msat {
+ awaiting_retry = true;
+ }
+ }
+ }
+ awaiting_retry
+ });
+
let mut all_paths_failed = false;
let mut full_failure_ev = None;
- let mut pending_retry_ev = None;
+ let mut pending_retry_ev = false;
let mut retry = None;
let attempts_remaining = if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(*payment_id) {
if !payment.get_mut().remove(&session_priv_bytes, Some(&path)) {
log_trace!(logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
- return
+ return false
}
if payment.get().is_fulfilled() {
log_trace!(logger, "Received failure of HTLC with payment_hash {} after payment completion", log_bytes!(payment_hash.0));
- return
+ return false
}
- let is_retryable_now = payment.get().is_auto_retryable_now();
+ let mut is_retryable_now = payment.get().is_auto_retryable_now();
if let Some(scid) = short_channel_id {
payment.get_mut().insert_previously_failed_scid(scid);
}
});
}
+ if payment_is_probe || !is_retryable_now || !payment_retryable || retry.is_none() {
+ let _ = payment.get_mut().mark_abandoned(); // we'll only Err if it's a legacy payment
+ is_retryable_now = false;
+ }
if payment.get().remaining_parts() == 0 {
all_paths_failed = true;
if payment.get().abandoned() {
- full_failure_ev = Some(events::Event::PaymentFailed {
- payment_id: *payment_id,
- payment_hash: payment.get().payment_hash().expect("PendingOutboundPayments::RetriesExceeded always has a payment hash set"),
- });
+ if !payment_is_probe {
+ full_failure_ev = Some(events::Event::PaymentFailed {
+ payment_id: *payment_id,
+ payment_hash: payment.get().payment_hash().expect("PendingOutboundPayments::RetriesExceeded always has a payment hash set"),
+ });
+ }
payment.remove();
}
}
is_retryable_now
} else {
log_trace!(logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
- return
+ return false
};
core::mem::drop(outbounds);
log_trace!(logger, "Failing outbound payment HTLC with payment_hash {}", log_bytes!(payment_hash.0));
let path_failure = {
- if payment_is_probe(payment_hash, &payment_id, probing_cookie_secret) {
+ if payment_is_probe {
if !payment_retryable {
events::Event::ProbeSuccessful {
payment_id: *payment_id,
if let Some(scid) = short_channel_id {
retry.as_mut().map(|r| r.payment_params.previously_failed_channels.push(scid));
}
- if payment_retryable && attempts_remaining && retry.is_some() {
+ // If we miss abandoning the payment above, we *must* generate an event here or else the
+ // payment will sit in our outbounds forever.
+ if attempts_remaining && !already_awaiting_retry {
debug_assert!(full_failure_ev.is_none());
- pending_retry_ev = Some(events::Event::PendingHTLCsForwardable {
- time_forwardable: Duration::from_millis(MIN_HTLC_RELAY_HOLDING_CELL_MILLIS),
- });
+ pending_retry_ev = true;
}
events::Event::PaymentPathFailed {
payment_id: Some(*payment_id),
let mut pending_events = pending_events.lock().unwrap();
pending_events.push(path_failure);
if let Some(ev) = full_failure_ev { pending_events.push(ev); }
- if let Some(ev) = pending_retry_ev { pending_events.push(ev); }
+ pending_retry_ev
}
- pub(super) fn abandon_payment(&self, payment_id: PaymentId) -> Option<events::Event> {
- let mut failed_ev = None;
+ pub(super) fn abandon_payment(
+ &self, payment_id: PaymentId, pending_events: &Mutex<Vec<events::Event>>
+ ) {
let mut outbounds = self.pending_outbound_payments.lock().unwrap();
if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) {
if let Ok(()) = payment.get_mut().mark_abandoned() {
if payment.get().remaining_parts() == 0 {
- failed_ev = Some(events::Event::PaymentFailed {
+ pending_events.lock().unwrap().push(events::Event::PaymentFailed {
payment_id,
payment_hash: payment.get().payment_hash().expect("PendingOutboundPayments::RetriesExceeded always has a payment hash set"),
});
}
}
}
- failed_ev
}
#[cfg(test)]
#[cfg(feature = "std")]
use {
crate::util::time::tests::SinceEpoch,
- std::time::{SystemTime, Duration}
+ std::time::{SystemTime, Instant, Duration}
};
-#[test]
-fn retry_single_path_payment() {
- 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);
-
- let _chan_0 = create_announced_chan_between_nodes(&nodes, 0, 1);
- let chan_1 = create_announced_chan_between_nodes(&nodes, 2, 1);
- // Rebalance to find a route
- send_payment(&nodes[2], &vec!(&nodes[1])[..], 3_000_000);
-
- let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], 100_000);
-
- // Rebalance so that the first hop fails.
- send_payment(&nodes[1], &vec!(&nodes[2])[..], 2_000_000);
-
- // Make sure the payment fails on the first hop.
- let payment_id = PaymentId(payment_hash.0);
- nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), payment_id).unwrap();
- check_added_monitors!(nodes[0], 1);
- let mut events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- let mut payment_event = SendEvent::from_event(events.pop().unwrap());
- nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
- check_added_monitors!(nodes[1], 0);
- commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false);
- expect_pending_htlcs_forwardable!(nodes[1]);
- expect_pending_htlcs_forwardable_and_htlc_handling_failed!(&nodes[1], vec![HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: chan_1.2 }]);
- let htlc_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
- assert!(htlc_updates.update_add_htlcs.is_empty());
- assert_eq!(htlc_updates.update_fail_htlcs.len(), 1);
- assert!(htlc_updates.update_fulfill_htlcs.is_empty());
- assert!(htlc_updates.update_fail_malformed_htlcs.is_empty());
- check_added_monitors!(nodes[1], 1);
- nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &htlc_updates.update_fail_htlcs[0]);
- commitment_signed_dance!(nodes[0], nodes[1], htlc_updates.commitment_signed, false);
- expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new().mpp_parts_remain());
-
- // Rebalance the channel so the retry succeeds.
- send_payment(&nodes[2], &vec!(&nodes[1])[..], 3_000_000);
-
- // Mine two blocks (we expire retries after 3, so this will check that we don't expire early)
- connect_blocks(&nodes[0], 2);
-
- // Retry the payment and make sure it succeeds.
- nodes[0].node.retry_payment(&route, payment_id).unwrap();
- check_added_monitors!(nodes[0], 1);
- let mut events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- pass_along_path(&nodes[0], &[&nodes[1], &nodes[2]], 100_000, payment_hash, Some(payment_secret), events.pop().unwrap(), true, None);
- claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], false, payment_preimage);
-}
-
#[test]
fn mpp_failure() {
let chanmon_cfgs = create_chanmon_cfgs(4);
// Rebalance
send_payment(&nodes[3], &vec!(&nodes[2])[..], 1_500_000);
- let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[3], 1_000_000);
+ let amt_msat = 1_000_000;
+ let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[3], amt_msat);
let path = route.paths[0].clone();
route.paths.push(path);
route.paths[0][0].pubkey = nodes[1].node.get_our_node_id();
// Initiate the MPP payment.
let payment_id = PaymentId(payment_hash.0);
- nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), payment_id).unwrap();
+ let mut route_params = RouteParameters {
+ payment_params: route.payment_params.clone().unwrap(),
+ final_value_msat: amt_msat,
+ final_cltv_expiry_delta: TEST_FINAL_CLTV,
+ };
+
+ nodes[0].router.expect_find_route(route_params.clone(), Ok(route.clone()));
+ nodes[0].node.send_payment_with_retry(payment_hash, &Some(payment_secret), payment_id, route_params.clone(), Retry::Attempts(1)).unwrap();
check_added_monitors!(nodes[0], 2); // one monitor per path
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 2);
check_added_monitors!(nodes[2], 1);
nodes[0].node.handle_update_fail_htlc(&nodes[2].node.get_our_node_id(), &htlc_updates.update_fail_htlcs[0]);
commitment_signed_dance!(nodes[0], nodes[2], htlc_updates.commitment_signed, false);
- expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new().mpp_parts_remain());
+ let mut events = nodes[0].node.get_and_clear_pending_events();
+ match events[1] {
+ Event::PendingHTLCsForwardable { .. } => {},
+ _ => panic!("Unexpected event")
+ }
+ events.remove(1);
+ expect_payment_failed_conditions_event(events, payment_hash, false, PaymentFailedConditions::new().mpp_parts_remain());
// Rebalance the channel so the second half of the payment can succeed.
send_payment(&nodes[3], &vec!(&nodes[2])[..], 1_500_000);
- // Make sure it errors as expected given a too-large amount.
- if let Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError { err })) = nodes[0].node.retry_payment(&route, payment_id) {
- assert!(err.contains("over total_payment_amt_msat"));
- } else { panic!("Unexpected error"); }
-
- // Make sure it errors as expected given the wrong payment_id.
- if let Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError { err })) = nodes[0].node.retry_payment(&route, PaymentId([0; 32])) {
- assert!(err.contains("not found"));
- } else { panic!("Unexpected error"); }
-
// Retry the second half of the payment and make sure it succeeds.
- let mut path = route.clone();
- path.paths.remove(0);
- nodes[0].node.retry_payment(&path, payment_id).unwrap();
+ route.paths.remove(0);
+ route_params.final_value_msat = 1_000_000;
+ route_params.payment_params.previously_failed_channels.push(chan_4_update.contents.short_channel_id);
+ nodes[0].router.expect_find_route(route_params, Ok(route));
+ nodes[0].node.process_pending_htlc_forwards();
check_added_monitors!(nodes[0], 1);
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 1);
do_mpp_receive_timeout(false);
}
-#[test]
-fn retry_expired_payment() {
- 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);
-
- let _chan_0 = create_announced_chan_between_nodes(&nodes, 0, 1);
- let chan_1 = create_announced_chan_between_nodes(&nodes, 2, 1);
- // Rebalance to find a route
- send_payment(&nodes[2], &vec!(&nodes[1])[..], 3_000_000);
-
- let (route, payment_hash, _, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], 100_000);
-
- // Rebalance so that the first hop fails.
- send_payment(&nodes[1], &vec!(&nodes[2])[..], 2_000_000);
-
- // Make sure the payment fails on the first hop.
- nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), PaymentId(payment_hash.0)).unwrap();
- check_added_monitors!(nodes[0], 1);
- let mut events = nodes[0].node.get_and_clear_pending_msg_events();
- assert_eq!(events.len(), 1);
- let mut payment_event = SendEvent::from_event(events.pop().unwrap());
- nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
- check_added_monitors!(nodes[1], 0);
- commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false);
- expect_pending_htlcs_forwardable!(nodes[1]);
- expect_pending_htlcs_forwardable_and_htlc_handling_failed!(&nodes[1], vec![HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: chan_1.2 }]);
- let htlc_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
- assert!(htlc_updates.update_add_htlcs.is_empty());
- assert_eq!(htlc_updates.update_fail_htlcs.len(), 1);
- assert!(htlc_updates.update_fulfill_htlcs.is_empty());
- assert!(htlc_updates.update_fail_malformed_htlcs.is_empty());
- check_added_monitors!(nodes[1], 1);
- nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &htlc_updates.update_fail_htlcs[0]);
- commitment_signed_dance!(nodes[0], nodes[1], htlc_updates.commitment_signed, false);
- expect_payment_failed!(nodes[0], payment_hash, false);
-
- // Mine blocks so the payment will have expired.
- connect_blocks(&nodes[0], 3);
-
- // Retry the payment and make sure it errors as expected.
- if let Err(PaymentSendFailure::ParameterError(APIError::APIMisuseError { err })) = nodes[0].node.retry_payment(&route, PaymentId(payment_hash.0)) {
- assert!(err.contains("not found"));
- } else {
- panic!("Unexpected error");
- }
-}
-
#[test]
fn no_pending_leak_on_initial_send_failure() {
// In an earlier version of our payment tracking, we'd have a retry entry even when the initial
// Send two payments - one which will get to nodes[2] and will be claimed, one which we'll time
// out and retry.
- let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], 1_000_000);
+ let amt_msat = 1_000_000;
+ let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[2], amt_msat);
let (payment_preimage_1, payment_hash_1, _, payment_id_1) = send_along_route(&nodes[0], route.clone(), &[&nodes[1], &nodes[2]], 1_000_000);
- nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), PaymentId(payment_hash.0)).unwrap();
+ let route_params = RouteParameters {
+ payment_params: route.payment_params.clone().unwrap(),
+ final_value_msat: amt_msat,
+ final_cltv_expiry_delta: TEST_FINAL_CLTV,
+ };
+ nodes[0].node.send_payment_with_retry(payment_hash, &Some(payment_secret), PaymentId(payment_hash.0), route_params, Retry::Attempts(1)).unwrap();
check_added_monitors!(nodes[0], 1);
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
confirm_transaction(&nodes[0], &first_htlc_timeout_tx);
}
nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear();
- expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new().mpp_parts_remain());
+ expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new());
// Finally, retry the payment (which was reloaded from the ChannelMonitor when nodes[0] was
// reloaded) via a route over the new channel, which work without issue and eventually be
nodes[1].node.timer_tick_occurred();
}
- assert!(nodes[0].node.retry_payment(&new_route, payment_id_1).is_err()); // Shouldn't be allowed to retry a fulfilled payment
- nodes[0].node.retry_payment(&new_route, PaymentId(payment_hash.0)).unwrap();
+ assert!(nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id_1).is_err()); // Shouldn't be allowed to retry a fulfilled payment
+ nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), PaymentId(payment_hash.0)).unwrap();
check_added_monitors!(nodes[0], 1);
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 1);
// If we attempt to retry prior to the HTLC-Timeout (or commitment transaction, for dust HTLCs)
// confirming, we will fail as it's considered still-pending...
let (new_route, _, _, _) = get_route_and_payment_hash!(nodes[0], nodes[2], if use_dust { 1_000 } else { 1_000_000 });
- assert!(nodes[0].node.retry_payment(&new_route, payment_id).is_err());
+ match nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id) {
+ Err(PaymentSendFailure::DuplicatePayment) => {},
+ _ => panic!("Unexpected error")
+ }
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
// After ANTI_REORG_DELAY confirmations, the HTLC should be failed and we can try the payment
// (which should also still work).
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
- // We set mpp_parts_remain to avoid having abandon_payment called
- expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new().mpp_parts_remain());
+ expect_payment_failed_conditions(&nodes[0], payment_hash, false, PaymentFailedConditions::new());
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
let chan_1_monitor_serialized = get_monitor!(nodes[0], chan_id_3).encode();
nodes_0_serialized = nodes[0].node.encode();
- assert!(nodes[0].node.retry_payment(&new_route, payment_id).is_ok());
+ // After the payment failed, we're free to send it again.
+ assert!(nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id).is_ok());
assert!(!nodes[0].node.get_and_clear_pending_msg_events().is_empty());
reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], second_persister, second_new_chain_monitor, second_nodes_0_deserialized);
// Now resend the payment, delivering the HTLC and actually claiming it this time. This ensures
// the payment is not (spuriously) listed as still pending.
- assert!(nodes[0].node.retry_payment(&new_route, payment_id).is_ok());
+ assert!(nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id).is_ok());
check_added_monitors!(nodes[0], 1);
pass_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], if use_dust { 1_000 } else { 1_000_000 }, payment_hash, payment_secret);
claim_payment(&nodes[0], &[&nodes[1], &nodes[2]], payment_preimage);
- assert!(nodes[0].node.retry_payment(&new_route, payment_id).is_err());
+ match nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id) {
+ Err(PaymentSendFailure::DuplicatePayment) => {},
+ _ => panic!("Unexpected error")
+ }
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
let chan_1_monitor_serialized = get_monitor!(nodes[0], chan_id_3).encode();
nodes_0_serialized = nodes[0].node.encode();
- // Ensure that after reload we cannot retry the payment.
+ // Check that after reload we can send the payment again (though we shouldn't, since it was
+ // claimed previously).
reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], third_persister, third_new_chain_monitor, third_nodes_0_deserialized);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
- assert!(nodes[0].node.retry_payment(&new_route, payment_id).is_err());
+ match nodes[0].node.send_payment(&new_route, payment_hash, &Some(payment_secret), payment_id) {
+ Err(PaymentSendFailure::DuplicatePayment) => {},
+ _ => panic!("Unexpected error")
+ }
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
}
},
_ => panic!(),
};
+ assert!(!nodes[0].node.has_pending_payments());
}
#[test]
},
_ => panic!(),
};
+ assert!(!nodes[0].node.has_pending_payments());
}
#[test]
}
}
assert!(found_probe_failed);
+ assert!(!nodes[0].node.has_pending_payments());
}
#[test]
nodes[1].node.fail_htlc_backwards(&first_payment_hash);
expect_pending_htlcs_forwardable_and_htlc_handling_failed!(nodes[1], [HTLCDestination::FailedPayment { payment_hash: first_payment_hash }]);
- pass_failed_payment_back_no_abandon(&nodes[0], &[&[&nodes[1]]], false, first_payment_hash);
- check_send_rejected!();
-
- // Until we abandon the payment, no matter how many timer ticks pass, we still cannot reuse the
+ // Until we abandon the payment upon path failure, no matter how many timer ticks pass, we still cannot reuse the
// PaymentId.
for _ in 0..=IDEMPOTENCY_TIMEOUT_TICKS {
nodes[0].node.timer_tick_occurred();
}
check_send_rejected!();
- nodes[0].node.abandon_payment(payment_id);
- get_event!(nodes[0], Event::PaymentFailed);
+ pass_failed_payment_back(&nodes[0], &[&[&nodes[1]]], false, first_payment_hash);
- // However, we can reuse the PaymentId immediately after we `abandon_payment`.
+ // However, we can reuse the PaymentId immediately after we `abandon_payment` upon passing the
+ // failed payment back.
nodes[0].node.send_payment(&route, second_payment_hash, &Some(second_payment_secret), payment_id).unwrap();
check_added_monitors!(nodes[0], 1);
pass_along_route(&nodes[0], &[&[&nodes[1]]], 100_000, second_payment_hash, second_payment_secret);
FailAttempts,
FailTimeout,
FailOnRestart,
+ FailOnRetry,
}
#[test]
do_automatic_retries(AutoRetry::FailAttempts);
do_automatic_retries(AutoRetry::FailTimeout);
do_automatic_retries(AutoRetry::FailOnRestart);
+ do_automatic_retries(AutoRetry::FailOnRetry);
}
fn do_automatic_retries(test: AutoRetry) {
// Test basic automatic payment retries in ChannelManager. See individual `test` variant comments
check_added_monitors!(&nodes[1], 1);
assert!(update_1.update_fail_htlcs.len() == 1);
let fail_msg = update_1.update_fail_htlcs[0].clone();
-
nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &fail_msg);
commitment_signed_dance!(nodes[0], nodes[1], update_1.commitment_signed, false);
// Ensure the attempt fails and a new PendingHTLCsForwardable event is generated for the retry
let mut events = nodes[0].node.get_and_clear_pending_events();
+ assert_eq!(events.len(), 2);
match events[0] {
Event::PaymentPathFailed { payment_hash: ev_payment_hash, payment_failed_permanently, .. } => {
assert_eq!(payment_hash, ev_payment_hash);
_ => panic!("Unexpected event"),
}
if $expect_pending_htlcs_forwardable {
- assert_eq!(events.len(), 2);
match events[1] {
Event::PendingHTLCsForwardable { .. } => {},
_ => panic!("Unexpected event"),
}
- } else { assert_eq!(events.len(), 1) }
+ } else {
+ match events[1] {
+ Event::PaymentFailed { payment_hash: ev_payment_hash, .. } => {
+ assert_eq!(payment_hash, ev_payment_hash);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ }
}
}
nodes[0].node.process_pending_htlc_forwards();
let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(msg_events.len(), 0);
-
- nodes[0].node.abandon_payment(PaymentId(payment_hash.0));
- let events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- match events[0] {
- Event::PaymentFailed { payment_hash: ref ev_payment_hash, payment_id: ref ev_payment_id } => {
- assert_eq!(payment_hash, *ev_payment_hash);
- assert_eq!(PaymentId(payment_hash.0), *ev_payment_id);
- },
- _ => panic!("Unexpected event"),
- }
} else if test == AutoRetry::FailTimeout {
#[cfg(not(feature = "no-std"))] {
// Ensure ChannelManager will not retry a payment if it times out due to Retry::Timeout.
let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(msg_events.len(), 0);
- nodes[0].node.abandon_payment(PaymentId(payment_hash.0));
let mut events = nodes[0].node.get_and_clear_pending_events();
assert_eq!(events.len(), 1);
match events[0] {
let chan_1_monitor_serialized = get_monitor!(nodes[0], channel_id_1).encode();
reload_node!(nodes[0], node_encoded, &[&chan_1_monitor_serialized], persister, new_chain_monitor, node_0_deserialized);
+ let mut events = nodes[0].node.get_and_clear_pending_events();
+ expect_pending_htlcs_forwardable_from_events!(nodes[0], events, true);
// Make sure we don't retry again.
+ let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+ assert_eq!(msg_events.len(), 0);
+
+ let mut events = nodes[0].node.get_and_clear_pending_events();
+ assert_eq!(events.len(), 1);
+ match events[0] {
+ Event::PaymentFailed { payment_hash: ref ev_payment_hash, payment_id: ref ev_payment_id } => {
+ assert_eq!(payment_hash, *ev_payment_hash);
+ assert_eq!(PaymentId(payment_hash.0), *ev_payment_id);
+ },
+ _ => panic!("Unexpected event"),
+ }
+ } else if test == AutoRetry::FailOnRetry {
+ nodes[0].node.send_payment_with_retry(payment_hash, &Some(payment_secret), PaymentId(payment_hash.0), route_params, Retry::Attempts(1)).unwrap();
+ pass_failed_attempt_with_retry_along_path!(channel_id_2, true);
+
+ // We retry payments in `process_pending_htlc_forwards`. Since our channel closed, we should
+ // fail to find a route.
nodes[0].node.process_pending_htlc_forwards();
let mut msg_events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(msg_events.len(), 0);
- nodes[0].node.abandon_payment(PaymentId(payment_hash.0));
let mut events = nodes[0].node.get_and_clear_pending_events();
assert_eq!(events.len(), 1);
match events[0] {
// by adding the `PaymentFailed` event.
//
// Because we now retry payments as a batch, we simply return a single-path route in the
- // second, batched, request, have that fail, then complete the payment via `abandon_payment`.
+ // second, batched, request, have that fail, ensure the payment was abandoned.
let mut events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 4);
+ assert_eq!(events.len(), 3);
match events[0] {
Event::PaymentPathFailed { payment_hash: ev_payment_hash, payment_failed_permanently, .. } => {
assert_eq!(payment_hash, ev_payment_hash);
},
_ => panic!("Unexpected event"),
}
- match events[3] {
- Event::PendingHTLCsForwardable { .. } => {},
- _ => panic!("Unexpected event"),
- }
nodes[0].node.process_pending_htlc_forwards();
let retry_htlc_updates = SendEvent::from_node(&nodes[0]);
commitment_signed_dance!(nodes[0], nodes[1], &bs_fail_update.commitment_signed, false, true);
let mut events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
+ assert_eq!(events.len(), 2);
match events[0] {
Event::PaymentPathFailed { payment_hash: ev_payment_hash, payment_failed_permanently, .. } => {
assert_eq!(payment_hash, ev_payment_hash);
},
_ => panic!("Unexpected event"),
}
- nodes[0].node.abandon_payment(PaymentId(payment_hash.0));
- events = nodes[0].node.get_and_clear_pending_events();
- assert_eq!(events.len(), 1);
- match events[0] {
+ match events[1] {
Event::PaymentFailed { payment_hash: ref ev_payment_hash, payment_id: ref ev_payment_id } => {
assert_eq!(payment_hash, *ev_payment_hash);
assert_eq!(PaymentId(payment_hash.0), *ev_payment_id);
expect_pending_htlcs_forwardable!(nodes[2]);
expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat);
}
+
+#[test]
+#[cfg(feature = "std")]
+fn test_threaded_payment_retries() {
+ // In the first version of the in-`ChannelManager` payment retries, retries weren't limited to
+ // a single thread and would happily let multiple threads run retries at the same time. Because
+ // retries are done by first calculating the amount we need to retry, then dropping the
+ // relevant lock, then actually sending, we would happily let multiple threads retry the same
+ // amount at the same time, overpaying our original HTLC!
+ let chanmon_cfgs = create_chanmon_cfgs(4);
+ let node_cfgs = create_node_cfgs(4, &chanmon_cfgs);
+ let node_chanmgrs = create_node_chanmgrs(4, &node_cfgs, &[None, None, None, None]);
+ let nodes = create_network(4, &node_cfgs, &node_chanmgrs);
+
+ // There is one mitigating guardrail when retrying payments - we can never over-pay by more
+ // than 10% of the original value. Thus, we want all our retries to be below that. In order to
+ // keep things simple, we route one HTLC for 0.1% of the payment over channel 1 and the rest
+ // out over channel 3+4. This will let us ignore 99% of the payment value and deal with only
+ // our channel.
+ let chan_1_scid = create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 10_000_000, 0).0.contents.short_channel_id;
+ create_announced_chan_between_nodes_with_value(&nodes, 1, 3, 10_000_000, 0);
+ let chan_3_scid = create_announced_chan_between_nodes_with_value(&nodes, 0, 2, 10_000_000, 0).0.contents.short_channel_id;
+ let chan_4_scid = create_announced_chan_between_nodes_with_value(&nodes, 2, 3, 10_000_000, 0).0.contents.short_channel_id;
+
+ let amt_msat = 100_000_000;
+ let (_, payment_hash, _, payment_secret) = get_route_and_payment_hash!(&nodes[0], nodes[2], amt_msat);
+ #[cfg(feature = "std")]
+ let payment_expiry_secs = SystemTime::UNIX_EPOCH.elapsed().unwrap().as_secs() + 60 * 60;
+ #[cfg(not(feature = "std"))]
+ let payment_expiry_secs = 60 * 60;
+ let mut invoice_features = InvoiceFeatures::empty();
+ invoice_features.set_variable_length_onion_required();
+ invoice_features.set_payment_secret_required();
+ invoice_features.set_basic_mpp_optional();
+ let payment_params = PaymentParameters::from_node_id(nodes[1].node.get_our_node_id(), TEST_FINAL_CLTV)
+ .with_expiry_time(payment_expiry_secs as u64)
+ .with_features(invoice_features);
+ let mut route_params = RouteParameters {
+ payment_params,
+ final_value_msat: amt_msat,
+ final_cltv_expiry_delta: TEST_FINAL_CLTV,
+ };
+
+ let mut route = Route {
+ paths: vec![
+ vec![RouteHop {
+ pubkey: nodes[1].node.get_our_node_id(),
+ node_features: nodes[1].node.node_features(),
+ short_channel_id: chan_1_scid,
+ channel_features: nodes[1].node.channel_features(),
+ fee_msat: 0,
+ cltv_expiry_delta: 100,
+ }, RouteHop {
+ pubkey: nodes[3].node.get_our_node_id(),
+ node_features: nodes[2].node.node_features(),
+ short_channel_id: 42, // Set a random SCID which nodes[1] will fail as unknown
+ channel_features: nodes[2].node.channel_features(),
+ fee_msat: amt_msat / 1000,
+ cltv_expiry_delta: 100,
+ }],
+ vec![RouteHop {
+ pubkey: nodes[2].node.get_our_node_id(),
+ node_features: nodes[2].node.node_features(),
+ short_channel_id: chan_3_scid,
+ channel_features: nodes[2].node.channel_features(),
+ fee_msat: 100_000,
+ cltv_expiry_delta: 100,
+ }, RouteHop {
+ pubkey: nodes[3].node.get_our_node_id(),
+ node_features: nodes[3].node.node_features(),
+ short_channel_id: chan_4_scid,
+ channel_features: nodes[3].node.channel_features(),
+ fee_msat: amt_msat - amt_msat / 1000,
+ cltv_expiry_delta: 100,
+ }]
+ ],
+ payment_params: Some(PaymentParameters::from_node_id(nodes[2].node.get_our_node_id(), TEST_FINAL_CLTV)),
+ };
+ nodes[0].router.expect_find_route(route_params.clone(), Ok(route.clone()));
+
+ nodes[0].node.send_payment_with_retry(payment_hash, &Some(payment_secret), PaymentId(payment_hash.0), route_params.clone(), Retry::Attempts(0xdeadbeef)).unwrap();
+ check_added_monitors!(nodes[0], 2);
+ let mut send_msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+ assert_eq!(send_msg_events.len(), 2);
+ send_msg_events.retain(|msg|
+ if let MessageSendEvent::UpdateHTLCs { node_id, .. } = msg {
+ // Drop the commitment update for nodes[2], we can just let that one sit pending
+ // forever.
+ *node_id == nodes[1].node.get_our_node_id()
+ } else { panic!(); }
+ );
+
+ // from here on out, the retry `RouteParameters` amount will be amt/1000
+ route_params.final_value_msat /= 1000;
+ route.paths.pop();
+
+ let end_time = Instant::now() + Duration::from_secs(1);
+ macro_rules! thread_body { () => { {
+ // We really want std::thread::scope, but its not stable until 1.63. Until then, we get unsafe.
+ let node_ref = NodePtr::from_node(&nodes[0]);
+ move || {
+ let node_a = unsafe { &*node_ref.0 };
+ while Instant::now() < end_time {
+ node_a.node.get_and_clear_pending_events(); // wipe the PendingHTLCsForwardable
+ // Ignore if we have any pending events, just always pretend we just got a
+ // PendingHTLCsForwardable
+ node_a.node.process_pending_htlc_forwards();
+ }
+ }
+ } } }
+ let mut threads = Vec::new();
+ for _ in 0..16 { threads.push(std::thread::spawn(thread_body!())); }
+
+ // Back in the main thread, poll pending messages and make sure that we never have more than
+ // one HTLC pending at a time. Note that the commitment_signed_dance will fail horribly if
+ // there are HTLC messages shoved in while its running. This allows us to test that we never
+ // generate an additional update_add_htlc until we've fully failed the first.
+ let mut previously_failed_channels = Vec::new();
+ loop {
+ assert_eq!(send_msg_events.len(), 1);
+ let send_event = SendEvent::from_event(send_msg_events.pop().unwrap());
+ assert_eq!(send_event.msgs.len(), 1);
+
+ nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
+ commitment_signed_dance!(nodes[1], nodes[0], send_event.commitment_msg, false, true);
+
+ // Note that we only push one route into `expect_find_route` at a time, because that's all
+ // the retries (should) need. If the bug is reintroduced "real" routes may be selected, but
+ // we should still ultimately fail for the same reason - because we're trying to send too
+ // many HTLCs at once.
+ let mut new_route_params = route_params.clone();
+ previously_failed_channels.push(route.paths[0][1].short_channel_id);
+ new_route_params.payment_params.previously_failed_channels = previously_failed_channels.clone();
+ route.paths[0][1].short_channel_id += 1;
+ nodes[0].router.expect_find_route(new_route_params, Ok(route.clone()));
+
+ let bs_fail_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+ nodes[0].node.handle_update_fail_htlc(&nodes[1].node.get_our_node_id(), &bs_fail_updates.update_fail_htlcs[0]);
+ // The "normal" commitment_signed_dance delivers the final RAA and then calls
+ // `check_added_monitors` to ensure only the one RAA-generated monitor update was created.
+ // This races with our other threads which may generate an add-HTLCs commitment update via
+ // `process_pending_htlc_forwards`. Instead, we defer the monitor update check until after
+ // *we've* called `process_pending_htlc_forwards` when its guaranteed to have two updates.
+ let last_raa = commitment_signed_dance!(nodes[0], nodes[1], bs_fail_updates.commitment_signed, false, true, false, true);
+ nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &last_raa);
+
+ let cur_time = Instant::now();
+ if cur_time > end_time {
+ for thread in threads.drain(..) { thread.join().unwrap(); }
+ }
+
+ // Make sure we have some events to handle when we go around...
+ nodes[0].node.get_and_clear_pending_events(); // wipe the PendingHTLCsForwardable
+ nodes[0].node.process_pending_htlc_forwards();
+ send_msg_events = nodes[0].node.get_and_clear_pending_msg_events();
+ check_added_monitors!(nodes[0], 2);
+
+ if cur_time > end_time {
+ break;
+ }
+ }
+}
use bitcoin::hashes::sha256::HashEngine as Sha256Engine;
use bitcoin::hashes::{HashEngine, Hash};
-/// Handler for BOLT1-compliant messages.
+/// A handler provided to [`PeerManager`] for reading and handling custom messages.
+///
+/// [BOLT 1] specifies a custom message type range for use with experimental or application-specific
+/// messages. `CustomMessageHandler` allows for user-defined handling of such types. See the
+/// [`lightning_custom_message`] crate for tools useful in composing more than one custom handler.
+///
+/// [BOLT 1]: https://github.com/lightning/bolts/blob/master/01-messaging.md
+/// [`lightning_custom_message`]: https://docs.rs/lightning_custom_message/latest/lightning_custom_message
pub trait CustomMessageHandler: wire::CustomMessageReader {
- /// Called with the message type that was received and the buffer to be read.
- /// Can return a `MessageHandlingError` if the message could not be handled.
+ /// Handles the given message sent from `sender_node_id`, possibly producing messages for
+ /// [`CustomMessageHandler::get_and_clear_pending_msg`] to return and thus for [`PeerManager`]
+ /// to send.
fn handle_custom_message(&self, msg: Self::CustomMessage, sender_node_id: &PublicKey) -> Result<(), LightningError>;
- /// Gets the list of pending messages which were generated by the custom message
- /// handler, clearing the list in the process. The first tuple element must
- /// correspond to the intended recipients node ids. If no connection to one of the
- /// specified node does not exist, the message is simply not sent to it.
+ /// Returns the list of pending messages that were generated by the handler, clearing the list
+ /// in the process. Each message is paired with the node id of the intended recipient. If no
+ /// connection to the node exists, then the message is simply not sent.
fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)>;
}
/// # extern crate bitcoin;
/// # use bitcoin::hashes::_export::_core::time::Duration;
/// # use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
-/// # use lightning::chain::keysinterface::{InMemorySigner, KeysManager};
-/// # use lightning::ln::msgs::DecodeError;
+/// # use lightning::chain::keysinterface::KeysManager;
/// # use lightning::ln::peer_handler::IgnoringMessageHandler;
/// # use lightning::onion_message::{BlindedPath, CustomOnionMessageContents, Destination, OnionMessageContents, OnionMessenger};
/// # use lightning::util::logger::{Logger, Record};
/// # use lightning::util::ser::{Writeable, Writer};
/// # use lightning::io;
/// # use std::sync::Arc;
-/// # struct FakeLogger {};
+/// # struct FakeLogger;
/// # impl Logger for FakeLogger {
/// # fn log(&self, record: &Record) { unimplemented!() }
/// # }
/// # let your_custom_message_handler = IgnoringMessageHandler {};
/// // Create the onion messenger. This must use the same `keys_manager` as is passed to your
/// // ChannelManager.
-/// let onion_messenger = OnionMessenger::new(&keys_manager, &keys_manager, logger, your_custom_message_handler);
+/// let onion_messenger = OnionMessenger::new(&keys_manager, &keys_manager, logger, &your_custom_message_handler);
///
/// # struct YourCustomMessage {}
/// impl Writeable for YourCustomMessage {
}
fn get_next_channel_announcement(&self, starting_point: u64) -> Option<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> {
- let channels = self.network_graph.channels.read().unwrap();
+ let mut channels = self.network_graph.channels.write().unwrap();
for (_, ref chan) in channels.range(starting_point..) {
if chan.announcement_message.is_some() {
let chan_announcement = chan.announcement_message.clone().unwrap();
}
fn get_next_node_announcement(&self, starting_point: Option<&NodeId>) -> Option<NodeAnnouncement> {
- let nodes = self.network_graph.nodes.read().unwrap();
+ let mut nodes = self.network_graph.nodes.write().unwrap();
let iter = if let Some(node_id) = starting_point {
nodes.range((Bound::Excluded(node_id), Bound::Unbounded))
} else {
// (has at least one update). A peer may still want to know the channel
// exists even if its not yet routable.
let mut batches: Vec<Vec<u64>> = vec![Vec::with_capacity(MAX_SCIDS_PER_REPLY)];
- let channels = self.network_graph.channels.read().unwrap();
+ let mut channels = self.network_graph.channels.write().unwrap();
for (_, ref chan) in channels.range(inclusive_start_scid.unwrap()..exclusive_end_scid.unwrap()) {
if let Some(chan_announcement) = &chan.announcement_message {
// Construct a new batch if last one is full
/// Parameters needed to find a [`Route`].
///
/// Passed to [`find_route`] and [`build_route_from_hops`], but also provided in
-/// [`Event::PaymentPathFailed`] for retrying a failed payment path.
+/// [`Event::PaymentPathFailed`].
///
/// [`Event::PaymentPathFailed`]: crate::util::events::Event::PaymentPathFailed
#[derive(Clone, Debug, PartialEq, Eq)]
use crate::prelude::HashMap;
+use super::{LockTestExt, LockHeldState};
+
#[cfg(feature = "backtrace")]
use {crate::prelude::hash_map, backtrace::Backtrace, std::sync::Once};
fn pre_lock(this: &Arc<LockMetadata>) { Self::_pre_lock(this, false); }
fn pre_read_lock(this: &Arc<LockMetadata>) -> bool { Self::_pre_lock(this, true) }
+ fn held_by_thread(this: &Arc<LockMetadata>) -> LockHeldState {
+ let mut res = LockHeldState::NotHeldByThread;
+ LOCKS_HELD.with(|held| {
+ for (locked_idx, _locked) in held.borrow().iter() {
+ if *locked_idx == this.lock_idx {
+ res = LockHeldState::HeldByThread;
+ }
+ }
+ });
+ res
+ }
+
fn try_locked(this: &Arc<LockMetadata>) {
LOCKS_HELD.with(|held| {
// Since a try-lock will simply fail if the lock is held already, we do not
}
}
+impl <T> LockTestExt for Mutex<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState {
+ LockMetadata::held_by_thread(&self.deps)
+ }
+}
+
pub struct RwLock<T: Sized> {
inner: StdRwLock<T>,
deps: Arc<LockMetadata>,
}
}
+impl <T> LockTestExt for RwLock<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState {
+ LockMetadata::held_by_thread(&self.deps)
+ }
+}
+
pub type FairRwLock<T> = RwLock<T>;
--- /dev/null
+use std::sync::{LockResult, RwLock, RwLockReadGuard, RwLockWriteGuard, TryLockResult};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use super::{LockHeldState, LockTestExt};
+
+/// Rust libstd's RwLock does not provide any fairness guarantees (and, in fact, when used on
+/// Linux with pthreads under the hood, readers trivially and completely starve writers).
+/// Because we often hold read locks while doing message processing in multiple threads which
+/// can use significant CPU time, with write locks being time-sensitive but relatively small in
+/// CPU time, we can end up with starvation completely blocking incoming connections or pings,
+/// especially during initial graph sync.
+///
+/// Thus, we need to block readers when a writer is pending, which we do with a trivial RwLock
+/// wrapper here. Its not particularly optimized, but provides some reasonable fairness by
+/// blocking readers (by taking the write lock) if there are writers pending when we go to take
+/// a read lock.
+pub struct FairRwLock<T> {
+ lock: RwLock<T>,
+ waiting_writers: AtomicUsize,
+}
+
+impl<T> FairRwLock<T> {
+ pub fn new(t: T) -> Self {
+ Self { lock: RwLock::new(t), waiting_writers: AtomicUsize::new(0) }
+ }
+
+ // Note that all atomic accesses are relaxed, as we do not rely on the atomics here for any
+ // ordering at all, instead relying on the underlying RwLock to provide ordering of unrelated
+ // memory.
+ pub fn write(&self) -> LockResult<RwLockWriteGuard<T>> {
+ self.waiting_writers.fetch_add(1, Ordering::Relaxed);
+ let res = self.lock.write();
+ self.waiting_writers.fetch_sub(1, Ordering::Relaxed);
+ res
+ }
+
+ pub fn read(&self) -> LockResult<RwLockReadGuard<T>> {
+ if self.waiting_writers.load(Ordering::Relaxed) != 0 {
+ let _write_queue_lock = self.lock.write();
+ }
+ // Note that we don't consider ensuring that an underlying RwLock allowing writers to
+ // starve readers doesn't exhibit the same behavior here. I'm not aware of any
+ // libstd-backing RwLock which exhibits this behavior, and as documented in the
+ // struct-level documentation, it shouldn't pose a significant issue for our current
+ // codebase.
+ self.lock.read()
+ }
+
+ pub fn try_write(&self) -> TryLockResult<RwLockWriteGuard<'_, T>> {
+ self.lock.try_write()
+ }
+}
+
+impl<T> LockTestExt for FairRwLock<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState {
+ // fairrwlock is only built in non-test modes, so we should never support tests.
+ LockHeldState::Unsupported
+ }
+}
+#[allow(dead_code)] // Depending on the compilation flags some variants are never used
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum LockHeldState {
+ HeldByThread,
+ NotHeldByThread,
+ #[cfg(any(feature = "_bench_unstable", not(test)))]
+ Unsupported,
+}
+
+pub(crate) trait LockTestExt {
+ fn held_by_thread(&self) -> LockHeldState;
+}
+
#[cfg(all(feature = "std", not(feature = "_bench_unstable"), test))]
mod debug_sync;
#[cfg(all(feature = "std", not(feature = "_bench_unstable"), test))]
mod test_lockorder_checks;
#[cfg(all(feature = "std", any(feature = "_bench_unstable", not(test))))]
-pub use ::std::sync::{Arc, Mutex, Condvar, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
+pub(crate) mod fairrwlock;
+#[cfg(all(feature = "std", any(feature = "_bench_unstable", not(test))))]
+pub use {std::sync::{Arc, Mutex, Condvar, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard}, fairrwlock::FairRwLock};
+
#[cfg(all(feature = "std", any(feature = "_bench_unstable", not(test))))]
-pub use crate::util::fairrwlock::FairRwLock;
+mod ext_impl {
+ use super::*;
+ impl<T> LockTestExt for Mutex<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState { LockHeldState::Unsupported }
+ }
+ impl<T> LockTestExt for RwLock<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState { LockHeldState::Unsupported }
+ }
+}
#[cfg(not(feature = "std"))]
mod nostd_sync;
use core::ops::{Deref, DerefMut};
use core::time::Duration;
use core::cell::{RefCell, Ref, RefMut};
+use super::{LockTestExt, LockHeldState};
pub type LockResult<Guard> = Result<Guard, ()>;
}
}
+impl<T> LockTestExt for Mutex<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState {
+ if self.lock().is_err() { return LockHeldState::HeldByThread; }
+ else { return LockHeldState::NotHeldByThread; }
+ }
+}
+
pub struct RwLock<T: ?Sized> {
inner: RefCell<T>
}
}
}
+impl<T> LockTestExt for RwLock<T> {
+ #[inline]
+ fn held_by_thread(&self) -> LockHeldState {
+ if self.write().is_err() { return LockHeldState::HeldByThread; }
+ else { return LockHeldState::NotHeldByThread; }
+ }
+}
+
pub type FairRwLock<T> = RwLock<T>;
use crate::sync::debug_sync::{RwLock, Mutex};
+use super::{LockHeldState, LockTestExt};
+
+use std::sync::Arc;
+use std::thread;
+
#[test]
#[should_panic]
#[cfg(not(feature = "backtrace"))]
let _a = a.write().unwrap();
}
}
+
+#[test]
+fn test_thread_locked_state() {
+ let mtx = Arc::new(Mutex::new(()));
+ let mtx_ref = Arc::clone(&mtx);
+ assert_eq!(mtx.held_by_thread(), LockHeldState::NotHeldByThread);
+
+ let lck = mtx.lock().unwrap();
+ assert_eq!(mtx.held_by_thread(), LockHeldState::HeldByThread);
+
+ let thrd = std::thread::spawn(move || {
+ assert_eq!(mtx_ref.held_by_thread(), LockHeldState::NotHeldByThread);
+ });
+ thrd.join().unwrap();
+ assert_eq!(mtx.held_by_thread(), LockHeldState::HeldByThread);
+
+ std::mem::drop(lck);
+ assert_eq!(mtx.held_by_thread(), LockHeldState::NotHeldByThread);
+}
/// Note for MPP payments: in rare cases, this event may be preceded by a `PaymentPathFailed`
/// event. In this situation, you SHOULD treat this payment as having succeeded.
PaymentSent {
- /// The id returned by [`ChannelManager::send_payment`] and used with
- /// [`ChannelManager::retry_payment`].
+ /// The id returned by [`ChannelManager::send_payment`].
///
/// [`ChannelManager::send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
payment_id: Option<PaymentId>,
/// The preimage to the hash given to ChannelManager::send_payment.
/// Note that this serves as a payment receipt, if you wish to have such a thing, you must
/// provide failure information for each MPP part in the payment.
///
/// This event is provided once there are no further pending HTLCs for the payment and the
- /// payment is no longer retryable due to [`ChannelManager::abandon_payment`] having been
- /// called for the corresponding payment.
+ /// payment is no longer retryable, due either to the [`Retry`] provided or
+ /// [`ChannelManager::abandon_payment`] having been called for the corresponding payment.
///
+ /// [`Retry`]: crate::ln::channelmanager::Retry
/// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
PaymentFailed {
/// The id returned by [`ChannelManager::send_payment`] and used with
- /// [`ChannelManager::retry_payment`] and [`ChannelManager::abandon_payment`].
+ /// [`ChannelManager::abandon_payment`].
///
/// [`ChannelManager::send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
/// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
payment_id: PaymentId,
/// The hash that was given to [`ChannelManager::send_payment`].
/// Always generated after [`Event::PaymentSent`] and thus useful for scoring channels. See
/// [`Event::PaymentSent`] for obtaining the payment preimage.
PaymentPathSuccessful {
- /// The id returned by [`ChannelManager::send_payment`] and used with
- /// [`ChannelManager::retry_payment`].
+ /// The id returned by [`ChannelManager::send_payment`].
///
/// [`ChannelManager::send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
payment_id: PaymentId,
/// The hash that was given to [`ChannelManager::send_payment`].
///
/// May contain a closed channel if the HTLC sent along the path was fulfilled on chain.
path: Vec<RouteHop>,
},
- /// Indicates an outbound HTLC we sent failed. Probably some intermediary node dropped
- /// something. You may wish to retry with a different route.
- ///
- /// If you have given up retrying this payment and wish to fail it, you MUST call
- /// [`ChannelManager::abandon_payment`] at least once for a given [`PaymentId`] or memory
- /// related to payment tracking will leak.
+ /// Indicates an outbound HTLC we sent failed, likely due to an intermediary node being unable to
+ /// handle the HTLC.
///
/// Note that this does *not* indicate that all paths for an MPP payment have failed, see
/// [`Event::PaymentFailed`] and [`all_paths_failed`].
///
+ /// See [`ChannelManager::abandon_payment`] for giving up on this payment before its retries have
+ /// been exhausted.
+ ///
/// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
/// [`all_paths_failed`]: Self::PaymentPathFailed::all_paths_failed
PaymentPathFailed {
/// The id returned by [`ChannelManager::send_payment`] and used with
- /// [`ChannelManager::retry_payment`] and [`ChannelManager::abandon_payment`].
+ /// [`ChannelManager::abandon_payment`].
///
/// [`ChannelManager::send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
- /// [`ChannelManager::retry_payment`]: crate::ln::channelmanager::ChannelManager::retry_payment
/// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
payment_id: Option<PaymentId>,
/// The hash that was given to [`ChannelManager::send_payment`].
/// [`ChannelManager::send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
payment_hash: PaymentHash,
/// Indicates the payment was rejected for some reason by the recipient. This implies that
- /// the payment has failed, not just the route in question. If this is not set, you may
- /// retry the payment via a different route.
+ /// the payment has failed, not just the route in question. If this is not set, the payment may
+ /// be retried via a different route.
payment_failed_permanently: bool,
/// Any failure information conveyed via the Onion return packet by a node along the failed
/// payment route.
/// For both single-path and multi-path payments, this is set if all paths of the payment have
/// failed. This will be set to false if (1) this is an MPP payment and (2) other parts of the
/// larger MPP payment were still in flight when this event was generated.
- ///
- /// Note that if you are retrying individual MPP parts, using this value to determine if a
- /// payment has fully failed is race-y. Because multiple failures can happen prior to events
- /// being processed, you may retry in response to a first failure, with a second failure
- /// (with `all_paths_failed` set) still pending. Then, when the second failure is processed
- /// you will see `all_paths_failed` set even though the retry of the first failure still
- /// has an associated in-flight HTLC. See (1) for an example of such a failure.
- ///
- /// If you wish to retry individual MPP parts and learn when a payment has failed, you must
- /// call [`ChannelManager::abandon_payment`] and wait for a [`Event::PaymentFailed`] event.
- ///
- /// (1) <https://github.com/lightningdevkit/rust-lightning/issues/1164>
- ///
- /// [`ChannelManager::abandon_payment`]: crate::ln::channelmanager::ChannelManager::abandon_payment
all_paths_failed: bool,
/// The payment path that failed.
path: Vec<RouteHop>,
/// If this is `Some`, then the corresponding channel should be avoided when the payment is
/// retried. May be `None` for older [`Event`] serializations.
short_channel_id: Option<u64>,
- /// Parameters needed to compute a new [`Route`] when retrying the failed payment path.
- ///
- /// See [`find_route`] for details.
+ /// Parameters used by LDK to compute a new [`Route`] when retrying the failed payment path.
///
/// [`Route`]: crate::routing::router::Route
- /// [`find_route`]: crate::routing::router::find_route
retry: Option<RouteParameters>,
#[cfg(test)]
error_code: Option<u16>,
+++ /dev/null
-use std::sync::{LockResult, RwLock, RwLockReadGuard, RwLockWriteGuard, TryLockResult};
-use std::sync::atomic::{AtomicUsize, Ordering};
-
-/// Rust libstd's RwLock does not provide any fairness guarantees (and, in fact, when used on
-/// Linux with pthreads under the hood, readers trivially and completely starve writers).
-/// Because we often hold read locks while doing message processing in multiple threads which
-/// can use significant CPU time, with write locks being time-sensitive but relatively small in
-/// CPU time, we can end up with starvation completely blocking incoming connections or pings,
-/// especially during initial graph sync.
-///
-/// Thus, we need to block readers when a writer is pending, which we do with a trivial RwLock
-/// wrapper here. Its not particularly optimized, but provides some reasonable fairness by
-/// blocking readers (by taking the write lock) if there are writers pending when we go to take
-/// a read lock.
-pub struct FairRwLock<T> {
- lock: RwLock<T>,
- waiting_writers: AtomicUsize,
-}
-
-impl<T> FairRwLock<T> {
- pub fn new(t: T) -> Self {
- Self { lock: RwLock::new(t), waiting_writers: AtomicUsize::new(0) }
- }
-
- // Note that all atomic accesses are relaxed, as we do not rely on the atomics here for any
- // ordering at all, instead relying on the underlying RwLock to provide ordering of unrelated
- // memory.
- pub fn write(&self) -> LockResult<RwLockWriteGuard<T>> {
- self.waiting_writers.fetch_add(1, Ordering::Relaxed);
- let res = self.lock.write();
- self.waiting_writers.fetch_sub(1, Ordering::Relaxed);
- res
- }
-
- pub fn read(&self) -> LockResult<RwLockReadGuard<T>> {
- if self.waiting_writers.load(Ordering::Relaxed) != 0 {
- let _write_queue_lock = self.lock.write();
- }
- // Note that we don't consider ensuring that an underlying RwLock allowing writers to
- // starve readers doesn't exhibit the same behavior here. I'm not aware of any
- // libstd-backing RwLock which exhibits this behavior, and as documented in the
- // struct-level documentation, it shouldn't pose a significant issue for our current
- // codebase.
- self.lock.read()
- }
-
- pub fn try_write(&self) -> TryLockResult<RwLockWriteGuard<'_, T>> {
- self.lock.try_write()
- }
-}
//! This module has a map which can be iterated in a deterministic order. See the [`IndexedMap`].
use crate::prelude::{HashMap, hash_map};
-use alloc::collections::{BTreeSet, btree_set};
+use alloc::vec::Vec;
+use alloc::slice::Iter;
use core::hash::Hash;
use core::cmp::Ord;
-use core::ops::RangeBounds;
+use core::ops::{Bound, RangeBounds};
/// A map which can be iterated in a deterministic order.
///
/// keys in the order defined by [`Ord`].
///
/// [`BTreeMap`]: alloc::collections::BTreeMap
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, Eq)]
pub struct IndexedMap<K: Hash + Ord, V> {
map: HashMap<K, V>,
- // TODO: Explore swapping this for a sorted vec (that is only sorted on first range() call)
- keys: BTreeSet<K>,
+ keys: Vec<K>,
}
impl<K: Clone + Hash + Ord, V> IndexedMap<K, V> {
pub fn new() -> Self {
Self {
map: HashMap::new(),
- keys: BTreeSet::new(),
+ keys: Vec::new(),
}
}
pub fn remove(&mut self, key: &K) -> Option<V> {
let ret = self.map.remove(key);
if let Some(_) = ret {
- assert!(self.keys.remove(key), "map and keys must be consistent");
+ let idx = self.keys.iter().position(|k| k == key).expect("map and keys must be consistent");
+ self.keys.remove(idx);
}
ret
}
pub fn insert(&mut self, key: K, value: V) -> Option<V> {
let ret = self.map.insert(key.clone(), value);
if ret.is_none() {
- assert!(self.keys.insert(key), "map and keys must be consistent");
+ self.keys.push(key);
}
ret
}
}
/// Returns an iterator which iterates over the `key`/`value` pairs in a given range.
- pub fn range<R: RangeBounds<K>>(&self, range: R) -> Range<K, V> {
+ pub fn range<R: RangeBounds<K>>(&mut self, range: R) -> Range<K, V> {
+ self.keys.sort_unstable();
+ let start = match range.start_bound() {
+ Bound::Unbounded => 0,
+ Bound::Included(key) => self.keys.binary_search(key).unwrap_or_else(|index| index),
+ Bound::Excluded(key) => self.keys.binary_search(key).and_then(|index| Ok(index + 1)).unwrap_or_else(|index| index),
+ };
+ let end = match range.end_bound() {
+ Bound::Unbounded => self.keys.len(),
+ Bound::Included(key) => self.keys.binary_search(key).and_then(|index| Ok(index + 1)).unwrap_or_else(|index| index),
+ Bound::Excluded(key) => self.keys.binary_search(key).unwrap_or_else(|index| index),
+ };
+
Range {
- inner_range: self.keys.range(range),
+ inner_range: self.keys[start..end].iter(),
map: &self.map,
}
}
}
}
+impl<K: Hash + Ord + PartialEq, V: PartialEq> PartialEq for IndexedMap<K, V> {
+ fn eq(&self, other: &Self) -> bool {
+ self.map == other.map
+ }
+}
+
/// An iterator over a range of values in an [`IndexedMap`]
pub struct Range<'a, K: Hash + Ord, V> {
- inner_range: btree_set::Range<'a, K>,
+ inner_range: Iter<'a, K>,
map: &'a HashMap<K, V>,
}
impl<'a, K: Hash + Ord, V: 'a> Iterator for Range<'a, K, V> {
#[cfg(not(feature = "hashbrown"))]
underlying_entry: hash_map::VacantEntry<'a, K, V>,
key: K,
- keys: &'a mut BTreeSet<K>,
+ keys: &'a mut Vec<K>,
}
/// An [`Entry`] for an existing key-value pair
underlying_entry: hash_map::OccupiedEntry<'a, K, V, hash_map::DefaultHashBuilder>,
#[cfg(not(feature = "hashbrown"))]
underlying_entry: hash_map::OccupiedEntry<'a, K, V>,
- keys: &'a mut BTreeSet<K>,
+ keys: &'a mut Vec<K>,
}
/// A mutable reference to a position in the map. This can be used to reference, add, or update the
impl<'a, K: Hash + Ord, V> VacantEntry<'a, K, V> {
/// Insert a value into the position described by this entry.
pub fn insert(self, value: V) -> &'a mut V {
- assert!(self.keys.insert(self.key), "map and keys must be consistent");
+ self.keys.push(self.key);
self.underlying_entry.insert(value)
}
}
/// Remove the value at the position described by this entry.
pub fn remove_entry(self) -> (K, V) {
let res = self.underlying_entry.remove_entry();
- assert!(self.keys.remove(&res.0), "map and keys must be consistent");
+ let idx = self.keys.iter().position(|k| k == &res.0).expect("map and keys must be consistent");
+ self.keys.remove(idx);
res
}
pub(crate) mod atomic_counter;
pub(crate) mod byte_utils;
pub(crate) mod chacha20;
-#[cfg(all(any(feature = "_bench_unstable", not(test)), feature = "std"))]
-pub(crate) mod fairrwlock;
#[cfg(fuzzing)]
pub mod zbase32;
#[cfg(not(fuzzing))]
}
}
-#[cfg(feature = "std")] // If we put this on the `if`, we get "attributes are not yet allowed on `if` expressions" on 1.41.1
impl<'a> Drop for TestRouter<'a> {
fn drop(&mut self) {
- if std::thread::panicking() {
- return;
+ #[cfg(feature = "std")] {
+ if std::thread::panicking() {
+ return;
+ }
}
assert!(self.next_routes.lock().unwrap().is_empty());
}
projects / rust-lightning / commitdiff