feat(rolldb): allow optionally overlap of WAL over immutable chain (#419)
This commit is contained in:
parent
b640ce4097
commit
394f14e835
3 changed files with 44 additions and 4 deletions
|
|
@ -167,10 +167,15 @@ pub struct Store {
|
||||||
pub tip_change: Arc<tokio::sync::Notify>,
|
pub tip_change: Arc<tokio::sync::Notify>,
|
||||||
wal_seq: u64,
|
wal_seq: u64,
|
||||||
k_param: u64,
|
k_param: u64,
|
||||||
|
immutable_overlap: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Store {
|
impl Store {
|
||||||
pub fn open(path: impl AsRef<Path>, k_param: u64) -> Result<Self, Error> {
|
pub fn open(
|
||||||
|
path: impl AsRef<Path>,
|
||||||
|
k_param: u64,
|
||||||
|
immutable_overlap: Option<u64>,
|
||||||
|
) -> Result<Self, Error> {
|
||||||
let mut opts = Options::default();
|
let mut opts = Options::default();
|
||||||
opts.create_if_missing(true);
|
opts.create_if_missing(true);
|
||||||
opts.create_missing_column_families(true);
|
opts.create_missing_column_families(true);
|
||||||
|
|
@ -184,6 +189,7 @@ impl Store {
|
||||||
tip_change: Arc::new(tokio::sync::Notify::new()),
|
tip_change: Arc::new(tokio::sync::Notify::new()),
|
||||||
wal_seq,
|
wal_seq,
|
||||||
k_param,
|
k_param,
|
||||||
|
immutable_overlap: immutable_overlap.unwrap_or(0),
|
||||||
};
|
};
|
||||||
|
|
||||||
Ok(out)
|
Ok(out)
|
||||||
|
|
@ -378,7 +384,7 @@ impl Store {
|
||||||
// get the number of slots that have passed since the wal point
|
// get the number of slots that have passed since the wal point
|
||||||
let slot_delta = tip - value.slot().unwrap_or(0);
|
let slot_delta = tip - value.slot().unwrap_or(0);
|
||||||
|
|
||||||
if slot_delta <= self.k_param {
|
if slot_delta <= self.k_param + self.immutable_overlap {
|
||||||
break;
|
break;
|
||||||
} else {
|
} else {
|
||||||
WalKV::stage_delete(&self.db, wal_key, &mut batch);
|
WalKV::stage_delete(&self.db, wal_key, &mut batch);
|
||||||
|
|
|
||||||
|
|
@ -46,7 +46,7 @@ mod tests {
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_stream_waiting() {
|
async fn test_stream_waiting() {
|
||||||
let path = tempfile::tempdir().unwrap().into_path();
|
let path = tempfile::tempdir().unwrap().into_path();
|
||||||
let mut db = Store::open(path.clone(), 30).unwrap();
|
let mut db = Store::open(path.clone(), 30, None).unwrap();
|
||||||
|
|
||||||
for i in 0..=100 {
|
for i in 0..=100 {
|
||||||
let (slot, hash, body) = dummy_block(i * 10);
|
let (slot, hash, body) = dummy_block(i * 10);
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,16 @@ use super::{BlockBody, BlockHash, BlockSlot, Store};
|
||||||
|
|
||||||
fn with_tmp_db<T>(k_param: u64, op: fn(store: Store) -> T) {
|
fn with_tmp_db<T>(k_param: u64, op: fn(store: Store) -> T) {
|
||||||
let path = tempfile::tempdir().unwrap().into_path();
|
let path = tempfile::tempdir().unwrap().into_path();
|
||||||
let store = Store::open(path.clone(), k_param).unwrap();
|
let store = Store::open(path.clone(), k_param, None).unwrap();
|
||||||
|
|
||||||
|
op(store);
|
||||||
|
|
||||||
|
Store::destroy(path).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
fn with_tmp_db_overlap<T>(k_param: u64, overlap: u64, op: fn(store: Store) -> T) {
|
||||||
|
let path = tempfile::tempdir().unwrap().into_path();
|
||||||
|
let store = Store::open(path.clone(), k_param, Some(overlap)).unwrap();
|
||||||
|
|
||||||
op(store);
|
op(store);
|
||||||
|
|
||||||
|
|
@ -109,6 +118,8 @@ fn test_prune_linear() {
|
||||||
db.roll_forward(slot, hash, body).unwrap();
|
db.roll_forward(slot, hash, body).unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// db contains slots: 0, 10, ..., 980, 990
|
||||||
|
// this should prune slots less than (990 - 30) = 960
|
||||||
db.prune_wal().unwrap();
|
db.prune_wal().unwrap();
|
||||||
|
|
||||||
let mut wal = db.crawl_after(None);
|
let mut wal = db.crawl_after(None);
|
||||||
|
|
@ -122,6 +133,29 @@ fn test_prune_linear() {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_prune_linear_with_overlap() {
|
||||||
|
with_tmp_db_overlap(30, 20, |mut db| {
|
||||||
|
for i in 0..100 {
|
||||||
|
let (slot, hash, body) = dummy_block(i * 10);
|
||||||
|
db.roll_forward(slot, hash, body).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
// db contains slots: 0, 10, ..., 980, 990
|
||||||
|
// this should prune slots less than (990 - 30 - 20) = 940
|
||||||
|
db.prune_wal().unwrap();
|
||||||
|
|
||||||
|
let mut wal = db.crawl_after(None);
|
||||||
|
|
||||||
|
for i in 94..100 {
|
||||||
|
let (_, val) = wal.next().unwrap().unwrap();
|
||||||
|
assert_eq!(val.slot().unwrap(), i * 10);
|
||||||
|
}
|
||||||
|
|
||||||
|
assert!(wal.next().is_none());
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_prune_with_rollback() {
|
fn test_prune_with_rollback() {
|
||||||
with_tmp_db(30, |mut db| {
|
with_tmp_db(30, |mut db| {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue