Repository: qdrant/qdrant
File: lib/segment/src/common/flags/buffered_dynamic_flags.rs (Rust)

Code before:
    }

    pub fn flusher(&self) -> Flusher {
        // take pending changes
        let (updates, required_len) = {
            let mut buffer_guard = self.buffer.write();
            let updates = std::mem::take(&mut *buffer_guard);
            let Some(required_len) = updates.keys().max().map(|&max_id| max_id as usize + 1) else {
                return Box::new(|| Ok(()));
            };
            (updates, required_len)
        };

        let flags_arc = self.storage.clone();
        Box::new(move || {
            // lock for the entire flushing process
            let mut flags_guard = flags_arc.lock();

            // resize if needed
            if required_len > flags_guard.len() {
                flags_guard.set_len(required_len)?;
            }

            for (index, value) in updates {
                flags_guard.set(index as usize, value);
            }

            flags_guard.flusher()()?;

            Ok(())
        })
    }
}

#[cfg(test)]
mod tests {

    use common::types::PointOffsetType;
    use rand::rngs::StdRng;
    use rand::{Rng, SeedableRng};

    use crate::common::flags::buffered_dynamic_flags::BufferedDynamicFlags;
    use crate::common::flags::dynamic_mmap_flags::DynamicMmapFlags;

    #[test]
    fn test_buffered_flags_growth_persistence() {
        let dir = tempfile::Builder::new()
            .prefix("buffered_flags_growth")
            .tempdir()
            .unwrap();


Review comment (refactor):
We should be able to avoid this check by also using a weak reference to the lock. Then it would also simplify the mutex to just hold a unit struct `()`

Code after:
    }

    pub fn files(&self) -> Vec<PathBuf> {
        self.storage.lock().files()
    }

    pub fn flusher(&self) -> Flusher {
        // take pending changes
        let (updates, required_len) = {
            let mut buffer_guard = self.buffer.write();
            let updates = std::mem::take(&mut *buffer_guard);
            let Some(required_len) = updates.keys().max().map(|&max_id| max_id as usize + 1) else {
                return Box::new(|| Ok(()));
            };
            (updates, required_len)
        };

        // Weak reference to detect when the storage has been deleted
        let flags_arc = Arc::downgrade(&self.storage);
        let is_alive_flush_lock = self.is_alive_flush_lock.clone();

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

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

            let Some(flags_arc) = flags_arc.upgrade() else {
                log::debug!("skipping flushing on deleted storage");
                return Ok(());
            };

            // lock for the entire flushing process
            let mut flags_guard = flags_arc.lock();

            // resize if needed
            if required_len > flags_guard.len() {
                flags_guard.set_len(required_len)?;
            }

            for (index, value) in updates {
                flags_guard.set(index as usize, value);
            }

            flags_guard.flusher()()?;

            Ok(())
        })