Repository: meilisearch/meilisearch
File: crates/milli/src/update/new/indexer/write.rs (Rust)

Code before:
                writer.del_items(wtxn, *dimensions, docid)?;
                writer.add_items(wtxn, docid, &embeddings)?;
            }
        }

        // Every time the is a message in the channel we search
        // for new entries in the BBQueue buffers.
        write_from_bbqueue(
            &mut writer_receiver,
            index,
            wtxn,
            arroy_writers,
            &mut aligned_embedding,
        )?;
    }

    write_from_bbqueue(&mut writer_receiver, index, wtxn, arroy_writers, &mut aligned_embedding)?;

    let direct_attempts = writer_receiver.sent_messages_attempts();
    let blocking_attempts = writer_receiver.blocking_sent_messages_attempts();
    let congestion_pct = (blocking_attempts as f64 / direct_attempts as f64) * 100.0;
    tracing::debug!(
        "Channel congestion metrics - \
        Direct send attempts: {direct_attempts}, \
        Blocking send attempts: {blocking_attempts} \
        ({congestion_pct:.1}% congestion)"
    );

    Ok(())
}

#[tracing::instrument(level = "trace", skip_all, target = "indexing::vectors")]
pub(super) fn build_vectors<MSP>(
    index: &Index,
    wtxn: &mut RwTxn<'_>,
    index_embeddings: Vec<IndexEmbeddingConfig>,
    arroy_writers: &mut HashMap<u8, (&str, &Embedder, ArroyWrapper, usize)>,
    must_stop_processing: &MSP,
) -> Result<()>
where
    MSP: Fn() -> bool + Sync + Send,
{
    if index_embeddings.is_empty() {
        return Ok(());
    }

    let mut rng = rand::rngs::StdRng::seed_from_u64(42);
    for (_index, (_embedder_name, _embedder, writer, dimensions)) in arroy_writers {
        let dimensions = *dimensions;
        writer.build_and_quantize(wtxn, &mut rng, dimensions, false, must_stop_processing)?;
    }

Review comment (suggestion):
```suggestion
        Attempts: {direct_attempts}, \
        Blocked attempts: {blocking_attempts} \
        ({congestion_pct:.1}% congestion)"
```

Code after:
                writer.del_items(wtxn, *dimensions, docid)?;
                writer.add_items(wtxn, docid, &embeddings)?;
            }
        }

        // Every time the is a message in the channel we search
        // for new entries in the BBQueue buffers.
        write_from_bbqueue(
            &mut writer_receiver,
            index,
            wtxn,
            arroy_writers,
            &mut aligned_embedding,
        )?;
    }

    write_from_bbqueue(&mut writer_receiver, index, wtxn, arroy_writers, &mut aligned_embedding)?;

    let direct_attempts = writer_receiver.sent_messages_attempts();
    let blocking_attempts = writer_receiver.blocking_sent_messages_attempts();
    let congestion_pct = (blocking_attempts as f64 / direct_attempts as f64) * 100.0;
    tracing::debug!(
        "Channel congestion metrics - \
        Attempts: {direct_attempts}, \
        Blocked attempts: {blocking_attempts} \
        ({congestion_pct:.1}% congestion)"
    );

    Ok(())
}

#[tracing::instrument(level = "trace", skip_all, target = "indexing::vectors")]
pub(super) fn build_vectors<MSP>(
    index: &Index,
    wtxn: &mut RwTxn<'_>,
    index_embeddings: Vec<IndexEmbeddingConfig>,
    arroy_writers: &mut HashMap<u8, (&str, &Embedder, ArroyWrapper, usize)>,
    must_stop_processing: &MSP,
) -> Result<()>
where
    MSP: Fn() -> bool + Sync + Send,
{
    if index_embeddings.is_empty() {
        return Ok(());
    }

    let mut rng = rand::rngs::StdRng::seed_from_u64(42);
    for (_index, (_embedder_name, _embedder, writer, dimensions)) in arroy_writers {
        let dimensions = *dimensions;
        writer.build_and_quantize(wtxn, &mut rng, dimensions, false, must_stop_processing)?;
    }