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

Code before:
                                draining = true;
                            }
                            continue;
                        }
                        active += 1;
                        join_set.spawn(commit_batch::<H>(
                            handler.clone(),
                            db.clone(),
                            metrics.clone(),
                            checkpoint_lag_reporter.clone(),
                            batch,
                        ));
                    }
                    Err(TryRecvError::Empty) => break,
                    Err(TryRecvError::Disconnected) => {
                        draining = true;
                        break;
                    }
                }
            }

            if draining && active == 0 {
                break;
            }

            tokio::select! {
                biased;

                // Prioritize completions to free capacity
                Some(res) = join_set.join_next() => {
                    active -= 1;
                    match res {
                        Ok(Ok(watermark)) => {
                            if tx.send(watermark).await.is_err() {
                                info!(pipeline = H::NAME, "Watermark closed channel");
                                draining = true;
                            }
                        }
                        Ok(Err(e)) => {
                            fatal_error = Some(e);
                            draining = true;
                        }
                        Err(e) => {
                            fatal_error = Some(anyhow::anyhow!("Committer task panicked: {e}"));
                            draining = true;
                        }
                    }
                }

                // Then read new items
                result = rx.recv(), if !draining && active < limit => {

Review comment (performance):
so there are three exit conditions

- `draining` is true (channel disconnected, watermark channel closed)
- `active >= limit` all concurrency slots full
- `try_recv` channel has no further batches

At the select, for most of the exit conditions `join_next` is the only valid branch, which drains `active` and then re-loops

except for `try_recv` empty, then both branches are valid, and either we finish a commit or a new batch arrives first, either way we loop back ... or if no tasks are in flight, only `rx.recv()` is live, so we wait until collector sends something

Code after:
                    // Repeatedly try to get a connection to the DB and write the batch. Use an
                    // exponential backoff in case the failure is due to contention over the DB
                    // connection pool.
                    let backoff = ExponentialBackoff {
                        initial_interval: INITIAL_RETRY_INTERVAL,
                        current_interval: INITIAL_RETRY_INTERVAL,
                        max_interval: MAX_RETRY_INTERVAL,
                        max_elapsed_time: None,
                        ..Default::default()
                    };

                    let highest_checkpoint = watermark.iter().map(|w| w.checkpoint()).max();
                    let highest_checkpoint_timestamp =
                        watermark.iter().map(|w| w.timestamp_ms()).max();

                    use backoff::Error as BE;
                    let commit = move || {
                        let batch = batch.clone();
                        let handler = handler.clone();
                        let db = db.clone();
                        let metrics = metrics.clone();
                        let checkpoint_lag_reporter = checkpoint_lag_reporter.clone();
                        async move {
                            if batch_len == 0 {
                                return Ok(());
                            }

                            metrics
                                .total_committer_batches_attempted
                                .with_label_values(&[H::NAME])
                                .inc();

                            let guard = metrics
                                .committer_commit_latency
                                .with_label_values(&[H::NAME])
                                .start_timer();

                            let mut conn = db.connect().await.map_err(|e| {
                                warn!(
                                    pipeline = H::NAME,
                                    "Committed failed to get connection for DB"
                                );

                                metrics
                                    .total_committer_batches_failed
                                    .with_label_values(&[H::NAME])
                                    .inc();

                                BE::transient(Break::Err(e))
                            })?;