refactor(chain): use single queue for anchored_txs in canonicalization

- Unify both `unprocessed_anchored_txs` and `pending_anchored_txs` in a
  single `unprocessed_anchored_txs` queue.
- Changes the `unprocessed_anchored_txs from `Iterator` to `VecDeque`.
- Removes the `pending_anchored_txs` field and it's usage.
- Collects all `anchored_txs` upfront instead of lazy iteration.

Author: oleonardolima

crates/chain/src/canonical_task.rs

@@ -27,16 +27,13 @@ pub struct CanonicalizationTask<'g, A> {
     chain_tip: BlockId,
 
     unprocessed_assumed_txs: Box<dyn Iterator<Item = (Txid, Arc<Transaction>)> + 'g>,
-    unprocessed_anchored_txs:
-        Box<dyn Iterator<Item = (Txid, Arc<Transaction>, &'g BTreeSet<A>)> + 'g>,
+    unprocessed_anchored_txs: VecDeque<(Txid, Arc<Transaction>, &'g BTreeSet<A>)>,
     unprocessed_seen_txs: Box<dyn Iterator<Item = (Txid, Arc<Transaction>, u64)> + 'g>,
     unprocessed_leftover_txs: VecDeque<(Txid, Arc<Transaction>, u32)>,
 
     canonical: CanonicalMap<A>,
     not_canonical: NotCanonicalSet,
 
-    pending_anchor_checks: VecDeque<(Txid, Arc<Transaction>, Vec<A>)>,
-
     // Store canonical transactions in order
     canonical_order: Vec<Txid>,
 
@@ -48,22 +45,19 @@ impl<'g, A: Anchor> ChainQuery for CanonicalizationTask<'g, A> {
     type Output = CanonicalView<A>;
 
     fn next_query(&mut self) -> Option<ChainRequest> {
-        // Check if we have pending anchor checks
-        if let Some((_, _, anchors)) = self.pending_anchor_checks.front() {
-            // Convert anchors to BlockIds for the ChainRequest
+        // Get the next unprocessed anchored tx that needs to query a chain oracle.
+        if let Some((_txid, _tx, anchors)) = self.unprocessed_anchored_txs.front() {
             let block_ids = anchors.iter().map(|anchor| anchor.anchor_block()).collect();
             return Some(ChainRequest {
                 chain_tip: self.chain_tip,
                 block_ids,
             });
         }
-
-        // Process more anchored transactions if available
-        self.process_anchored_txs()
+        None
     }
 
     fn resolve_query(&mut self, response: ChainResponse) {
-        if let Some((txid, tx, anchors)) = self.pending_anchor_checks.pop_front() {
+        if let Some((txid, tx, anchors)) = self.unprocessed_anchored_txs.pop_front() {
             // Find the anchor that matches the confirmed BlockId
             let best_anchor = response.and_then(|block_id| {
                 anchors
@@ -97,7 +91,7 @@ impl<'g, A: Anchor> ChainQuery for CanonicalizationTask<'g, A> {
     }
 
     fn is_finished(&mut self) -> bool {
-        self.pending_anchor_checks.is_empty() && self.unprocessed_anchored_txs.size_hint().0 == 0
+        self.unprocessed_anchored_txs.is_empty()
     }
 
     fn finish(mut self) -> Self::Output {
@@ -199,11 +193,10 @@ impl<'g, A: Anchor> CanonicalizationTask<'g, A> {
                 .rev()
                 .filter_map(|txid| Some((txid, tx_graph.get_tx(txid)?))),
         );
-        let unprocessed_anchored_txs = Box::new(
-            tx_graph
-                .txids_by_descending_anchor_height()
-                .filter_map(|(_, txid)| Some((txid, tx_graph.get_tx(txid)?, anchors.get(&txid)?))),
-        );
+        let unprocessed_anchored_txs: VecDeque<_> = tx_graph
+            .txids_by_descending_anchor_height()
+            .filter_map(|(_, txid)| Some((txid, tx_graph.get_tx(txid)?, anchors.get(&txid)?)))
+            .collect();
         let unprocessed_seen_txs = Box::new(
             tx_graph
                 .txids_by_descending_last_seen()
@@ -222,8 +215,6 @@ impl<'g, A: Anchor> CanonicalizationTask<'g, A> {
             canonical: HashMap::new(),
             not_canonical: HashSet::new(),
 
-            pending_anchor_checks: VecDeque::new(),
-
             canonical_order: Vec::new(),
             confirmed_anchors: HashMap::new(),
         };
@@ -246,17 +237,6 @@ impl<'g, A: Anchor> CanonicalizationTask<'g, A> {
         }
     }
 
-    fn process_anchored_txs(&mut self) -> Option<ChainRequest> {
-        while let Some((txid, tx, anchors)) = self.unprocessed_anchored_txs.next() {
-            if !self.is_canonicalized(txid) {
-                self.pending_anchor_checks
-                    .push_back((txid, tx, anchors.iter().cloned().collect()));
-                return self.next_query();
-            }
-        }
-        None
-    }
-
     fn process_seen_txs(&mut self) {
         while let Some((txid, tx, last_seen)) = self.unprocessed_seen_txs.next() {
             debug_assert!(
@@ -372,11 +352,8 @@ impl<'g, A: Anchor> CanonicalizationTask<'g, A> {
                     if let Some(anchors) = self.tx_graph.all_anchors().get(txid) {
                         // only check anchors we haven't already confirmed
                         if !self.confirmed_anchors.contains_key(txid) {
-                            self.pending_anchor_checks.push_back((
-                                *txid,
-                                tx.clone(),
-                                anchors.iter().cloned().collect(),
-                            ));
+                            self.unprocessed_anchored_txs
+                                .push_back((*txid, tx.clone(), anchors));
                         }
                     }
                 }