fix: deduplicate multi-wallet BDK events and correct onchain payment direction

When a transaction touches multiple address-type wallets, each wallet
independently emits a BdkWalletEvent, causing duplicate onchain events
per sync cycle. Add per-type HashSet deduplication in process_wallet_events
so each txid produces at most one event of each type per sync.

Also fix PaymentDetails::update to apply direction corrections as secondary
wallets sync. When the primary wallet sees only a change output it records
the payment as Inbound; once secondary wallets reveal the spent inputs the
direction must update to Outbound. Add a never-downgrade guard so a
change-only wallet syncing before an input-holding wallet cannot transiently
flip a correct Outbound back to Inbound.

Add unit tests for the direction logic and integration tests covering event
deduplication, direction correction, send_all direction, secondary wallet
receives, and reorg deduplication.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ben-kaufman

src/chain/mod.rs

@@ -227,6 +227,8 @@ pub(super) fn apply_additional_sync_results(
 }
 
 // Process BDK wallet events and emit corresponding ldk-node events via the event queue.
+// When a transaction touches multiple address-type wallets, each wallet emits its own
+// BdkWalletEvent, so we deduplicate by txid before forwarding to the event queue.
 async fn process_wallet_events<L2: Deref>(
 	wallet_events: Vec<BdkWalletEvent>, wallet: &crate::wallet::Wallet,
 	event_queue: &EventQueue<L2>, logger: &Arc<Logger>,
@@ -235,9 +237,19 @@ async fn process_wallet_events<L2: Deref>(
 where
 	L2::Target: LdkLogger,
 {
+	// Use per-type sets so that two wallets with different prior state can each contribute
+	// their event type for the same txid without suppressing the other.
+	let mut seen_received_txids = std::collections::HashSet::new();
+	let mut seen_confirmed_txids = std::collections::HashSet::new();
+	let mut seen_reorged_txids = std::collections::HashSet::new();
+	let mut seen_replaced_txids = std::collections::HashSet::new();
+
 	for wallet_event in wallet_events {
 		match wallet_event {
 			BdkWalletEvent::TxConfirmed { txid, block_time, .. } => {
+				if !seen_confirmed_txids.insert(txid) {
+					continue;
+				}
 				let details = get_transaction_details(&txid, wallet, channel_manager)
 					.unwrap_or_else(|| {
 						log_error!(logger, "Transaction {} not found in wallet", txid);
@@ -270,6 +282,9 @@ where
 			BdkWalletEvent::TxUnconfirmed { txid, old_block_time, .. } => {
 				match old_block_time {
 					Some(_) => {
+						if !seen_reorged_txids.insert(txid) {
+							continue;
+						}
 						// Transaction was previously confirmed but is now unconfirmed (reorg)
 						log_info!(
 							logger,
@@ -283,6 +298,9 @@ where
 						})?;
 					},
 					None => {
+						if !seen_received_txids.insert(txid) {
+							continue;
+						}
 						// New unconfirmed transaction detected in mempool
 						let details = get_transaction_details(&txid, wallet, channel_manager)
 							.unwrap_or_else(|| {
@@ -321,6 +339,9 @@ where
 				// We don't emit an event for chain tip changes as this is too noisy
 			},
 			BdkWalletEvent::TxReplaced { txid, conflicts, .. } => {
+				if !seen_replaced_txids.insert(txid) {
+					continue;
+				}
 				let conflict_txids: Vec<Txid> =
 					conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
 				log_info!(
@@ -336,7 +357,7 @@ where
 				})?;
 			},
 			_ => {
-				// Ignore other event types
+				// TxDropped is handled via check_and_emit_evicted_transactions; skip here.
 			},
 		}
 	}

src/payment/store.rs

@@ -293,6 +293,16 @@ impl StorableObject for PaymentDetails {
 			}
 		}
 
+		// Direction can change for onchain payments as wallets sync; never downgrade from
+		// Outbound since a pure receive always has sent=0 and can never compute as Outbound.
+		if let Some(direction) = update.direction {
+			let downgrade = self.direction == PaymentDirection::Outbound
+				&& direction == PaymentDirection::Inbound;
+			if !downgrade {
+				update_if_necessary!(self.direction, direction);
+			}
+		}
+
 		if updated {
 			self.latest_update_timestamp = SystemTime::now()
 				.duration_since(UNIX_EPOCH)
@@ -619,6 +629,7 @@ impl StorableObjectUpdate<PaymentDetails> for PaymentDetailsUpdate {
 
 #[cfg(test)]
 mod tests {
+	use bitcoin::hashes::Hash;
 	use bitcoin::io::Cursor;
 	use lightning::util::ser::Readable;
 
@@ -790,4 +801,65 @@ mod tests {
 			}
 		}
 	}
+
+	#[test]
+	fn onchain_direction_never_downgrades_from_outbound() {
+		// A change-only wallet syncing before an input-holding wallet can temporarily
+		// make received > partial-sent, producing a spurious Inbound update. The guard
+		// must reject it and leave direction as Outbound throughout.
+		let txid = Txid::from_slice(&[1u8; 32]).unwrap();
+		let id = PaymentId(txid.to_byte_array());
+		let kind = PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed };
+
+		// Primary wallet has inputs, so initial direction is Outbound.
+		let mut payment = PaymentDetails::new(
+			id,
+			kind.clone(),
+			Some(5_000 * 1000),
+			Some(100 * 1000),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+		assert_eq!(payment.direction, PaymentDirection::Outbound);
+
+		// Change-only secondary syncs: partial view would produce Inbound; guard blocks it.
+		let mut update = PaymentDetailsUpdate::new(id);
+		update.direction = Some(PaymentDirection::Inbound);
+		update.amount_msat = Some(Some(8_000 * 1000));
+		payment.update(&update);
+		assert_eq!(payment.direction, PaymentDirection::Outbound);
+
+		// Input secondary syncs: full picture, Outbound is confirmed.
+		let mut update = PaymentDetailsUpdate::new(id);
+		update.direction = Some(PaymentDirection::Outbound);
+		update.amount_msat = Some(Some(1_000 * 1000));
+		payment.update(&update);
+		assert_eq!(payment.direction, PaymentDirection::Outbound);
+	}
+
+	#[test]
+	fn onchain_direction_corrects_inbound_to_outbound() {
+		// Primary wallet sees only the change output initially (Inbound); once a
+		// secondary wallet syncs and reveals the spent inputs, direction must update.
+		let txid = Txid::from_slice(&[2u8; 32]).unwrap();
+		let id = PaymentId(txid.to_byte_array());
+		let kind = PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed };
+
+		let mut payment = PaymentDetails::new(
+			id,
+			kind.clone(),
+			Some(9_000 * 1000),
+			Some(0),
+			PaymentDirection::Inbound,
+			PaymentStatus::Pending,
+		);
+		assert_eq!(payment.direction, PaymentDirection::Inbound);
+
+		// Secondary wallet syncs with full view; direction must be corrected.
+		let mut update = PaymentDetailsUpdate::new(id);
+		update.direction = Some(PaymentDirection::Outbound);
+		update.amount_msat = Some(Some(1_000 * 1000));
+		payment.update(&update);
+		assert_eq!(payment.direction, PaymentDirection::Outbound);
+	}
 }