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

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

        let flush_op = move || {
            // 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| {
                OperationError::service_error(format!("Failed to flush id_tracker versions: {err}"))
            })?;

            let mut current_persisted_version_guard = persisted_version.lock();
            let persisted_version_value_opt = *current_persisted_version_guard;

            if persisted_version_value_opt > state.version {
                debug_assert!(

Review comment (suggestion):
I'd prefer a separate `Mutex<()>` and stopped flag, I think. It's a mechanism we use already, and then we can more easily propagate the flag throughout the rest of the flusher operation.

But it's not critical. Let's merge this and revisit it if it ever becomes important.

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

        let is_alive_flush_lock = self.is_alive_flush_lock.clone();

        let flush_op = move || {
            // Keep the guard till the end of the flush to prevent concurrent flushes
            let is_alive_flush_guard = is_alive_flush_lock.lock();

            if !*is_alive_flush_guard {
                // Segment is removed, skip flush
                return Ok(());
            }

            // 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.