Add payjoin v2 receiver flow

Implements the receiver side of the BIP 77 Payjoin v2 protocol, allowing
LDK Node users to receive payjoin payments via a payjoin directory and
OHTTP relay.

- Adds a `PayjoinPayment` handler exposing a `receive()` method that returns
  a BIP 21 URI the sender can use to initiate the payjoin flow. The full
  receiver state machine is implemented covering all `ReceiveSession` states:
  polling the directory, validating the sender's proposal, contributing
  inputs, finalizing the PSBT, and monitoring the mempool.

- Session state is persisted via `KVStorePayjoinReceiverPersister` and
  survives node restarts through event log replay. Sender inputs are tracked
  by `OutPoint` across polling attempts to prevent replay attacks. The
  sender's fallback transaction is broadcast on cancellation or failure to
  ensure the receiver still gets paid.

- Adds `PaymentKind::Payjoin` to the payment store, `PayjoinConfig` for
  configuring the payjoin directory and OHTTP relay via
  `Builder::set_payjoin_config`, and background tasks for session resumption
  every 15 seconds and cleanup of terminal sessions after 24 hours.

Author: Camillarhi

Cargo.toml

@@ -81,6 +81,8 @@ prost = { version = "0.11.6", default-features = false}
 #bitcoin-payment-instructions = { version = "0.6" }
 bitcoin-payment-instructions = { git = "https://github.com/jkczyz/bitcoin-payment-instructions", rev = "869fd348c3ca0c78f439d2f31181f4d798c6b20e" }
 
+payjoin = { git = "https://github.com/payjoin/rust-payjoin.git", rev= "dd21835e55116c681fddae29bf3cd363dd177745", package = "payjoin", default-features = false, features = ["v2", "io"] }
+
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["winbase"] }
 

bindings/ldk_node.udl

@@ -93,6 +93,7 @@ interface Node {
 	SpontaneousPayment spontaneous_payment();
 	OnchainPayment onchain_payment();
 	UnifiedPayment unified_payment();
+	PayjoinPayment payjoin_payment();
 	LSPS1Liquidity lsps1_liquidity();
 	[Throws=NodeError]
 	void lnurl_auth(string lnurl);
@@ -157,6 +158,8 @@ interface FeeRate {
 
 typedef interface UnifiedPayment;
 
+typedef interface PayjoinPayment;
+
 typedef interface LSPS1Liquidity;
 
 [Error]
@@ -184,6 +187,8 @@ enum NodeError {
 	"OnchainTxSigningFailed",
 	"TxSyncFailed",
 	"TxSyncTimeout",
+	"TxLookupFailed",
+	"TxLookupTimeout",
 	"GossipUpdateFailed",
 	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
@@ -221,6 +226,10 @@ enum NodeError {
 	"LnurlAuthFailed",
 	"LnurlAuthTimeout",
 	"InvalidLnurl",
+	"PayjoinNotConfigured",
+	"PayjoinSessionCreationFailed",
+	"PayjoinSessionFailed",
+	"UnsupportedChainSource"
 };
 
 typedef dictionary NodeStatus;

src/builder.rs

@@ -45,7 +45,7 @@ use vss_client::headers::VssHeaderProvider;
 use crate::chain::ChainSource;
 use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
-	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig,
+	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, PayjoinConfig,
 	DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
 };
 use crate::connection::ConnectionManager;
@@ -56,12 +56,13 @@ use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
 	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
-	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_pending_payments,
-	read_scorer, write_node_metrics,
+	read_node_metrics, read_output_sweeper, read_payjoin_sessions, read_payments, read_peer_info,
+	read_pending_payments, read_scorer, write_node_metrics,
 };
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
-	self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	self, PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE, PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE,
+	PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 	PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 };
@@ -72,13 +73,14 @@ use crate::lnurl_auth::LnurlAuth;
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
+use crate::payment::payjoin::manager::PayjoinManager;
 use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
-	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
-	Persister, SyncAndAsyncKVStore,
+	KeysManager, MessageRouter, OnionMessenger, PayjoinSessionStore, PaymentStore, PeerManager,
+	PendingPaymentStore, Persister, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -549,6 +551,15 @@ impl NodeBuilder {
 		Ok(self)
 	}
 
+	/// Configures the [`Node`] instance to enable payjoin payments.
+	///
+	/// The `payjoin_config` specifies the PayJoin directory and OHTTP relay URLs required
+	/// for payjoin V2 protocol.
+	pub fn set_payjoin_config(&mut self, payjoin_config: PayjoinConfig) -> &mut Self {
+		self.config.payjoin_config = Some(payjoin_config);
+		self
+	}
+
 	/// Configures the [`Node`] to resync chain data from genesis on first startup, recovering any
 	/// historical wallet funds.
 	///
@@ -972,6 +983,14 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_async_payments_role(role).map(|_| ())
 	}
 
+	/// Configures the [`Node`] instance to enable payjoin payments.
+	///
+	/// The `payjoin_config` specifies the PayJoin directory and OHTTP relay URLs required
+	/// for payjoin V2 protocol.
+	pub fn set_payjoin_config(&self, payjoin_config: PayjoinConfig) {
+		self.inner.write().unwrap().set_payjoin_config(payjoin_config);
+	}
+
 	/// Configures the [`Node`] to resync chain data from genesis on first startup, recovering any
 	/// historical wallet funds.
 	///
@@ -1151,12 +1170,13 @@ 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, payjoin_session_store_res) =
 		runtime.block_on(async move {
 			tokio::join!(
 				read_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
 				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
+				read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
+				read_payjoin_sessions(&*kv_store_ref, Arc::clone(&logger_ref))
 			)
 		});
 
@@ -1841,6 +1861,34 @@ fn build_with_store_internal(
 
 	let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());
 
+	let payjoin_session_store = match payjoin_session_store_res {
+		Ok(payjoin_sessions) => Arc::new(PayjoinSessionStore::new(
+			payjoin_sessions,
+			PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE.to_string(),
+			PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read payjoin session data from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
+	let payjoin_manager = Arc::new(PayjoinManager::new(
+		Arc::clone(&payjoin_session_store),
+		Arc::clone(&logger),
+		Arc::clone(&config),
+		Arc::clone(&wallet),
+		Arc::clone(&fee_estimator),
+		Arc::clone(&chain_source),
+		Arc::clone(&channel_manager),
+		stop_sender.subscribe(),
+		Arc::clone(&payment_store),
+		Arc::clone(&pending_payment_store),
+		Arc::clone(&tx_broadcaster),
+	));
+
 	#[cfg(cycle_tests)]
 	let mut _leak_checker = crate::LeakChecker(Vec::new());
 	#[cfg(cycle_tests)]
@@ -1888,6 +1936,7 @@ fn build_with_store_internal(
 		hrn_resolver,
 		#[cfg(cycle_tests)]
 		_leak_checker,
+		payjoin_manager,
 	})
 }
 

src/chain/bitcoind.rs

@@ -33,7 +33,7 @@ use serde::Serialize;
 use super::WalletSyncStatus;
 use crate::config::{
 	BitcoindRestClientConfig, Config, DEFAULT_FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS,
-	DEFAULT_TX_BROADCAST_TIMEOUT_SECS,
+	DEFAULT_TX_BROADCAST_TIMEOUT_SECS, DEFAULT_TX_LOOKUP_TIMEOUT_SECS,
 };
 use crate::fee_estimator::{
 	apply_post_estimation_adjustments, get_all_conf_targets, get_num_block_defaults_for_target,
@@ -620,6 +620,57 @@ impl BitcoindChainSource {
 			}
 		}
 	}
+
+	pub(crate) async fn can_broadcast_transaction(&self, tx: &Transaction) -> Result<bool, Error> {
+		let timeout_fut = tokio::time::timeout(
+			Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS),
+			self.api_client.test_mempool_accept(tx),
+		);
+
+		match timeout_fut.await {
+			Ok(res) => res.map_err(|e| {
+				log_error!(
+					self.logger,
+					"Failed to test mempool accept for transaction {}: {}",
+					tx.compute_txid(),
+					e
+				);
+				Error::WalletOperationFailed
+			}),
+			Err(e) => {
+				log_error!(
+					self.logger,
+					"Failed to test mempool accept for transaction {} due to timeout: {}",
+					tx.compute_txid(),
+					e
+				);
+				log_trace!(
+					self.logger,
+					"Failed test mempool accept transaction bytes: {}",
+					log_bytes!(tx.encode())
+				);
+				Err(Error::WalletOperationTimeout)
+			},
+		}
+	}
+
+	pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+		let timeout_fut = tokio::time::timeout(
+			Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS),
+			self.api_client.get_raw_transaction(txid),
+		);
+
+		match timeout_fut.await {
+			Ok(res) => res.map_err(|e| {
+				log_error!(self.logger, "Failed to get transaction {}: {}", txid, e);
+				Error::TxLookupFailed
+			}),
+			Err(e) => {
+				log_error!(self.logger, "Failed to get transaction {} due to timeout: {}", txid, e);
+				Err(Error::TxLookupTimeout)
+			},
+		}
+	}
 }
 
 #[derive(Clone)]
@@ -1179,6 +1230,34 @@ impl BitcoindClient {
 			.collect();
 		Ok(evicted_txids)
 	}
+
+	/// Tests whether the provided transaction would be accepted by the mempool.
+	pub(crate) async fn test_mempool_accept(
+		&self, tx: &Transaction,
+	) -> Result<bool, RpcClientError> {
+		match self {
+			BitcoindClient::Rpc { rpc_client, .. } => {
+				Self::test_mempool_accept_inner(Arc::clone(rpc_client), tx).await
+			},
+			BitcoindClient::Rest { rpc_client, .. } => {
+				// We rely on the internal RPC client to make this call, as this
+				// operation is not supported by Bitcoin Core's REST interface.
+				Self::test_mempool_accept_inner(Arc::clone(rpc_client), tx).await
+			},
+		}
+	}
+
+	async fn test_mempool_accept_inner(
+		rpc_client: Arc<RpcClient>, tx: &Transaction,
+	) -> Result<bool, RpcClientError> {
+		let tx_serialized = bitcoin::consensus::encode::serialize_hex(tx);
+		let tx_array = serde_json::json!([tx_serialized]);
+
+		rpc_client
+			.call_method::<TestMempoolAcceptResponse>("testmempoolaccept", &[tx_array])
+			.await
+			.map(|resp| resp.0)
+	}
 }
 
 impl BlockSource for BitcoindClient {
@@ -1334,6 +1413,23 @@ impl TryInto<GetMempoolEntryResponse> for JsonResponse {
 	}
 }
 
+pub(crate) struct TestMempoolAcceptResponse(pub bool);
+
+impl TryInto<TestMempoolAcceptResponse> for JsonResponse {
+	type Error = String;
+	fn try_into(self) -> Result<TestMempoolAcceptResponse, String> {
+		let array =
+			self.0.as_array().ok_or("Failed to parse testmempoolaccept response".to_string())?;
+		let first =
+			array.first().ok_or("Empty array response from testmempoolaccept".to_string())?;
+		let allowed = first
+			.get("allowed")
+			.and_then(|v| v.as_bool())
+			.ok_or("Missing 'allowed' field in testmempoolaccept response".to_string())?;
+		Ok(TestMempoolAcceptResponse(allowed))
+	}
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct MempoolEntry {
 	/// The transaction id

src/chain/electrum.rs

@@ -18,13 +18,16 @@ use bdk_wallet::{KeychainKind as BdkKeyChainKind, Update as BdkUpdate};
 use bitcoin::{FeeRate, Network, Script, ScriptBuf, Transaction, Txid};
 use electrum_client::{
 	Batch, Client as ElectrumClient, ConfigBuilder as ElectrumConfigBuilder, ElectrumApi,
+	Error as ElectrumError,
 };
 use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
 use lightning_transaction_sync::ElectrumSyncClient;
 
 use super::WalletSyncStatus;
-use crate::config::{Config, ElectrumSyncConfig, BDK_CLIENT_STOP_GAP};
+use crate::config::{
+	Config, ElectrumSyncConfig, BDK_CLIENT_STOP_GAP, DEFAULT_TX_LOOKUP_TIMEOUT_SECS,
+};
 use crate::error::Error;
 use crate::fee_estimator::{
 	apply_post_estimation_adjustments, get_all_conf_targets, get_num_block_defaults_for_target,
@@ -288,6 +291,21 @@ impl ElectrumChainSource {
 			electrum_client.broadcast(tx).await;
 		}
 	}
+
+	pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+		let electrum_client: Arc<ElectrumRuntimeClient> =
+			if let Some(client) = self.electrum_runtime_status.read().unwrap().client().as_ref() {
+				Arc::clone(client)
+			} else {
+				debug_assert!(
+					false,
+					"We should have started the chain source before getting transactions"
+				);
+				return Err(Error::TxLookupTimeout);
+			};
+
+		electrum_client.get_transaction(txid).await
+	}
 }
 
 impl Filter for ElectrumChainSource {
@@ -652,6 +670,37 @@ impl ElectrumRuntimeClient {
 
 		Ok(new_fee_rate_cache)
 	}
+
+	async fn get_transaction(&self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+		let electrum_client = Arc::clone(&self.electrum_client);
+		let txid_copy = *txid;
+
+		let spawn_fut =
+			self.runtime.spawn_blocking(move || electrum_client.transaction_get(&txid_copy));
+		let timeout_fut =
+			tokio::time::timeout(Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS), spawn_fut);
+
+		match timeout_fut.await {
+			Ok(res) => match res {
+				Ok(inner_res) => match inner_res {
+					Ok(tx) => Ok(Some(tx)),
+					Err(ElectrumError::Protocol(_)) => Ok(None),
+					Err(e) => {
+						log_error!(self.logger, "Failed to get transaction {}: {}", txid, e);
+						Err(Error::TxLookupFailed)
+					},
+				},
+				Err(e) => {
+					log_error!(self.logger, "Failed to get transaction {}: {}", txid, e);
+					Err(Error::TxLookupFailed)
+				},
+			},
+			Err(e) => {
+				log_error!(self.logger, "Failed to get transaction {} due to timeout: {}", txid, e);
+				Err(Error::TxLookupTimeout)
+			},
+		}
+	}
 }
 
 impl Filter for ElectrumRuntimeClient {

src/chain/esplora.rs

@@ -422,6 +422,13 @@ impl EsploraChainSource {
 			}
 		}
 	}
+
+	pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+		self.esplora_client.get_tx(txid).await.map_err(|e| {
+			log_error!(self.logger, "Failed to get transaction {}: {}", txid, e);
+			Error::TxLookupFailed
+		})
+	}
 }
 
 impl Filter for EsploraChainSource {