Refactor: improve the read and write logic for the RocksSnapshot

[Phoenix500526]

Currently, the read and write methods in RocksSnapshot look like this:

impl RocksSnapshot {
   /// Read data from the snapshot
    async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        if self.meta.is_current {
            let n = self.meta.data.read(buf).await?;
            if n == 0 {
                self.meta.is_current = false;
            } else {
                return Ok(n);
            }
        }

        while self.snap_file_idx < self.snap_files.len() {
            let f = if let Some(ref mut f) = self.current_file {
                f
            } else {
                let path = self.current_file_path(false);
                let reader = File::open(path).await?;
                self.current_file = Some(reader);
                self.current_file
                    .as_mut()
                    .unwrap_or_else(|| unreachable!("current_file must be `Some` here"))
            };
            let n = f.read(buf).await?;
            if n == 0 {
                let _ignore = self.current_file.take();
                self.snap_file_idx = self.snap_file_idx.overflow_add(1);
            } else {
                return Ok(n);
            }
        }
        Ok(0)
    }

    /// Write snapshot data
    async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        let (mut next_buf, mut written_bytes) = (buf, 0);

        #[allow(clippy::indexing_slicing)] // length is checked when reading meta
        if self.meta.is_current {
            let Some(meta_len_slice) = next_buf.get(0..8) else {
                return Err(io::Error::new(
                    ErrorKind::UnexpectedEof,
                    "cannot read meta length from buffer",
                ));
            };
            let meta_len_bytes: [u8; 8] = meta_len_slice
                .try_into()
                .map_err(|e| io::Error::new(ErrorKind::Other, e))?;
            let meta_len = u64::from_le_bytes(meta_len_bytes);

            let Some(meta_slice) = next_buf.get(8..meta_len.overflow_add(8).numeric_cast()) else {
                return Err(io::Error::new(
                    ErrorKind::UnexpectedEof,
                    "cannot read meta from buffer",
                ));
            };
            let meta_bytes: Vec<u8> = meta_slice
                .try_into()
                .unwrap_or_else(|_e| unreachable!("infallible"));
            let meta = bincode::deserialize(&meta_bytes)
                .map_err(|e| io::Error::new(ErrorKind::Other, e))?;

            self.apply_snap_meta(meta);
            let mut data = Vec::new();
            data.extend(meta_len_bytes);
            data.extend(meta_bytes);
            *self.meta.data.get_mut() = data;
            self.meta.is_current = false;
            written_bytes = written_bytes
                .overflow_add(meta_len.numeric_cast())
                .overflow_add(8);
            next_buf = &next_buf[meta_len.overflow_add(8).numeric_cast()..];
        }

        // the snap_file_idx has checked with while's pattern
        // written_len is calculated by next_buf.len() so it must be less than next_buf's length
        #[allow(clippy::indexing_slicing)]
        while self.snap_file_idx < self.snap_files.len() {
            let snap_file = &mut self.snap_files[self.snap_file_idx];
            if snap_file.size == 0 {
                return Err(io::Error::new(
                    ErrorKind::Other,
                    format!("snap file {} size is 0", snap_file.filename),
                ));
            }
            let left = snap_file.remain_size().numeric_cast();
            let (write_len, switch, finished) = match next_buf.len().cmp(&left) {
                Ordering::Greater => (left, true, false),
                Ordering::Equal => (left, true, true),
                Ordering::Less => (next_buf.len(), false, true),
            };
            snap_file.written_size = snap_file
                .written_size
                .overflow_add(write_len.numeric_cast());
            written_bytes = written_bytes.overflow_add(write_len);

            let buffer = &next_buf[0..write_len];

            let f = if let Some(ref mut f) = self.current_file {
                f
            } else {
                let path = self.current_file_path(true);
                let writer = File::create(path).await?;
                self.current_file = Some(writer);
                self.current_file
                    .as_mut()
                    .unwrap_or_else(|| unreachable!("current_file must be `Some` here"))
            };
            f.write_all(buffer).await?;

            if switch {
                next_buf = &next_buf[write_len..];
                let old = self.current_file.take();
                if let Some(mut old_f) = old {
                    old_f.flush().await?;
                    let path = self.current_file_path(false);
                    let tmp_path = self.current_file_path(true);
                    fs::rename(tmp_path, path)?;
                }
                self.snap_file_idx = self.snap_file_idx.overflow_add(1);
            }
            if finished {
                break;
            }
        }
        Ok(written_bytes)
    }
}

In this implementation, we should maintain the slice index manually. IMO, we can use the bytes crate to refine it.

[bsbds]

May I take this issue, thanks.