[Feature] Add support for endBlock in data sources (#4787)

* graph,chain,store/test-store : Allow new param `endBlock` in manifest

* core,graph,store: ignore end_block reached datasources in match_and_decode, include them in processed datasources

* tests : add runner tests for end-block

* core: move TriggerFilter construction into SubgraphRunner.run_inner

* core: filter out endBlock reached subgraphs when constructing TriggerFilter

* chain,core: refactor endBlock implementation

* refactor `SubgraphRunner.run_inner` to extract `build_filter`

* core : handle reverts for endBlock

* chain,graph: set min_spec_version requirements for endBlock

* core: refaction `build_filter`

* tests: runner test for endblock on reorg

* core: restart block stream in the next block for endblock reached ds

* graph: bump specVersion requirement for endBlock

* core: refactor build_filter logic

* core, tests, graph : make TriggerFilters testable

* chain/startknet: endBlock support for starknet

* chain,core,graph: refactor end_block implementation

* core: refactor build_filter

* Add comments for end-block runner tests

.gitignore

@@ -24,6 +24,7 @@ lcov.info
 /tests/**/generated
 /tests/**/node_modules
 /tests/**/yarn-error.log
+/tests/**/pnpm-lock.yaml
 
 # Built solidity contracts.
 /tests/**/bin

chain/arweave/src/adapter.rs

@@ -239,7 +239,11 @@ mod test {
             kind: "".into(),
             network: None,
             name: "".into(),
-            source: Source { owner, start_block },
+            source: Source {
+                owner,
+                start_block,
+                end_block: None,
+            },
             mapping: Mapping {
                 api_version: Version::new(1, 2, 3),
                 language: "".into(),

chain/arweave/src/data_source.rs

@@ -63,6 +63,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         kinds
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        self.source.end_block
+    }
+
     fn match_and_decode(
         &self,
         trigger: &<Chain as Blockchain>::TriggerData,
@@ -392,9 +396,11 @@ pub struct TransactionHandler {
 }
 
 #[derive(Clone, Debug, Hash, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub(crate) struct Source {
     // A data source that does not have an owner can only have block handlers.
     pub(crate) owner: Option<String>,
-    #[serde(rename = "startBlock", default)]
+    #[serde(default)]
     pub(crate) start_block: BlockNumber,
+    pub(crate) end_block: Option<BlockNumber>,
 }

chain/cosmos/src/data_source.rs

@@ -83,6 +83,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         kinds
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        self.source.end_block
+    }
+
     fn match_and_decode(
         &self,
         trigger: &<Chain as Blockchain>::TriggerData,
@@ -502,9 +506,11 @@ pub struct MappingMessageHandler {
 }
 
 #[derive(Clone, Debug, Hash, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct Source {
-    #[serde(rename = "startBlock", default)]
+    #[serde(default)]
     pub start_block: BlockNumber,
+    pub(crate) end_block: Option<BlockNumber>,
 }
 
 #[derive(Clone, Copy, Debug, Hash, Eq, PartialEq, Deserialize)]
@@ -657,7 +663,10 @@ mod tests {
                 kind: "cosmos".to_string(),
                 network: None,
                 name: "Test".to_string(),
-                source: Source { start_block: 1 },
+                source: Source {
+                    start_block: 1,
+                    end_block: None,
+                },
                 mapping: Mapping {
                     api_version: semver::Version::new(0, 0, 0),
                     language: "".to_string(),
@@ -679,7 +688,10 @@ mod tests {
                 kind: "cosmos".to_string(),
                 network: None,
                 name: "Test".to_string(),
-                source: Source { start_block: 1 },
+                source: Source {
+                    start_block: 1,
+                    end_block: None,
+                },
                 mapping: Mapping {
                     api_version: semver::Version::new(0, 0, 0),
                     language: "".to_string(),

chain/ethereum/src/adapter.rs

@@ -121,6 +121,21 @@ impl TriggerFilter {
     pub(crate) fn requires_traces(&self) -> bool {
         !self.call.is_empty() || self.block.requires_traces()
     }
+
+    #[cfg(debug_assertions)]
+    pub fn log(&self) -> &EthereumLogFilter {
+        &self.log
+    }
+
+    #[cfg(debug_assertions)]
+    pub fn call(&self) -> &EthereumCallFilter {
+        &self.call
+    }
+
+    #[cfg(debug_assertions)]
+    pub fn block(&self) -> &EthereumBlockFilter {
+        &self.block
+    }
 }
 
 impl bc::TriggerFilter<Chain> for TriggerFilter {
@@ -185,7 +200,7 @@ impl bc::TriggerFilter<Chain> for TriggerFilter {
 }
 
 #[derive(Clone, Debug, Default)]
-pub(crate) struct EthereumLogFilter {
+pub struct EthereumLogFilter {
     /// Log filters can be represented as a bipartite graph between contracts and events. An edge
     /// exists between a contract and an event if a data source for the contract has a trigger for
     /// the event.
@@ -382,10 +397,20 @@ impl EthereumLogFilter {
         }
         filters.into_iter()
     }
+
+    #[cfg(debug_assertions)]
+    pub fn contract_addresses(&self) -> impl Iterator<Item = Address> + '_ {
+        self.contracts_and_events_graph
+            .nodes()
+            .filter_map(|node| match node {
+                LogFilterNode::Contract(address) => Some(address),
+                LogFilterNode::Event(_) => None,
+            })
+    }
 }
 
 #[derive(Clone, Debug, Default)]
-pub(crate) struct EthereumCallFilter {
+pub struct EthereumCallFilter {
     // Each call filter has a map of filters keyed by address, each containing a tuple with
     // start_block and the set of function signatures
     pub contract_addresses_function_signatures:
@@ -583,7 +608,7 @@ impl From<&EthereumBlockFilter> for EthereumCallFilter {
 }
 
 #[derive(Clone, Debug, Default)]
-pub(crate) struct EthereumBlockFilter {
+pub struct EthereumBlockFilter {
     /// Used for polling block handlers, a hashset of (start_block, polling_interval)
     pub polling_intervals: HashSet<(BlockNumber, i32)>,
     pub contract_addresses: HashSet<(BlockNumber, Address)>,

chain/ethereum/src/data_source.rs

@@ -51,6 +51,7 @@ pub struct DataSource {
     pub manifest_idx: u32,
     pub address: Option<Address>,
     pub start_block: BlockNumber,
+    pub end_block: Option<BlockNumber>,
     pub mapping: Mapping,
     pub context: Arc<Option<DataSourceContext>>,
     pub creation_block: Option<BlockNumber>,
@@ -99,6 +100,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             manifest_idx: template.manifest_idx,
             address: Some(address),
             start_block: creation_block,
+            end_block: None,
             mapping: template.mapping,
             context: Arc::new(context),
             creation_block: Some(creation_block),
@@ -137,6 +139,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         self.start_block
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        self.end_block
+    }
+
     fn match_and_decode(
         &self,
         trigger: &<Chain as Blockchain>::TriggerData,
@@ -176,12 +182,12 @@ impl blockchain::DataSource<Chain> for DataSource {
             address,
             mapping,
             context,
-
             // The creation block is ignored for detection duplicate data sources.
             // Contract ABI equality is implicit in `mapping.abis` equality.
             creation_block: _,
             contract_abi: _,
             start_block: _,
+            end_block: _,
         } = self;
 
         // mapping_request_sender, host_metrics, and (most of) host_exports are operational structs
@@ -247,6 +253,7 @@ impl blockchain::DataSource<Chain> for DataSource {
             manifest_idx,
             address,
             start_block: creation_block.unwrap_or(0),
+            end_block: None,
             mapping: template.mapping.clone(),
             context: Arc::new(context),
             creation_block,
@@ -382,6 +389,7 @@ impl DataSource {
             manifest_idx,
             address: source.address,
             start_block: source.start_block,
+            end_block: source.end_block,
             mapping,
             context: Arc::new(context),
             creation_block,

chain/near/src/chain.rs

@@ -1010,6 +1010,7 @@ mod test {
             source: crate::data_source::Source {
                 account,
                 start_block: 10,
+                end_block: None,
                 accounts: partial_accounts,
             },
             mapping: Mapping {

chain/near/src/data_source.rs

@@ -92,6 +92,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         kinds
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        self.source.end_block
+    }
+
     fn match_and_decode(
         &self,
         trigger: &<Chain as Blockchain>::TriggerData,
@@ -493,10 +497,12 @@ impl PartialAccounts {
 }
 
 #[derive(Clone, Debug, Hash, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub(crate) struct Source {
     // A data source that does not have an account or accounts can only have block handlers.
     pub(crate) account: Option<String>,
-    #[serde(rename = "startBlock", default)]
+    #[serde(default)]
     pub(crate) start_block: BlockNumber,
+    pub(crate) end_block: Option<BlockNumber>,
     pub(crate) accounts: Option<PartialAccounts>,
 }

chain/starknet/src/data_source.rs

@@ -48,6 +48,7 @@ pub struct UnresolvedDataSource {
 #[serde(rename_all = "camelCase")]
 pub struct Source {
     pub start_block: BlockNumber,
+    pub end_block: Option<BlockNumber>,
     #[serde(default, deserialize_with = "deserialize_address")]
     pub address: Option<FieldElement>,
 }
@@ -98,6 +99,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         self.source.start_block
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        self.source.end_block
+    }
+
     fn handler_kinds(&self) -> HashSet<&str> {
         let mut kinds = HashSet::new();
 

chain/substreams/src/data_source.rs

@@ -47,6 +47,10 @@ impl blockchain::DataSource<Chain> for DataSource {
         self.initial_block.unwrap_or(0)
     }
 
+    fn end_block(&self) -> Option<BlockNumber> {
+        None
+    }
+
     fn name(&self) -> &str {
         &self.name
     }

core/src/subgraph/context.rs

@@ -66,24 +66,23 @@ where
 {
     instance: SubgraphInstance<C, T>,
     pub instances: SubgraphKeepAlive,
-    pub filter: C::TriggerFilter,
     pub offchain_monitor: OffchainMonitor,
+    pub filter: Option<C::TriggerFilter>,
     trigger_processor: Box<dyn TriggerProcessor<C, T>>,
 }
 
 impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
     pub fn new(
         instance: SubgraphInstance<C, T>,
         instances: SubgraphKeepAlive,
-        filter: C::TriggerFilter,
         offchain_monitor: OffchainMonitor,
         trigger_processor: Box<dyn TriggerProcessor<C, T>>,
     ) -> Self {
         Self {
             instance,
             instances,
-            filter,
             offchain_monitor,
+            filter: None,
             trigger_processor,
         }
     }
@@ -182,7 +181,6 @@ impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
         self.instance.causality_region_next_value()
     }
 
-    #[cfg(debug_assertions)]
     pub fn instance(&self) -> &SubgraphInstance<C, T> {
         &self.instance
     }

core/src/subgraph/context/instance.rs

@@ -15,7 +15,9 @@ pub struct SubgraphInstance<C: Blockchain, T: RuntimeHostBuilder<C>> {
     subgraph_id: DeploymentHash,
     network: String,
     host_builder: T,
-    templates: Arc<Vec<DataSourceTemplate<C>>>,
+    pub templates: Arc<Vec<DataSourceTemplate<C>>>,
+    /// The data sources declared in the subgraph manifest. This does not include dynamic data sources.
+    pub data_sources: Arc<Vec<DataSource<C>>>,
     host_metrics: Arc<HostMetrics>,
 
     /// The hosts represent the data sources in the subgraph. There is one host per data source.
@@ -33,9 +35,12 @@ where
     C: Blockchain,
     T: RuntimeHostBuilder<C>,
 {
+    /// Create a new subgraph instance from the given manifest and data sources.
+    /// `data_sources` must contain all data sources declared in the manifest + all dynamic data sources.
     pub fn from_manifest(
         logger: &Logger,
         manifest: SubgraphManifest<C>,
+        data_sources: Vec<DataSource<C>>,
         host_builder: T,
         host_metrics: Arc<HostMetrics>,
         offchain_monitor: &mut OffchainMonitor,
@@ -49,6 +54,7 @@ where
             host_builder,
             subgraph_id,
             network,
+            data_sources: Arc::new(manifest.data_sources),
             hosts: Hosts::new(),
             module_cache: HashMap::new(),
             templates,
@@ -59,7 +65,7 @@ where
         // Create a new runtime host for each data source in the subgraph manifest;
         // we use the same order here as in the subgraph manifest to make the
         // event processing behavior predictable
-        for ds in manifest.data_sources {
+        for ds in data_sources {
             // TODO: This is duplicating code from `IndexingContext::add_dynamic_data_source` and
             // `SubgraphInstance::add_dynamic_data_source`. Ideally this should be refactored into
             // `IndexingContext`.
@@ -215,7 +221,6 @@ where
         self.causality_region_seq.next_val()
     }
 
-    #[cfg(debug_assertions)]
     pub fn hosts(&self) -> &[Arc<T::Host>] {
         &self.hosts.hosts()
     }

core/src/subgraph/inputs.rs

@@ -15,6 +15,7 @@ pub struct IndexingInputs<C: Blockchain> {
     pub deployment: DeploymentLocator,
     pub features: BTreeSet<SubgraphFeature>,
     pub start_blocks: Vec<BlockNumber>,
+    pub end_blocks: BTreeSet<BlockNumber>,
     pub stop_block: Option<BlockNumber>,
     pub store: Arc<dyn WritableStore>,
     pub debug_fork: Option<Arc<dyn SubgraphFork>>,
@@ -37,6 +38,7 @@ impl<C: Blockchain> IndexingInputs<C> {
             deployment,
             features,
             start_blocks,
+            end_blocks,
             stop_block,
             store: _,
             debug_fork,
@@ -53,6 +55,7 @@ impl<C: Blockchain> IndexingInputs<C> {
             deployment: deployment.clone(),
             features: features.clone(),
             start_blocks: start_blocks.clone(),
+            end_blocks: end_blocks.clone(),
             stop_block: stop_block.clone(),
             store,
             debug_fork: debug_fork.clone(),