armdb 0.2.0

sharded bitcask key-value storage optimized for NVMe
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

use std::fs::{self, OpenOptions};
use std::path::Path;

use crate::error::{DbError, DbResult};

const MAGIC: [u8; 4] = *b"RCUR";
const VERSION: u32 = 2; // bumped from 1; old (v1, 24-byte) files are unreadable
const CURSOR_SIZE: usize = 16;

#[derive(Debug, Clone, Copy, Default)]
pub struct ReplicationCursor {
    pub last_gsn: u64,
}

impl ReplicationCursor {
    pub fn load(path: &Path) -> DbResult<Option<Self>> {
        let bytes = match fs::read(path) {
            Ok(b) => b,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(DbError::Io(e)),
        };
        if bytes.len() != CURSOR_SIZE {
            return Ok(None); // wrong size — treat as missing, replaced on next save
        }
        if bytes[..4] != MAGIC {
            return Ok(None);
        }
        let version = u32::from_le_bytes(bytes[4..8].try_into().expect("len 4"));
        if version != VERSION {
            tracing::warn!(
                path = %path.display(),
                found_version = version,
                expected_version = VERSION,
                "replication cursor has unexpected version, discarding"
            );
            return Ok(None); // version mismatch — discard (v1 files are caught earlier by the size check)
        }
        let last_gsn = u64::from_le_bytes(bytes[8..16].try_into().expect("len 8"));
        Ok(Some(Self { last_gsn }))
    }

    pub fn save(&self, path: &Path) -> DbResult<()> {
        use std::io::Write;
        let mut buf = [0u8; CURSOR_SIZE];
        buf[..4].copy_from_slice(&MAGIC);
        buf[4..8].copy_from_slice(&VERSION.to_le_bytes());
        buf[8..16].copy_from_slice(&self.last_gsn.to_le_bytes());
        let mut f = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(path)
            .map_err(DbError::Io)?;
        f.write_all(&buf).map_err(DbError::Io)?;
        f.sync_data().map_err(DbError::Io)?;
        // C11: also fsync the parent directory so the new file entry is durable.
        if let Some(parent) = path.parent()
            && let Ok(dir) = std::fs::File::open(parent)
        {
            // Best-effort: directory fsync is not portable on all filesystems; ignore Err.
            let _ = dir.sync_all();
        }
        Ok(())
    }

    pub fn advance(&mut self, last_gsn: u64) {
        self.last_gsn = last_gsn;
    }
}

#[cfg(test)]
mod cursor_tests {
    use super::*;

    #[test]
    fn from_gsn_saturates_at_u64_max() {
        let cursor = ReplicationCursor { last_gsn: u64::MAX };
        assert_eq!(cursor.last_gsn.saturating_add(1), u64::MAX);
    }
}