armdb 0.1.12

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

use std::fs;
use std::path::Path;

use crate::error::DbResult;

/// Per-shard replication cursor. Tracks the last replicated GSN
/// and the file/offset position for efficient seeking.
pub struct ReplicationCursor {
    pub shard_id: u8,
    /// Last GSN successfully replicated for this shard.
    pub last_gsn: u64,
    /// Data file ID where we last read.
    pub file_id: u32,
    /// Byte offset within the file — points to the next unread entry.
    pub file_offset: u64,
}

const CURSOR_FILE: &str = "repl.cursor";
// Wire format: gsn:u64 + file_id:u32 + file_offset:u64 = 20 bytes, padded to 24.
const CURSOR_SIZE: usize = 24;

impl ReplicationCursor {
    pub fn new(shard_id: u8) -> Self {
        Self {
            shard_id,
            last_gsn: 0,
            file_id: 0,
            file_offset: 0,
        }
    }

    /// Load cursor from shard directory. Returns None if no cursor file exists.
    pub fn load(shard_id: u8, shard_dir: &Path) -> DbResult<Option<Self>> {
        let path = shard_dir.join(CURSOR_FILE);
        if !path.exists() {
            return Ok(None);
        }
        let data = fs::read(&path)?;
        if data.len() < CURSOR_SIZE {
            return Ok(None);
        }
        let last_gsn = u64::from_le_bytes(data[0..8].try_into().expect("8 bytes"));
        let file_id = u32::from_le_bytes(data[8..12].try_into().expect("4 bytes"));
        let file_offset = u64::from_le_bytes(data[12..20].try_into().expect("8 bytes"));
        Ok(Some(Self {
            shard_id,
            last_gsn,
            file_id,
            file_offset,
        }))
    }

    /// Persist cursor to shard directory.
    pub fn save(&self, shard_dir: &Path) -> DbResult<()> {
        let path = shard_dir.join(CURSOR_FILE);
        let mut buf = [0u8; CURSOR_SIZE];
        buf[0..8].copy_from_slice(&self.last_gsn.to_le_bytes());
        buf[8..12].copy_from_slice(&self.file_id.to_le_bytes());
        buf[12..20].copy_from_slice(&self.file_offset.to_le_bytes());
        fs::write(&path, buf)?;
        Ok(())
    }

    /// Advance cursor to a new position.
    pub fn advance(&mut self, gsn: u64, file_id: u32, file_offset: u64) {
        self.last_gsn = gsn;
        self.file_id = file_id;
        self.file_offset = file_offset;
    }
}