Issue when opening multiple `raft_engine::Engine`'s

[Dan-J-D]

I'm testing my store which uses raft_engine as its log store. I was trying to unit test my integration against the same test cases of MemStorage from raft-rs, in specific test_storage_append. I have found that each test case when ran individually, works. But when i test it all in one #[test]. It appears to behave weirdly, it fails on the second test case tests[1] but when i swap the first and second test case, it fails on the second test case. And when I added logging in between the initialization of the storage and writing of ents. It causes the third case to fail and the first 2 to succeed. Any ideas?

[ti-chi-bot]

Welcome @Dan-J-D! It looks like this is your first issue to tikv/raft-engine 🎉

[Dan-J-D]

Here is a full demo of what I am talking about.

use raft::{
    Error as RaftError, RaftState, Result as RaftResult, StorageError as RaftStorageError,
    prelude::*,
};

pub struct RaftEntryExt(Entry);

impl From<Entry> for RaftEntryExt {
    fn from(e: Entry) -> Self {
        RaftEntryExt(e)
    }
}

impl raft_engine::MessageExt for RaftEntryExt {
    type Entry = raft::prelude::Entry;

    fn index(e: &Self::Entry) -> u64 {
        e.index
    }
}

struct RaftStoreFsmInner {
    engine: raft_engine::Engine,
    raft_id: u64,
    raft_state: RaftState,
}

impl RaftStoreFsmInner {
    fn first_index(&self) -> u64 {
        match self.engine.first_index(self.raft_id) {
            Some(idx) => idx,
            None => 0,
        }
    }

    fn last_index(&self) -> u64 {
        match self.engine.last_index(self.raft_id) {
            Some(idx) => idx,
            None => 0,
        }
    }

    pub fn append(&mut self, entries: &[Entry], sync: bool) -> RaftResult<()> {
        if entries.is_empty() {
            return Ok(());
        }

        if self.first_index() > entries[0].index {
            panic!(
                "overwrite compacted raft logs, compacted: {}, append: {}",
                self.first_index() - 1,
                entries[0].index,
            );
        }

        if self.last_index() + 1 < entries[0].index {
            panic!(
                "raft logs should be continuous, last index: {}, new appended: {}",
                self.last_index(),
                entries[0].index,
            );
        }

        let mut log_batch = raft_engine::LogBatch::with_capacity(entries.len());

        log_batch
            .add_entries::<RaftEntryExt>(self.raft_id, entries)
            .map_err(|e| RaftError::Store(RaftStorageError::Other(Box::new(e))))?;

        self.engine
            .write(&mut log_batch, sync)
            .map_err(|e| RaftError::Store(RaftStorageError::Other(Box::new(e))))?;

        Ok(())
    }
}

impl Storage for RaftStoreFsmInner {
    fn initial_state(&self) -> RaftResult<RaftState> {
        Ok(self.raft_state.clone())
    }

    fn entries(
        &self,
        low: u64,
        high: u64,
        max_size: impl Into<Option<u64>>,
        _context: raft::GetEntriesContext,
    ) -> RaftResult<Vec<Entry>> {
        let mut entries = Vec::new();
        let res = self.engine.fetch_entries_to::<RaftEntryExt>(
            self.raft_id,
            low,
            high,
            max_size.into().map(|s| s as usize),
            &mut entries,
        );

        match res {
            Ok(_) => Ok(entries),
            Err(e) => Err(RaftError::Store(RaftStorageError::Other(Box::new(e)))),
        }
    }

    fn term(&self, _idx: u64) -> RaftResult<u64> {
        todo!();
    }

    fn first_index(&self) -> RaftResult<u64> {
        Ok(self.first_index())
    }

    fn last_index(&self) -> RaftResult<u64> {
        Ok(self.last_index())
    }

    fn snapshot(&self, _request_index: u64, _to: u64) -> RaftResult<Snapshot> {
        todo!();
    }
}

#[cfg(test)]
mod test {
    use std::panic::{self, AssertUnwindSafe};

    use raft::GetEntriesContext;

    use super::*;

    impl Default for RaftStoreFsmInner {
        fn default() -> Self {
            let tempdir = tempfile::tempdir().unwrap();
            let conf = raft_engine::Config {
                dir: tempdir.path().to_str().unwrap().to_string(),
                ..Default::default()
            };
            let engine = raft_engine::Engine::open(conf).unwrap();
            RaftStoreFsmInner {
                engine,
                raft_id: 1,
                raft_state: RaftState::default(),
            }
        }
    }

    fn new_entry(index: u64, term: u64) -> Entry {
        let mut e = Entry::default();
        e.term = term;
        e.index = index;
        e
    }

    #[test]
    fn test_storage_append() {
        let ents = vec![new_entry(3, 3), new_entry(4, 4), new_entry(5, 5)];
        let mut tests = vec![
            (
                vec![new_entry(3, 3), new_entry(4, 4), new_entry(5, 5)],
                Some(vec![new_entry(3, 3), new_entry(4, 4), new_entry(5, 5)]),
            ),
            (
                vec![new_entry(3, 3), new_entry(4, 6), new_entry(5, 6)],
                Some(vec![new_entry(3, 3), new_entry(4, 6), new_entry(5, 6)]),
            ),
            (
                vec![
                    new_entry(3, 3),
                    new_entry(4, 4),
                    new_entry(5, 5),
                    new_entry(6, 5),
                ],
                Some(vec![
                    new_entry(3, 3),
                    new_entry(4, 4),
                    new_entry(5, 5),
                    new_entry(6, 5),
                ]),
            ),
            // overwrite compacted raft logs is not allowed
            (
                vec![new_entry(2, 3), new_entry(3, 3), new_entry(4, 5)],
                None,
            ),
            // truncate the existing entries and append
            (
                vec![new_entry(4, 5)],
                Some(vec![new_entry(3, 3), new_entry(4, 5)]),
            ),
            // direct append
            (
                vec![new_entry(6, 6)],
                Some(vec![
                    new_entry(3, 3),
                    new_entry(4, 4),
                    new_entry(5, 5),
                    new_entry(6, 6),
                ]),
            ),
        ];
        for (i, (entries, wentries)) in tests.drain(..).enumerate() {
            let mut storage = RaftStoreFsmInner::default();

            // Uncomment this to see it work for test[1] but fail on test[2]
            // if let Ok(e) = storage.entries(3, 6, None, GetEntriesContext::empty(false)) {
            //     println!("before write #{}: {:?}", i, e);
            // }

            {
                let mut log_batch = raft_engine::LogBatch::with_capacity(ents.len());
                log_batch
                    .add_entries::<RaftEntryExt>(storage.raft_id, &ents)
                    .unwrap();
                storage.engine.write(&mut log_batch, true).unwrap();
            }

            // Uncomment this aswell
            // if let Ok(e) = storage.entries(3, 6, None, GetEntriesContext::empty(false)) {
            //     println!("before append #{}: {:?}", i, e);
            // }

            let res = panic::catch_unwind(AssertUnwindSafe(|| storage.append(&entries, true)));
            if let Some(wentries) = wentries {
                let _ = res.unwrap();
                let e = storage
                    .entries(3, 6, None, GetEntriesContext::empty(false))
                    .unwrap();
                if *e != wentries {
                    panic!("#{}: want {:?}, entries {:?}", i, wentries, e);
                }
            } else {
                res.unwrap_err();
            }
        }
    }
}