Persist the peer before initiating a channel

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)
 			},
 		}

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<DynStore> = 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);
+	}
 }

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<InMemoryStore>,
+	fail_peer_writes: Arc<AtomicBool>,
+}
+
+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<Output = Result<Vec<u8>, 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<u8>,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + '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<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, 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<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, lightning::io::Error>> + '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();