Merge pull request #2966 from G8XSU/2647-distribute

Optimize ChannelMonitor persistence on block connections.

diff --combined lightning/src/chain/channelmonitor.rs
index a037a965a8fbaac1f5419ed5acc41bb83617a594,c6d1bcdd2158025fa11984aa32f5207380d459cd..0e87f3569e605d0c25736a9d203bc1fe6170655f
--- 1/lightning/src/chain/channelmonitor.rs
--- 2/lightning/src/chain/channelmonitor.rs
+++ b/lightning/src/chain/channelmonitor.rs
@@@ -1812,6 -1812,12 +1812,12 @@@ impl<Signer: EcdsaChannelSigner> Channe
                );
        }
  
+       /// Returns true if the monitor has pending claim requests that are not fully confirmed yet.
+       pub fn has_pending_claims(&self) -> bool
+       {
+               self.inner.lock().unwrap().onchain_tx_handler.has_pending_claims()
+       }
+ 
        /// Triggers rebroadcasts of pending claims from a force-closed channel after a transaction
        /// signature generation failure.
        pub fn signer_unblocked<B: Deref, F: Deref, L: Deref>(
@@@ -1924,9 -1930,9 +1930,9 @@@
        }
  
        #[cfg(test)]
 -      pub fn do_signer_call<F: FnMut(&Signer) -> ()>(&self, mut f: F) {
 -              let inner = self.inner.lock().unwrap();
 -              f(&inner.onchain_tx_handler.signer);
 +      pub fn do_mut_signer_call<F: FnMut(&mut Signer) -> ()>(&self, mut f: F) {
 +              let mut inner = self.inner.lock().unwrap();
 +              f(&mut inner.onchain_tx_handler.signer);
        }
  }
  

diff --combined lightning/src/util/test_utils.rs
index 2a11a91f29480675ea585badb8ed40068de28658,2d00844772aca60eb841b2eb5bde57f7dcae8556..4d17f8fd17ddcd9a919acb4a4abddd9aac673c17
--- 1/lightning/src/util/test_utils.rs
--- 2/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@@ -79,8 -79,6 +79,8 @@@ use std::time::{SystemTime, UNIX_EPOCH}
  use bitcoin::psbt::Psbt;
  use bitcoin::Sequence;
  
 +use super::test_channel_signer::SignerOp;
 +
  pub fn pubkey(byte: u8) -> PublicKey {
        let secp_ctx = Secp256k1::new();
        PublicKey::from_secret_key(&secp_ctx, &privkey(byte))
@@@ -547,12 -545,16 +547,16 @@@ pub struct TestPersister 
        ///
        /// [`ChannelMonitor`]: channelmonitor::ChannelMonitor
        pub offchain_monitor_updates: Mutex<HashMap<OutPoint, HashSet<u64>>>,
+       /// When we get an update_persisted_channel call with no ChannelMonitorUpdate, we insert the
+       /// monitor's funding outpoint here.
+       pub chain_sync_monitor_persistences: Mutex<VecDeque<OutPoint>>
  }
  impl TestPersister {
        pub fn new() -> Self {
                Self {
                        update_rets: Mutex::new(VecDeque::new()),
                        offchain_monitor_updates: Mutex::new(new_hash_map()),
+                       chain_sync_monitor_persistences: Mutex::new(VecDeque::new())
                }
        }
  
@@@ -575,15 -577,18 +579,18 @@@ impl<Signer: sign::ecdsa::EcdsaChannelS
                        ret = update_ret;
                }
  
-               if let Some(update) = update  {
+               if let Some(update) = update {
                        self.offchain_monitor_updates.lock().unwrap().entry(funding_txo).or_insert(new_hash_set()).insert(update.update_id);
+               } else {
+                       self.chain_sync_monitor_persistences.lock().unwrap().push_back(funding_txo);
                }
                ret
        }
  
        fn archive_persisted_channel(&self, funding_txo: OutPoint) {
-               // remove the channel from the offchain_monitor_updates map
+               // remove the channel from the offchain_monitor_updates and chain_sync_monitor_persistences.
                self.offchain_monitor_updates.lock().unwrap().remove(&funding_txo);
+               self.chain_sync_monitor_persistences.lock().unwrap().retain(|x| x != &funding_txo);
        }
  }
  
@@@ -1217,7 -1222,6 +1224,7 @@@ pub struct TestKeysInterface 
        enforcement_states: Mutex<HashMap<[u8;32], Arc<Mutex<EnforcementState>>>>,
        expectations: Mutex<Option<VecDeque<OnGetShutdownScriptpubkey>>>,
        pub unavailable_signers: Mutex<HashSet<[u8; 32]>>,
 +      pub unavailable_signers_ops: Mutex<HashMap<[u8; 32], HashSet<SignerOp>>>,
  }
  
  impl EntropySource for TestKeysInterface {
@@@ -1276,11 -1280,9 +1283,11 @@@ impl SignerProvider for TestKeysInterfa
        fn derive_channel_signer(&self, channel_value_satoshis: u64, channel_keys_id: [u8; 32]) -> TestChannelSigner {
                let keys = self.backing.derive_channel_signer(channel_value_satoshis, channel_keys_id);
                let state = self.make_enforcement_state_cell(keys.commitment_seed);
 -              let signer = TestChannelSigner::new_with_revoked(keys, state, self.disable_revocation_policy_check);
 -              if self.unavailable_signers.lock().unwrap().contains(&channel_keys_id) {
 -                      signer.set_available(false);
 +              let mut signer = TestChannelSigner::new_with_revoked(keys, state, self.disable_revocation_policy_check);
 +              if let Some(ops) = self.unavailable_signers_ops.lock().unwrap().get(&channel_keys_id) {
 +                      for &op in ops {
 +                              signer.disable_op(op);
 +                      }
                }
                signer
        }
@@@ -1321,7 -1323,6 +1328,7 @@@ impl TestKeysInterface 
                        enforcement_states: Mutex::new(new_hash_map()),
                        expectations: Mutex::new(None),
                        unavailable_signers: Mutex::new(new_hash_set()),
 +                      unavailable_signers_ops: Mutex::new(new_hash_map()),
                }
        }