fix: dependency events not being applied to the correct contract

Author: joshstevens19

core/src/indexer/dependency.rs

@@ -272,6 +272,17 @@ impl ContractEventDependencies {
                     if has_dependency {
                         return Some(d.contract_name.to_string());
                     }
+
+                    // check if it's a filter event
+                    let has_dependency =
+                        d.event_dependencies.has_dependency(&ContractEventMapping {
+                            // TODO - this is a hacky way to check if it's a filter event
+                            contract_name: contract_name.to_string().replace("Filter", ""),
+                            event_name: event_name.to_string(),
+                        });
+                    if has_dependency {
+                        return Some(d.contract_name.to_string());
+                    }
                 }
                 None
             })
@@ -293,24 +304,23 @@ impl ContractEventsDependenciesConfig {
     }
 
     pub fn add_to_event_or_new_entry(
-        contract_name: &str,
         dependency_event_processing_configs: &mut Vec<ContractEventsDependenciesConfig>,
         event_processing_config: Arc<EventProcessingConfig>,
         dependencies: &[ContractEventDependencies],
     ) {
         match dependency_event_processing_configs
             .iter_mut()
-            .find(|c| c.contract_name == contract_name)
+            .find(|c| c.contract_name == event_processing_config.contract_name)
         {
             Some(contract_events_config) => {
                 contract_events_config.add_event_config(event_processing_config)
             }
             None => {
                 dependency_event_processing_configs.push(ContractEventsDependenciesConfig {
-                    contract_name: contract_name.to_string(),
+                    contract_name: event_processing_config.contract_name.clone(),
                     event_dependencies: dependencies
                         .iter()
-                        .find(|d| d.contract_name == contract_name)
+                        .find(|d| d.contract_name == event_processing_config.contract_name)
                         .expect("Failed to find contract dependencies")
                         .event_dependencies
                         .clone(),

core/src/indexer/process.rs

@@ -107,8 +107,8 @@ pub enum ProcessContractEventsWithDependenciesError {
     #[error("Could not build filter: {0}")]
     BuildFilterError(#[from] BuildRindexerFilterError),
 
-    #[error("Event config not found")]
-    EventConfigNotFound,
+    #[error("Event config not found for contract: {0} and event: {1}")]
+    EventConfigNotFound(String, String),
 
     #[error("Could not run all the logs processes {0}")]
     JoinError(#[from] JoinError),
@@ -118,6 +118,7 @@ pub enum ProcessContractEventsWithDependenciesError {
 pub struct OrderedLiveIndexingDetails {
     pub filter: RindexerEventFilter,
     pub last_seen_block_number: U64,
+    pub last_no_new_block_log_time: Instant,
 }
 
 async fn process_contract_events_with_dependencies(
@@ -141,10 +142,15 @@ async fn process_contract_events_with_dependencies(
                 let event_processing_config = event_processing_config
                     .iter()
                     .find(|e| {
-                        e.contract_name == dependency.contract_name &&
+                        // TODO - this is a hacky way to check if it's a filter event
+                        (e.contract_name == dependency.contract_name ||
+                            e.contract_name.replace("Filter", "") == dependency.contract_name) &&
                             e.event_name == dependency.event_name
                     })
-                    .ok_or(ProcessContractEventsWithDependenciesError::EventConfigNotFound)?;
+                    .ok_or(ProcessContractEventsWithDependenciesError::EventConfigNotFound(
+                        dependency.contract_name,
+                        dependency.event_name,
+                    ))?;
 
                 // forces live indexing off as it has to handle it a bit differently
                 process_event_logs(Arc::clone(event_processing_config), true).await?;
@@ -165,9 +171,18 @@ async fn process_contract_events_with_dependencies(
 
         let results = join_all(tasks).await;
         for result in results {
-            if let Err(e) = result {
-                error!("Error processing logs: {:?}", e);
-                return Err(ProcessContractEventsWithDependenciesError::JoinError(e));
+            match result {
+                Ok(result) => match result {
+                    Ok(_) => {}
+                    Err(e) => {
+                        error!("Error processing logs due to dependencies error: {:?}", e);
+                        return Err(e);
+                    }
+                },
+                Err(e) => {
+                    error!("Error processing logs: {:?}", e);
+                    return Err(ProcessContractEventsWithDependenciesError::JoinError(e));
+                }
             }
         }
 
@@ -210,12 +225,15 @@ async fn live_indexing_for_contract_event_dependencies<'a>(
 
         ordering_live_indexing_details_map.insert(
             config.topic_id,
-            Arc::new(Mutex::new(OrderedLiveIndexingDetails { filter, last_seen_block_number })),
+            Arc::new(Mutex::new(OrderedLiveIndexingDetails {
+                filter,
+                last_seen_block_number,
+                last_no_new_block_log_time: Instant::now(),
+            })),
         );
     }
 
     // this is used for less busy chains to make sure they know rindexer is still alive
-    let mut last_no_new_block_log_time = Instant::now();
     let log_no_new_block_interval = Duration::from_secs(20);
 
     loop {
@@ -243,15 +261,24 @@ async fn live_indexing_for_contract_event_dependencies<'a>(
                                     &config.info_log_name,
                                     IndexingEventProgressStatus::Live.log()
                                 );
-                                if last_no_new_block_log_time.elapsed() >= log_no_new_block_interval
+                                if ordering_live_indexing_details
+                                    .last_no_new_block_log_time
+                                    .elapsed() >=
+                                    log_no_new_block_interval
                                 {
                                     info!(
                                         "{} - {} - No new blocks published in the last 20 seconds - latest block number {}",
                                         &config.info_log_name,
                                         IndexingEventProgressStatus::Live.log(),
                                         latest_block_number
                                     );
-                                    last_no_new_block_log_time = Instant::now();
+                                    ordering_live_indexing_details.last_no_new_block_log_time =
+                                        Instant::now();
+                                    *ordering_live_indexing_details_map
+                                        .get(&config.topic_id)
+                                        .expect("Failed to get ordering_live_indexing_details_map")
+                                        .lock()
+                                        .await = ordering_live_indexing_details;
                                 }
                                 continue;
                             }

core/src/indexer/start.rs

@@ -107,6 +107,7 @@ pub async fn start_indexing(
             .iter()
             .find(|c| c.name == event.contract.name)
             .and_then(|c| c.streams.as_ref());
+
         for network_contract in event.contract.details.iter() {
             let config = SyncConfig {
                 project_path,
@@ -209,8 +210,8 @@ pub async fn start_indexing(
             };
 
             let dependencies_status = ContractEventDependencies::dependencies_status(
-                &event.contract.name,
-                &event.event_name,
+                &event_processing_config.contract_name,
+                &event_processing_config.event_name,
                 dependencies,
             );
 
@@ -232,7 +233,6 @@ pub async fn start_indexing(
                 }
 
                 ContractEventsDependenciesConfig::add_to_event_or_new_entry(
-                    &event.contract.name,
                     &mut dependency_event_processing_configs,
                     event_processing_config_arc,
                     dependencies,
@@ -247,7 +247,6 @@ pub async fn start_indexing(
     // apply dependency events config after processing to avoid ordering issues
     for apply in apply_cross_contract_dependency_events_config_after_processing {
         let (dependency_in_other_contract, event_processing_config) = apply;
-
         ContractEventsDependenciesConfig::add_to_event_or_panic(
             &dependency_in_other_contract,
             &mut dependency_event_processing_configs,

core/src/manifest/yaml.rs

@@ -39,6 +39,9 @@ pub enum ValidateManifestError {
     #[error("Contract names {0} must be unique")]
     ContractNameMustBeUnique(String),
 
+    #[error("Contract name {0} can not include 'Filter' in the name as it is a reserved word")]
+    ContractNameCanNotIncludeFilter(String),
+
     #[error("Invalid network mapped to contract: network - {0} contract - {1}")]
     InvalidNetworkMappedToContract(String, String),
 
@@ -88,6 +91,12 @@ fn validate_manifest(
     }
 
     for contract in &manifest.contracts {
+        if contract.name.to_lowercase().contains("filter") {
+            return Err(ValidateManifestError::ContractNameCanNotIncludeFilter(
+                contract.name.clone(),
+            ));
+        }
+
         let events = ABIItem::read_abi_items(project_path, contract)
             .map_err(|e| ValidateManifestError::InvalidABI(contract.name.clone(), e.to_string()))?;
 
@@ -266,7 +275,7 @@ pub fn read_manifest(file_path: &PathBuf) -> Result<Manifest, ReadManifestError>
     let manifest_before_transform: Manifest = serde_yaml::from_str(&contents)?;
 
     contents = substitute_env_variables(&contents)?;
-    
+
     let mut manifest_after_transform: Manifest = serde_yaml::from_str(&contents)?;
 
     // as we don't want to inject the RPC URL in rust projects in clear text we should change

documentation/docs/pages/docs/changelog.mdx

@@ -9,6 +9,7 @@
 
 ### Bug fixes
 -------------------------------------------------
+- fix: dependency events not being applied to the correct contract
 
 ### Breaking changes
 -------------------------------------------------