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

Code before:
            .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));

            loop {
                interval.tick().await;
                process_below.send(allow_process).ok();

                let current = metrics
                    .collector_reader_lo
                    .with_label_values(&[ControllableHandler::NAME])
                    .get();
                if current == 250 {
                    break;
                }
                allow_process += 1;
            }
        })
        .await
        .expect("Timed out waiting for collector to observe new reader_lo");

        // At this point, the collector has observed reader_lo = 250. Release all remaining
        // checkpoints. Guarantees:
        // - [0, 10]: committed (before reader_lo was set)
        // - [11, allow_process]: some committed, some skipped (timing-dependent during detection)
        // - (allow_process, 250): skipped (in-flight, filtered by collector)
        // - [250, 500]: committed (>= reader_lo)
        process_below.send(500).ok();

        let _ = indexer_handle.await;

        let data = store.data.get(ControllableHandler::NAME).unwrap();

        // In-flight checkpoints (allow_process, 250) must be skipped.
        let num_must_skipped = 250 - (allow_process + 1);
        let skipped = metrics
            .total_collector_skipped_checkpoints
            .with_label_values(&[ControllableHandler::NAME])
            .get();
        assert!(
            skipped >= num_must_skipped,
            "Expected at least {num_must_skipped} skipped, got {skipped}"

Review comment (suggestion):
```rust
let reader_lo = metrics
    .collector_reader_lo
    .with_label_values(&[ControllableHandler::NAME]);

while (reader_lo.get() < 250) {
    interval.tick().await;
    process_below.send(allow_process).ok();
    allow_process += 1;
}
```

Code after:

        // 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]);

        // Send checkpoints one at a time at 10ms intervals. The tasked indexer has a reader refresh
        // interval of 10ms as well, so the collector should pick up the new reader_lo after a few
        // checkpoints have been processed.
        let mut interval = tokio::time::interval(std::time::Duration::from_millis(10));
        while reader_lo.get() != 250 {
            interval.tick().await;
            // allow_process is initialized to 11, bump to 11 for the next checkpoint
            allow_process += 1;
            assert!(
                allow_process <= 500,
                "Released all checkpoints but collector never observed new reader_lo"
            );
            process_below.send(allow_process).ok();
        }

        // At this point, the collector has observed reader_lo = 250. Release all remaining
        // checkpoints. Guarantees:
        // - [0, 10]: committed (before reader_lo was set)
        // - [11, allow_process]: some committed, some skipped (timing-dependent during detection)
        // - (allow_process, 250): skipped (in-flight, filtered by collector)
        // - [250, 500]: committed (>= reader_lo)
        process_below.send(500).ok();

        s_indexer.join().await.unwrap();

        let data = store.data.get(ControllableHandler::NAME).unwrap();

        // Checkpoints (allow_process, 250) must be skipped.
        for chkpt in (allow_process + 1)..250 {
            assert!(
                data.get(&chkpt).is_none(),
                "Checkpoint {chkpt} should have been skipped"
            );
        }

        // Checkpoints >= reader_lo must be committed.
        for chkpt in 250..=500 {
            assert!(