mrecordlog 0.2.0

Quickwit's shared record log.
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use super::{RecordReader, RecordWriter};
use crate::block_read_write::ArrayReader;
use crate::error::ReadRecordError;
use crate::frame::HEADER_LEN;
use crate::BLOCK_NUM_BYTES;

#[tokio::test]
async fn test_no_data() {
    let data = vec![0u8; BLOCK_NUM_BYTES * 4];
    let mut reader = RecordReader::open(ArrayReader::from(&data[..]));
    assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
}

#[tokio::test]
async fn test_empty_record() {
    let mut writer = RecordWriter::in_memory();
    writer.write_record("").await.unwrap();
    writer.flush().await.unwrap();
    let buf: Vec<u8> = writer.into_writer().into();
    let mut reader = RecordReader::open(ArrayReader::from(&buf[..]));
    assert_eq!(reader.read_record::<&str>().await.unwrap(), Some(""));
    assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
}

#[tokio::test]
async fn test_simple_record() {
    let mut writer = RecordWriter::in_memory();
    let record = "hello";
    writer.write_record(record).await.unwrap();
    writer.flush().await.unwrap();
    let buf: Vec<u8> = writer.into_writer().into();
    let mut reader = RecordReader::open(ArrayReader::from(&buf[..]));
    assert!(matches!(
        reader.read_record::<&str>().await,
        Ok(Some("hello"))
    ));
    assert!(matches!(reader.read_record::<&str>().await, Ok(None)));
}

fn make_long_entry(len: usize) -> String {
    "A".repeat(len)
}

#[tokio::test]
async fn test_spans_over_more_than_one_block() {
    let long_entry: String = make_long_entry(80_000);
    let mut writer = RecordWriter::in_memory();
    writer.write_record(long_entry.as_str()).await.unwrap();
    writer.flush().await.unwrap();
    let buf: Vec<u8> = writer.into_writer().into();
    let mut reader = RecordReader::open(ArrayReader::from(&buf[..]));
    let record_payload: &str = reader.read_record().await.unwrap().unwrap();
    assert_eq!(record_payload, &long_entry);
    assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
}

#[tokio::test]
async fn test_block_requires_padding() {
    // We'll miss 1 byte to be able to fit our next chunk header in the
    // first block.
    let long_record = make_long_entry(BLOCK_NUM_BYTES - HEADER_LEN - HEADER_LEN - 1 - 8);
    let short_record = "hello";
    let mut writer = RecordWriter::in_memory();
    writer.write_record(long_record.as_str()).await.unwrap();
    writer.write_record(short_record).await.unwrap();
    writer.flush().await.unwrap();
    let buffer: Vec<u8> = writer.into_writer().into();
    let mut reader = RecordReader::open(ArrayReader::from(&buffer[..]));
    assert_eq!(
        reader.read_record::<&str>().await.unwrap(),
        Some(long_record.as_str())
    );
    assert_eq!(
        reader.read_record::<&str>().await.unwrap(),
        Some(short_record)
    );
    assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
}

#[tokio::test]
async fn test_first_chunk_empty() {
    // We'll miss 1 byte to be able to fit our next chunk header in the
    // first block.
    let long_record = make_long_entry(BLOCK_NUM_BYTES - HEADER_LEN - HEADER_LEN);
    let short_record = "hello";
    let mut writer = RecordWriter::in_memory();
    writer.write_record(&long_record[..]).await.unwrap();
    writer.write_record(short_record).await.unwrap();
    writer.flush().await.unwrap();
    let buf: Vec<u8> = writer.into_writer().into();
    let mut reader = RecordReader::open(ArrayReader::from(&buf[..]));
    assert_eq!(
        reader.read_record::<&str>().await.unwrap(),
        Some(long_record.as_str())
    );
    assert_eq!(
        reader.read_record::<&str>().await.unwrap(),
        Some(short_record)
    );
    assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
}

#[tokio::test]
async fn test_behavior_upon_corruption() {
    let records: Vec<String> = (0..1_000).map(|i| format!("hello{i}")).collect();
    let mut writer = RecordWriter::in_memory();
    for record in &records {
        writer.write_record(record.as_str()).await.unwrap();
    }
    writer.flush().await.unwrap();
    let mut buffer: Vec<u8> = writer.into_writer().into();
    {
        let mut reader = RecordReader::open(ArrayReader::from(&buffer[..]));
        for record in &records {
            assert_eq!(
                reader.read_record::<&str>().await.unwrap(),
                Some(record.as_str())
            );
        }
        assert_eq!(reader.read_record::<&str>().await.unwrap(), None);
    }
    // Introducing a corruption.
    buffer[1_000] = 3;
    {
        let mut reader = RecordReader::open(ArrayReader::from(&buffer[..]));
        for record in &records[0..72] {
            // bug at i=72
            assert_eq!(
                reader.read_record::<&str>().await.unwrap(),
                Some(record.as_str())
            );
        }
        assert!(matches!(
            reader.read_record::<&str>().await,
            Err(ReadRecordError::Corruption)
        ));
    }
}