iota_data_ingestion_core/reader/
v2.rs

1// Copyright (c) 2025 IOTA Stiftung
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    num::NonZeroUsize,
6    path::{Path, PathBuf},
7    sync::Arc,
8    time::Duration,
9};
10
11use backoff::backoff::Backoff;
12use futures::{StreamExt, TryStreamExt};
13use iota_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
14use iota_grpc_client::GrpcClient;
15use iota_metrics::spawn_monitored_task;
16use iota_types::{
17    full_checkpoint_content::CheckpointData, messages_checkpoint::CheckpointSequenceNumber,
18};
19use object_store::ObjectStore;
20use serde::{Deserialize, Serialize};
21use tap::Pipe;
22use tokio::{
23    sync::{mpsc, watch},
24    task::JoinHandle,
25    time::timeout,
26};
27use tokio_util::sync::CancellationToken;
28use tracing::{debug, error};
29
30#[cfg(not(target_os = "macos"))]
31use crate::reader::fetch::init_watcher;
32use crate::{
33    IngestionError, IngestionResult, MAX_CHECKPOINTS_IN_PROGRESS,
34    config::CheckpointReaderConfigExt,
35    create_remote_store_client,
36    history::reader::{HistoricalReader, HistoricalReaderConfig},
37    reader::{
38        ReaderOptions,
39        common::DataLimiter,
40        fetch::{
41            GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES, LocalRead, ReadSource, fetch_from_object_store,
42        },
43        filters::fullnode::TransactionFilter,
44    },
45};
46
47/// Available sources for checkpoint streams supported by the ingestion
48/// framework.
49///
50/// This enum represents the different types of remote sources from which
51/// checkpoint data can be fetched. Each variant corresponds to a supported
52/// backend or combination of backends for checkpoint retrieval.
53#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
54pub enum RemoteUrl {
55    /// The URL to the Fullnode server that exposes
56    /// checkpoint data streaming through gRPC.
57    ///
58    /// # Example
59    /// ```text
60    /// "http://127.0.0.1:50051"
61    /// ```
62    Fullnode(String),
63    /// A hybrid source combining historical object store and optional live
64    /// object store.
65    HybridHistoricalStore {
66        /// The URL path to the historical object store that contains `*.chk`,
67        /// `*.sum` & `MANIFEST` files.
68        ///
69        /// # Example
70        /// ```text
71        /// "https://checkpoints.mainnet.iota.cafe/ingestion/historical"
72        /// ```
73        historical_url: String,
74        /// The URL path to the live object store that contains `*.chk`
75        /// checkpoint files.
76        ///
77        /// # Example
78        /// ```text
79        /// "https://checkpoints.mainnet.iota.cafe/ingestion/live"
80        /// ```
81        live_url: Option<String>,
82    },
83}
84
85/// Represents a remote backend for checkpoint data retrieval.
86///
87/// This enum encapsulates the supported remote storage mechanisms that can be
88/// used by the ingestion framework to fetch checkpoint data. Each variant
89/// corresponds to a different type of remote source.
90enum RemoteStore {
91    Fullnode(GrpcClient),
92    HybridHistoricalStore {
93        historical: HistoricalReader,
94        live: Option<Box<dyn ObjectStore>>,
95    },
96}
97
98impl RemoteStore {
99    async fn new(
100        remote_url: RemoteUrl,
101        batch_size: usize,
102        timeout_secs: u64,
103    ) -> IngestionResult<Self> {
104        let store = match remote_url {
105            RemoteUrl::Fullnode(ref url) => {
106                let grpc_client = GrpcClient::new(url).map(|client| {
107                    client.with_max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES)
108                })?;
109                RemoteStore::Fullnode(grpc_client)
110            }
111            RemoteUrl::HybridHistoricalStore {
112                historical_url,
113                live_url,
114            } => {
115                let remote_store_config = if let Some(dir) = historical_url.strip_prefix("file://")
116                {
117                    ObjectStoreConfig {
118                        object_store: Some(ObjectStoreType::File),
119                        directory: Some(PathBuf::from(dir)),
120                        ..Default::default()
121                    }
122                } else {
123                    ObjectStoreConfig {
124                        object_store: Some(ObjectStoreType::S3),
125                        object_store_connection_limit: 20,
126                        aws_endpoint: Some(historical_url),
127                        aws_virtual_hosted_style_request: true,
128                        no_sign_request: true,
129                        ..Default::default()
130                    }
131                };
132                let config = HistoricalReaderConfig {
133                    download_concurrency: NonZeroUsize::new(batch_size)
134                        .expect("batch size must be greater than zero"),
135                    remote_store_config,
136                };
137                let historical = HistoricalReader::new(config)
138                    .inspect_err(|e| error!("unable to instantiate historical reader: {e}"))?;
139
140                let live = live_url
141                    .map(|url| create_remote_store_client(url, Default::default(), timeout_secs))
142                    .transpose()?;
143
144                RemoteStore::HybridHistoricalStore { historical, live }
145            }
146        };
147        Ok(store)
148    }
149}
150
151/// Configuration options to control the behavior of a checkpoint
152/// reader.
153#[derive(Default, Clone)]
154pub struct CheckpointReaderConfig {
155    /// Config the checkpoint reader behavior for downloading new checkpoints.
156    pub reader_options: ReaderOptions,
157    /// Local path for checkpoint ingestion. If not provided, checkpoints will
158    /// be ingested from a temporary directory.
159    pub ingestion_path: Option<PathBuf>,
160    /// Remote source for checkpoint data stream.
161    pub remote_store_url: Option<RemoteUrl>,
162}
163
164/// Internal actor responsible for reading and streaming checkpoints.
165///
166/// `CheckpointReaderActor` is the core background task that manages the logic
167/// for fetching, batching, and streaming checkpoint data from local or remote
168/// sources. It handles checkpoint discovery, garbage collection signals, and
169/// coordinates with remote fetchers as needed.
170///
171/// This struct is intended to be run as an asynchronous task and is not
172/// typically interacted with directly. Instead, users should use
173/// [`CheckpointReader`], which provides a safe and ergonomic API for
174/// interacting with the running actor, such as receiving checkpoints, sending
175/// GC signals, or triggering shutdown.
176///
177/// # Responsibilities
178/// - Periodically scans for new checkpoints from configured sources.
179/// - Streams checkpoints to consumers via channels.
180/// - Handles garbage collection signals to prune processed checkpoints.
181/// - Coordinates with remote fetchers for batch downloads and retries.
182///
183/// # Usage
184/// Users should not construct or manage `CheckpointReader` directly. Instead,
185/// use [`CheckpointReader::new`] to spawn the actor and obtain a handle
186/// for interaction.
187struct CheckpointReaderActor {
188    /// Filesystem path to the local checkpoint directory.
189    path: PathBuf,
190    /// Start fetch from the current checkpoint sequence.
191    current_checkpoint_number: CheckpointSequenceNumber,
192    /// Keeps tracks the last processed checkpoint sequence number, used to
193    /// delete checkpoint files from ingestion path.
194    last_pruned_watermark: CheckpointSequenceNumber,
195    /// Channel for sending checkpoints to WorkerPools.
196    checkpoint_tx: mpsc::Sender<Arc<CheckpointData>>,
197    /// Receives the latest garbage collection (GC) signal to prune checkpoint files below
198    /// the specified watermark.
199    gc_signal_rx: watch::Receiver<CheckpointSequenceNumber>,
200    /// Remote checkpoint reader for fetching checkpoints from the network.
201    remote_store: Option<Arc<RemoteStore>>,
202    /// Shutdown signal for the actor.
203    token: CancellationToken,
204    /// Configures the behavior of the checkpoint reader.
205    reader_options: ReaderOptions,
206    /// Limit the amount of downloaded checkpoints held in memory to avoid OOM.
207    data_limiter: DataLimiter,
208    /// Filter applied to transactions within a checkpoint.
209    fullnode_transaction_filter: Option<TransactionFilter>,
210}
211
212impl LocalRead for CheckpointReaderActor {
213    fn exceeds_capacity(&self, checkpoint_number: CheckpointSequenceNumber) -> bool {
214        ((MAX_CHECKPOINTS_IN_PROGRESS as u64 + self.last_pruned_watermark) <= checkpoint_number)
215            || self.data_limiter.exceeds()
216    }
217
218    fn path(&self) -> &Path {
219        &self.path
220    }
221
222    fn current_checkpoint_number(&self) -> CheckpointSequenceNumber {
223        self.current_checkpoint_number
224    }
225
226    fn update_last_pruned_watermark(&mut self, watermark: CheckpointSequenceNumber) {
227        self.last_pruned_watermark = watermark;
228    }
229}
230
231impl CheckpointReaderActor {
232    fn should_fetch_from_remote(&self, checkpoints: &[Arc<CheckpointData>]) -> bool {
233        self.remote_store.is_some()
234            && (checkpoints.is_empty()
235                || self.is_checkpoint_ahead(&checkpoints[0], self.current_checkpoint_number))
236    }
237
238    /// Fetches checkpoints from the historical object store and streams them to
239    /// a channel.
240    async fn relay_from_historical(
241        &mut self,
242        historical_reader: &HistoricalReader,
243    ) -> IngestionResult<()> {
244        // Only sync the manifest when needed to avoid unnecessary network calls.
245        // If the requested checkpoint is beyond what's currently available in our
246        // cached manifest, we need to refresh it to check for newer checkpoints.
247        if self.current_checkpoint_number > historical_reader.latest_available_checkpoint().await? {
248            timeout(
249                Duration::from_secs(self.reader_options.timeout_secs),
250                historical_reader.sync_manifest_once(),
251            )
252            .await
253            .map_err(|_| {
254                IngestionError::HistoryRead("reading manifest exceeded the timeout".into())
255            })??;
256
257            // Verify the requested checkpoint is now available after the manifest refresh.
258            // If it's still not available, the checkpoint hasn't been published yet.
259            if self.current_checkpoint_number
260                > historical_reader.latest_available_checkpoint().await?
261            {
262                return Err(IngestionError::CheckpointNotAvailableYet);
263            }
264        }
265
266        let manifest = historical_reader.get_manifest().await;
267
268        let files = historical_reader.verify_and_get_manifest_files(manifest)?;
269
270        let start_index = match files.binary_search_by_key(&self.current_checkpoint_number, |s| {
271            s.checkpoint_seq_range.start
272        }) {
273            Ok(index) => index,
274            Err(index) => index - 1,
275        };
276
277        for metadata in files
278            .into_iter()
279            .enumerate()
280            .filter_map(|(index, metadata)| (index >= start_index).then_some(metadata))
281        {
282            // We check capacity before downloading the file. This way, once its
283            // checkpoints are collected into memory we ensure that all will be
284            // sent through the channel.
285            if self.exceeds_capacity(self.current_checkpoint_number) {
286                return Err(IngestionError::MaxCheckpointsCapacityReached);
287            }
288
289            let checkpoints = timeout(
290                Duration::from_secs(self.reader_options.timeout_secs),
291                historical_reader.iter_for_file(metadata.file_path()),
292            )
293            .await
294            .map_err(|_| {
295                IngestionError::HistoryRead(format!(
296                    "reading checkpoint {} exceeded the timeout",
297                    metadata.file_path()
298                ))
299            })??
300            .filter(|c| c.checkpoint_summary.sequence_number >= self.current_checkpoint_number)
301            .collect::<Vec<CheckpointData>>();
302
303            // A file that fails to decode part way ends early, so check that it
304            // holds every checkpoint the manifest lists for it before sending any.
305            let expected = self.current_checkpoint_number..metadata.checkpoint_seq_range.end;
306            if !checkpoints
307                .iter()
308                .map(|c| c.checkpoint_summary.sequence_number)
309                .eq(expected.clone())
310            {
311                return Err(IngestionError::HistoryRead(format!(
312                    "checkpoint file {} does not hold checkpoints {expected:?}",
313                    metadata.file_path()
314                )));
315            }
316
317            for checkpoint in checkpoints {
318                let size = bcs::serialized_size(&checkpoint)?;
319                self.data_limiter.add(&checkpoint, size);
320                self.send_checkpoint_to_channel(Arc::new(checkpoint))
321                    .await?;
322            }
323        }
324
325        Ok(())
326    }
327
328    /// Fetches checkpoints from the live object store and streams them to a
329    /// channel.
330    async fn relay_from_live(
331        &mut self,
332        batch_size: usize,
333        live: &dyn ObjectStore,
334    ) -> IngestionResult<()> {
335        let mut checkpoint_stream = (self.current_checkpoint_number..u64::MAX)
336            .map(|checkpoint_number| fetch_from_object_store(live, checkpoint_number))
337            .pipe(futures::stream::iter)
338            .buffered(batch_size);
339        while let Some((checkpoint, size)) = self
340            .token
341            .run_until_cancelled(checkpoint_stream.try_next())
342            .await
343            .transpose()?
344            .flatten()
345        {
346            self.send_remote_checkpoint_with_capacity_check(checkpoint, size)
347                .await?;
348        }
349        Ok(())
350    }
351
352    /// Fetches checkpoints from the fullnode through a gRPC streaming
353    /// connection and streams them to a channel.
354    async fn relay_from_fullnode(&mut self, client: &mut GrpcClient) -> IngestionResult<()> {
355        let mut query = client
356            .checkpoints_stream()
357            .start_sequence_number(self.current_checkpoint_number)
358            .read_mask(iota_grpc_client::CHECKPOINT_RESPONSE_CHECKPOINT_DATA);
359        if let Some(filter) = self.fullnode_transaction_filter.clone() {
360            query = query.transactions_filter(filter.into());
361        }
362        let mut checkpoints_stream = query
363            .await
364            .map_err(|e| {
365                IngestionError::Grpc(format!("failed to initialize the checkpoint stream: {e}"))
366            })?
367            .into_inner();
368
369        while let Some(grpc_checkpoint) = self
370            .token
371            .run_until_cancelled(checkpoints_stream.try_next())
372            .await
373            .transpose()?
374            .flatten()
375        {
376            let checkpoint = grpc_checkpoint.checkpoint_data()?.try_into()?;
377            let size = bcs::serialized_size(&checkpoint)?;
378            self.send_remote_checkpoint_with_capacity_check(Arc::new(checkpoint), size)
379                .await?;
380        }
381
382        Ok(())
383    }
384
385    /// Fetches remote checkpoints from the remote store and streams them to the
386    /// channel.
387    ///
388    /// For every successfully fetched checkpoint, this function updates the
389    /// current checkpoint number and the data limiter. If an error occurs while
390    /// fetching a checkpoint, the function returns immediately with that error.
391    async fn fetch_and_send_to_channel(&mut self) -> IngestionResult<()> {
392        let Some(remote_store) = self.remote_store.as_ref().map(Arc::clone) else {
393            return Ok(());
394        };
395        let batch_size = self.reader_options.batch_size;
396        match remote_store.as_ref() {
397            RemoteStore::Fullnode(client) => {
398                self.relay_from_fullnode(&mut client.clone()).await?;
399            }
400            RemoteStore::HybridHistoricalStore { historical, live } => {
401                if let Some(Err(err)) = self
402                    .token
403                    .clone()
404                    .run_until_cancelled(self.relay_from_historical(historical))
405                    .await
406                {
407                    if matches!(err, IngestionError::CheckpointNotAvailableYet) {
408                        let live = live.as_ref().ok_or(err)?;
409                        return self.relay_from_live(batch_size, live).await;
410                    }
411                    return Err(err);
412                }
413            }
414        };
415        Ok(())
416    }
417
418    /// Fetches and sends checkpoints to the channel with retry logic.
419    ///
420    /// Uses an exponential backoff strategy to retry failed requests.
421    async fn fetch_and_send_to_channel_with_retry(&mut self) {
422        let mut backoff = backoff::ExponentialBackoff::default();
423        backoff.max_elapsed_time = Some(Duration::from_secs(60));
424        backoff.initial_interval = Duration::from_millis(100);
425        backoff.current_interval = backoff.initial_interval;
426        backoff.multiplier = 1.0;
427
428        loop {
429            match self.fetch_and_send_to_channel().await {
430                Ok(_) => break,
431                Err(IngestionError::MaxCheckpointsCapacityReached) => break,
432                Err(IngestionError::CheckpointNotAvailableYet) => {
433                    break debug!("historical reader does not have the requested checkpoint yet");
434                }
435                Err(err) => match backoff.next_backoff() {
436                    Some(duration) => {
437                        if !err.to_string().to_lowercase().contains("not found") {
438                            debug!(
439                                "remote reader retry in {} ms. Error is {err:?}",
440                                duration.as_millis(),
441                            );
442                        }
443                        if self
444                            .token
445                            .run_until_cancelled(tokio::time::sleep(duration))
446                            .await
447                            .is_none()
448                        {
449                            break;
450                        }
451                    }
452                    None => {
453                        break error!("remote reader transient error {err:?}");
454                    }
455                },
456            }
457        }
458    }
459
460    /// Attempts to send a checkpoint from remote source to the channel if
461    /// capacity allows.
462    ///
463    /// If the checkpoint's sequence number would exceed the allowed capacity,
464    /// returns `IngestionError::MaxCheckpointsCapacityReached` and does not
465    /// send. Otherwise, adds the checkpoint to the data limiter and sends
466    /// it to the channel.
467    async fn send_remote_checkpoint_with_capacity_check(
468        &mut self,
469        checkpoint: Arc<CheckpointData>,
470        size: usize,
471    ) -> IngestionResult<()> {
472        if self.exceeds_capacity(checkpoint.checkpoint_summary.sequence_number) {
473            return Err(IngestionError::MaxCheckpointsCapacityReached);
474        }
475        self.data_limiter.add(&checkpoint, size);
476        self.send_checkpoint_to_channel(checkpoint).await
477    }
478
479    /// Sends a batch of local checkpoints to the channel in order.
480    ///
481    /// Each checkpoint is sent sequentially until a gap is detected (i.e., a
482    /// checkpoint with a sequence number greater than the current
483    /// checkpoint number). If a gap is found, the function breaks early. If
484    /// sending fails, returns the error immediately.
485    async fn send_local_checkpoints_to_channel(
486        &mut self,
487        checkpoints: Vec<Arc<CheckpointData>>,
488    ) -> IngestionResult<()> {
489        for checkpoint in checkpoints {
490            if self.is_checkpoint_ahead(&checkpoint, self.current_checkpoint_number) {
491                break;
492            }
493            self.send_checkpoint_to_channel(checkpoint).await?;
494        }
495        Ok(())
496    }
497
498    /// Sends a single checkpoint to the channel and advances the current
499    /// checkpoint number.
500    ///
501    /// Asserts that the checkpoint's sequence number matches the expected
502    /// current number. Increments the current checkpoint number after
503    /// sending.
504    async fn send_checkpoint_to_channel(
505        &mut self,
506        checkpoint: Arc<CheckpointData>,
507    ) -> IngestionResult<()> {
508        assert_eq!(
509            checkpoint.checkpoint_summary.sequence_number,
510            self.current_checkpoint_number
511        );
512        self.checkpoint_tx.send(checkpoint).await.map_err(|_| {
513            IngestionError::Channel(
514                "unable to send checkpoint to executor, receiver half closed".to_owned(),
515            )
516        })?;
517        self.current_checkpoint_number += 1;
518        Ok(())
519    }
520
521    /// Sync from either local or remote source new checkpoints to be processed
522    /// by the executor.
523    async fn sync(&mut self) -> IngestionResult<()> {
524        let mut remote_source = ReadSource::Local;
525        let checkpoints = self.read_local_files_with_retry().await?;
526        let should_fetch_from_remote = self.should_fetch_from_remote(&checkpoints);
527
528        if should_fetch_from_remote {
529            remote_source = ReadSource::Remote;
530            self.fetch_and_send_to_channel_with_retry().await;
531        } else {
532            self.send_local_checkpoints_to_channel(checkpoints).await?;
533        }
534
535        debug!(
536            "Read from {remote_source}. Current checkpoint number: {}, pruning watermark: {}",
537            self.current_checkpoint_number, self.last_pruned_watermark,
538        );
539
540        Ok(())
541    }
542
543    /// Run the main loop of the checkpoint reader actor.
544    async fn run(mut self) {
545        let (_inotify_tx, mut inotify_rx) = mpsc::channel::<()>(1);
546        std::fs::create_dir_all(self.path()).expect("failed to create a directory");
547
548        #[cfg(not(target_os = "macos"))]
549        let _watcher = init_watcher(_inotify_tx, self.path());
550
551        self.data_limiter.gc(self.last_pruned_watermark);
552        self.gc_processed_files(self.last_pruned_watermark)
553            .expect("failed to clean the directory");
554
555        loop {
556            tokio::select! {
557                _ = self.token.cancelled() => break,
558                Ok(()) = self.gc_signal_rx.changed() => {
559                    let watermark = *self.gc_signal_rx.borrow_and_update();
560                    self.data_limiter.gc(watermark);
561                    self.gc_processed_files(watermark).expect("failed to clean the directory");
562                }
563                Ok(Some(_)) | Err(_) = timeout(Duration::from_millis(self.reader_options.tick_interval_ms), inotify_rx.recv())  => {
564                    self.sync().await.expect("failed to read checkpoint files");
565                }
566            }
567        }
568    }
569}
570
571/// Public API for interacting with the checkpoint reader actor.
572///
573/// It provides methods to receive streamed checkpoints, send garbage collection
574/// signals, and gracefully shut down the background checkpoint reading task.
575/// Internally, it communicates with a [`CheckpointReaderActor`], which manages
576/// the actual checkpoint fetching and streaming logic.
577pub(crate) struct CheckpointReader {
578    handle: JoinHandle<()>,
579    gc_signal_tx: watch::Sender<CheckpointSequenceNumber>,
580    checkpoint_rx: mpsc::Receiver<Arc<CheckpointData>>,
581    token: CancellationToken,
582}
583
584impl CheckpointReader {
585    pub(crate) async fn new(
586        starting_checkpoint_number: CheckpointSequenceNumber,
587        config: CheckpointReaderConfigExt,
588    ) -> IngestionResult<Self> {
589        if config.fullnode_transaction_filter.is_some()
590            && !matches!(config.base.remote_store_url, Some(RemoteUrl::Fullnode(_)))
591        {
592            return Err(IngestionError::Unsupported(
593                "filter is only supported on `RemoteUrl::Fullnode` connections".into(),
594            ));
595        }
596
597        let (checkpoint_tx, checkpoint_rx) = mpsc::channel(MAX_CHECKPOINTS_IN_PROGRESS);
598        let (gc_signal_tx, gc_signal_rx) = watch::channel(starting_checkpoint_number);
599
600        let remote_store = if let Some(url) = config.base.remote_store_url {
601            Some(Arc::new(
602                RemoteStore::new(
603                    url,
604                    config.base.reader_options.batch_size,
605                    config.base.reader_options.timeout_secs,
606                )
607                .await?,
608            ))
609        } else {
610            None
611        };
612
613        let path = match config.base.ingestion_path {
614            Some(p) => p,
615            None => tempfile::tempdir()?.keep(),
616        };
617        let token = CancellationToken::new();
618        let reader = CheckpointReaderActor {
619            path,
620            current_checkpoint_number: starting_checkpoint_number,
621            last_pruned_watermark: starting_checkpoint_number,
622            checkpoint_tx,
623            gc_signal_rx,
624            remote_store,
625            token: token.clone(),
626            data_limiter: DataLimiter::new(config.base.reader_options.data_limit),
627            reader_options: config.base.reader_options,
628            fullnode_transaction_filter: config.fullnode_transaction_filter,
629        };
630
631        let handle = spawn_monitored_task!(reader.run());
632
633        Ok(Self {
634            handle,
635            gc_signal_tx,
636            checkpoint_rx,
637            token,
638        })
639    }
640
641    /// Read downloaded checkpoints from the queue.
642    pub(crate) async fn checkpoint(&mut self) -> Option<Arc<CheckpointData>> {
643        self.checkpoint_rx.recv().await
644    }
645
646    /// Sends a garbage collection (GC) signal to the checkpoint reader.
647    ///
648    /// Transmits a watermark to the checkpoint reader, indicating that all
649    /// checkpoints below this watermark can be safely pruned or cleaned up.
650    /// The new watermark replaces the previous one.
651    ///
652    /// # Errors
653    ///
654    /// Returns [`IngestionError::Channel`] if the checkpoint reader task
655    /// stopped.
656    pub(crate) fn send_gc_signal(
657        &self,
658        watermark: CheckpointSequenceNumber,
659    ) -> IngestionResult<()> {
660        self.gc_signal_tx.send(watermark).map_err(|_| {
661            IngestionError::Channel(
662                "unable to send GC operation to checkpoint reader, receiver half closed".into(),
663            )
664        })
665    }
666
667    /// Gracefully shuts down the checkpoint reader task.
668    ///
669    /// It signals the background checkpoint reader actor to terminate, then
670    /// awaits its completion. Any in-progress checkpoint reading or streaming
671    /// operations will be stopped as part of the shutdown process.
672    pub(crate) async fn shutdown(self) -> IngestionResult<()> {
673        self.token.cancel();
674        self.handle.await.map_err(|err| IngestionError::Shutdown {
675            component: "checkpoint reader".into(),
676            msg: err.to_string(),
677        })
678    }
679}
iota_data_ingestion_core/reader/v2.rs

iota_data_ingestion_core/reader/
v2.rs
