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

Code before:
        })
    }

    fn save(&self, op_num: u64) -> CollectionResult<()> {
        self.file.write(|current| {
            let timestamp = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs();
            debug_assert!(current.timestamp <= timestamp);
            current.op_num = op_num;
            current.timestamp = timestamp;
        })?;
        Ok(())
    }

    /// Update the underlying file every `APPLIED_SEQ_SAVE_INTERVAL` updates.
    /// Always update memory representation
    pub fn update(&self, op_num: u64) -> CollectionResult<()> {
        // update in-memory
        self.op_num.store(op_num, Ordering::Relaxed);
        let prev_count = self.update_count.fetch_add(1, Ordering::Relaxed);
        // update on disk according to interval to amortize fsync
        if prev_count + 1 >= APPLIED_SEQ_SAVE_INTERVAL {
            self.save(op_num)?;
            self.update_count.store(0, Ordering::Relaxed);
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use tempfile::TempDir;

    use crate::update_workers::applied_seq::{APPLIED_SEQ_SAVE_INTERVAL, AppliedSeqHandler};

    #[test]
    fn nothing_persisted_on_init() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path()).unwrap();
        assert_eq!(handler.op_num(), 0);
        // nothing persisted yet because no updates observed
        assert!(!handler.path().exists());
    }

    #[test]
    fn persists_at_interval() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path()).unwrap();
        for i in 1..APPLIED_SEQ_SAVE_INTERVAL {

Review comment (bug):
I'd rather remove this line and call save each `prev_count % 64 == 0` instead.

That separate store could data race in a case where we'd call this concurrently. I understand that should never happen in the current implementation, but you never know how things change in the future. Because of it I'd like the above suggestion better with just a single atomic call.

Code after:
    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}");
                    // delete file as it is malformed
                    if let Err(err) = fs::remove_file(&path) {
                        log::error!("Could not delete malformed applied_seq file {path:?} {err}");
                        Self {
                            file: None,
                            path,
                            op_num: AtomicU64::new(wal_last_index),
                            update_count,
                        }
                    } else {
                        // try again to create the file from scratch
                        Self::load_or_init(shard_path, wal_last_index)
                    }
                } else {
                    log::error!("Error while creating new applied_seq at {path:?} {err}");
                    Self {
                        file: None,
                        path,
                        op_num: AtomicU64::new(wal_last_index),
                        update_count,