raft_log/raft_log/
dump.rs

1use std::io;
2use std::io::Error;
3use std::sync::Arc;
4
5use crate::chunk::Chunk;
6use crate::file_lock;
7use crate::raft_log::dump_api::DumpApi;
8use crate::types::Segment;
9use crate::ChunkId;
10use crate::Config;
11use crate::RaftLog;
12use crate::Types;
13use crate::WALRecord;
14
15/// A dump utility that reads WAL records from disk.
16///
17/// It acquires an exclusive lock on the directory to prevent concurrent writes
18/// while reading.
19pub struct Dump<T> {
20    config: Arc<Config>,
21
22    /// Acquire the dir exclusive lock when writing to the log.
23    _dir_lock: file_lock::FileLock,
24
25    _p: std::marker::PhantomData<T>,
26}
27
28impl<T: Types> DumpApi<T> for Dump<T> {
29    /// Reads all WAL records from disk and passes them to the provided callback
30    /// function.
31    ///
32    /// The callback receives:
33    /// - `chunk_id`: The ID of the chunk containing the record
34    /// - `index`: The 0-based index of the record within its chunk
35    /// - `result`: The result containing either the record data or an IO error
36    ///
37    /// # Errors
38    /// Returns an IO error if reading the chunks fails or if the callback
39    /// returns an error.
40    fn write_with<D>(&self, mut write_record: D) -> Result<(), io::Error>
41    where D: FnMut(
42            ChunkId,
43            u64,
44            Result<(Segment, WALRecord<T>), io::Error>,
45        ) -> Result<(), io::Error> {
46        let config = self.config.as_ref();
47
48        let chunk_ids = RaftLog::<T>::load_chunk_ids(config)?;
49        for chunk_id in chunk_ids {
50            let it = Chunk::<T>::dump(config, chunk_id)?;
51            for (i, res) in it.into_iter().enumerate() {
52                write_record(chunk_id, i as u64, res)?;
53            }
54        }
55        Ok(())
56    }
57}
58
59/// A dump utility that reads WAL records from an existing RaftLog instance.
60///
61/// Unlike [`Dump`], this does not acquire a directory lock since it operates on
62/// an already initialized RaftLog.
63pub struct RefDump<'a, T: Types> {
64    pub(crate) config: Arc<Config>,
65    pub(crate) raft_log: &'a RaftLog<T>,
66}
67
68impl<T: Types> DumpApi<T> for RefDump<'_, T> {
69    /// Reads all WAL records from the RaftLog and passes them to the provided
70    /// callback function.
71    ///
72    /// The callback receives:
73    /// - `chunk_id`: The ID of the chunk containing the record
74    /// - `index`: The 0-based index of the record within its chunk
75    /// - `result`: The result containing either the record data or an IO error
76    ///
77    /// # Errors
78    /// Returns an IO error if reading the chunks fails or if the callback
79    /// returns an error.
80    fn write_with<D>(&self, mut write_record: D) -> Result<(), Error>
81    where D: FnMut(
82            ChunkId,
83            u64,
84            Result<(Segment, WALRecord<T>), Error>,
85        ) -> Result<(), Error> {
86        let closed =
87            self.raft_log.wal.closed.values().map(|c| c.chunk.chunk_id());
88
89        let chunk_ids = closed.chain([self.raft_log.wal.open.chunk.chunk_id()]);
90
91        for chunk_id in chunk_ids {
92            let f =
93                Chunk::<T>::open_chunk_file(self.config.as_ref(), chunk_id)?;
94
95            let it = Chunk::load_records_iter(
96                self.config.as_ref(),
97                Arc::new(f),
98                chunk_id,
99            )?;
100
101            for (i, res) in it.enumerate() {
102                write_record(chunk_id, i as u64, res)?;
103            }
104        }
105
106        Ok(())
107    }
108}
109
110impl<T: Types> Dump<T> {
111    /// Creates a new Dump instance with the given configuration.
112    ///
113    /// # Errors
114    /// Returns an IO error if acquiring the directory lock fails.
115    pub fn new(config: Arc<Config>) -> Result<Self, io::Error> {
116        let dir_lock = file_lock::FileLock::new(config.clone())?;
117
118        Ok(Self {
119            config,
120            _dir_lock: dir_lock,
121            _p: std::marker::PhantomData,
122        })
123    }
124}
raft_log/raft_log/dump.rs

raft_log/raft_log/
dump.rs
