block-db 0.2.0

Local, multi-threaded, durable byte DB.
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

// Authors: Robert Lopez

use super::{error::Error, BlockDB};
use serde::{Deserialize, Serialize};
use std::{
    collections::HashSet,
    ops::{Deref, DerefMut},
    path::PathBuf,
    sync::Arc,
};
use tokio::sync::Mutex;
use walr::{options::WALOptionsBuilder, ReplayHandlerResult, WAL};

#[derive(Debug, Serialize, Deserialize)]
pub enum BlockDBLog {
    AddDataFile(String),
    AddDataFiles(HashSet<String>),
    DeleteDataFile(String),
    DeleteDataFiles,
}

#[derive(Debug, Default, Serialize, Deserialize)]
pub struct BlockDBState {
    pub valid_data_files: HashSet<String>,
}

#[derive(Debug)]
pub struct BlockDBWAL(WAL<BlockDBLog, BlockDBState, ()>);

impl Deref for BlockDBWAL {
    type Target = WAL<BlockDBLog, BlockDBState, ()>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl DerefMut for BlockDBWAL {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

fn replay_handler(
    logs: Vec<BlockDBLog>,
    checkpoint: Arc<Mutex<BlockDBState>>,
    _context: Option<()>,
) -> ReplayHandlerResult {
    Box::pin(async move {
        let mut checkpoint = checkpoint.lock().await;

        for log in logs {
            match log {
                BlockDBLog::AddDataFile(id) => {
                    checkpoint.valid_data_files.insert(id);
                }
                BlockDBLog::AddDataFiles(ids) => {
                    checkpoint.valid_data_files.extend(ids);
                }
                BlockDBLog::DeleteDataFile(ref id) => {
                    checkpoint.valid_data_files.remove(id);
                }
                BlockDBLog::DeleteDataFiles => {
                    checkpoint.valid_data_files.clear();
                }
            }
        }

        Ok(())
    })
}

impl BlockDBWAL {
    pub async fn open(data_path: PathBuf) -> Result<Self, Error> {
        Ok(Self(
            WAL::open(
                data_path,
                Box::new(replay_handler),
                Some(
                    WALOptionsBuilder::new()
                        .max_size((size_of::<BlockDBLog>() + 2) * 1_000)
                        .build(),
                ),
            )
            .await?,
        ))
    }
}

impl BlockDB {
    pub async fn wal_checkpoint_handler(
        &self,
        wal_writer: &mut WAL<BlockDBLog, BlockDBState, ()>,
    ) -> Result<(), Error> {
        if wal_writer.should_checkpoint() {
            wal_writer
                .checkpoint(&BlockDBState {
                    valid_data_files: self.data_files.read().await.keys().cloned().collect(),
                })
                .await?;
        }

        Ok(())
    }
}