Repository: qdrant/qdrant
File: lib/collection/src/update_workers/update_worker.rs (Rust)

Code before:
impl UpdateWorkers {
    #[allow(clippy::too_many_arguments)]
    pub async fn update_worker_fn(
        collection_name: CollectionId,
        mut receiver: Receiver<UpdateSignal>,
        optimize_sender: Sender<OptimizerSignal>,
        wal: LockedWal,
        segments: LockedSegmentHolder,
        update_operation_lock: Arc<tokio::sync::RwLock<()>>,
        update_tracker: UpdateTracker,
        prevent_unoptimized_threshold_kb: Option<usize>,
        optimization_handles: Arc<TokioMutex<Vec<StoppableTaskHandle<bool>>>>,
        mut optimization_finished_receiver: watch::Receiver<()>,
        applied_seq_handler: Arc<AppliedSeqHandler>,
        skip_uploads: Arc<AtomicBool>,
    ) {
        while let Some(signal) = receiver.recv().await {
            match signal {
                UpdateSignal::Operation(OperationData {
                    op_num,
                    operation,
                    sender,
                    hw_measurements,
                }) => {
                    if skip_uploads.load(std::sync::atomic::Ordering::Relaxed) {
                        continue;
                    }

                    let collection_name_clone = collection_name.clone();
                    let wal_clone = wal.clone();
                    let segments_clone = segments.clone();
                    let update_operation_lock_clone = update_operation_lock.clone();
                    let update_tracker_clone = update_tracker.clone();

                    let operation = if let Some(operation) = operation {
                        *operation
                    } else {
                        let record = wal.lock().await.read_single_record(op_num);
                        match record {
                            Ok(Some(op)) => op.operation,
                            Ok(None) => {
                                if let Some(feedback) = sender {
                                    feedback.send(Err(CollectionError::service_error(
                                        format!("Operation {op_num} not found in WAL"),
                                    ))).unwrap_or_else(|_| {
                                        log::debug!("Can't report operation {op_num} result. Assume already not required");
                                    });
                                }
                                continue;
                            }
                            Err(err) => {

Review comment (suggestion):
The idea is that we drop the entire operation, including the `sender`. And other places that wait on such sender should therefore error and return early. Am I correct?

If that is true, I vote to add a comment describing this.


```suggestion
                    if skip_uploads.load(std::sync::atomic::Ordering::Relaxed) {
                        // This will also drop `sender`, unblocking any operations waiting on it
                        continue;
```

Code after:
impl UpdateWorkers {
    #[allow(clippy::too_many_arguments)]
    pub async fn update_worker_fn(
        collection_name: CollectionId,
        mut receiver: Receiver<UpdateSignal>,
        optimize_sender: Sender<OptimizerSignal>,
        wal: LockedWal,
        segments: LockedSegmentHolder,
        update_operation_lock: Arc<tokio::sync::RwLock<()>>,
        update_tracker: UpdateTracker,
        prevent_unoptimized_threshold_kb: Option<usize>,
        optimization_handles: Arc<TokioMutex<Vec<StoppableTaskHandle<bool>>>>,
        mut optimization_finished_receiver: watch::Receiver<()>,
        applied_seq_handler: Arc<AppliedSeqHandler>,
        skip_updates: Arc<AtomicBool>,
    ) {
        // First skipped op_num while `skip_updates` is set
        let mut first_skipped: Option<SeqNumberType> = None;

        while let Some(signal) = receiver.recv().await {
            match signal {
                UpdateSignal::Operation(OperationData {
                    op_num,
                    operation,
                    sender,
                    hw_measurements,
                }) => {
                    if skip_updates.load(std::sync::atomic::Ordering::Relaxed) {
                        first_skipped.get_or_insert(op_num);

                        if let Some(feedback) = sender {
                            feedback.send(Err(CollectionError::service_error(
                                format!("Operation {op_num} skipped due to ongoing WAL truncation"),
                            ))).unwrap_or_else(|_| {
                                log::debug!("Can't report operation {op_num} result. Assume already not required");
                            });
                        }
                        continue;
                    }

                    first_skipped = None;

                    let collection_name_clone = collection_name.clone();
                    let wal_clone = wal.clone();
                    let segments_clone = segments.clone();
                    let update_operation_lock_clone = update_operation_lock.clone();
                    let update_tracker_clone = update_tracker.clone();

                    let operation = if let Some(operation) = operation {
                        *operation
                    } else {