4283961027

Build Type

push

github

Committed by GitHub

Commit Message

feat: Blue green cache (#1061)

Run Details

645 of 645 new or added lines in 45 files covered. (100.0%)

27779 of 39307 relevant lines covered (70.67%)

52489.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.18

/dozer-cache/src/cache/lmdb/cache/checkpoint_database.rs

use dozer_storage::{
    lmdb::{Cursor, Database, DatabaseFlags, RwTransaction, Transaction, WriteFlags},
    lmdb_storage::LmdbEnvironmentManager,
};
use dozer_types::node::{NodeHandle, OpIdentifier, SourceStates};

use crate::errors::{CacheError, QueryError};

#[derive(Debug, Clone, Copy)]
pub struct CheckpointDatabase(Database);

impl CheckpointDatabase {
    pub fn new(env: &mut LmdbEnvironmentManager) -> Result<Self, CacheError> {
        Ok(Self(env.create_database(
            Some("checkpoint"),
            Some(DatabaseFlags::empty()),
        )?))
    }

    pub fn write(
        &self,
        txn: &mut RwTransaction,
        source_states: &SourceStates,
    ) -> Result<(), CacheError> {
        txn.clear_db(self.0)
            .map_err(|e| CacheError::Internal(Box::new(e)))?;
        for (node_handle, op_identifier) in source_states {
            let key = node_handle.to_bytes();
            let value = op_identifier.to_bytes();
            txn.put(self.0, &key, &value, WriteFlags::empty())
                .map_err(|e| CacheError::Query(QueryError::InsertValue(e)))?;
        }
        Ok(())
    }

    pub fn read<T: Transaction>(&self, txn: &T) -> Result<SourceStates, CacheError> {
        let mut cursor = txn
            .open_ro_cursor(self.0)
            .map_err(|e| CacheError::Internal(Box::new(e)))?;
        let mut result = SourceStates::new();
        for key_value_pair in cursor.iter_start() {
            let (key, value) = key_value_pair.map_err(|e| CacheError::Internal(Box::new(e)))?;
            let node_handle = NodeHandle::from_bytes(key);
            let op_identifier = OpIdentifier::from_bytes(
                value
                    .try_into()
                    .expect("We only write op identifier to this database"),
            );
            result.insert(node_handle, op_identifier);
        }
        Ok(result)
    }
}

#[cfg(test)]
mod tests {
    use crate::cache::lmdb::utils::{init_env, CacheOptions};

    use super::*;

    #[test]
    fn test_checkpoint_database() {
        let mut env = init_env(&CacheOptions::default()).unwrap().0;
        let db = CheckpointDatabase::new(&mut env).unwrap();
        let txn = env.create_txn().unwrap();
        let mut txn = txn.write();

        // Write to empty database.
        let mut source_states_1 = SourceStates::new();
        source_states_1.insert(
            NodeHandle::new(Some(1), "1".to_string()),
            OpIdentifier::new(1, 1),
        );
        db.write(txn.txn_mut(), &source_states_1).unwrap();
        assert_eq!(db.read(txn.txn()).unwrap(), source_states_1);

        // Write to existing database.
        let mut source_states_2 = SourceStates::new();
        source_states_2.insert(
            NodeHandle::new(Some(2), "2".to_string()),
            OpIdentifier::new(2, 2),
        );
        db.write(txn.txn_mut(), &source_states_2).unwrap();
        assert_eq!(db.read(txn.txn()).unwrap(), source_states_2);
    }
}

1	use dozer_storage::{
2	lmdb::{Cursor, Database, DatabaseFlags, RwTransaction, Transaction, WriteFlags},
3	lmdb_storage::LmdbEnvironmentManager,
4	};
5	use dozer_types::node::{NodeHandle, OpIdentifier, SourceStates};
6
7	use crate::errors::{CacheError, QueryError};
8
9	#[derive(Debug, Clone, Copy)]	×
10	pub struct CheckpointDatabase(Database);
11
12	impl CheckpointDatabase {
13	pub fn new(env: &mut LmdbEnvironmentManager) -> Result<Self, CacheError> {	150✔
14	Ok(Self(env.create_database(	150✔
15	Some("checkpoint"),	150✔
16	Some(DatabaseFlags::empty()),	150✔
17	)?))	150✔
18	}	150✔
19
20	pub fn write(
21	&self,
22	txn: &mut RwTransaction,
23	source_states: &SourceStates,
24	) -> Result<(), CacheError> {
25	txn.clear_db(self.0)	141✔
26	.map_err(\|e\| CacheError::Internal(Box::new(e)))?;	141✔
27	for (node_handle, op_identifier) in source_states {	161✔
28	let key = node_handle.to_bytes();	20✔
29	let value = op_identifier.to_bytes();	20✔
30	txn.put(self.0, &key, &value, WriteFlags::empty())	20✔
31	.map_err(\|e\| CacheError::Query(QueryError::InsertValue(e)))?;	20✔
32	}
33	Ok(())	141✔
34	}	141✔
35
36	pub fn read<T: Transaction>(&self, txn: &T) -> Result<SourceStates, CacheError> {	2✔
37	let mut cursor = txn	2✔
38	.open_ro_cursor(self.0)	2✔
39	.map_err(\|e\| CacheError::Internal(Box::new(e)))?;	2✔
40	let mut result = SourceStates::new();	2✔
41	for key_value_pair in cursor.iter_start() {	2✔
42	let (key, value) = key_value_pair.map_err(\|e\| CacheError::Internal(Box::new(e)))?;	2✔
43	let node_handle = NodeHandle::from_bytes(key);	2✔
44	let op_identifier = OpIdentifier::from_bytes(	2✔
45	value	2✔
46	.try_into()	2✔
47	.expect("We only write op identifier to this database"),	2✔
48	);	2✔
49	result.insert(node_handle, op_identifier);	2✔
50	}
51	Ok(result)	2✔
52	}	2✔
53	}
54
55	#[cfg(test)]
56	mod tests {
57	use crate::cache::lmdb::utils::{init_env, CacheOptions};
58
59	use super::*;
60
61	#[test]	1✔
62	fn test_checkpoint_database() {	1✔
63	let mut env = init_env(&CacheOptions::default()).unwrap().0;	1✔
64	let db = CheckpointDatabase::new(&mut env).unwrap();	1✔
65	let txn = env.create_txn().unwrap();	1✔
66	let mut txn = txn.write();	1✔
67		1✔
68	// Write to empty database.	1✔
69	let mut source_states_1 = SourceStates::new();	1✔
70	source_states_1.insert(	1✔
71	NodeHandle::new(Some(1), "1".to_string()),	1✔
72	OpIdentifier::new(1, 1),	1✔
73	);	1✔
74	db.write(txn.txn_mut(), &source_states_1).unwrap();	1✔
75	assert_eq!(db.read(txn.txn()).unwrap(), source_states_1);	1✔
76
77	// Write to existing database.
78	let mut source_states_2 = SourceStates::new();	1✔
79	source_states_2.insert(	1✔
80	NodeHandle::new(Some(2), "2".to_string()),	1✔
81	OpIdentifier::new(2, 2),	1✔
82	);	1✔
83	db.write(txn.txn_mut(), &source_states_2).unwrap();	1✔
84	assert_eq!(db.read(txn.txn()).unwrap(), source_states_2);	1✔
85	}	1✔
86	}

getdozer / dozer / 4283961027

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous