Merge PR 'Use preferred sPK of watched txn in electrum, not rand ones' (#4867)

### lightning-transaction-sync/src/common.rs
@@ -6,18 +6,21 @@
 // accordance with one or both of these licenses.
 
 use bitcoin::block::Header;
-use bitcoin::{BlockHash, OutPoint, Transaction, Txid};
+use bitcoin::{BlockHash, OutPoint, ScriptBuf, Transaction, Txid};
 use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 use lightning::chain::{Confirm, WatchedOutput};
 
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::ops::Deref;
 
 // Represents the current state.
 pub(crate) struct SyncState {
 	// Transactions that were previously processed, but must not be forgotten
-	// yet since they still need to be monitored for confirmation on-chain.
-	pub watched_transactions: HashSet<Txid>,
+	// yet since they still need to be monitored for confirmation on-chain,
+	// mapped to the list of `script_pubkey`s we might use to check for their
+	// confirmation status. Note a watcher may re-register a transaction with a
+	// bogus `script_pubkey` which must not override a previously-registered one.
+	pub watched_transactions: HashMap<Txid, Vec<ScriptBuf>>,
 	// Outputs that were previously processed, but must not be forgotten yet as
 	// as we still need to monitor any spends on-chain.
 	pub watched_outputs: HashMap<OutPoint, WatchedOutput>,
@@ -33,7 +36,7 @@ pub(crate) struct SyncState {
 impl SyncState {
 	pub fn new() -> Self {
 		Self {
-			watched_transactions: HashSet::new(),
+			watched_transactions: HashMap::new(),
 			watched_outputs: HashMap::new(),
 			outputs_spends_pending_threshold_conf: Vec::new(),
 			last_sync_hash: None,
@@ -50,7 +53,7 @@ impl SyncState {
 				c.transaction_unconfirmed(&txid);
 			}
 
-			self.watched_transactions.insert(txid);
+			self.watched_transactions.entry(txid).or_default();
 
 			// If a previously-confirmed output spend is unconfirmed, re-add the watched output to
 			// the tracking map.
@@ -81,12 +84,11 @@ impl SyncState {
 				);
 			}
 
-			self.watched_transactions.remove(&ctx.tx.compute_txid());
+			self.watched_transactions.remove(&ctx.txid);
 
 			for input in &ctx.tx.input {
 				if let Some(output) = self.watched_outputs.remove(&input.previous_output) {
-					let spent =
-						(ctx.tx.compute_txid(), ctx.block_height, input.previous_output, output);
+					let spent = (ctx.txid, ctx.block_height, input.previous_output, output);
 					self.outputs_spends_pending_threshold_conf.push(spent);
 				}
 			}
@@ -101,15 +103,18 @@ impl SyncState {
 
 // A queue that is to be filled by `Filter` and drained during the next syncing round.
 pub(crate) struct FilterQueue {
-	// Transactions that were registered via the `Filter` interface and have to be processed.
-	pub transactions: HashSet<Txid>,
+	// Transactions that were registered via the `Filter` interface and have to be processed,
+	// mapped to the `script_pubkey`s they were registered with. A transaction may be
+	// registered multiple times with different `script_pubkey`s, in which case we keep all of
+	// them, as some may be bogus.
+	pub transactions: HashMap<Txid, Vec<ScriptBuf>>,
 	// Outputs that were registered via the `Filter` interface and have to be processed.
 	pub outputs: HashMap<OutPoint, WatchedOutput>,
 }
 
 impl FilterQueue {
 	pub fn new() -> Self {
-		Self { transactions: HashSet::new(), outputs: HashMap::new() }
+		Self { transactions: HashMap::new(), outputs: HashMap::new() }
 	}
 
 	// Processes the transaction and output queues and adds them to the given [`SyncState`].
@@ -121,7 +126,14 @@ impl FilterQueue {
 		if !self.transactions.is_empty() {
 			pending_registrations = true;
 
-			sync_state.watched_transactions.extend(self.transactions.drain());
+			for (txid, script_pubkeys) in self.transactions.drain() {
+				let watched = sync_state.watched_transactions.entry(txid).or_default();
+				for script_pubkey in script_pubkeys {
+					if !watched.contains(&script_pubkey) {
+						watched.push(script_pubkey);
+					}
+				}
+			}
 		}
 
 		if !self.outputs.is_empty() {

### lightning-transaction-sync/src/electrum.rs
@@ -274,26 +274,36 @@ impl<L: Logger> ElectrumSyncClient<L> {
 		);
 		let mut watched_txs = Vec::with_capacity(sync_state.watched_transactions.len());
 
-		for txid in &sync_state.watched_transactions {
+		for (txid, watch_script_pubkeys) in &sync_state.watched_transactions {
 			match self.client.transaction_get(&txid) {
 				Ok(tx) => {
 					if tx.compute_txid() != *txid {
 						log_error!(self.logger, "Retrieved transaction for txid {} doesn't match expectations. This should not happen. Please verify server integrity.", txid);
 						return Err(InternalError::Failed);
 					}
 
-					// Skip before using an arbitrary returned output to look up the
+					// Skip before potentially using an arbitrary returned output to look up the
 					// transaction's script history.
 					if is_potentially_unsafe_merkle_leaf(&tx) {
 						log_error!(self.logger, "Skipping transaction {} due to retrieving potentially invalid tx data.", txid);
 						continue;
 					}
 
 					watched_txs.push((txid, tx.clone()));
-					if let Some(tx_out) = tx.output.first() {
-						// We watch an arbitrary output of the transaction of interest in order to
-						// retrieve the associated script history, before narrowing down our search
-						// through `filter`ing by `txid` below.
+					// We watch an output's script_pubkey of the transaction of interest in order
+					// to retrieve the associated script history, before narrowing down our search
+					// through `filter`ing by `txid` below. Prefer a script_pubkey the transaction
+					// was registered with that actually appears in the transaction (as registered
+					// script_pubkeys may be bogus), and otherwise pick an arbitrary one, noting
+					// that electrum servers may not index OP_RETURN script_pubkeys at all.
+					let candidate_outputs =
+						tx.output.iter().filter(|output| !output.script_pubkey.is_op_return());
+					let watch_output = candidate_outputs
+						.clone()
+						.find(|output| watch_script_pubkeys.contains(&output.script_pubkey))
+						.or_else(|| candidate_outputs.clone().next())
+						.or_else(|| tx.output.iter().next());
+					if let Some(tx_out) = watch_output {
 						watched_script_pubkeys.push(tx_out.script_pubkey.clone());
 					} else {
 						debug_assert!(false, "Failed due to retrieving invalid tx data.");
@@ -529,9 +539,12 @@ impl<L: Logger> ElectrumSyncClient<L> {
 }
 
 impl<L: Logger> Filter for ElectrumSyncClient<L> {
-	fn register_tx(&self, txid: &Txid, _script_pubkey: &Script) {
+	fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
 		let mut locked_queue = self.queue.lock().unwrap();
-		locked_queue.transactions.insert(*txid);
+		let script_pubkeys = locked_queue.transactions.entry(*txid).or_default();
+		if script_pubkeys.iter().all(|spk| spk != script_pubkey) {
+			script_pubkeys.push(script_pubkey.to_owned());
+		}
 	}
 
 	fn register_output(&self, output: WatchedOutput) {

### lightning-transaction-sync/src/esplora.rs
@@ -298,7 +298,7 @@ impl<L: Logger> EsploraSyncClient<L> {
 
 		let mut confirmed_txs: Vec<ConfirmedTx> = Vec::new();
 
-		for txid in &sync_state.watched_transactions {
+		for txid in sync_state.watched_transactions.keys() {
 			if confirmed_txs.iter().any(|ctx| ctx.txid == *txid) {
 				continue;
 			}
@@ -478,9 +478,12 @@ type EsploraClientType = AsyncClient;
 type EsploraClientType = BlockingClient;
 
 impl<L: Logger> Filter for EsploraSyncClient<L> {
-	fn register_tx(&self, txid: &Txid, _script_pubkey: &Script) {
+	fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
 		let mut locked_queue = self.queue.lock().unwrap();
-		locked_queue.transactions.insert(*txid);
+		let script_pubkeys = locked_queue.transactions.entry(*txid).or_default();
+		if script_pubkeys.iter().all(|spk| spk != script_pubkey) {
+			script_pubkeys.push(script_pubkey.to_owned());
+		}
 	}
 
 	fn register_output(&self, output: WatchedOutput) {