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

Code before:

impl AppliedSeqHandler {
    /// Get the current in-memory op_num.
    pub fn op_num(&self) -> Option<u64> {
        if self.file.is_some() {
            Some(self.op_num.load(Ordering::Relaxed))
        } else {
            None
        }
    }

    /// Path for the applied_seq json file
    /// Not included in snapshots
    pub fn path(&self) -> &Path {
        self.path.as_path()
    }

    #[cfg(test)]
    fn persisted_op_num(&self) -> u64 {
        read_json::<AppliedSeq>(&self.path).unwrap().op_num
    }

    /// Load or create the underlying applied seq file.
    ///
    /// `wal_first_index` is expected to be the next used WAL index.
    pub fn load_or_init(shard_path: &Path, wal_first_index: u64) -> Self {
        let last_applied_wal_index = wal_first_index.saturating_sub(1);
        let update_count = AtomicU64::new(0);
        let path = shard_path.join(APPLIED_SEQ_FILE);
        let file_was_already_present = path.exists();

        let loaded_file: Result<SaveOnDisk<AppliedSeq>, _> =
            SaveOnDisk::load_or_init(&path, || AppliedSeq::new(last_applied_wal_index));
        match loaded_file {
            Ok(file) => {
                let existing_op_num = file.read().op_num;
                debug_assert!(
                    existing_op_num <= last_applied_wal_index,
                    "persisted applied_seq:{existing_op_num} cannot be larger than the last_applied_wal_index:{last_applied_wal_index}"
                );
                let op_num = AtomicU64::new(existing_op_num);
                Self {
                    file: Some(file),
                    path,
                    op_num,
                    update_count,
                }
            }
            Err(err) => {
                if file_was_already_present {
                    log::error!("Error while loading existing applied_seq at {path:?} {err}");

Review comment (bug):
Though this is a little complex, I like how we handle edge cases now 👍 

E.g. if disk is full, we don't crash. If the data is malformed, we try to recreate it replaying 100% of the WAL.

Code after:
impl AppliedSeqHandler {
    /// Get the current in-memory op_num for the last_applied_seq.
    /// The value is likely larger than what is persisted in `file`.
    ///
    ///
    /// Returns None if the handler is not active.
    pub fn op_num(&self) -> Option<u64> {
        if self.file.is_some() {
            Some(self.op_num.load(Ordering::Relaxed))
        } else {
            None
        }
    }

    /// Path for the applied_seq json file
    pub fn path(&self) -> &Path {
        self.path.as_path()
    }

    #[cfg(test)]
    fn persisted_op_num(&self) -> u64 {
        let persisted = read_json::<AppliedSeq>(&self.path).unwrap().op_num;
        debug_assert!(persisted <= self.op_num.load(Ordering::Relaxed));
        persisted
    }

    /// Load or create the underlying applied seq file.
    pub fn load_or_init(shard_path: &Path, wal_last_index: u64) -> Self {
        let update_count = AtomicU64::new(0);
        let path = shard_path.join(APPLIED_SEQ_FILE);
        let file_was_already_present = path.exists();

        let loaded_file: Result<SaveOnDisk<AppliedSeq>, _> =
            SaveOnDisk::load_or_init(&path, || AppliedSeq::new(wal_last_index));
        match loaded_file {
            Ok(file) => {
                let persisted_applied_seq = file.read().op_num;
                debug_assert!(
                    persisted_applied_seq <= wal_last_index,
                    "last_applied_seq:{persisted_applied_seq} cannot be larger than last_wal_index:{wal_last_index}"
                );
                Self {
                    file: Some(file),
                    path,
                    op_num: AtomicU64::new(persisted_applied_seq),
                    update_count,
                }
            }
            Err(err) => {
                if file_was_already_present {
                    log::error!("Error while loading existing applied_seq at {path:?} {err}");