Replace specialized firehose_triggers_in_block by direct usage of TriggerAdapter::triggers_in_blocks (#3082)

With the revamped `FirehoseBlockStream`, it's now possible to directly re-use the `TriggerAdapter::triggers_in_blocks` code instead of relying on our specialized `firehose_triggers_in_block`. This make the code more re-usable.

Eliminated at the same time a useless close in `NearChair::TriggersAdapter`. Also make some `NearChain` now panics instead of silently returning something that is not good.

Linked to #3080

Co-authored-by: Matthieu Vachon <matt@streamingfast.io>

Author: Matthieu Vachon

chain/ethereum/src/chain.rs

@@ -181,8 +181,6 @@ impl Chain {
         start_blocks: Vec<BlockNumber>,
         adapter: Arc<TriggersAdapter>,
         filter: Arc<TriggerFilter>,
-        _metrics: Arc<BlockStreamMetrics>,
-        _unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let firehose_endpoint = match self.firehose_endpoints.random() {
             Some(e) => e.clone(),
@@ -303,15 +301,8 @@ impl Blockchain for Chain {
             ));
 
         if self.firehose_endpoints.len() > 0 {
-            self.new_firehose_block_stream(
-                deployment,
-                start_blocks,
-                adapter,
-                filter,
-                metrics,
-                unified_api_version,
-            )
-            .await
+            self.new_firehose_block_stream(deployment, start_blocks, adapter, filter)
+                .await
         } else {
             self.new_polling_block_stream(
                 deployment,
@@ -579,12 +570,13 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
 
 pub struct FirehoseMapper {}
 
+#[async_trait]
 impl FirehoseMapperTrait<Chain> for FirehoseMapper {
-    fn to_block_stream_event(
+    async fn to_block_stream_event(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         response: &bstream::BlockResponseV2,
-        _adapter: &TriggersAdapter,
+        adapter: &TriggersAdapter,
         filter: &TriggerFilter,
     ) -> Result<BlockStreamEvent<Chain>, FirehoseError> {
         let step = bstream::ForkStep::from_i32(response.step).unwrap_or_else(|| {
@@ -610,8 +602,9 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
         match step {
             bstream::ForkStep::StepNew => {
                 let ethereum_block: EthereumBlockWithCalls = (&block).into();
-                let block_with_triggers =
-                    self.firehose_triggers_in_block(ethereum_block, filter)?;
+                let block_with_triggers = adapter
+                    .triggers_in_block(logger, BlockFinality::NonFinal(ethereum_block), filter)
+                    .await?;
 
                 Ok(BlockStreamEvent::ProcessBlock(
                     block_with_triggers,
@@ -642,29 +635,6 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
     }
 }
 
-impl FirehoseMapper {
-    // FIXME: This should be replaced by using the `TriggersAdapter` struct directly. However, the TriggersAdapter trait
-    //        is async. It's actual async usage is done inside a manual `poll` implementation in `firehose_block_stream#poll_next`
-    //        value. An upcoming improvement will be to remove this `poll_next`. Once the refactor occurs, this should be
-    //        removed and TriggersAdapter::triggers_in_block should be use straight.
-    fn firehose_triggers_in_block(
-        &self,
-        block: EthereumBlockWithCalls,
-        filter: &TriggerFilter,
-    ) -> Result<BlockWithTriggers<Chain>, FirehoseError> {
-        let mut triggers = Vec::new();
-
-        triggers.append(&mut parse_log_triggers(&filter.log, &block.ethereum_block));
-        triggers.append(&mut parse_call_triggers(&filter.call, &block)?);
-        triggers.append(&mut parse_block_triggers(filter.block.clone(), &block));
-
-        Ok(BlockWithTriggers::new(
-            BlockFinality::NonFinal(block),
-            triggers,
-        ))
-    }
-}
-
 pub struct IngestorAdapter {
     logger: Logger,
     ancestor_count: i32,

chain/near/src/chain.rs

@@ -199,33 +199,69 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         _to: BlockNumber,
         _filter: &TriggerFilter,
     ) -> Result<Vec<BlockWithTriggers<Chain>>, Error> {
-        // FIXME (NEAR): Scanning triggers makes little sense in Firehose approach, let's see
-        Ok(vec![])
+        panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
     async fn triggers_in_block(
         &self,
         _logger: &Logger,
-        _block: codec::Block,
+        block: codec::Block,
         _filter: &TriggerFilter,
     ) -> Result<BlockWithTriggers<Chain>, Error> {
-        // FIXME (NEAR): Share implementation with FirehoseMapper::firehose_triggers_in_block version.
-        // This is currently unreachable since Near does not yet support dynamic data sources.
-        todo!()
+        // TODO: Find the best place to introduce an `Arc` and avoid this clone.
+        let shared_block = Arc::new(block.clone());
+
+        // Filter non-successful or non-action receipts.
+        let receipts = block.shards.iter().flat_map(|shard| {
+            shard
+                .receipt_execution_outcomes
+                .iter()
+                .filter_map(|outcome| {
+                    if !outcome
+                        .execution_outcome
+                        .as_ref()?
+                        .outcome
+                        .as_ref()?
+                        .status
+                        .as_ref()?
+                        .is_success()
+                    {
+                        return None;
+                    }
+                    if !matches!(
+                        outcome.receipt.as_ref()?.receipt,
+                        Some(codec::receipt::Receipt::Action(_))
+                    ) {
+                        return None;
+                    }
+
+                    Some(trigger::ReceiptWithOutcome {
+                        outcome: outcome.execution_outcome.as_ref()?.clone(),
+                        receipt: outcome.receipt.as_ref()?.clone(),
+                        block: shared_block.cheap_clone(),
+                    })
+                })
+        });
+
+        let mut trigger_data: Vec<_> = receipts
+            .map(|r| NearTrigger::Receipt(Arc::new(r)))
+            .collect();
+
+        trigger_data.push(NearTrigger::Block(shared_block.cheap_clone()));
+
+        Ok(BlockWithTriggers::new(block, trigger_data))
     }
 
     async fn is_on_main_chain(&self, _ptr: BlockPtr) -> Result<bool, Error> {
-        // FIXME (NEAR): Might not be necessary for NEAR support for now
-        Ok(true)
+        panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
     fn ancestor_block(
         &self,
         _ptr: BlockPtr,
         _offset: BlockNumber,
     ) -> Result<Option<codec::Block>, Error> {
-        // FIXME (NEAR):  Might not be necessary for NEAR support for now
-        Ok(None)
+        panic!("Should never be called since FirehoseBlockStream cannot resolve it")
     }
 
     /// Panics if `block` is genesis.
@@ -241,12 +277,13 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
 
 pub struct FirehoseMapper {}
 
+#[async_trait]
 impl FirehoseMapperTrait<Chain> for FirehoseMapper {
-    fn to_block_stream_event(
+    async fn to_block_stream_event(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         response: &bstream::BlockResponseV2,
-        _adapter: &TriggersAdapter,
+        adapter: &TriggersAdapter,
         filter: &TriggerFilter,
     ) -> Result<BlockStreamEvent<Chain>, FirehoseError> {
         let step = bstream::ForkStep::from_i32(response.step).unwrap_or_else(|| {
@@ -271,7 +308,7 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 
         match step {
             bstream::ForkStep::StepNew => Ok(BlockStreamEvent::ProcessBlock(
-                self.firehose_triggers_in_block(&block, filter)?,
+                adapter.triggers_in_block(logger, block, filter).await?,
                 Some(response.cursor.clone()),
             )),
 
@@ -299,62 +336,6 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
     }
 }
 
-impl FirehoseMapper {
-    // FIXME: This should be replaced by using the `TriggersAdapter` struct directly. However, the TriggersAdapter trait
-    //        is async. It's actual async usage is done inside a manual `poll` implementation in `firehose_block_stream#poll_next`
-    //        value. An upcoming improvement will be to remove this `poll_next`. Once the refactor occurs, this should be
-    //        removed and TriggersAdapter::triggers_in_block should be use straight.
-    fn firehose_triggers_in_block(
-        &self,
-        block: &codec::Block,
-        _filter: &TriggerFilter,
-    ) -> Result<BlockWithTriggers<Chain>, FirehoseError> {
-        // TODO: Find the best place to introduce an `Arc` and avoid this clone.
-        let block = Arc::new(block.clone());
-
-        // Filter non-successful or non-action receipts.
-        let receipts = block.shards.iter().flat_map(|shard| {
-            shard
-                .receipt_execution_outcomes
-                .iter()
-                .filter_map(|outcome| {
-                    if !outcome
-                        .execution_outcome
-                        .as_ref()?
-                        .outcome
-                        .as_ref()?
-                        .status
-                        .as_ref()?
-                        .is_success()
-                    {
-                        return None;
-                    }
-                    if !matches!(
-                        outcome.receipt.as_ref()?.receipt,
-                        Some(codec::receipt::Receipt::Action(_))
-                    ) {
-                        return None;
-                    }
-
-                    Some(trigger::ReceiptWithOutcome {
-                        outcome: outcome.execution_outcome.as_ref()?.clone(),
-                        receipt: outcome.receipt.as_ref()?.clone(),
-                        block: block.cheap_clone(),
-                    })
-                })
-        });
-
-        let mut trigger_data: Vec<_> = receipts
-            .map(|r| NearTrigger::Receipt(Arc::new(r)))
-            .collect();
-
-        trigger_data.push(NearTrigger::Block(block.cheap_clone()));
-
-        // TODO: `block` should probably be an `Arc` in `BlockWithTriggers` to avoid this clone.
-        Ok(BlockWithTriggers::new(block.as_ref().clone(), trigger_data))
-    }
-}
-
 pub struct IngestorAdapter {
     logger: Logger,
 }

graph/src/blockchain/block_stream.rs

@@ -77,8 +77,9 @@ pub trait TriggersAdapter<C: Blockchain>: Send + Sync {
     async fn parent_ptr(&self, block: &BlockPtr) -> Result<Option<BlockPtr>, Error>;
 }
 
+#[async_trait]
 pub trait FirehoseMapper<C: Blockchain>: Send + Sync {
-    fn to_block_stream_event(
+    async fn to_block_stream_event(
         &self,
         logger: &Logger,
         response: &bstream::BlockResponseV2,

graph/src/blockchain/firehose_block_stream.rs

@@ -94,7 +94,7 @@ fn stream_blocks<C: Blockchain, F: FirehoseMapper<C>>(
                     for await response in stream {
                         match response {
                             Ok(v) => {
-                                match mapper.to_block_stream_event(&logger, &v, &adapter, &filter) {
+                                match mapper.to_block_stream_event(&logger, &v, &adapter, &filter).await {
                                     Ok(event) => {
                                         yield event;