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

Code before:
    metrics: Arc<IndexerMetrics>,
) -> Service {
    Service::new().spawn_aborting(async move {
        // The `poll` interval controls the maximum time to wait between collecting batches,
        // regardless of number of rows pending.
        let mut poll = interval(config.collect_interval());
        poll.set_missed_tick_behavior(MissedTickBehavior::Delay);

        let checkpoint_lag_reporter = CheckpointLagMetricReporter::new_for_pipeline::<H>(
            &metrics.collected_checkpoint_timestamp_lag,
            &metrics.latest_collected_checkpoint_timestamp_lag_ms,
            &metrics.latest_collected_checkpoint,
        );

        // Data for checkpoints that are ready to be sent but haven't been written yet.
        let mut pending: BTreeMap<u64, PendingCheckpoint<H>> = BTreeMap::new();
        let mut pending_rows = 0;

        info!(pipeline = H::NAME, "Starting collector");

        // Wait for main_reader_lo to be initialized before processing any checkpoints.
        let reader_lo_atomic = main_reader_lo.wait().await;

        loop {
            // Eager inner loop: drain processor channel while under backpressure limit.
            while pending_rows < H::MAX_PENDING_ROWS {
                match rx.try_recv() {
                    Ok(mut indexed) => {
                        let reader_lo = reader_lo_atomic.load(Ordering::Relaxed);
                        if indexed.checkpoint() < reader_lo {
                            indexed.values.clear();
                            metrics
                                .total_collector_skipped_checkpoints
                                .with_label_values(&[H::NAME])
                                .inc();
                        }

                        metrics
                            .total_collector_rows_received
                            .with_label_values(&[H::NAME])
                            .inc_by(indexed.len() as u64);
                        metrics
                            .total_collector_checkpoints_received
                            .with_label_values(&[H::NAME])
                            .inc();
                        metrics
                            .collector_reader_lo
                            .with_label_values(&[H::NAME])
                            .set(reader_lo as i64);

                        pending_rows += indexed.len();

Review comment (suggestion):
ok, similarly here, previously we receive one checkpoint from the channel, now we greedily drain

`try_recv()` is the equivalent to `now_or_never()` in the stream code? non-blocking check to grab everything already available

Code after:
    metrics: Arc<IndexerMetrics>,
) -> Service {
    Service::new().spawn_aborting(async move {
        // The `poll` interval controls the maximum time to wait between collecting batches,
        // regardless of number of rows pending.
        let mut poll = interval(config.collect_interval());
        poll.set_missed_tick_behavior(MissedTickBehavior::Delay);

        let checkpoint_lag_reporter = CheckpointLagMetricReporter::new_for_pipeline::<H>(
            &metrics.collected_checkpoint_timestamp_lag,
            &metrics.latest_collected_checkpoint_timestamp_lag_ms,
            &metrics.latest_collected_checkpoint,
        );

        // Data for checkpoints that are ready to be sent but haven't been written yet.
        let mut pending: BTreeMap<u64, PendingCheckpoint<H>> = BTreeMap::new();
        let mut pending_rows = 0;

        info!(pipeline = H::NAME, "Starting collector");

        // Wait for main_reader_lo to be initialized before processing any checkpoints.
        let reader_lo_atomic = main_reader_lo.wait().await;

        loop {
            // Eager inner loop: drain processor channel while under backpressure limit.
            while pending_rows < H::MAX_PENDING_ROWS {
                match rx.try_recv() {
                    Ok(indexed) => {
                        pending_rows += receive_checkpoint::<H>(
                            indexed,
                            &mut pending,
                            reader_lo_atomic,
                            &metrics,
                        );
                    }
                    Err(_) => break,
                }
            }

            // Eagerly build and send batches if enough data is pending. Once we start
            // batching, flush everything including any small tail below the threshold.
            if pending_rows >= H::MIN_EAGER_ROWS {
                loop {
                    let (batch, watermark, batch_len) = build_batch::<H>(
                        &*handler,
                        &mut pending,
                        &checkpoint_lag_reporter,
                        &metrics,
                    );
                    pending_rows -= batch_len;