From cff971336f0de4df6892954b2f638df66cb75137 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 23 Jun 2026 16:34:35 -0500 Subject: [PATCH 1/6] Add a per-channel record store Introduce durable, per-channel state keyed by the channel's user-facing identifier and tracking where the channel sits in its lifecycle (unfunded, funded, or closed), persisted alongside the node's other stores and loaded at startup. Wiring it in now lets upcoming per-channel features land independently of one another rather than stacking on a single change. A record currently holds only the channel's identity; the per-feature state and the transitions between lifecycle states are added by the work that needs them. Co-Authored-By: Claude Opus 4.8 --- src/builder.rs | 33 ++++++++++-- src/channel/mod.rs | 10 ++++ src/channel/store.rs | 125 +++++++++++++++++++++++++++++++++++++++++++ src/io/mod.rs | 4 ++ src/lib.rs | 11 ++-- src/types.rs | 5 +- 6 files changed, 179 insertions(+), 9 deletions(-) create mode 100644 src/channel/mod.rs create mode 100644 src/channel/store.rs diff --git a/src/builder.rs b/src/builder.rs index 8b575cc3f..b1d9ec655 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -64,7 +64,9 @@ use crate::io::utils::{ }; use crate::io::vss_store::VssStoreBuilder; use crate::io::{ - self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + self, CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -77,9 +79,9 @@ use crate::peer_store::PeerStore; use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ - AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper, - GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore, - PeerManager, PendingPaymentStore, + AsyncPersister, ChainMonitor, ChannelManager, ChannelRecordStore, DynStore, DynStoreRef, + DynStoreWrapper, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, + PaymentStore, PeerManager, PendingPaymentStore, }; use crate::wallet::persist::KVStoreWalletPersister; use crate::wallet::Wallet; @@ -1397,7 +1399,7 @@ fn build_with_store_internal( let kv_store_ref = Arc::clone(&kv_store); let logger_ref = Arc::clone(&logger); - let (payment_store_res, node_metris_res, pending_payment_store_res) = + let (payment_store_res, node_metris_res, pending_payment_store_res, channel_record_store_res) = runtime.block_on(async move { tokio::join!( read_all_objects( @@ -1412,6 +1414,12 @@ fn build_with_store_internal( PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), ) ) }); @@ -1699,6 +1707,20 @@ fn build_with_store_internal( }, }; + let channel_record_store = match channel_record_store_res { + Ok(channel_records) => Arc::new(ChannelRecordStore::new( + channel_records, + CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read channel record data from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let wallet = Arc::new(Wallet::new( bdk_wallet, wallet_persister, @@ -2245,6 +2267,7 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + channel_record_store, lnurl_auth, is_running, node_metrics, diff --git a/src/channel/mod.rs b/src/channel/mod.rs new file mode 100644 index 000000000..95a9f5224 --- /dev/null +++ b/src/channel/mod.rs @@ -0,0 +1,10 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Per-channel state tracking. + +pub(crate) mod store; diff --git a/src/channel/store.rs b/src/channel/store.rs new file mode 100644 index 000000000..8619be009 --- /dev/null +++ b/src/channel/store.rs @@ -0,0 +1,125 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use bitcoin::secp256k1::PublicKey; +use lightning::impl_writeable_tlv_based_enum; +use lightning::ln::types::ChannelId; + +use crate::data_store::{StorableObject, StorableObjectId, StorableObjectUpdate}; +use crate::hex_utils; +use crate::types::UserChannelId; + +/// Persistent per-channel state tracked by LDK Node, keyed by [`UserChannelId`]. +/// +/// Tracks where a channel sits in its lifecycle. Each variant currently holds only the channel's +/// identity; the per-feature state and the transitions between lifecycle states are added by the +/// work that needs them. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum ChannelRecord { + /// A channel whose funding transaction has not yet been observed. + Unfunded { + user_channel_id: UserChannelId, + counterparty_node_id: PublicKey, + channel_id: ChannelId, + }, + /// A channel whose funding transaction exists. + Funded { + user_channel_id: UserChannelId, + counterparty_node_id: PublicKey, + channel_id: ChannelId, + }, + /// A channel that has been closed. + Closed { + user_channel_id: UserChannelId, + counterparty_node_id: PublicKey, + channel_id: ChannelId, + }, +} + +impl_writeable_tlv_based_enum!(ChannelRecord, + (0, Unfunded) => { + (0, user_channel_id, required), + (2, counterparty_node_id, required), + (4, channel_id, required), + }, + (2, Funded) => { + (0, user_channel_id, required), + (2, counterparty_node_id, required), + (4, channel_id, required), + }, + (4, Closed) => { + (0, user_channel_id, required), + (2, counterparty_node_id, required), + (4, channel_id, required), + }, +); + +impl StorableObjectId for UserChannelId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0.to_be_bytes()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ChannelRecordUpdate { + pub user_channel_id: UserChannelId, +} + +impl StorableObject for ChannelRecord { + type Id = UserChannelId; + type Update = ChannelRecordUpdate; + + fn id(&self) -> Self::Id { + match self { + ChannelRecord::Unfunded { user_channel_id, .. } + | ChannelRecord::Funded { user_channel_id, .. } + | ChannelRecord::Closed { user_channel_id, .. } => *user_channel_id, + } + } + + fn update(&mut self, _update: Self::Update) -> bool { + // Records currently carry only the channel's identity, so there is nothing to mutate in + // place. Lifecycle transitions are performed by replacing the record once a consumer of + // this store exists. + false + } + + fn to_update(&self) -> Self::Update { + Self::Update { user_channel_id: self.id() } + } +} + +impl StorableObjectUpdate for ChannelRecordUpdate { + fn id(&self) -> ::Id { + self.user_channel_id + } +} + +#[cfg(test)] +mod tests { + use lightning::util::ser::{Readable, Writeable}; + + use super::*; + + #[test] + fn channel_record_is_serializable() { + let user_channel_id = UserChannelId(42); + let counterparty_node_id = bitcoin::secp256k1::PublicKey::from_slice(&[2u8; 33]).unwrap(); + let channel_id = ChannelId([3u8; 32]); + + for record in [ + ChannelRecord::Unfunded { user_channel_id, counterparty_node_id, channel_id }, + ChannelRecord::Funded { user_channel_id, counterparty_node_id, channel_id }, + ChannelRecord::Closed { user_channel_id, counterparty_node_id, channel_id }, + ] { + let encoded = record.encode(); + let decoded = ChannelRecord::read(&mut &encoded[..]).unwrap(); + assert_eq!(record, decoded); + assert_eq!(decoded.id(), user_channel_id); + } + } +} diff --git a/src/io/mod.rs b/src/io/mod.rs index a01aa59a8..01c7f4492 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -84,3 +84,7 @@ pub(crate) const BDK_WALLET_INDEXER_KEY: &str = "indexer"; /// /// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice pub(crate) const STATIC_INVOICE_STORE_PRIMARY_NAMESPACE: &str = "static_invoices"; + +/// The per-channel records will be persisted under this prefix. +pub(crate) const CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE: &str = "channel_records"; +pub(crate) const CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/lib.rs b/src/lib.rs index c97e16fe6..e96afe754 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -83,6 +83,7 @@ mod balance; mod builder; mod chain; +mod channel; pub mod config; mod connection; mod data_store; @@ -175,9 +176,9 @@ use peer_store::{PeerInfo, PeerStore}; use runtime::Runtime; pub use tokio; use types::{ - Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph, - HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper, - Wallet, + Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, ChannelRecordStore, + DynStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, + Scorer, Sweeper, Wallet, }; pub use types::{ ChannelCounterparty, ChannelDetails, CustomTlvRecord, PeerDetails, ReserveType, UserChannelId, @@ -244,6 +245,10 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + // Foundational per-channel record store, loaded and persisted here so that upcoming + // per-channel features can build on it. Not yet read within this crate. + #[allow(dead_code)] + channel_record_store: Arc, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, diff --git a/src/types.rs b/src/types.rs index e24db4d25..6dcab55dc 100644 --- a/src/types.rs +++ b/src/types.rs @@ -42,6 +42,7 @@ use lightning_net_tokio::SocketDescriptor; use crate::chain::bitcoind::UtxoSourceClient; use crate::chain::ChainSource; +use crate::channel::store::ChannelRecord; use crate::config::{AnchorChannelsConfig, ChannelConfig}; use crate::data_store::DataStore; use crate::fee_estimator::OnchainFeeEstimator; @@ -376,7 +377,7 @@ pub(crate) type PaymentStore = DataStore>; /// A local, potentially user-provided, identifier of a channel. /// /// By default, this will be randomly generated for the user to ensure local uniqueness. -#[derive(Debug, Copy, Clone, PartialEq, Eq)] +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] pub struct UserChannelId(pub u128); impl Writeable for UserChannelId { @@ -756,3 +757,5 @@ impl From<&(u64, Vec)> for CustomTlvRecord { } pub(crate) type PendingPaymentStore = DataStore>; + +pub(crate) type ChannelRecordStore = DataStore>; From a1fea8a5fc0552f011ba2c841c931b32f96fbbde Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Mon, 20 Apr 2026 11:13:58 -0500 Subject: [PATCH 2/6] Add persistent closed channel history and `list_closed_channels()` Introduce `ClosedChannelDetails`, a new record type persisted to the KV store under the `"closed_channels"` namespace whenever a channel closes. Records are written in the `ChannelClosed` event handler and loaded back at startup in parallel with other stores via `tokio::join!`. Add `Node::list_closed_channels()` to expose the full history of closed channels across restarts. Track outbound channel direction via an in-memory `outbound_channel_ids` set seeded from `channel_manager.list_channels()` at startup and updated on `ChannelPending` events, since `ChannelClosed` does not carry that information directly. --- bindings/ldk_node.udl | 3 + src/builder.rs | 86 +++++++++----- src/channel/store.rs | 113 +++++++++--------- src/closed_channel.rs | 89 +++++++++++++++ src/event.rs | 196 +++++++++++++++++++++++++++++--- src/io/mod.rs | 4 + src/lib.rs | 19 +++- src/types.rs | 3 + tests/common/mod.rs | 17 +++ tests/integration_tests_rust.rs | 110 +++++++++++++++++- 10 files changed, 536 insertions(+), 104 deletions(-) create mode 100644 src/closed_channel.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index 7c0edc535..38cbb1f27 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -139,6 +139,7 @@ interface Node { sequence list_payments(); sequence list_peers(); sequence list_channels(); + sequence list_closed_channels(); NetworkGraph network_graph(); string sign_message([ByRef]sequence msg); boolean verify_signature([ByRef]sequence msg, [ByRef]string sig, [ByRef]PublicKey pkey); @@ -322,6 +323,8 @@ dictionary OutPoint { typedef dictionary ChannelDetails; +typedef dictionary ClosedChannelDetails; + typedef dictionary PeerDetails; [Remote] diff --git a/src/builder.rs b/src/builder.rs index b1d9ec655..65539a71f 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -65,8 +65,10 @@ use crate::io::utils::{ use crate::io::vss_store::VssStoreBuilder; use crate::io::{ self, CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE, - CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, + CLOSED_CHANNEL_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + CLOSED_CHANNEL_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -79,9 +81,9 @@ use crate::peer_store::PeerStore; use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ - AsyncPersister, ChainMonitor, ChannelManager, ChannelRecordStore, DynStore, DynStoreRef, - DynStoreWrapper, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, - PaymentStore, PeerManager, PendingPaymentStore, + AsyncPersister, ChainMonitor, ChannelManager, ChannelRecordStore, ClosedChannelStore, DynStore, + DynStoreRef, DynStoreWrapper, GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, + OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, }; use crate::wallet::persist::KVStoreWalletPersister; use crate::wallet::Wallet; @@ -1399,30 +1401,41 @@ fn build_with_store_internal( let kv_store_ref = Arc::clone(&kv_store); let logger_ref = Arc::clone(&logger); - let (payment_store_res, node_metris_res, pending_payment_store_res, channel_record_store_res) = - runtime.block_on(async move { - tokio::join!( - read_all_objects( - &*kv_store_ref, - PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ), - read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), - read_all_objects( - &*kv_store_ref, - PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ), - read_all_objects( - &*kv_store_ref, - CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE, - CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ) - ) - }); + let ( + payment_store_res, + node_metris_res, + pending_payment_store_res, + closed_channel_store_res, + channel_record_store_res, + ) = runtime.block_on(async move { + tokio::join!( + read_all_objects( + &*kv_store_ref, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), + read_all_objects( + &*kv_store_ref, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + CLOSED_CHANNEL_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + CLOSED_CHANNEL_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_all_objects( + &*kv_store_ref, + CHANNEL_RECORD_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_RECORD_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + ) + }); // Initialize the status fields. let node_metrics = match node_metris_res { @@ -1451,6 +1464,20 @@ fn build_with_store_internal( }, }; + let closed_channel_store = match closed_channel_store_res { + Ok(channels) => Arc::new(ClosedChannelStore::new( + channels, + CLOSED_CHANNEL_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + CLOSED_CHANNEL_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read closed channel data from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + let (chain_source, chain_tip_opt) = match chain_data_source_config { Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => { let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default()); @@ -2267,6 +2294,7 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + closed_channel_store, channel_record_store, lnurl_auth, is_running, diff --git a/src/channel/store.rs b/src/channel/store.rs index 8619be009..0d2969218 100644 --- a/src/channel/store.rs +++ b/src/channel/store.rs @@ -15,111 +15,116 @@ use crate::types::UserChannelId; /// Persistent per-channel state tracked by LDK Node, keyed by [`UserChannelId`]. /// -/// Tracks where a channel sits in its lifecycle. Each variant currently holds only the channel's -/// identity; the per-feature state and the transitions between lifecycle states are added by the -/// work that needs them. +/// Durably stores channel flags at `ChannelPending` time so they remain accessible when the +/// channel closes, even after a restart or a [`ReplayEvent`]. The `Funded` variant is designed +/// to be extended with a `pending_splice` field in a future PR to support splice retry across +/// restarts and peer disconnects. +/// +/// [`ReplayEvent`]: lightning::events::ReplayEvent #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum ChannelRecord { - /// A channel whose funding transaction has not yet been observed. - Unfunded { - user_channel_id: UserChannelId, - counterparty_node_id: PublicKey, - channel_id: ChannelId, - }, - /// A channel whose funding transaction exists. + /// State for a live channel whose funding transaction exists. Funded { user_channel_id: UserChannelId, + /// The node ID of the channel counterparty. counterparty_node_id: PublicKey, + /// The channel's ID at the time the `ChannelPending` event fired. channel_id: ChannelId, - }, - /// A channel that has been closed. - Closed { - user_channel_id: UserChannelId, - counterparty_node_id: PublicKey, - channel_id: ChannelId, + /// Whether we opened the channel (outbound) or the counterparty did (inbound). + is_outbound: bool, + /// Whether the channel was publicly announced. + is_announced: bool, }, } impl_writeable_tlv_based_enum!(ChannelRecord, - (0, Unfunded) => { - (0, user_channel_id, required), - (2, counterparty_node_id, required), - (4, channel_id, required), - }, - (2, Funded) => { - (0, user_channel_id, required), - (2, counterparty_node_id, required), - (4, channel_id, required), - }, - (4, Closed) => { + (0, Funded) => { (0, user_channel_id, required), (2, counterparty_node_id, required), (4, channel_id, required), + (6, is_outbound, required), + (8, is_announced, required), }, ); -impl StorableObjectId for UserChannelId { - fn encode_to_hex_str(&self) -> String { - hex_utils::to_string(&self.0.to_be_bytes()) - } -} - #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct ChannelRecordUpdate { pub user_channel_id: UserChannelId, } +impl StorableObjectUpdate for ChannelRecordUpdate { + fn id(&self) -> UserChannelId { + self.user_channel_id + } +} + impl StorableObject for ChannelRecord { type Id = UserChannelId; type Update = ChannelRecordUpdate; - fn id(&self) -> Self::Id { + fn id(&self) -> UserChannelId { match self { - ChannelRecord::Unfunded { user_channel_id, .. } - | ChannelRecord::Funded { user_channel_id, .. } - | ChannelRecord::Closed { user_channel_id, .. } => *user_channel_id, + ChannelRecord::Funded { user_channel_id, .. } => *user_channel_id, } } fn update(&mut self, _update: Self::Update) -> bool { - // Records currently carry only the channel's identity, so there is nothing to mutate in - // place. Lifecycle transitions are performed by replacing the record once a consumer of - // this store exists. + // ChannelRecord fields are immutable once written in this version. Returning false + // makes insert_or_update a no-op when the record already exists, ensuring idempotency + // on ChannelPending replay. false } fn to_update(&self) -> Self::Update { - Self::Update { user_channel_id: self.id() } + ChannelRecordUpdate { user_channel_id: self.id() } } } -impl StorableObjectUpdate for ChannelRecordUpdate { - fn id(&self) -> ::Id { - self.user_channel_id +impl StorableObjectId for UserChannelId { + fn encode_to_hex_str(&self) -> String { + hex_utils::to_string(&self.0.to_be_bytes()) } } #[cfg(test)] mod tests { + use lightning::ln::types::ChannelId; use lightning::util::ser::{Readable, Writeable}; use super::*; - #[test] - fn channel_record_is_serializable() { + fn make_record(is_outbound: bool, is_announced: bool) -> ChannelRecord { let user_channel_id = UserChannelId(42); - let counterparty_node_id = bitcoin::secp256k1::PublicKey::from_slice(&[2u8; 33]).unwrap(); + // A valid compressed public key: prefix 0x02 followed by 32 bytes. + let counterparty_node_id = PublicKey::from_slice(&[2u8; 33]).expect("valid pubkey"); let channel_id = ChannelId([3u8; 32]); + ChannelRecord::Funded { + user_channel_id, + counterparty_node_id, + channel_id, + is_outbound, + is_announced, + } + } - for record in [ - ChannelRecord::Unfunded { user_channel_id, counterparty_node_id, channel_id }, - ChannelRecord::Funded { user_channel_id, counterparty_node_id, channel_id }, - ChannelRecord::Closed { user_channel_id, counterparty_node_id, channel_id }, - ] { + #[test] + fn channel_record_roundtrips() { + for (is_outbound, is_announced) in + [(true, false), (false, true), (true, true), (false, false)] + { + let record = make_record(is_outbound, is_announced); let encoded = record.encode(); - let decoded = ChannelRecord::read(&mut &encoded[..]).unwrap(); + let decoded = ChannelRecord::read(&mut &encoded[..]).expect("decode succeeds"); assert_eq!(record, decoded); - assert_eq!(decoded.id(), user_channel_id); + assert_eq!(decoded.id(), UserChannelId(42)); + assert!(matches!( + decoded, + ChannelRecord::Funded { + is_outbound: dec_out, + is_announced: dec_ann, + .. + } if dec_out == is_outbound && dec_ann == is_announced + )); } } } diff --git a/src/closed_channel.rs b/src/closed_channel.rs new file mode 100644 index 000000000..84a48c75a --- /dev/null +++ b/src/closed_channel.rs @@ -0,0 +1,89 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use bitcoin::secp256k1::PublicKey; +use bitcoin::OutPoint; +use lightning::events::ClosureReason; +use lightning::impl_writeable_tlv_based; +use lightning::ln::types::ChannelId; + +use crate::data_store::{StorableObject, StorableObjectUpdate}; +use crate::types::UserChannelId; + +/// Details of a closed channel. +/// +/// Returned by [`Node::list_closed_channels`]. +/// +/// [`Node::list_closed_channels`]: crate::Node::list_closed_channels +#[derive(Clone, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct ClosedChannelDetails { + /// The channel's ID at the time it was closed. + pub channel_id: ChannelId, + /// The local identifier of the channel. + pub user_channel_id: UserChannelId, + /// The node ID of the channel's counterparty. + pub counterparty_node_id: PublicKey, + /// The channel's funding transaction outpoint. + pub funding_txo: Option, + /// The channel's capacity in satoshis. + pub channel_capacity_sats: Option, + /// Our local balance in millisatoshis at the time of channel closure. + pub last_local_balance_msat: Option, + /// Indicates whether we initiated the channel opening. + /// + /// `true` if the channel was opened by us (outbound), `false` if opened by the counterparty + /// (inbound). This will be `false` for channels opened prior to this field being tracked. + pub is_outbound: bool, + /// Indicates whether the channel was publicly announced. + /// + /// This will be `false` for channels opened prior to this field being tracked. + pub is_announced: bool, + /// The reason for the channel closure. + pub closure_reason: Option, + /// The timestamp, in seconds since start of the UNIX epoch, when the channel was closed. + pub closed_at: u64, +} + +impl_writeable_tlv_based!(ClosedChannelDetails, { + (0, channel_id, required), + (2, user_channel_id, required), + (4, counterparty_node_id, required), + (6, funding_txo, option), + (8, channel_capacity_sats, option), + (10, last_local_balance_msat, option), + (12, is_outbound, required), + (14, closure_reason, upgradable_option), + (16, closed_at, required), + (18, is_announced, required), +}); + +pub(crate) struct ClosedChannelDetailsUpdate(pub UserChannelId); + +impl StorableObjectUpdate for ClosedChannelDetailsUpdate { + fn id(&self) -> UserChannelId { + self.0 + } +} + +impl StorableObject for ClosedChannelDetails { + type Id = UserChannelId; + type Update = ClosedChannelDetailsUpdate; + + fn id(&self) -> UserChannelId { + self.user_channel_id + } + + fn update(&mut self, _update: Self::Update) -> bool { + // Closed channel records are immutable once written. + false + } + + fn to_update(&self) -> Self::Update { + ClosedChannelDetailsUpdate(self.user_channel_id) + } +} diff --git a/src/event.rs b/src/event.rs index 93d274ff7..4dd003fdf 100644 --- a/src/event.rs +++ b/src/event.rs @@ -7,9 +7,10 @@ use core::future::Future; use core::task::{Poll, Waker}; -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use std::ops::Deref; use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bitcoin::blockdata::locktime::absolute::LockTime; use bitcoin::secp256k1::PublicKey; @@ -33,6 +34,8 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_liquidity::lsps2::utils::compute_opening_fee; use lightning_types::payment::{PaymentHash, PaymentPreimage}; +use crate::channel::store::ChannelRecord; +use crate::closed_channel::ClosedChannelDetails; use crate::config::{may_announce_channel, Config}; use crate::connection::ConnectionManager; use crate::data_store::DataStoreUpdateResult; @@ -53,7 +56,8 @@ use crate::payment::store::{ use crate::payment::PaymentMetadata; use crate::runtime::Runtime; use crate::types::{ - CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, Sweeper, Wallet, + ChannelRecordStore, ClosedChannelStore, CustomTlvRecord, DynStore, KeysManager, OnionMessenger, + PaymentStore, Sweeper, Wallet, }; use crate::{ hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore, @@ -270,6 +274,18 @@ pub enum Event { counterparty_node_id: Option, /// This will be `None` for events serialized by LDK Node v0.2.1 and prior. reason: Option, + /// The channel's capacity in satoshis. + /// + /// This will be `None` for events serialized by LDK Node v0.8.0 and prior. + channel_capacity_sats: Option, + /// The channel's funding transaction outpoint. + /// + /// This will be `None` for events serialized by LDK Node v0.8.0 and prior. + channel_funding_txo: Option, + /// Our local balance in millisatoshis at the time of channel closure. + /// + /// This will be `None` for events serialized by LDK Node v0.8.0 and prior. + last_local_balance_msat: Option, }, /// A channel splice has been negotiated and the funding transaction is pending /// confirmation on-chain. @@ -331,6 +347,9 @@ impl_writeable_tlv_based_enum!(Event, (1, counterparty_node_id, option), (2, user_channel_id, required), (3, reason, upgradable_option), + (5, channel_capacity_sats, option), + (7, channel_funding_txo, option), + (9, last_local_balance_msat, option), }, (6, PaymentClaimable) => { (0, payment_hash, required), @@ -536,6 +555,14 @@ where liquidity_source: Arc>>, payment_store: Arc, peer_store: Arc>, + closed_channel_store: Arc, + channel_record_store: Arc, + // Tracks which user_channel_ids correspond to outbound channels. Populated at startup from + // list_channels() and updated on ChannelPending events. Consumed on ChannelClosed events. + outbound_channel_ids: Mutex>, + // Tracks which user_channel_ids correspond to announced channels. Populated at startup from + // list_channels() and updated on ChannelPending events. Consumed on ChannelClosed events. + announced_channel_ids: Mutex>, keys_manager: Arc, runtime: Arc, logger: L, @@ -555,11 +582,28 @@ where channel_manager: Arc, connection_manager: Arc>, output_sweeper: Arc, network_graph: Arc, liquidity_source: Arc>>, payment_store: Arc, - peer_store: Arc>, keys_manager: Arc, + peer_store: Arc>, closed_channel_store: Arc, + channel_record_store: Arc, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, runtime: Arc, logger: L, config: Arc, ) -> Self { + // Seed outbound_channel_ids and announced_channel_ids from currently open channels so we + // correctly classify channels that were already open when this node started. + let (outbound_channel_ids, announced_channel_ids) = { + let mut outbound = HashSet::new(); + let mut announced = HashSet::new(); + for chan in channel_manager.list_channels() { + if chan.is_outbound { + outbound.insert(UserChannelId(chan.user_channel_id)); + } + if chan.is_announced { + announced.insert(UserChannelId(chan.user_channel_id)); + } + } + (Mutex::new(outbound), Mutex::new(announced)) + }; + Self { event_queue, wallet, @@ -571,6 +615,10 @@ where liquidity_source, payment_store, peer_store, + closed_channel_store, + channel_record_store, + outbound_channel_ids, + announced_channel_ids, keys_manager, logger, runtime, @@ -1527,15 +1575,39 @@ where }, }; - let peer_to_store = { + let (record_opt, peer_to_store) = { let network_graph = self.network_graph.read_only(); let channels = self.channel_manager.list_channels_with_counterparty(&counterparty_node_id); - channels - .into_iter() - .find(|c| c.channel_id == channel_id) - .filter(|pending_channel| { - !pending_channel.is_outbound + let pending_channel = channels.into_iter().find(|c| c.channel_id == channel_id); + + let record_opt = if let Some(ref ch) = pending_channel { + if ch.is_outbound { + self.outbound_channel_ids + .lock() + .expect("Lock poisoned") + .insert(UserChannelId(user_channel_id)); + } + if ch.is_announced { + self.announced_channel_ids + .lock() + .expect("Lock poisoned") + .insert(UserChannelId(user_channel_id)); + } + Some(ChannelRecord::Funded { + user_channel_id: UserChannelId(user_channel_id), + counterparty_node_id, + channel_id, + is_outbound: ch.is_outbound, + is_announced: ch.is_announced, + }) + } else { + None + }; + + let peer_to_store = pending_channel + .filter(|ch| { + !ch.is_outbound && self.peer_store.get_peer(&counterparty_node_id).is_none() }) .and_then(|_| { @@ -1548,8 +1620,22 @@ where node_id: counterparty_node_id, address: address.clone(), }) - }) - }; + }); + + (record_opt, peer_to_store) + }; // network_graph is dropped here, before any await + + if let Some(record) = record_opt { + if let Err(e) = self.channel_record_store.insert_or_update(record).await { + log_error!( + self.logger, + "Failed to persist channel record for {}: {}", + channel_id, + e + ); + return Err(ReplayEvent()); + } + } if let Some(peer) = peer_to_store { self.peer_store.add_peer(peer).await.unwrap_or_else(|e| { log_error!( @@ -1609,19 +1695,99 @@ where reason, user_channel_id, counterparty_node_id, - .. + channel_capacity_sats, + channel_funding_txo, + last_local_balance_msat, } => { log_info!(self.logger, "Channel {} closed due to: {}", channel_id, reason); - let event = Event::ChannelClosed { + let user_channel_id = UserChannelId(user_channel_id); + let is_outbound_from_set = self + .outbound_channel_ids + .lock() + .expect("Lock poisoned") + .remove(&user_channel_id); + let is_announced_from_set = self + .announced_channel_ids + .lock() + .expect("Lock poisoned") + .remove(&user_channel_id); + + // Primary: use the durably-persisted ChannelRecord written at ChannelPending + // time. Falls back to in-memory sets (populated at startup or on + // ChannelPending), then to any already-persisted ClosedChannelDetails record + // (for the replay case where insert_or_update already succeeded but + // add_event failed and the ChannelRecord was already cleaned up). + let (is_outbound, is_announced) = self + .channel_record_store + .get(&user_channel_id) + .and_then(|record| match record { + ChannelRecord::Funded { is_outbound, is_announced, .. } => { + Some((is_outbound, is_announced)) + }, + }) + .or_else(|| { + self.closed_channel_store + .get(&user_channel_id) + .map(|existing| (existing.is_outbound, existing.is_announced)) + }) + .unwrap_or((is_outbound_from_set, is_announced_from_set)); + + let closed_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::ZERO) + .as_secs(); + + let funding_txo = channel_funding_txo.map(|op| op.into_bitcoin_outpoint()); + + // Since LDK Node v0.2 this is expected to always be set. See + // CHANGELOG.md for details on the serialization compatibility break. + let counterparty_node_id = counterparty_node_id + .expect("counterparty_node_id must be set for closed channels"); + + let record = ClosedChannelDetails { channel_id, - user_channel_id: UserChannelId(user_channel_id), + user_channel_id, counterparty_node_id, + funding_txo, + channel_capacity_sats, + last_local_balance_msat, + is_outbound, + is_announced, + closure_reason: Some(reason.clone()), + closed_at, + }; + + if let Err(e) = self.closed_channel_store.insert(record).await { + log_error!( + self.logger, + "Failed to persist closed channel {}: {}", + channel_id, + e + ); + } + + let event = Event::ChannelClosed { + channel_id, + user_channel_id, + counterparty_node_id: Some(counterparty_node_id), reason: Some(reason), + channel_capacity_sats, + channel_funding_txo: funding_txo, + last_local_balance_msat, }; match self.event_queue.add_event(event).await { - Ok(_) => {}, + Ok(_) => { + if let Err(e) = self.channel_record_store.remove(&user_channel_id).await { + log_error!( + self.logger, + "Failed to remove channel record for {}: {}", + channel_id, + e + ); + } + }, Err(e) => { log_error!(self.logger, "Failed to push to event queue: {}", e); return Err(ReplayEvent()); diff --git a/src/io/mod.rs b/src/io/mod.rs index 01c7f4492..3b169a93f 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -33,6 +33,10 @@ pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pending_payments"; pub(crate) const PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; +/// The closed channel information will be persisted under this prefix. +pub(crate) const CLOSED_CHANNEL_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "closed_channels"; +pub(crate) const CLOSED_CHANNEL_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = ""; + /// The node metrics will be persisted under this key. pub(crate) const NODE_METRICS_PRIMARY_NAMESPACE: &str = ""; pub(crate) const NODE_METRICS_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/lib.rs b/src/lib.rs index e96afe754..bed52bc4c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -84,6 +84,7 @@ mod balance; mod builder; mod chain; mod channel; +pub(crate) mod closed_channel; pub mod config; mod connection; mod data_store; @@ -127,6 +128,7 @@ pub use builder::BuildError; #[cfg(not(feature = "uniffi"))] pub use builder::NodeBuilder as Builder; use chain::ChainSource; +pub use closed_channel::ClosedChannelDetails; use config::{ default_user_config, may_announce_channel, AsyncPaymentsRole, ChannelConfig, Config, LNURL_AUTH_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL, PEER_RECONNECTION_INTERVAL, @@ -177,8 +179,8 @@ use runtime::Runtime; pub use tokio; use types::{ Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, ChannelRecordStore, - DynStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, - Scorer, Sweeper, Wallet, + ClosedChannelStore, DynStore, Graph, HRNResolver, KeysManager, OnionMessenger, PaymentStore, + PeerManager, Router, Scorer, Sweeper, Wallet, }; pub use types::{ ChannelCounterparty, ChannelDetails, CustomTlvRecord, PeerDetails, ReserveType, UserChannelId, @@ -245,9 +247,7 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, - // Foundational per-channel record store, loaded and persisted here so that upcoming - // per-channel features can build on it. Not yet read within this crate. - #[allow(dead_code)] + closed_channel_store: Arc, channel_record_store: Arc, lnurl_auth: Arc, is_running: Arc>, @@ -611,6 +611,8 @@ impl Node { Arc::clone(&self.liquidity_source), Arc::clone(&self.payment_store), Arc::clone(&self.peer_store), + Arc::clone(&self.closed_channel_store), + Arc::clone(&self.channel_record_store), Arc::clone(&self.keys_manager), static_invoice_store, Arc::clone(&self.onion_messenger), @@ -1159,6 +1161,13 @@ impl Node { .collect() } + /// Retrieve a list of closed channels. + /// + /// Channels are added to this list when they are closed and will be persisted across restarts. + pub fn list_closed_channels(&self) -> Vec { + self.closed_channel_store.list_filter(|_| true) + } + /// Connect to a node on the peer-to-peer network. /// /// If `persist` is set to `true`, we'll remember the peer and reconnect to it on restart. diff --git a/src/types.rs b/src/types.rs index 6dcab55dc..66bed3153 100644 --- a/src/types.rs +++ b/src/types.rs @@ -43,6 +43,7 @@ use lightning_net_tokio::SocketDescriptor; use crate::chain::bitcoind::UtxoSourceClient; use crate::chain::ChainSource; use crate::channel::store::ChannelRecord; +use crate::closed_channel::ClosedChannelDetails; use crate::config::{AnchorChannelsConfig, ChannelConfig}; use crate::data_store::DataStore; use crate::fee_estimator::OnchainFeeEstimator; @@ -758,4 +759,6 @@ impl From<&(u64, Vec)> for CustomTlvRecord { pub(crate) type PendingPaymentStore = DataStore>; +pub(crate) type ClosedChannelStore = DataStore>; + pub(crate) type ChannelRecordStore = DataStore>; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f0148da8a..b68fb5be1 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1651,6 +1651,23 @@ pub(crate) async fn do_channel_full_cycle( assert_eq!(node_a.next_event(), None); assert_eq!(node_b.next_event(), None); + // Check that the closed channel record was persisted. + let closed_a = node_a.list_closed_channels(); + let closed_b = node_b.list_closed_channels(); + assert_eq!(closed_a.len(), 1); + assert_eq!(closed_b.len(), 1); + assert!(closed_a[0].channel_capacity_sats.is_some()); + assert!(closed_b[0].channel_capacity_sats.is_some()); + assert!(closed_a[0].is_outbound, "node_a opened the channel, should be outbound"); + assert!(!closed_b[0].is_outbound, "node_b received the channel, should be inbound"); + assert!(closed_a[0].closure_reason.is_some()); + assert!(closed_b[0].closure_reason.is_some()); + assert!(closed_a[0].funding_txo.is_some()); + assert!(closed_b[0].funding_txo.is_some()); + assert_eq!(closed_a[0].funding_txo, closed_b[0].funding_txo); + assert_eq!(closed_a[0].counterparty_node_id, node_b.node_id()); + assert_eq!(closed_b[0].counterparty_node_id, node_a.node_id()); + node_a.stop().unwrap(); println!("\nA stopped"); node_b.stop().unwrap(); diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index eedd62ebe..0e7d3d6e8 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -36,7 +36,7 @@ use ldk_node::payment::{ ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, TransactionType, UnifiedPaymentResult, }; -use ldk_node::{BuildError, Builder, Event, Node, NodeError}; +use ldk_node::{BuildError, Builder, ClosedChannelDetails, Event, Node, NodeError}; use lightning::ln::channelmanager::PaymentId; use lightning::routing::gossip::{NodeAlias, NodeId}; use lightning::routing::router::RouteParametersConfig; @@ -4106,3 +4106,111 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { cheap.stop().unwrap(); expensive.stop().unwrap(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn closed_channel_history_persists_after_restart() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + println!("== Node A =="); + let mut config_a = random_config(true); + config_a.store_type = TestStoreType::Sqlite; + + println!("\n== Node B =="); + let mut config_b = random_config(true); + config_b.store_type = TestStoreType::Sqlite; + + let channel_amount_sat = 1_000_000; + let premine_amount_sat = 2_125_000; + + let closed_channel_before: ClosedChannelDetails; + + { + let node_a = setup_node(&chain_source, config_a.clone()); + let node_b = setup_node(&chain_source, config_b.clone()); + + let addr_a = node_a.onchain_payment().new_address().unwrap(); + let addr_b = node_b.onchain_payment().new_address().unwrap(); + + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![addr_a, addr_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + // Open channel from A to B. + let funding_txo = open_channel(&node_a, &node_b, channel_amount_sat, true, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Cooperatively close. + let user_channel_id = node_a + .list_channels() + .into_iter() + .find(|c| c.counterparty_node_id == node_b.node_id()) + .map(|c| c.user_channel_id) + .expect("open channel not found"); + node_a.close_channel(&user_channel_id, node_b.node_id()).unwrap(); + expect_event!(node_a, ChannelClosed); + expect_event!(node_b, ChannelClosed); + + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + // Verify the record is present before restart. + let closed_a = node_a.list_closed_channels(); + assert_eq!(closed_a.len(), 1); + let record = closed_a.into_iter().next().unwrap(); + assert_eq!(record.channel_capacity_sats, Some(channel_amount_sat)); + assert!(record.is_outbound); + assert!(record.is_announced); + assert_eq!(record.counterparty_node_id, node_b.node_id()); + assert!(record.funding_txo.is_some()); + assert_eq!(record.funding_txo.unwrap().txid, funding_txo.txid); + assert!(record.closure_reason.is_some()); + + closed_channel_before = record; + + let closed_b = node_b.list_closed_channels(); + assert_eq!(closed_b.len(), 1); + let record_b = closed_b.into_iter().next().unwrap(); + assert_eq!(record_b.channel_capacity_sats, Some(channel_amount_sat)); + assert!(!record_b.is_outbound); + assert!(record_b.is_announced); + assert_eq!(record_b.counterparty_node_id, node_a.node_id()); + assert!(record_b.funding_txo.is_some()); + assert_eq!(record_b.funding_txo.unwrap().txid, funding_txo.txid); + assert!(record_b.closure_reason.is_some()); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); + } + + // Restart node_a with the same config and verify the record survived. + println!("\nRestarting node A..."); + let restarted_node_a = setup_node(&chain_source, config_a); + + let closed_after = restarted_node_a.list_closed_channels(); + assert_eq!(closed_after.len(), 1, "closed channel record should survive restart"); + + let record = &closed_after[0]; + assert_eq!(record.channel_id, closed_channel_before.channel_id); + assert_eq!(record.user_channel_id, closed_channel_before.user_channel_id); + assert_eq!(record.channel_capacity_sats, closed_channel_before.channel_capacity_sats); + assert_eq!(record.funding_txo, closed_channel_before.funding_txo); + assert_eq!(record.counterparty_node_id, closed_channel_before.counterparty_node_id); + assert_eq!(record.is_outbound, closed_channel_before.is_outbound); + assert_eq!(record.is_announced, closed_channel_before.is_announced); + assert_eq!(record.closed_at, closed_channel_before.closed_at); + assert_eq!(record.closure_reason, closed_channel_before.closure_reason); + + restarted_node_a.stop().unwrap(); +} From 41f23500e7158eab1c5945370c0bcf0a11810c30 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Wed, 13 May 2026 13:20:02 -0500 Subject: [PATCH 3/6] Make counterparty_node_id non-optional on ChannelReady and ChannelClosed event We no longer need to maintain backwards compatibility with LDK Node v0.1.0, so we can make this a required field and avoid propagating the Option through the API. --- CHANGELOG.md | 5 +++++ src/event.rs | 27 +++++++++++++++------------ tests/common/mod.rs | 5 ++--- tests/integration_tests_rust.rs | 6 +++--- tests/reorg_test.rs | 6 +++--- 5 files changed, 28 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e7e012a14..9306677a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ - Users of the VSS storage backend must upgrade their VSS server to at least version `v0.1.0-alpha.0` before upgrading LDK Node. +## Serialization Compatibility +- The `counterparty_node_id` field of the `ChannelReady` and `ChannelClosed` events is now + required. Events persisted by LDK Node v0.1.0 and prior that are missing this field will + fail to deserialize. + ## Feature and API updates - The Bitcoin Core RPC and REST chain-source builder methods now accept an optional `wallet_rescan_from_height` argument. Passing a height lets fresh wallets rescan from a known diff --git a/src/event.rs b/src/event.rs index 4dd003fdf..cf845b773 100644 --- a/src/event.rs +++ b/src/event.rs @@ -250,9 +250,7 @@ pub enum Event { /// The `user_channel_id` of the channel. user_channel_id: UserChannelId, /// The `node_id` of the channel counterparty. - /// - /// This will be `None` for events serialized by LDK Node v0.1.0 and prior. - counterparty_node_id: Option, + counterparty_node_id: PublicKey, /// The outpoint of the channel's funding transaction. /// /// This represents the channel's current funding output, which may change when the @@ -269,9 +267,7 @@ pub enum Event { /// The `user_channel_id` of the channel. user_channel_id: UserChannelId, /// The `node_id` of the channel counterparty. - /// - /// This will be `None` for events serialized by LDK Node v0.1.0 and prior. - counterparty_node_id: Option, + counterparty_node_id: PublicKey, /// This will be `None` for events serialized by LDK Node v0.2.1 and prior. reason: Option, /// The channel's capacity in satoshis. @@ -331,7 +327,7 @@ impl_writeable_tlv_based_enum!(Event, }, (3, ChannelReady) => { (0, channel_id, required), - (1, counterparty_node_id, option), + (1, counterparty_node_id, required), (2, user_channel_id, required), (3, funding_txo, option), }, @@ -344,7 +340,7 @@ impl_writeable_tlv_based_enum!(Event, }, (5, ChannelClosed) => { (0, channel_id, required), - (1, counterparty_node_id, option), + (1, counterparty_node_id, required), (2, user_channel_id, required), (3, reason, upgradable_option), (5, channel_capacity_sats, option), @@ -1679,7 +1675,7 @@ where let event = Event::ChannelReady { channel_id, user_channel_id: UserChannelId(user_channel_id), - counterparty_node_id: Some(counterparty_node_id), + counterparty_node_id, funding_txo, }; match self.event_queue.add_event(event).await { @@ -1770,7 +1766,7 @@ where let event = Event::ChannelClosed { channel_id, user_channel_id, - counterparty_node_id: Some(counterparty_node_id), + counterparty_node_id, reason: Some(reason), channel_capacity_sats, channel_funding_txo: funding_txo, @@ -2080,6 +2076,7 @@ mod tests { use std::sync::atomic::{AtomicU16, Ordering}; use std::time::Duration; + use bitcoin::secp256k1::{Secp256k1, SecretKey}; use lightning::util::test_utils::TestLogger; use super::*; @@ -2143,10 +2140,13 @@ mod tests { let event_queue = Arc::new(EventQueue::new(Arc::clone(&store), Arc::clone(&logger))); assert_eq!(event_queue.next_event(), None); + let secp = Secp256k1::new(); + let counterparty_node_id = + PublicKey::from_secret_key(&secp, &SecretKey::from_slice(&[1u8; 32]).unwrap()); let expected_event = Event::ChannelReady { channel_id: ChannelId([23u8; 32]), user_channel_id: UserChannelId(2323), - counterparty_node_id: None, + counterparty_node_id, funding_txo: None, }; event_queue.add_event(expected_event.clone()).await.unwrap(); @@ -2181,10 +2181,13 @@ mod tests { let event_queue = Arc::new(EventQueue::new(Arc::clone(&store), Arc::clone(&logger))); assert_eq!(event_queue.next_event(), None); + let secp = Secp256k1::new(); + let counterparty_node_id = + PublicKey::from_secret_key(&secp, &SecretKey::from_slice(&[1u8; 32]).unwrap()); let expected_event = Event::ChannelReady { channel_id: ChannelId([23u8; 32]), user_channel_id: UserChannelId(2323), - counterparty_node_id: None, + counterparty_node_id, funding_txo: None, }; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index b68fb5be1..6412dd16e 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -133,7 +133,7 @@ macro_rules! expect_channel_ready_event { match event { ref e @ Event::ChannelReady { user_channel_id, counterparty_node_id, .. } => { println!("{} got event {:?}", $node.node_id(), e); - assert_eq!(counterparty_node_id, Some($counterparty_node_id)); + assert_eq!(counterparty_node_id, $counterparty_node_id); $node.event_handled().unwrap(); user_channel_id }, @@ -170,8 +170,7 @@ macro_rules! expect_channel_ready_events { } } assert!( - ids.contains(&Some($counterparty_node_id_a)) - && ids.contains(&Some($counterparty_node_id_b)), + ids.contains(&$counterparty_node_id_a) && ids.contains(&$counterparty_node_id_b), "Expected ChannelReady events from {:?} and {:?}, but got {:?}", $counterparty_node_id_a, $counterparty_node_id_b, diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 0e7d3d6e8..e8d452c48 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -1838,7 +1838,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // Verify the candidate that locked is the one that confirmed, not necessarily the last broadcast. match node_a.next_event_async().await { Event::ChannelReady { funding_txo, counterparty_node_id, .. } => { - assert_eq!(counterparty_node_id, Some(node_b.node_id())); + assert_eq!(counterparty_node_id, node_b.node_id()); assert_eq!(funding_txo, Some(winning_txo)); node_a.event_handled().unwrap(); }, @@ -1846,7 +1846,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { } match node_b.next_event_async().await { Event::ChannelReady { funding_txo, counterparty_node_id, .. } => { - assert_eq!(counterparty_node_id, Some(node_a.node_id())); + assert_eq!(counterparty_node_id, node_a.node_id()); assert_eq!(funding_txo, Some(winning_txo)); node_b.event_handled().unwrap(); }, @@ -4154,7 +4154,7 @@ async fn closed_channel_history_persists_after_restart() { let user_channel_id = node_a .list_channels() .into_iter() - .find(|c| c.counterparty_node_id == node_b.node_id()) + .find(|c| c.counterparty.node_id == node_b.node_id()) .map(|c| c.user_channel_id) .expect("open channel not found"); node_a.close_channel(&user_channel_id, node_b.node_id()).unwrap(); diff --git a/tests/reorg_test.rs b/tests/reorg_test.rs index 295d9fdd2..f2c3b6586 100644 --- a/tests/reorg_test.rs +++ b/tests/reorg_test.rs @@ -112,11 +112,11 @@ proptest! { let next_node = nodes.get((i + 1) % nodes.len()).unwrap(); let prev_node = nodes.get((i + nodes.len() - 1) % nodes.len()).unwrap(); - assert!(user_channels.get(&Some(next_node.node_id())) != None); - assert!(user_channels.get(&Some(prev_node.node_id())) != None); + assert!(user_channels.get(&next_node.node_id()) != None); + assert!(user_channels.get(&prev_node.node_id()) != None); let user_channel_id = - user_channels.get(&Some(next_node.node_id())).expect("Missing user channel for node"); + user_channels.get(&next_node.node_id()).expect("Missing user channel for node"); node_channels_id.insert(node.node_id(), *user_channel_id); } From 1d8fc59795f75eea1a84c8addd412cfd9ce27207 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Fri, 22 May 2026 11:16:26 -0500 Subject: [PATCH 4/6] Make ChannelClosed persistence idempotent and retry on failure Fall back to the already-persisted record for is_outbound/is_announced when in-memory sets are empty on replay, use insert_or_update to avoid overwriting correct values, and propagate persist failures as ReplayEvent. --- src/event.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/event.rs b/src/event.rs index cf845b773..6930d6c45 100644 --- a/src/event.rs +++ b/src/event.rs @@ -1754,13 +1754,14 @@ where closed_at, }; - if let Err(e) = self.closed_channel_store.insert(record).await { + if let Err(e) = self.closed_channel_store.insert_or_update(record).await { log_error!( self.logger, "Failed to persist closed channel {}: {}", channel_id, e ); + return Err(ReplayEvent()); } let event = Event::ChannelClosed { From 959992f8c25e7aedc118fccdf36f51e5c0f0c0b1 Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Fri, 22 May 2026 11:17:47 -0500 Subject: [PATCH 5/6] Document when optional fields of ClosedChannelDetails can be None --- src/closed_channel.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/closed_channel.rs b/src/closed_channel.rs index 84a48c75a..0bd00665e 100644 --- a/src/closed_channel.rs +++ b/src/closed_channel.rs @@ -29,10 +29,16 @@ pub struct ClosedChannelDetails { /// The node ID of the channel's counterparty. pub counterparty_node_id: PublicKey, /// The channel's funding transaction outpoint. + /// + /// Will be `None` if the channel was closed before a funding transaction was established. pub funding_txo: Option, /// The channel's capacity in satoshis. + /// + /// Will be `None` if the channel was closed before the capacity was known. pub channel_capacity_sats: Option, /// Our local balance in millisatoshis at the time of channel closure. + /// + /// Will be `None` if the local balance was not available at the time of closure. pub last_local_balance_msat: Option, /// Indicates whether we initiated the channel opening. /// @@ -44,6 +50,9 @@ pub struct ClosedChannelDetails { /// This will be `false` for channels opened prior to this field being tracked. pub is_announced: bool, /// The reason for the channel closure. + /// + /// Will be `None` if the closure reason could not be decoded, e.g., if it was written by a + /// future version of LDK Node using a closure reason variant not yet known to this version. pub closure_reason: Option, /// The timestamp, in seconds since start of the UNIX epoch, when the channel was closed. pub closed_at: u64, From 1e7dec984ba9b3d49e5d5067c0fe196e12068d5b Mon Sep 17 00:00:00 2001 From: Fernando Ledesma Date: Wed, 8 Jul 2026 13:32:43 -0500 Subject: [PATCH 6/6] fixup! Make ChannelClosed persistence idempotent and retry on failure --- src/event.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/event.rs b/src/event.rs index 6930d6c45..fb4fd70d8 100644 --- a/src/event.rs +++ b/src/event.rs @@ -1702,12 +1702,12 @@ where .outbound_channel_ids .lock() .expect("Lock poisoned") - .remove(&user_channel_id); + .contains(&user_channel_id); let is_announced_from_set = self .announced_channel_ids .lock() .expect("Lock poisoned") - .remove(&user_channel_id); + .contains(&user_channel_id); // Primary: use the durably-persisted ChannelRecord written at ChannelPending // time. Falls back to in-memory sets (populated at startup or on @@ -1776,6 +1776,14 @@ where match self.event_queue.add_event(event).await { Ok(_) => { + self.outbound_channel_ids + .lock() + .expect("Lock poisoned") + .remove(&user_channel_id); + self.announced_channel_ids + .lock() + .expect("Lock poisoned") + .remove(&user_channel_id); if let Err(e) = self.channel_record_store.remove(&user_channel_id).await { log_error!( self.logger,