Persist the peer before initiating a channel

src/lib.rs

@@ -1449,6 +1449,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()))?;
+
 		// Register the fee-rate override before creating the channel, as the
 		// `FundingGenerationReady` event may be handled before `create_channel` returns.
 		if let Some(fee_rate) = fee_rate {
@@ -1485,7 +1491,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) => {
@@ -1496,6 +1501,19 @@ impl Node {
 					e
 				);
 				self.pending_funding_fee_rates.lock().expect("lock").remove(&user_channel_id);
+				// 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)
 			},
 		}

tests/integration_tests_rust.rs

@@ -1049,6 +1049,134 @@ 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, None)
+	);
+	assert!(node_a.list_channels().is_empty());
+
+	// 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, 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();
+}
+
 // The handler answers a failed write of a channel's facts two ways. The funding this node
 // generates is withheld from LDK until its facts are on record, since nothing may broadcast a
 // transaction this node could not classify: the event is replayed. Every other report accompanies