iota_indexer/processors/
processor_orchestrator.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use tokio::task::JoinSet;
6use tokio_util::sync::CancellationToken;
7use tracing::{error, info};
8
9use super::{
10    address_metrics_processor::AddressMetricsProcessor,
11    move_call_metrics_processor::MoveCallMetricsProcessor,
12    network_metrics_processor::NetworkMetricsProcessor,
13};
14use crate::{metrics::IndexerMetrics, store::IndexerAnalyticalStore};
15
16pub struct ProcessorOrchestrator<S> {
17    store: S,
18    metrics: IndexerMetrics,
19    cancel: CancellationToken,
20}
21
22impl<S> ProcessorOrchestrator<S>
23where
24    S: IndexerAnalyticalStore + Clone + Send + Sync + 'static,
25{
26    pub fn new(store: S, metrics: IndexerMetrics, cancel: CancellationToken) -> Self {
27        Self {
28            store,
29            metrics,
30            cancel,
31        }
32    }
33
34    pub async fn run_forever(&mut self) {
35        info!("Processor orchestrator started...");
36        let mut tasks = JoinSet::new();
37
38        let network_metrics_processor =
39            NetworkMetricsProcessor::new(self.store.clone(), self.metrics.clone());
40
41        tasks.spawn(async move {
42            loop {
43                let network_metrics_res = network_metrics_processor.start().await;
44                if let Err(e) = network_metrics_res {
45                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
46                    error!(
47                        "indexer network metrics processor failed with error {e:?}, retrying in 5s..."
48                    );
49                }
50            }
51        });
52
53        let addr_metrics_processor =
54            AddressMetricsProcessor::new(self.store.clone(), self.metrics.clone());
55
56        tasks.spawn(async move {
57            loop {
58                let addr_metrics_res = addr_metrics_processor.start().await;
59                if let Err(e) = addr_metrics_res {
60                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
61                    error!(
62                        "indexer address metrics processor failed with error {e:?}, retrying in 5s..."
63                    );
64                }
65            }
66        });
67
68        let move_call_metrics_processor =
69            MoveCallMetricsProcessor::new(self.store.clone(), self.metrics.clone());
70
71        tasks.spawn(async move {
72            loop {
73                let move_call_metrics_res = move_call_metrics_processor.start().await;
74                if let Err(e) = move_call_metrics_res {
75                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
76                    error!(
77                        "indexer move call metrics processor failed with error {e:?}, retrying in 5s..."
78                    );
79                }
80            }
81        });
82
83        tokio::select! {
84            _ = self.cancel.cancelled() => {
85                info!("Processor orchestrator shutting down...");
86                // Aborting mid-batch is safe, each batch commits in one transaction,
87                // so an interrupted one rolls back, and the processors resume from the
88                // last committed cursor on the next start.
89                tasks.shutdown().await;
90            }
91            _ = async {
92                while let Some(res) = tasks.join_next().await {
93                    res.expect("processor orchestrator should not run into errors.");
94                }
95            } => {}
96        }
97    }
98}
iota_indexer/processors/processor_orchestrator.rs

iota_indexer/processors/
processor_orchestrator.rs
