Repository: MystenLabs/sui
File: crates/sui-indexer-alt-framework/src/lib.rs (Rust)

Code before:
            indexer_args,
            client_args,
            ingestion_config,
            None,
            &registry,
        )
        .await
        .unwrap();
        let mut allow_process = 10;
        // Limit the pipeline to process only checkpoints `[0, 10]`.
        let (controllable_handler, process_below) = ControllableHandler::with_limit(allow_process);
        let _ = tasked_indexer
            .concurrent_pipeline(
                controllable_handler,
                ConcurrentConfig {
                    committer: CommitterConfig {
                        collect_interval_ms: 10,
                        watermark_interval_ms: 10,
                        ..Default::default()
                    },
                    ..Default::default()
                },
            )
            .await;
        let metrics = tasked_indexer.indexer_metrics().clone();
        let indexer_handle = tokio::spawn(async move {
            tasked_indexer.run().await.unwrap().join().await.unwrap();
        });

        // Wait for pipeline to commit up to configured checkpoint 10 inclusive. With the main
        // pipeline `reader_lo` currently unset, all checkpoints are allowed and should be
        // committed.
        store
            .wait_for_watermark(
                &pipeline_task::<MockStore>(ControllableHandler::NAME, Some("task")).unwrap(),
                10,
                std::time::Duration::from_secs(10),
            )
            .await;

        // Set the reader_lo to 250, simulating the main pipeline getting ahead. The
        // track_main_reader_lo task will eventually pick this up and update the atomic. The
        // collector reads from the atomic when it receives checkpoints, so we release checkpoints
        // one at a time until the collector_reader_lo metric shows the new value.
        conn.set_reader_watermark(ControllableHandler::NAME, 250)
            .await
            .unwrap();

        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            let mut interval = tokio::time::interval(std::time::Duration::from_millis(10));


Review comment (refactor):
Wrapping this in `tokio::spawn` somewhat defeats the point of the `Service` abstraction 🥲. You should just be able to call `let s_indexer = tasked_indexer.run().await.unwrap();`, here and then later when you want to wait for it to finish, call `s_indexer.join().await.unwrap();`.

Code after:
            indexer_args,
            client_args,
            ingestion_config,
            None,
            &registry,
        )
        .await
        .unwrap();
        let mut allow_process = 10;
        // Limit the pipeline to process only checkpoints `[0, 10]`.
        let (controllable_handler, process_below) = ControllableHandler::with_limit(allow_process);
        let _ = tasked_indexer
            .concurrent_pipeline(
                controllable_handler,
                ConcurrentConfig {
                    committer: CommitterConfig {
                        collect_interval_ms: 10,
                        watermark_interval_ms: 10,
                        ..Default::default()
                    },
                    ..Default::default()
                },
            )
            .await;
        let metrics = tasked_indexer.indexer_metrics().clone();

        let mut s_indexer = tasked_indexer.run().await.unwrap();

        // Wait for pipeline to commit up to configured checkpoint 10 inclusive. With the main
        // pipeline `reader_lo` currently unset, all checkpoints are allowed and should be
        // committed.
        store
            .wait_for_watermark(
                &pipeline_task::<MockStore>(ControllableHandler::NAME, Some("task")).unwrap(),
                10,
                std::time::Duration::from_secs(10),
            )
            .await;

        // Set the reader_lo to 250, simulating the main pipeline getting ahead. The
        // track_main_reader_lo task will eventually pick this up and update the atomic. The
        // collector reads from the atomic when it receives checkpoints, so we release checkpoints
        // one at a time until the collector_reader_lo metric shows the new value.
        conn.set_reader_watermark(ControllableHandler::NAME, 250)
            .await
            .unwrap();

        let reader_lo = metrics
            .collector_reader_lo
            .with_label_values(&[ControllableHandler::NAME]);