Skip to content

Commit db68d6d

Browse files
desferodesfero
authored
Add support for RINDEXER_LOG env variable (#323)
1 parent 0897d1e commit db68d6d

File tree

8 files changed

+101
-125
lines changed

8 files changed

+101
-125
lines changed

‎core/src/event/config.rs‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ pub struct ContractEventProcessingConfig {
2727
pub project_path: PathBuf,
2828
pub indexer_name: String,
2929
pub contract_name: String,
30-
pub info_log_name: String,
3130
pub topic_id: B256,
3231
pub event_name: String,
3332
pub config: Config,
@@ -46,6 +45,10 @@ pub struct ContractEventProcessingConfig {
4645
}
4746

4847
impl ContractEventProcessingConfig {
48+
pub fn info_log_name(&self) -> String {
49+
format!("{}::{}::{}", self.contract_name, self.event_name, self.network_contract.network)
50+
}
51+
4952
pub fn to_event_filter(&self) -> Result<RindexerEventFilter, BuildRindexerFilterError> {
5053
match &self.network_contract.indexing_contract_setup {
5154
IndexingContractSetup::Address(details) => RindexerEventFilter::new_address_filter(
@@ -153,7 +156,7 @@ impl FactoryEventProcessingConfig {
153156
}
154157

155158
pub fn info_log_name(&self) -> String {
156-
format!("{}::{}", self.contract_name, self.event.name)
159+
format!("{}::{}::{}", self.contract_name, self.event.name, self.network_contract.network)
157160
}
158161
}
159162

@@ -214,7 +217,7 @@ impl EventProcessingConfig {
214217

215218
pub fn info_log_name(&self) -> String {
216219
match self {
217-
Self::ContractEventProcessing(config) => config.info_log_name.clone(),
220+
Self::ContractEventProcessing(config) => config.info_log_name().clone(),
218221
Self::FactoryEventProcessing(config) => config.info_log_name(),
219222
}
220223
}

‎core/src/indexer/fetch_logs.rs‎

Lines changed: 35 additions & 63 deletions
Large diffs are not rendered by default.

‎core/src/indexer/native_transfer.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,11 +119,10 @@ pub async fn native_transfer_block_fetch(
119119
// Spawn a separate task to handle notifications
120120
if let Some(notifications) = chain_state_notification {
121121
// Subscribe to notifications for this network
122-
let network_clone = network.clone();
123122
let mut notifications_clone = notifications.subscribe();
124123
tokio::spawn(async move {
125124
while let Ok(notification) = notifications_clone.recv().await {
126-
handle_chain_notification(notification, "NativeTransfer", &network_clone);
125+
handle_chain_notification(notification, "NativeTransfer");
127126
}
128127
});
129128
}

‎core/src/indexer/process.rs‎

Lines changed: 44 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -42,13 +42,27 @@ pub enum ProcessEventError {
4242
ProviderCallError(#[from] ProviderError),
4343
}
4444

45-
pub async fn process_event(
45+
/// Processes an event that doesn't have dependencies.
46+
/// First processes historical logs, then starts live indexing if the event is configured for live indexing.
47+
/// This function returns immediately without waiting for the indexing to complete.
48+
pub async fn process_non_blocking_event(
4649
config: EventProcessingConfig,
47-
block_until_indexed: bool,
4850
) -> Result<(), ProcessEventError> {
49-
debug!("{} - Processing events", config.info_log_name());
51+
debug!("{} - Processing non blocking event", config.info_log_name());
52+
53+
process_event_logs(Arc::new(config), false, false).await?;
54+
55+
Ok(())
56+
}
57+
58+
/// Processes historical logs for a blocking event that has dependencies.
59+
/// This function waits for the indexing to complete before returning.
60+
pub async fn process_blocking_event_historical_data(
61+
config: Arc<EventProcessingConfig>,
62+
) -> Result<(), Box<ProviderError>> {
63+
debug!("{} - Processing blocking event historical data", config.info_log_name());
5064

51-
process_event_logs(Arc::new(config), false, block_until_indexed).await?;
65+
process_event_logs(config, true, true).await?;
5266

5367
Ok(())
5468
}
@@ -191,9 +205,10 @@ async fn process_contract_events_with_dependencies(
191205
let task = tokio::spawn({
192206
let live_indexing_events = Arc::clone(&live_indexing_events);
193207
async move {
194-
// forces live indexing off as it has to handle it a bit differently
195-
process_event_logs(Arc::clone(&event_processing_config), true, true)
196-
.await?;
208+
process_blocking_event_historical_data(Arc::clone(
209+
&event_processing_config,
210+
))
211+
.await?;
197212

198213
if event_processing_config.live_indexing() {
199214
let network_contract = event_processing_config.network_contract();
@@ -275,7 +290,7 @@ async fn live_indexing_for_contract_event_dependencies(
275290
EventDependenciesIndexingConfig { cached_provider, events, network }: EventDependenciesIndexingConfig,
276291
) {
277292
debug!(
278-
"Live indexing events on {} - {}",
293+
"Live indexing events on {} in order: {}",
279294
network,
280295
events
281296
.iter()
@@ -359,12 +374,11 @@ async fn live_indexing_for_contract_event_dependencies(
359374
>= log_no_new_block_interval
360375
{
361376
info!(
362-
"{}::{} - {} - No new blocks published in the last 5 minutes - latest block number {}",
363-
&config.info_log_name(),
364-
&config.network_contract().network,
365-
IndexingEventProgressStatus::Live.log(),
366-
latest_block_number
367-
);
377+
"{} - {} - No new blocks published in the last 5 minutes - latest block number {}",
378+
&config.info_log_name(),
379+
IndexingEventProgressStatus::Live.log(),
380+
latest_block_number
381+
);
368382
ordering_live_indexing_details.last_no_new_block_log_time = Instant::now();
369383
*ordering_live_indexing_details_map
370384
.get(&config.id())
@@ -396,18 +410,16 @@ async fn live_indexing_for_contract_event_dependencies(
396410
// therefore, we log an error as means RCP state is not in sync with the blockchain
397411
if is_outside_reorg_range {
398412
error!(
399-
"{}::{} - {} - RPC has gone back on latest block: rpc returned {}, last seen: {}",
413+
"{} - {} - RPC has gone back on latest block: rpc returned {}, last seen: {}",
400414
&config.info_log_name(),
401-
&config.network_contract().network,
402415
IndexingEventProgressStatus::Live.log(),
403416
latest_block_number,
404417
from_block
405418
);
406419
} else {
407420
info!(
408-
"{}::{} - {} - RPC has gone back on latest block: rpc returned {}, last seen: {}",
421+
"{} - {} - RPC has gone back on latest block: rpc returned {}, last seen: {}",
409422
&config.info_log_name(),
410-
&config.network_contract().network,
411423
IndexingEventProgressStatus::Live.log(),
412424
latest_block_number,
413425
from_block
@@ -417,9 +429,8 @@ async fn live_indexing_for_contract_event_dependencies(
417429
continue;
418430
} else {
419431
info!(
420-
"{}::{} - {} - not in safe reorg block range yet block: {} > range: {}",
432+
"{} - {} - not in safe reorg block range yet block: {} > range: {}",
421433
&config.info_log_name(),
422-
&config.network_contract().network,
423434
IndexingEventProgressStatus::Live.log(),
424435
from_block,
425436
safe_block_number
@@ -438,16 +449,14 @@ async fn live_indexing_for_contract_event_dependencies(
438449
)
439450
{
440451
debug!(
441-
"{}::{} - {} - Skipping block {} as it's not relevant",
452+
"{} - {} - Skipping block {} as it's not relevant",
442453
&config.info_log_name(),
443-
&config.network_contract().network,
444454
IndexingEventProgressStatus::Live.log(),
445455
from_block
446456
);
447457
debug!(
448-
"{}::{} - {} - Did not need to hit RPC as no events in {} block - LogsBloom for block checked",
458+
"{} - {} - Did not need to hit RPC as no events in {} block - LogsBloom for block checked",
449459
&config.info_log_name(),
450-
&config.network_contract().network,
451460
IndexingEventProgressStatus::Live.log(),
452461
from_block
453462
);
@@ -477,9 +486,8 @@ async fn live_indexing_for_contract_event_dependencies(
477486
match cached_provider.get_logs(&ordering_live_indexing_details.filter).await {
478487
Ok(logs) => {
479488
debug!(
480-
"{}::{} - {} - Live id {} topic_id {}, Logs: {} from {} to {}",
489+
"{} - {} - Live id {} topic_id {}, Logs: {} from {} to {}",
481490
&config.info_log_name(),
482-
&config.network_contract().network,
483491
IndexingEventProgressStatus::Live.log(),
484492
&config.id(),
485493
&config.topic_id(),
@@ -489,9 +497,8 @@ async fn live_indexing_for_contract_event_dependencies(
489497
);
490498

491499
debug!(
492-
"{}::{} - {} - Fetched {} event logs - blocks: {} - {}",
500+
"{} - {} - Fetched {} event logs - blocks: {} - {}",
493501
&config.info_log_name(),
494-
&config.network_contract().network,
495502
IndexingEventProgressStatus::Live.log(),
496503
logs.len(),
497504
from_block,
@@ -516,12 +523,11 @@ async fn live_indexing_for_contract_event_dependencies(
516523
let complete = task.await;
517524
if let Err(e) = complete {
518525
error!(
519-
"{}::{} - {} - Error indexing task: {} - will try again in 200ms",
520-
&config.info_log_name(),
521-
&config.network_contract().network,
522-
IndexingEventProgressStatus::Live.log(),
523-
e
524-
);
526+
"{} - {} - Error indexing task: {} - will try again in 200ms",
527+
&config.info_log_name(),
528+
IndexingEventProgressStatus::Live.log(),
529+
e
530+
);
525531
break;
526532
}
527533
ordering_live_indexing_details.last_seen_block_number = to_block;
@@ -531,9 +537,8 @@ async fn live_indexing_for_contract_event_dependencies(
531537
.filter
532538
.set_from_block(to_block + U64::from(1));
533539
debug!(
534-
"{}::{} - {} - No events found between blocks {} - {}",
540+
"{} - {} - No events found between blocks {} - {}",
535541
&config.info_log_name(),
536-
&config.network_contract().network,
537542
IndexingEventProgressStatus::Live.log(),
538543
from_block,
539544
to_block
@@ -557,9 +562,8 @@ async fn live_indexing_for_contract_event_dependencies(
557562
}
558563
Err(err) => {
559564
error!(
560-
"{}::{} - {} - Error fetching logs: {} - will try again in 200ms",
565+
"{} - {} - Error fetching logs: {} - will try again in 200ms",
561566
&config.info_log_name(),
562-
&config.network_contract().network,
563567
IndexingEventProgressStatus::Live.log(),
564568
err
565569
);
@@ -569,9 +573,8 @@ async fn live_indexing_for_contract_event_dependencies(
569573
}
570574
Err(err) => {
571575
error!(
572-
"{}::{} - {} - Error fetching logs: {} - will try again in 200ms",
576+
"{} - {} - Error fetching logs: {} - will try again in 200ms",
573577
&config.info_log_name(),
574-
&config.network_contract().network,
575578
IndexingEventProgressStatus::Live.log(),
576579
err
577580
);
@@ -621,7 +624,7 @@ async fn handle_logs_result(
621624
) -> Result<JoinHandle<()>, Box<dyn std::error::Error + Send>> {
622625
match result {
623626
Ok(result) => {
624-
debug!("Processing logs {} - length {}", config.event_name(), result.logs.len());
627+
debug!("{} - Processing {} logs", config.info_log_name(), result.logs.len());
625628

626629
let fn_data = result
627630
.logs

‎core/src/indexer/reorg.rs‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,7 @@ use tracing::{debug, warn};
44
use crate::notifications::ChainStateNotification;
55

66
/// Handles chain state notifications (reorgs, reverts, commits)
7-
pub fn handle_chain_notification(
8-
notification: ChainStateNotification,
9-
info_log_name: &str,
10-
network: &str,
11-
) {
7+
pub fn handle_chain_notification(notification: ChainStateNotification, info_log_name: &str) {
128
match notification {
139
ChainStateNotification::Reorged {
1410
revert_from_block,
@@ -18,9 +14,8 @@ pub fn handle_chain_notification(
1814
new_tip_hash,
1915
} => {
2016
warn!(
21-
"{}::{} - REORG DETECTED! Need to revert blocks {} to {} and re-index {} to {} (new tip: {})",
17+
"{} - REORG DETECTED! Need to revert blocks {} to {} and re-index {} to {} (new tip: {})",
2218
info_log_name,
23-
network,
2419
revert_from_block, revert_to_block,
2520
new_from_block, new_to_block,
2621
new_tip_hash
@@ -29,15 +24,15 @@ pub fn handle_chain_notification(
2924
}
3025
ChainStateNotification::Reverted { from_block, to_block } => {
3126
warn!(
32-
"{}::{} - CHAIN REVERTED! Blocks {} to {} have been reverted",
33-
info_log_name, network, from_block, to_block
27+
"{} - CHAIN REVERTED! Blocks {} to {} have been reverted",
28+
info_log_name, from_block, to_block
3429
);
3530
// TODO: In future PR, mark affected logs as removed in the database
3631
}
3732
ChainStateNotification::Committed { from_block, to_block, tip_hash } => {
3833
debug!(
39-
"{}::{} - Chain committed: blocks {} to {} (tip: {})",
40-
info_log_name, network, from_block, to_block, tip_hash
34+
"{} - Chain committed: blocks {} to {} (tip: {})",
35+
info_log_name, from_block, to_block, tip_hash
4136
);
4237
}
4338
}

‎core/src/indexer/start.rs‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ use crate::{
2525
last_synced::{get_last_synced_block_number, SyncConfig},
2626
native_transfer::{native_transfer_block_fetch, NATIVE_TRANSFER_CONTRACT_NAME},
2727
process::{
28-
process_contracts_events_with_dependencies, process_event,
28+
process_contracts_events_with_dependencies, process_non_blocking_event,
2929
ProcessContractsEventsWithDependenciesError, ProcessEventError,
3030
},
3131
progress::IndexingEventsProgressState,
@@ -451,7 +451,6 @@ pub async fn start_indexing_contract_events(
451451
project_path: project_path.clone(),
452452
indexer_name: event.indexer_name.clone(),
453453
contract_name: event.contract.name.clone(),
454-
info_log_name: event.info_log_name(),
455454
topic_id: event.topic_id,
456455
event_name: event.event_name.clone(),
457456
network_contract: Arc::new(network_contract.clone()),
@@ -505,7 +504,7 @@ pub async fn start_indexing_contract_events(
505504
&dependencies,
506505
);
507506
} else {
508-
let process_event = tokio::spawn(process_event(event_processing_config, false));
507+
let process_event = tokio::spawn(process_non_blocking_event(event_processing_config));
509508
non_blocking_process_events.push(process_event);
510509
}
511510
}

‎core/src/logger.rs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,8 +73,12 @@ impl tracing_subscriber::fmt::time::FormatTime for CustomTimer {
7373
}
7474
}
7575

76-
pub fn setup_logger(log_level: LevelFilter) {
77-
let filter = EnvFilter::from_default_env().add_directive(log_level.into());
76+
const LOG_LEVEL_ENV: &str = "RINDEXER_LOG";
77+
78+
pub fn setup_logger(default_log_level: LevelFilter) {
79+
let filter = EnvFilter::try_from_env(LOG_LEVEL_ENV).unwrap_or(
80+
EnvFilter::builder().with_default_directive(default_log_level.into()).parse_lossy(""),
81+
);
7882

7983
let format = Format::default().with_timer(CustomTimer).with_level(true).with_target(false);
8084

‎documentation/docs/pages/docs/changelog.mdx‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
### Features
77
-------------------------------------------------
88
- feat: check if the RPC chain id is matching the configured chain id in the yaml config on startup
9+
- feat: add support for `RINDEXER_LOG` environment variable to control the log level of rindexer
910

1011
### Bug fixes
1112
-------------------------------------------------

0 commit comments

Comments
 (0)