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

@@ -83,6 +83,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 = "679dac50cc0d81ec4d31da94b93d467e5308f16a" }
 
+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

@@ -97,6 +97,8 @@ interface Node {
 	SpontaneousPayment spontaneous_payment();
 	OnchainPayment onchain_payment();
 	UnifiedPayment unified_payment();
+	[Throws=NodeError]
+	PayjoinPayment payjoin_payment();
 	LSPS1Liquidity lsps1_liquidity();
 	[Throws=NodeError]
 	void lnurl_auth(string lnurl);
@@ -165,6 +167,8 @@ interface FeeRate {
 
 typedef interface UnifiedPayment;
 
+typedef interface PayjoinPayment;
+
 typedef interface LSPS1Liquidity;
 
 [Error]
@@ -192,6 +196,8 @@ enum NodeError {
 	"OnchainTxSigningFailed",
 	"TxSyncFailed",
 	"TxSyncTimeout",
+	"TxLookupFailed",
+	"TxLookupTimeout",
 	"GossipUpdateFailed",
 	"GossipUpdateTimeout",
 	"LiquidityRequestFailed",
@@ -229,6 +235,9 @@ enum NodeError {
 	"LnurlAuthFailed",
 	"LnurlAuthTimeout",
 	"InvalidLnurl",
+	"PayjoinNotConfigured",
+	"PayjoinSessionCreationFailed",
+	"PayjoinSessionFailed",
 };
 
 typedef dictionary NodeStatus;

src/builder.rs

@@ -50,7 +50,7 @@ use crate::chain::ChainSource;
 use crate::config::{
 	default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole,
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig,
-	TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
+	PayjoinConfig, TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
 };
 use crate::connection::ConnectionManager;
 use crate::entropy::NodeEntropy;
@@ -64,7 +64,8 @@ use crate::io::utils::{
 };
 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,
 };
@@ -75,13 +76,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, DynStoreRef, DynStoreWrapper,
-	GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore,
-	PeerManager, PendingPaymentStore, SyncAndAsyncKVStore,
+	GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger,
+	PayjoinSessionStore, PaymentStore, PeerManager, PendingPaymentStore, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -200,6 +202,8 @@ pub enum BuildError {
 	AsyncPaymentsConfigMismatch,
 	/// An attempt to setup a DNS Resolver failed.
 	DNSResolverSetupFailed,
+	/// The payjoin configuration requires a Bitcoin Core backend, but a different chain source was configured.
+	PayjoinConfigMismatch,
 }
 
 impl fmt::Display for BuildError {
@@ -237,6 +241,9 @@ impl fmt::Display for BuildError {
 			Self::DNSResolverSetupFailed => {
 				write!(f, "An attempt to setup a DNS resolver has failed.")
 			},
+			Self::PayjoinConfigMismatch => {
+				write!(f, "Payjoin requires a Bitcoin Core chain source, but a different one was configured.")
+			},
 		}
 	}
 }
@@ -615,6 +622,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.
 	///
@@ -1100,6 +1116,14 @@ impl ArcedNodeBuilder {
 		self.inner.write().expect("lock").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.
 	///
@@ -1288,7 +1312,7 @@ 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_all_objects(
@@ -1303,6 +1327,12 @@ fn build_with_store_internal(
 					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 					Arc::clone(&logger_ref),
+				),
+				read_all_objects(
+					&*kv_store_ref,
+					PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE,
+					PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE,
+					Arc::clone(&logger_ref),
 				)
 			)
 		});
@@ -2024,6 +2054,42 @@ fn build_with_store_internal(
 
 	let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());
 
+	let payjoin_manager = if config.payjoin_config.is_some() {
+		if !matches!(chain_data_source_config, Some(ChainDataSourceConfig::Bitcoind { .. })) {
+			return Err(BuildError::PayjoinConfigMismatch);
+		}
+
+		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);
+			},
+		};
+
+		Some(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),
+		)))
+	} else {
+		None
+	};
+
 	#[cfg(cycle_tests)]
 	let mut _leak_checker = crate::LeakChecker(Vec::new());
 	#[cfg(cycle_tests)]
@@ -2071,6 +2137,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,
@@ -610,6 +610,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)]
@@ -1169,6 +1220,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 {
@@ -1324,6 +1403,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