iota_data_ingestion_core/
lib.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5//! This library provides an easy way to create custom indexers.
6//! <br>
7//!
8//! ## Graceful shutdown
9//!
10//! The shutdown sequence in the data ingestion system ensures clean termination
11//! of all components while preserving data integrity. It is initiated via a
12//! [CancellationToken](tokio_util::sync::CancellationToken), which triggers a
13//! hierarchical and graceful shutdown process.
14//!
15//! The shutdown process follows a top-down hierarchy:
16//! 1. [`Worker`]: Individual workers within a [`WorkerPool`] detect the cancellation signal,
17//!    completes current checkpoint processing, sends final progress updates and signals completion
18//!    to parent [`WorkerPool`] via `WorkerStatus::Shutdown` message.
19//! 2. [`WorkerPool`]: Coordinates worker shutdowns, ensures all progress messages are processed,
20//!    waits for all workers' shutdown signals and notifies [`IndexerExecutor`] with
21//!    `WorkerPoolStatus::Shutdown` message when fully terminated.
22//! 3. [`IndexerExecutor`]: Orchestrates the shutdown of all worker pools and and finalizes system
23//!    termination.
24
25mod errors;
26mod executor;
27pub mod history;
28mod metrics;
29mod progress_store;
30pub mod reader;
31mod reducer;
32#[cfg(test)]
33mod tests;
34mod util;
35mod worker_pool;
36
37use std::{
38    fmt::{Debug, Display},
39    sync::Arc,
40};
41
42use async_trait::async_trait;
43pub use errors::{IngestionError, IngestionResult};
44pub use executor::{
45    IndexerExecutor, IngestionLimit, MAX_CHECKPOINTS_IN_PROGRESS, ShutdownAction,
46    setup_single_workflow,
47};
48use iota_types::full_checkpoint_content::CheckpointData;
49pub use metrics::DataIngestionMetrics;
50pub use progress_store::{FileProgressStore, ProgressStore, ShimProgressStore};
51pub use reader::{ReaderOptions, config, filters};
52pub use reducer::Reducer;
53pub use util::{create_remote_store_client, create_remote_store_client_with_ops};
54pub use worker_pool::WorkerPool;
55
56/// Processes individual checkpoints and produces messages for optional batch
57/// processing.
58///
59/// The Worker trait defines the core processing logic for checkpoint data.
60/// Workers run in parallel within a [`WorkerPool`] to process checkpoints and
61/// generate messages that can optionally be batched and processed by a
62/// [`Reducer`].
63///
64/// # Processing Modes
65///
66/// Workers support two processing modes:
67/// * **Direct Processing**: Messages are handled immediately without batching.
68/// * **Batch Processing**: Messages are accumulated and processed in batches by a [`Reducer`].
69///
70/// The processing mode is determined by the presence of a [`Reducer`] in the
71/// [`WorkerPool`] configuration.
72///
73/// # Concurrency
74///
75/// Multiple instances of a worker can run in parallel in the worker pool. The
76/// implementation must be thread-safe and handle checkpoint processing
77/// efficiently.
78///
79/// # Integration with Optional Reducer
80///
81/// Messages produced by the worker can be:
82/// * Processed directly without batching.
83/// * Accumulated and passed to a [`Reducer`] for batch processing.
84///
85/// The worker's [`Message`](Worker::Message) type must match the reducer's
86/// input type when batch processing is enabled.
87#[async_trait]
88pub trait Worker: Send + Sync {
89    type Error: Debug + Display;
90    type Message: Send + Sync;
91
92    /// Processes a single checkpoint and returns a message.
93    ///
94    /// This method contains the core logic for processing checkpoint data.
95    ///
96    /// # Note
97    /// - Checkpoints are processed in order when a single worker is used.
98    /// - Parallel processing with multiple workers does not guarantee checkpoint order.
99    async fn process_checkpoint(
100        &self,
101        checkpoint: Arc<CheckpointData>,
102    ) -> Result<Self::Message, Self::Error>;
103
104    /// A hook that allows preprocessing a checkpoint before it's fully
105    /// processed.
106    ///
107    /// This method can be used to perform actions like validation or data
108    /// transformation before the main
109    /// [`process_checkpoint`](Worker::process_checkpoint) logic is executed.
110    ///
111    /// # Default implementation
112    ///
113    /// By default it returns `Ok(())`.
114    fn preprocess_hook(&self, _: Arc<CheckpointData>) -> Result<(), Self::Error> {
115        Ok(())
116    }
117}
iota_data_ingestion_core/lib.rs

iota_data_ingestion_core/
lib.rs
