writeahead 0.4.1

A partitioned write-ahead log (WAL) for building append-only, high-throughput durability systems
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

//! Property test: arbitrary payload batches roundtrip through read-by-id and
//! full stream replay, across file rotations.

use futures::StreamExt;
use proptest::prelude::*;
use writeahead::logfile::FILE_HEADER_SIZE;
use writeahead::{SimpleFile, WriteAhead, WriteAheadOptions};

proptest! {
    #![proptest_config(ProptestConfig::with_cases(32))]

    #[test]
    fn roundtrip_arbitrary_batches(
        batches in prop::collection::vec(
            prop::collection::vec(prop::collection::vec(any::<u8>(), 0..300), 1..8),
            1..6,
        )
    ) {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let dir = tempfile::tempdir().unwrap();
            let mut wal = WriteAhead::<SimpleFile>::with_options(WriteAheadOptions {
                log_dir: dir.path().to_path_buf(),
                max_file_size: FILE_HEADER_SIZE + 512, // force rotations
                ..Default::default()
            });
            wal.start().unwrap();

            let mut expected: Vec<Vec<u8>> = Vec::new();
            let mut ids = Vec::new();
            for batch in &batches {
                ids.extend(wal.write_batch(batch.clone()).await.unwrap());
                expected.extend(batch.iter().cloned());
            }

            // Read back by id
            for (id, want) in ids.iter().zip(&expected) {
                let got = wal.read(id.file_id, id.file_offset).unwrap();
                prop_assert_eq!(&got, want);
            }

            // Full stream replay must yield the same data AND the same
            // addresses the writes were acknowledged with
            let mut stream = wal.create_stream().unwrap();
            let mut got = Vec::new();
            let mut got_ids = Vec::new();
            while let Some(r) = stream.next().await {
                let (id, data) = r.unwrap();
                got_ids.push(id);
                got.push(data);
            }
            prop_assert_eq!(got, expected);
            prop_assert_eq!(got_ids, ids);
            Ok(())
        })?;
    }
}