Fetch blocks from source in parallel during initial sync

diff --git a/lightning-block-sync/src/init.rs b/lightning-block-sync/src/init.rs
index 3f5abf9..cedd11e 100644
--- a/lightning-block-sync/src/init.rs
+++ b/lightning-block-sync/src/init.rs
@@ -1,8 +1,9 @@
 //! Utilities to assist in the initial sync required to initialize or reload Rust-Lightning objects
 //! from disk.
 
-use crate::poll::{ChainPoller, Validate, ValidatedBlockHeader};
-use crate::{BlockSource, BlockSourceResult, ChainNotifier, HeaderCache};
+use crate::async_poll::{MultiResultFuturePoller, ResultFuture};
+use crate::poll::{ChainPoller, Poll, Validate, ValidatedBlockHeader};
+use crate::{BlockData, BlockSource, BlockSourceResult, ChainNotifier, HeaderCache};
 
 use bitcoin::block::Header;
 use bitcoin::network::Network;
@@ -149,7 +150,6 @@ where
 	// Find differences and disconnect blocks for each listener individually.
 	let mut chain_poller = ChainPoller::new(block_source, network);
 	let mut chain_listeners_at_height = Vec::new();
-	let mut most_common_ancestor = None;
 	let mut most_connected_blocks = Vec::new();
 	let mut header_cache = HeaderCache::new();
 	for (old_best_block, chain_listener) in chain_listeners.drain(..) {
@@ -170,19 +170,59 @@ where
 		// Keep track of the most common ancestor and all blocks connected across all listeners.
 		chain_listeners_at_height.push((common_ancestor.height, chain_listener));
 		if connected_blocks.len() > most_connected_blocks.len() {
-			most_common_ancestor = Some(common_ancestor);
 			most_connected_blocks = connected_blocks;
 		}
 	}
 
-	// Connect new blocks for all listeners at once to avoid re-fetching blocks.
-	if let Some(common_ancestor) = most_common_ancestor {
-		let chain_listener = &ChainListenerSet(chain_listeners_at_height);
-		let mut chain_notifier = ChainNotifier { header_cache: &mut header_cache, chain_listener };
-		chain_notifier
-			.connect_blocks(common_ancestor, most_connected_blocks, &mut chain_poller)
-			.await
-			.map_err(|(e, _)| e)?;
+	while !most_connected_blocks.is_empty() {
+		#[cfg(not(test))]
+		const MAX_BLOCKS_AT_ONCE: usize = 6 * 6; // Six hours of blocks, 144MiB encoded
+		#[cfg(test)]
+		const MAX_BLOCKS_AT_ONCE: usize = 2;
+
+		let mut fetch_block_futures =
+			Vec::with_capacity(core::cmp::min(MAX_BLOCKS_AT_ONCE, most_connected_blocks.len()));
+		for header in most_connected_blocks.iter().rev().take(MAX_BLOCKS_AT_ONCE) {
+			let fetch_future = chain_poller.fetch_block(header);
+			fetch_block_futures
+				.push(ResultFuture::Pending(Box::pin(async move { (header, fetch_future.await) })));
+		}
+		let results = MultiResultFuturePoller::new(fetch_block_futures).await.into_iter();
+
+		const NO_BLOCK: Option<(u32, crate::poll::ValidatedBlock)> = None;
+		let mut fetched_blocks = [NO_BLOCK; MAX_BLOCKS_AT_ONCE];
+		for ((header, block_res), result) in results.into_iter().zip(fetched_blocks.iter_mut()) {
+			*result = Some((header.height, block_res?));
+		}
+		debug_assert!(fetched_blocks.iter().take(most_connected_blocks.len()).all(|r| r.is_some()));
+		// TODO: When our MSRV is 1.82, use is_sorted_by_key
+		debug_assert!(fetched_blocks.windows(2).all(|blocks| {
+			if let (Some(a), Some(b)) = (&blocks[0], &blocks[1]) {
+				a.0 < b.0
+			} else {
+				// Any non-None blocks have to come before any None entries
+				blocks[1].is_none()
+			}
+		}));
+
+		for (listener_height, listener) in chain_listeners_at_height.iter() {
+			// Connect blocks for this listener.
+			for (height, block_data) in fetched_blocks.iter().flatten() {
+				if *height > *listener_height {
+					match &**block_data {
+						BlockData::FullBlock(block) => {
+							listener.block_connected(&block, *height);
+						},
+						BlockData::HeaderOnly(header_data) => {
+							listener.filtered_block_connected(&header_data, &[], *height);
+						},
+					}
+				}
+			}
+		}
+
+		most_connected_blocks
+			.truncate(most_connected_blocks.len().saturating_sub(MAX_BLOCKS_AT_ONCE));
 	}
 
 	Ok((header_cache, best_header))
@@ -203,33 +243,6 @@ impl<'a, L: chain::Listen + ?Sized> chain::Listen for DynamicChainListener<'a, L
 	}
 }
 
-/// A set of dynamically sized chain listeners, each paired with a starting block height.
-struct ChainListenerSet<'a, L: chain::Listen + ?Sized>(Vec<(u32, &'a L)>);
-
-impl<'a, L: chain::Listen + ?Sized> chain::Listen for ChainListenerSet<'a, L> {
-	fn block_connected(&self, block: &bitcoin::Block, height: u32) {
-		for (starting_height, chain_listener) in self.0.iter() {
-			if height > *starting_height {
-				chain_listener.block_connected(block, height);
-			}
-		}
-	}
-
-	fn filtered_block_connected(
-		&self, header: &Header, txdata: &chain::transaction::TransactionData, height: u32,
-	) {
-		for (starting_height, chain_listener) in self.0.iter() {
-			if height > *starting_height {
-				chain_listener.filtered_block_connected(header, txdata, height);
-			}
-		}
-	}
-
-	fn blocks_disconnected(&self, _fork_point: BestBlock) {
-		unreachable!()
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;