Function tx_cache_clearing_loop

fendermint/eth/api/src/mpool.rs:98–149

Subscribe to notifications about new blocks and 1) remove all included transactions from the caches 2) broadcast buffered out-of-order transactions when they are unblocked Re-subscribe in the event of a subscription failure.

(
    client: C,
    chain_id: ChainID,
    tx_cache: TransactionCache,
    tx_buffer: TransactionBuffer,
)

///
/// Re-subscribe in the event of a subscription failure.
async fn tx_cache_clearing_loop<C>(
    client: C,
    chain_id: ChainID,
    tx_cache: TransactionCache,
    tx_buffer: TransactionBuffer,
) where
    C: Client + SubscriptionClient + Send + Sync,
{
    loop {
        let query = Query::from(EventType::NewBlock);

        match client.subscribe(query).await {
            Err(e) => {
                tracing::warn!(error=?e, "failed to subscribe to NewBlocks; retrying later...");
                tokio::time::sleep(Duration::from_secs(RETRY_SLEEP_SECS)).await;
            }
            Ok(mut subscription) => {
                while let Some(result) = subscription.next().await {
                    match result {
                        Err(e) => {
                            tracing::warn!(error=?e, "NewBlocks subscription failed; resubscribing...");
                            break;
                        }
                        Ok(event) => {
                            if let EventData::NewBlock {
                                block: Some(block), ..
                            } = event.data
                            {
                                let txs = collect_txs(&block, &chain_id);

                                if txs.is_empty() {
                                    continue;
                                }

                                let tx_hashes = txs.iter().map(|(h, _, _)| h);
                                let tx_nonces = || txs.iter().map(|(_, s, n)| (s, *n));

                                tx_cache.remove_many(tx_hashes);
                                // First remove all transactions which have been in the block (could be multiple from the same sender).
                                tx_buffer.remove_many(tx_nonces());
                                // Then collect whatever is unblocked on top of those, ie. anything that hasn't been included, but now can.
                                let unblocked_msgs = tx_buffer.remove_unblocked(tx_nonces());
                                // Send them all with best-effort.
                                send_msgs(&client, unblocked_msgs).await;
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Collect the identifiers of the transactions in the block.
fn collect_txs(block: &Block, chain_id: &ChainID) -> Vec<(et::TxHash, Address, Nonce)> {