diff --git a/src/lib.rs b/src/lib.rs index 1c88de2f7..a5ff6c435 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1415,6 +1415,12 @@ impl Node { .expect("a 16-byte slice should convert into a [u8; 16]"), ); + // Persist the peer before creating the channel. Once `create_channel` succeeds the + // channel exists, so failing afterwards would report a failed initiation for a live + // channel, and a caller retrying would open a second one. + let peer_was_stored = self.peer_store.get_peer(&peer_info.node_id).is_some(); + self.runtime.block_on(self.peer_store.add_peer(peer_info.clone()))?; + let result = if disable_counterparty_reserve { self.channel_manager.create_channel_to_trusted_peer_0reserve( peer_info.node_id, @@ -1445,7 +1451,6 @@ impl Node { zero_reserve_string, peer_info.node_id ); - self.runtime.block_on(self.peer_store.add_peer(peer_info))?; Ok(UserChannelId(user_channel_id)) }, Err(e) => { @@ -1455,6 +1460,19 @@ impl Node { zero_reserve_string, e ); + // Don't keep reconnecting to a peer we only stored for this attempt. + if !peer_was_stored { + if let Err(e) = + self.runtime.block_on(self.peer_store.remove_peer(&peer_info.node_id)) + { + log_error!( + self.logger, + "Failed to remove peer {} from peer store: {}", + peer_info.node_id, + e + ); + } + } Err(Error::ChannelCreationFailed) }, } diff --git a/src/peer_store.rs b/src/peer_store.rs index 8345bf711..a2df26f64 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -45,14 +45,19 @@ where pub(crate) async fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { let _guard = self.mutation_lock.lock().await; let data = { - let mut locked_peers = self.peers.write().expect("lock"); + let locked_peers = self.peers.read().expect("lock"); if locked_peers.contains_key(&peer_info.node_id) { return Ok(()); } - locked_peers.insert(peer_info.node_id, peer_info); - PeerStoreSerWrapper(&locked_peers).encode() + let mut updated_peers = locked_peers.clone(); + updated_peers.insert(peer_info.node_id, peer_info.clone()); + PeerStoreSerWrapper(&updated_peers).encode() }; - self.persist_peers(data).await + // Only update the in-memory set once the write succeeded, so a failed write can be + // retried: otherwise the next `add_peer` would find the peer and skip persisting it. + self.persist_peers(data).await?; + self.peers.write().expect("lock").insert(peer_info.node_id, peer_info); + Ok(()) } pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { @@ -277,4 +282,22 @@ mod tests { assert_eq!(Err(Error::PersistenceFailed), peer_store.remove_peer(&node_id).await); assert_eq!(Some(peer_info), peer_store.get_peer(&node_id)); } + + #[tokio::test] + async fn add_peer_does_not_mutate_memory_if_persist_fails() { + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let logger = Arc::new(TestLogger::new()); + let peer_store = PeerStore::new(store, logger); + let node_id = PublicKey::from_str( + "0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993", + ) + .unwrap(); + let peer_info = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() }; + + assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info.clone()).await); + assert_eq!(None, peer_store.get_peer(&node_id)); + // A retry must attempt the write again rather than treating the peer as stored. + assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info).await); + } } diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index f58cad690..67b2a00ed 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -600,6 +600,135 @@ async fn channel_full_cycle_0conf_0reserve() { .await; } +/// An [`InMemoryStore`] whose writes of the persisted peer list fail while `fail_peer_writes` is set. +#[derive(Clone)] +struct PeerWriteFailingStore { + inner: Arc, + fail_peer_writes: Arc, +} + +impl PeerWriteFailingStore { + fn new() -> Self { + Self { + inner: Arc::new(InMemoryStore::new()), + fail_peer_writes: Arc::new(AtomicBool::new(false)), + } + } +} + +impl KVStore for PeerWriteFailingStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, lightning::io::Error>> + 'static + Send { + KVStore::read(&*self.inner, primary_namespace, secondary_namespace, key) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> impl Future> + 'static + Send { + let inner = Arc::clone(&self.inner); + // The peer list lives at `("", "", "peers")`, see `PEER_INFO_PERSISTENCE_*`. + let fail = self.fail_peer_writes.load(Ordering::Acquire) + && primary_namespace.is_empty() + && secondary_namespace.is_empty() + && key == "peers"; + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + async move { + if fail { + return Err(lightning::io::Error::new( + lightning::io::ErrorKind::Other, + "injected peer store write failure", + )); + } + KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await + } + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> impl Future> + 'static + Send { + KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> impl Future, lightning::io::Error>> + 'static + Send { + KVStore::list(&*self.inner, primary_namespace, secondary_namespace) + } +} + +impl PaginatedKVStore for PeerWriteFailingStore { + fn list_paginated( + &self, primary_namespace: &str, secondary_namespace: &str, page_token: Option, + ) -> impl Future> + 'static + Send + { + PaginatedKVStore::list_paginated( + &*self.inner, + primary_namespace, + secondary_namespace, + page_token, + ) + } +} + +// If the peer cannot be persisted, `open_channel` must fail without having created a channel: +// reporting a failure for a live channel would make a retrying caller open a second one. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn channel_open_fails_cleanly_when_peer_persistence_fails() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + let chain_source = TestChainSource::Esplora(&electrsd); + + let config_a = random_config(); + setup_builder!(builder_a, config_a.node_config); + let mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + builder_a.set_chain_source_esplora(esplora_url, Some(sync_config)); + let store = PeerWriteFailingStore::new(); + let node_a = builder_a.build_with_store(config_a.node_entropy.into(), store.clone()).unwrap(); + node_a.start().unwrap(); + + let node_b = setup_node(&chain_source, random_config()); + + // Fund both nodes so node B passes the anchor reserve check on the accept path. + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(1_000_000), + ) + .await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let node_b_id = node_b.node_id(); + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + + // The peer store write fails: no channel may have been initiated. + store.fail_peer_writes.store(true, Ordering::Release); + assert_eq!( + Err(NodeError::PersistenceFailed), + node_a.open_channel(node_b_id, node_b_addr.clone(), 100_000, None, None) + ); + assert!(node_a.list_channels().is_empty()); + assert!(!node_a.list_peers().iter().any(|p| p.node_id == node_b_id && p.is_persisted)); + + // Once the store recovers, a retry opens exactly one channel and persists the peer. + store.fail_peer_writes.store(false, Ordering::Release); + node_a.open_channel(node_b_id, node_b_addr, 100_000, None, None).unwrap(); + expect_channel_pending_event!(node_a, node_b_id); + expect_channel_pending_event!(node_b, node_a.node_id()); + assert_eq!(node_a.list_channels().len(), 1); + assert!(node_a.list_peers().iter().any(|p| p.node_id == node_b_id && p.is_persisted)); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn channel_open_fails_when_funds_insufficient() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();