haematite 0.7.0

Content-addressed, branchable, actor-native storage engine
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
139

use super::expiry_index::{ArmError, DeadlineScheduler, Generation};
use super::{ShardActor, ShardError, sequence_key};
use crate::store::MemoryStore;
use crate::sync::ballot::{Ballot, Stamp};
use crate::sync::topology::SyncNodeId;
use crate::wal::{DurableWal, FsyncPolicy, WalError, WalRecovery};
use std::path::{Path, PathBuf};
use std::time::Duration;

#[derive(Debug)]
struct TempWal {
    dir: tempfile::TempDir,
    path: PathBuf,
}

impl TempWal {
    fn path(&self) -> &Path {
        debug_assert!(self.path.starts_with(self.dir.path()));
        &self.path
    }
}

fn temp_path(name: &str) -> Result<TempWal, WalError> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join(name);
    Ok(TempWal { dir, path })
}

fn stamped_seq_value(next_seq: u64, stamp_seq: u64) -> Result<Vec<u8>, WalError> {
    crate::ttl::entry::encode_stamped_optional_ttl(
        next_seq.to_be_bytes().to_vec(),
        Stamp::new(Ballot::new(9, SyncNodeId::new("stream-index")), stamp_seq),
        None,
    )
    .map_err(|error| WalError::TreeError(error.to_string()))
}

fn scan_error(error: &ShardError) -> WalError {
    WalError::TreeError(error.to_string())
}

#[derive(Default)]
struct ArmCount(usize);

impl DeadlineScheduler for ArmCount {
    fn schedule(&mut self, _delay: Duration, _generation: Generation) -> Result<(), ArmError> {
        self.0 = self.0.saturating_add(1);
        Ok(())
    }
}

#[test]
fn stream_index_matches_full_walk_after_commit_recovery_and_merge_adopt() -> Result<(), WalError> {
    let source = temp_path("actor-stream-index-source.wal")?;
    let target = temp_path("actor-stream-index-target.wal")?;
    let mut source_store = MemoryStore::new();
    let mut target_store = MemoryStore::new();

    let source_wal = DurableWal::new(source.path(), FsyncPolicy::CommitOnly)?;
    let mut source_actor = ShardActor::new(source_wal);
    source_actor.put(sequence_key(b"alpha"), stamped_seq_value(2, 0)?)?;
    source_actor.put(sequence_key(b"beta"), stamped_seq_value(3, 1)?)?;
    source_actor
        .apply_durable(
            b"expiring",
            None,
            b"value".to_vec(),
            Some(Duration::from_secs(3_600)),
            Stamp::new(Ballot::new(9, SyncNodeId::new("stream-index")), 9),
            &mut source_store,
        )
        .map_err(|error| WalError::TreeError(error.to_string()))?;
    source_actor.commit(&mut source_store)?;
    assert_eq!(
        source_actor
            .scan_sequences()
            .map_err(|error| scan_error(&error))?,
        source_actor
            .scan_sequences_full_walk(&source_store)
            .map_err(|error| scan_error(&error))?,
        "index must match a full walk after normal commit"
    );

    drop(source_actor);
    let recovered = WalRecovery::recover_path(source.path(), &source_store)?;
    let source_wal = DurableWal::new(source.path(), FsyncPolicy::CommitOnly)?;
    let mut recovered_actor = ShardActor::from_recovered(
        source_wal,
        recovered,
        &source_store,
        crate::tree::TreePolicy::V1_DEFAULT,
    )?;
    assert_eq!(
        recovered_actor
            .scan_sequences()
            .map_err(|error| scan_error(&error))?,
        recovered_actor
            .scan_sequences_full_walk(&source_store)
            .map_err(|error| scan_error(&error))?,
        "index must be rebuilt from the committed root on recovery"
    );

    recovered_actor.put(sequence_key(b"gamma"), stamped_seq_value(5, 2)?)?;
    recovered_actor.commit(&mut source_store)?;
    let export = recovered_actor.export_reachable(0, &source_store)?;

    let target_wal = DurableWal::new(target.path(), FsyncPolicy::CommitOnly)?;
    let mut target_actor = ShardActor::new(target_wal);
    target_actor.put(sequence_key(b"local"), stamped_seq_value(1, 3)?)?;
    target_actor.commit(&mut target_store)?;
    target_actor.merge_adopt(&[export], &mut target_store)?;
    let mut arm_count = ArmCount::default();
    target_actor
        .arm_expiry(&mut arm_count)
        .map_err(|error| WalError::TreeError(error.to_string()))?;
    assert_eq!(arm_count.0, 1, "merge-adopt arms the rebuilt minimum once");
    assert!(target_actor.expiry_metrics().merge_rebuild_entries >= 5);
    assert_eq!(
        target_actor
            .scan_sequences()
            .map_err(|error| scan_error(&error))?,
        target_actor
            .scan_sequences_full_walk(&target_store)
            .map_err(|error| scan_error(&error))?,
        "index must be rebuilt when merge_adopt swaps the committed root"
    );
    assert_eq!(
        target_actor
            .scan_sequences()
            .map_err(|error| scan_error(&error))?,
        vec![
            (b"alpha".to_vec(), 2),
            (b"beta".to_vec(), 3),
            (b"gamma".to_vec(), 5),
            (b"local".to_vec(), 1),
        ]
    );
    Ok(())
}