Repository: qdrant/qdrant
File: lib/segment/src/segment/entry.rs (Rust)

Code before:
        //                  └───┘       └───┘   └───┘   └───┘
        //  WAL:      ▲
        //            │                 ┌───┐   ┌───┐
        //  100───────┘      ┌────────► │200├───┤3  │
        //                   |          └───┘   └───┘
        //  200──────────────┘
        //
        //  300
        //
        //  400

        let is_alive_flush_lock = self.is_alive_flush_lock.handle();

        let flush_op = move || {
            let Some(is_alive_flush_guard) = is_alive_flush_lock.lock_if_alive() else {
                // Segment is removed, skip flush
                log::debug!("Segment was dropped, skip flush");
                return Ok(());
            };

            let flush_components = || {
                // Flush mapping first to prevent having orphan internal ids.
                id_tracker_mapping_flusher().map_err(|err| {
                    OperationError::service_error(format!(
                        "Failed to flush id_tracker mapping: {err}"
                    ))
                })?;
                for vector_storage_flusher in vector_storage_flushers {
                    vector_storage_flusher().map_err(|err| {
                        OperationError::service_error(format!(
                            "Failed to flush vector_storage: {err}"
                        ))
                    })?;
                }
                for quantization_flusher in quantization_flushers {
                    quantization_flusher().map_err(|err| {
                        OperationError::service_error(format!(
                            "Failed to flush quantized vectors: {err}"
                        ))
                    })?;
                }
                payload_index_flusher().map_err(|err| {
                    OperationError::service_error(format!("Failed to flush payload_index: {err}"))
                })?;
                // Id Tracker contains versions of points. We need to flush it after vector_storage and payload_index flush.
                // This is because vector_storage and payload_index flush are not atomic.
                // If payload or vector flush fails, we will be able to recover data from WAL.
                // If Id Tracker flush fails, we are also able to recover data from WAL
                //  by simply overriding data in vector and payload storages.
                // Once versions are saved - points are considered persisted.
                id_tracker_versions_flusher().map_err(|err| {

Review comment (suggestion):
We wrap potential cancellation errors here as service error. This breaks cancellation handling further below!

Code after:
        //                  └───┘       └───┘   └───┘   └───┘
        //  WAL:      ▲
        //            │                 ┌───┐   ┌───┐
        //  100───────┘      ┌────────► │200├───┤3  │
        //                   |          └───┘   └───┘
        //  200──────────────┘
        //
        //  300
        //
        //  400

        let is_alive_flush_lock = self.is_alive_flush_lock.handle();

        let flush_op = move || {
            let Some(is_alive_flush_guard) = is_alive_flush_lock.lock_if_alive() else {
                // Segment is removed, skip flush
                log::debug!("Segment was dropped, skip flush");
                return Ok(());
            };

            let flush_components = || {
                // Flush mapping first to prevent having orphan internal ids.
                id_tracker_mapping_flusher().map_err(|err| match err {
                    OperationError::Cancelled { .. } => err,
                    _ => OperationError::service_error(format!(
                        "Failed to flush id_tracker mapping: {err}"
                    )),
                })?;
                for vector_storage_flusher in vector_storage_flushers {
                    vector_storage_flusher().map_err(|err| match err {
                        OperationError::Cancelled { .. } => err,
                        _ => OperationError::service_error(format!(
                            "Failed to flush vector_storage: {err}"
                        )),
                    })?;
                }
                for quantization_flusher in quantization_flushers {
                    quantization_flusher().map_err(|err| match err {
                        OperationError::Cancelled { .. } => err,
                        _ => OperationError::service_error(format!(
                            "Failed to flush quantized vectors: {err}"
                        )),
                    })?;
                }
                payload_index_flusher().map_err(|err| match err {
                    OperationError::Cancelled { .. } => err,
                    _ => OperationError::service_error(format!(
                        "Failed to flush payload_index: {err}"
                    )),
                })?;
                // Id Tracker contains versions of points. We need to flush it after vector_storage and payload_index flush.