Fix unified payment falling back to on-chain after PersistenceFailed

In UnifiedPayment::send, the BOLT11 leg's bolt11_invoice.send only
returns Err(PersistenceFailed) *after* pay_for_bolt11_invoice has
already succeeded and the Lightning payment is in-flight. The previous
match treated every remaining error as a fall-through to the next
payment method, so a persistence failure after initiation would
broadcast an on-chain transaction for the same URI — a duplicate
payment.

Err(Error::PersistenceFailed) on the BOLT11 leg is now terminal,
mirroring how DuplicatePayment is handled, and aborts the unified
payment instead of falling back to on-chain.

This is a regression hazard raised during review of the DuplicatePayment
fix (PR #1038). It is pre-existing and orthogonal to #1033; tracked
here as the unified variant of the broader post-commit persistence
hazard.

Per review discussion: rather than a dedicated node/channel fixture,
build unified_send_receive_bip21_uri's node_a on a PaymentFailingStore
(inert until armed) from the start, and add a PersistenceFailed
assertion at the end of that test using a fresh BOLT11-only URI. This
reuses the funding/channel/announcement setup and the successful-send
flow the test already has, rather than duplicating it.

Adds PaymentFailingStore (a KVStore wrapper that fails writes to the
payments namespace on demand) and setup_two_nodes_with_failing_store_a
(mirrors setup_two_nodes, but node_a is built on PaymentFailingStore).

Author: elnafateh

src/payment/unified.rs

@@ -333,6 +333,14 @@ impl UnifiedPayment {
 							log_error!(self.logger, "Failed to send BOLT11 invoice: DuplicatePayment. This is part of a unified payment. Aborting to avoid duplicate payment.");
 							return Err(Error::DuplicatePayment);
 						},
+						// A persistence failure may occur after the Lightning payment has
+						// already been initiated with the ChannelManager. Falling back to
+						// the on-chain method in that case would double-pay, so we abort
+						// instead of proceeding to the next payment method.
+						Err(Error::PersistenceFailed) => {
+							log_error!(self.logger, "Failed to send BOLT11 invoice: PersistenceFailed. This is part of a unified payment. Aborting to avoid a potential duplicate payment.");
+							return Err(Error::PersistenceFailed);
+						},
 						Err(e) => {
 							log_error!(self.logger, "Failed to send BOLT11 invoice: {:?}. This is part of a unified payment. Falling back to the on-chain transaction.", e);
 						},

tests/integration_tests_rust.rs

@@ -22,15 +22,16 @@ use common::logging::{
 	init_log_logger, validate_log_entry, CollectingLogWriter, MultiNodeLogger, TestLogWriter,
 };
 use common::{
-	bump_fee_and_broadcast, distribute_funds_unconfirmed, do_channel_full_cycle,
-	expect_channel_pending_event, expect_channel_ready_event, expect_channel_ready_events,
-	expect_event, expect_payment_claimable_event, expect_payment_received_event,
-	expect_payment_successful_event, expect_splice_negotiated_event, generate_blocks_and_wait,
-	generate_listening_addresses, invalidate_blocks, open_channel, open_channel_no_wait,
-	open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks,
-	prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder,
-	setup_node, setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore,
-	NodePaymentExt, TestChainSource, TestConfig, TestStoreType, TestSyncStore,
+	bump_fee_and_broadcast, configure_chain_source, distribute_funds_unconfirmed,
+	do_channel_full_cycle, expect_channel_pending_event, expect_channel_ready_event,
+	expect_channel_ready_events, expect_event, expect_payment_claimable_event,
+	expect_payment_received_event, expect_payment_successful_event, expect_splice_negotiated_event,
+	generate_blocks_and_wait, generate_listening_addresses, invalidate_blocks, open_channel,
+	open_channel_no_wait, open_channel_push_amt, open_channel_with_all,
+	premine_and_distribute_funds, premine_blocks, prepare_rbf, random_chain_source, random_config,
+	setup_bitcoind_and_electrsd, setup_builder, setup_node, setup_two_nodes, splice_in_with_all,
+	wait_for_block, wait_for_tx, InMemoryStore, NodePaymentExt, TestChainSource, TestConfig,
+	TestNode, TestStoreType, TestSyncStore,
 };
 use electrsd::corepc_node::{self, Node as BitcoinD};
 use electrsd::ElectrsD;
@@ -3315,12 +3316,99 @@ async fn unified_receive_rejects_msat_overflow() {
 	);
 }
 
+/// A [`KVStore`] that fails every `write` to the payments namespace once `fail_writes` is set,
+/// while keeping everything else operational. Used to arm a `PersistenceFailed` regression case
+/// on top of an otherwise-normal node, without needing a dedicated node/channel fixture.
+struct PaymentFailingStore {
+	inner: Arc<InMemoryStore>,
+	fail_writes: Arc<AtomicBool>,
+}
+
+impl KVStore for PaymentFailingStore {
+	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);
+		let fail_writes = Arc::clone(&self.fail_writes);
+		let primary_namespace = primary_namespace.to_string();
+		let secondary_namespace = secondary_namespace.to_string();
+		let key = key.to_string();
+		async move {
+			// Only fail payment-store writes. Failing every write (e.g. channel monitor
+			// updates) would crash the background processor, defeating the test.
+			if fail_writes.load(Ordering::Acquire) && primary_namespace == "payments" {
+				return Err(lightning::io::Error::new(
+					lightning::io::ErrorKind::Other,
+					"injected payment persistence 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 PaymentFailingStore {
+	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,
+		)
+	}
+}
+
+/// Builds `node_a` on a [`PaymentFailingStore`] the caller can arm later via `fail_writes`, and
+/// `node_b` on the default store — otherwise identical to `setup_two_nodes`. Lets a single test
+/// flow cover the `PersistenceFailed` fallback hazard on top of the fixture it already needs for
+/// the normal unified-payment paths, instead of duplicating that fixture in a standalone test.
+fn setup_two_nodes_with_failing_store_a(
+	chain_source: &TestChainSource, fail_writes: Arc<AtomicBool>,
+) -> (TestNode, TestNode) {
+	let config_a = random_config();
+	setup_builder!(builder_a, config_a.node_config);
+	configure_chain_source(chain_source, &mut builder_a, &config_a);
+	builder_a.set_async_payments_role(config_a.async_payments_role).unwrap();
+	let failing_store = PaymentFailingStore { inner: Arc::new(InMemoryStore::new()), fail_writes };
+	let node_a = builder_a.build_with_store(config_a.node_entropy.into(), failing_store).unwrap();
+	node_a.start().unwrap();
+
+	let mut config_b = random_config();
+	config_b.node_config.manually_handle_unknown_bolt11_payments = true;
+	let node_b = setup_node(chain_source, config_b);
+
+	(node_a, node_b)
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn unified_send_receive_bip21_uri() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
 	let chain_source = random_chain_source(&bitcoind, &electrsd);
 
-	let (node_a, node_b) = setup_two_nodes(&chain_source, false, false);
+	let fail_writes = Arc::new(AtomicBool::new(false));
+	let (node_a, node_b) =
+		setup_two_nodes_with_failing_store_a(&chain_source, Arc::clone(&fail_writes));
 
 	let address_a = node_a.onchain_payment().new_address().unwrap();
 	let premined_sats = 5_000_000;
@@ -3441,6 +3529,36 @@ async fn unified_send_receive_bip21_uri() {
 
 	assert_eq!(node_b.list_balances().total_onchain_balance_sats, 800_000);
 	assert_eq!(node_b.list_balances().total_lightning_balance_sats, 200_000);
+
+	// Regression test: a payment-store persistence failure on the BOLT11 leg — which can occur
+	// after the Lightning payment has already been initiated with the ChannelManager — must also
+	// abort rather than fall back to on-chain. Arm node_a's store and send a fresh BOLT11-only
+	// URI to isolate the BOLT11 leg from the BOLT12/on-chain legs already exercised above.
+	fail_writes.store(true, Ordering::Release);
+
+	let fresh_amount_sats = 50_000;
+	let fresh_uri = node_b.unified_payment().receive(fresh_amount_sats, "asdf", 4_000).unwrap();
+	let fresh_uri_bolt11_only = fresh_uri.split("&lno=").next().unwrap();
+
+	let persistence_result = node_a.unified_payment().send(fresh_uri_bolt11_only, None, None).await;
+	match persistence_result {
+		Err(NodeError::PersistenceFailed) => {
+			// Expected — the unified payment must abort, not fall back to on-chain.
+		},
+		Ok(UnifiedPaymentResult::Onchain { txid }) => {
+			panic!("Regression: PersistenceFailed fell back to on-chain. txid={}", txid);
+		},
+		other => panic!("Expected PersistenceFailed error, got: {:?}", other),
+	}
+
+	let onchain_payments = node_a.list_all_payments().into_iter().any(|p| {
+		matches!(p.kind, PaymentKind::Onchain { .. })
+			&& p.amount_msat == Some(fresh_amount_sats as u64 * 1000)
+	});
+	assert!(
+		!onchain_payments,
+		"An on-chain payment for the fresh amount was broadcast despite PersistenceFailed"
+	);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]