iota_indexer/processors/
address_metrics_processor.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use tap::tap::TapFallible;
6use tracing::{error, info};
7
8use crate::{
9    metrics::IndexerMetrics,
10    store::{IndexerAnalyticalStore, diesel_macro::spawn_blocking_task},
11    types::IndexerResult,
12};
13
14const ADDRESS_PROCESSOR_BATCH_SIZE: usize = 80000;
15const PARALLELISM: usize = 10;
16
17pub struct AddressMetricsProcessor<S> {
18    pub store: S,
19    metrics: IndexerMetrics,
20    pub address_processor_batch_size: usize,
21    pub address_processor_parallelism: usize,
22}
23
24impl<S> AddressMetricsProcessor<S>
25where
26    S: IndexerAnalyticalStore + Clone + Sync + Send + 'static,
27{
28    pub fn new(store: S, metrics: IndexerMetrics) -> AddressMetricsProcessor<S> {
29        let address_processor_batch_size = std::env::var("ADDRESS_PROCESSOR_BATCH_SIZE")
30            .map(|s| s.parse::<usize>().unwrap_or(ADDRESS_PROCESSOR_BATCH_SIZE))
31            .unwrap_or(ADDRESS_PROCESSOR_BATCH_SIZE);
32        let address_processor_parallelism = std::env::var("ADDRESS_PROCESSOR_PARALLELISM")
33            .map(|s| s.parse::<usize>().unwrap_or(PARALLELISM))
34            .unwrap_or(PARALLELISM);
35        Self {
36            store,
37            metrics,
38            address_processor_batch_size,
39            address_processor_parallelism,
40        }
41    }
42
43    pub async fn start(&self) -> IndexerResult<()> {
44        info!("Indexer address metrics async processor started...");
45        let latest_tx_seq = self
46            .store
47            .get_address_metrics_last_processed_tx_seq()
48            .await?;
49        let mut last_processed_tx_seq = latest_tx_seq.unwrap_or_default().seq;
50        loop {
51            let mut latest_tx = self.store.get_latest_stored_transaction().await?;
52            while if let Some(tx) = latest_tx {
53                tx.tx_sequence_number
54                    < last_processed_tx_seq + self.address_processor_batch_size as i64
55            } else {
56                true
57            } {
58                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
59                latest_tx = self.store.get_latest_stored_transaction().await?;
60            }
61
62            let mut persist_tasks = vec![];
63            let batch_size = self.address_processor_batch_size;
64            let step_size = batch_size / self.address_processor_parallelism;
65            for chunk_start_tx_seq in (last_processed_tx_seq + 1
66                ..last_processed_tx_seq + batch_size as i64 + 1)
67                .step_by(step_size)
68            {
69                let address_store = self.store.clone();
70                persist_tasks.push(spawn_blocking_task(move || {
71                    address_store.persist_addresses_in_tx_range(
72                        chunk_start_tx_seq,
73                        chunk_start_tx_seq + step_size as i64,
74                    )
75                }));
76            }
77            futures::future::join_all(persist_tasks)
78                .await
79                .into_iter()
80                .collect::<Result<Vec<_>, _>>()
81                .tap_err(|e| {
82                    error!("error joining address persist tasks: {e:?}");
83                })?
84                .into_iter()
85                .collect::<Result<Vec<_>, _>>()
86                .tap_err(|e| {
87                    error!("error persisting addresses or active addresses: {e:?}");
88                })?;
89            last_processed_tx_seq += self.address_processor_batch_size as i64;
90            info!(
91                "Persisted addresses and active addresses for tx seq: {}",
92                last_processed_tx_seq,
93            );
94            self.metrics
95                .latest_address_metrics_tx_seq
96                .set(last_processed_tx_seq);
97
98            let mut last_processed_tx = self.store.get_tx(last_processed_tx_seq).await?;
99            while last_processed_tx.is_none() {
100                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
101                last_processed_tx = self.store.get_tx(last_processed_tx_seq).await?;
102            }
103            // unwrap is safe here b/c we just checked that it's not None
104            let last_processed_cp = last_processed_tx.unwrap().checkpoint_sequence_number;
105            self.store
106                .calculate_and_persist_address_metrics(last_processed_cp)
107                .await?;
108            info!(
109                "Persisted address metrics for checkpoint: {}",
110                last_processed_cp
111            );
112        }
113    }
114}
iota_indexer/processors/address_metrics_processor.rs

iota_indexer/processors/
address_metrics_processor.rs
