kcode_k1_launch_nodes/
lib.rs

1use std::{
2    collections::HashMap,
3    fs::{File, OpenOptions},
4    io::{self, Read, Write},
5    path::Path,
6    sync::Mutex,
7};
8
9pub use kcode_k1_kmap_format::NodeId;
10
11pub const TARGET_ID_BYTES: usize = 12;
12pub type TargetId = [u8; TARGET_ID_BYTES];
13
14const RECORD_BYTES: usize = TARGET_ID_BYTES + 12 + 4;
15
16type Bindings = HashMap<TargetId, NodeId>;
17
18struct State {
19    file: File,
20    bindings: Bindings,
21    poisoned: bool,
22}
23
24pub struct LaunchNodes {
25    state: Mutex<State>,
26}
27
28impl LaunchNodes {
29    pub fn open(path: &Path) -> Result<Self, String> {
30        let (mut file, created) = open_file(path).map_err(io_error)?;
31        if created {
32            sync_parent(path).map_err(io_error)?;
33        }
34        let bindings = replay(&mut file).map_err(io_error)?;
35        Ok(Self {
36            state: Mutex::new(State {
37                file,
38                bindings,
39                poisoned: false,
40            }),
41        })
42    }
43
44    pub fn get(&self, target_id: TargetId) -> Result<Option<NodeId>, String> {
45        Ok(self.lock()?.bindings.get(&target_id).copied())
46    }
47
48    pub fn set(&self, target_id: TargetId, node_id: NodeId) -> Result<(), String> {
49        let mut state = self.lock()?;
50        let result = state
51            .file
52            .write_all(&record(target_id, node_id))
53            .and_then(|_| state.file.sync_data())
54            .map_err(io_error);
55        if let Err(error) = result {
56            state.poisoned = true;
57            return Err(error);
58        }
59        state.bindings.insert(target_id, node_id);
60        Ok(())
61    }
62
63    fn lock(&self) -> Result<std::sync::MutexGuard<'_, State>, String> {
64        let state = self
65            .state
66            .lock()
67            .map_err(|_| "launch-node store lock poisoned".to_owned())?;
68        if state.poisoned {
69            Err("launch-node store is unavailable".to_owned())
70        } else {
71            Ok(state)
72        }
73    }
74}
75
76fn open_file(path: &Path) -> io::Result<(File, bool)> {
77    match OpenOptions::new()
78        .read(true)
79        .append(true)
80        .create_new(true)
81        .open(path)
82    {
83        Ok(file) => Ok((file, true)),
84        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => OpenOptions::new()
85            .read(true)
86            .append(true)
87            .open(path)
88            .map(|file| (file, false)),
89        Err(error) => Err(error),
90    }
91}
92
93fn replay(file: &mut File) -> io::Result<Bindings> {
94    let mut bytes = Vec::new();
95    file.read_to_end(&mut bytes)?;
96    let complete = bytes.len() / RECORD_BYTES * RECORD_BYTES;
97    if complete != bytes.len() {
98        file.set_len(complete as u64)?;
99        file.sync_data()?;
100    }
101
102    let mut bindings = HashMap::new();
103    for bytes in bytes[..complete].chunks_exact(RECORD_BYTES) {
104        let target_id = bytes[..TARGET_ID_BYTES]
105            .try_into()
106            .expect("record target has fixed width");
107        let node_bytes: [u8; 12] = bytes[TARGET_ID_BYTES..24]
108            .try_into()
109            .expect("record node has fixed width");
110        let expected = checksum(&bytes[..24]).to_le_bytes();
111        if bytes[24..] != expected {
112            return Err(io::Error::new(
113                io::ErrorKind::InvalidData,
114                "corrupt launch-node record",
115            ));
116        }
117        bindings.insert(target_id, NodeId(node_bytes));
118    }
119    Ok(bindings)
120}
121
122fn record(target_id: TargetId, node_id: NodeId) -> [u8; RECORD_BYTES] {
123    let mut record = [0; RECORD_BYTES];
124    record[..TARGET_ID_BYTES].copy_from_slice(&target_id);
125    record[TARGET_ID_BYTES..24].copy_from_slice(&node_id.0);
126    let checksum = checksum(&record[..24]).to_le_bytes();
127    record[24..].copy_from_slice(&checksum);
128    record
129}
130
131fn checksum(bytes: &[u8]) -> u32 {
132    bytes.iter().fold(2_166_136_261, |sum, byte| {
133        (sum ^ u32::from(*byte)).wrapping_mul(16_777_619)
134    })
135}
136
137fn sync_parent(path: &Path) -> io::Result<()> {
138    File::open(
139        path.parent()
140            .filter(|parent| !parent.as_os_str().is_empty())
141            .unwrap_or_else(|| Path::new(".")),
142    )?
143    .sync_all()
144}
145
146fn io_error(error: io::Error) -> String {
147    format!("launch-node store I/O: {error}")
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use std::{
154        fs,
155        io::{Seek, SeekFrom},
156        path::PathBuf,
157        sync::atomic::{AtomicU64, Ordering},
158    };
159
160    static NEXT: AtomicU64 = AtomicU64::new(0);
161
162    fn path() -> PathBuf {
163        let root = std::env::temp_dir().join(format!(
164            "launch-nodes-{}-{}",
165            std::process::id(),
166            NEXT.fetch_add(1, Ordering::Relaxed)
167        ));
168        fs::create_dir(&root).unwrap();
169        root.join("bindings")
170    }
171
172    fn target(byte: u8) -> TargetId {
173        [byte; TARGET_ID_BYTES]
174    }
175
176    fn node(byte: u8) -> NodeId {
177        NodeId([byte; 12])
178    }
179
180    #[test]
181    fn empty_store_has_no_bindings() {
182        let path = path();
183        let store = LaunchNodes::open(&path).unwrap();
184        assert_eq!(store.get(target(1)).unwrap(), None);
185        fs::remove_dir_all(path.parent().unwrap()).unwrap();
186    }
187
188    #[test]
189    fn set_is_last_write_wins_and_replays() {
190        let path = path();
191        let store = LaunchNodes::open(&path).unwrap();
192        store.set(target(1), node(2)).unwrap();
193        store.set(target(3), node(4)).unwrap();
194        store.set(target(1), node(5)).unwrap();
195        assert_eq!(store.get(target(1)).unwrap(), Some(node(5)));
196        assert_eq!(store.get(target(3)).unwrap(), Some(node(4)));
197        drop(store);
198
199        let store = LaunchNodes::open(&path).unwrap();
200        assert_eq!(store.get(target(1)).unwrap(), Some(node(5)));
201        assert_eq!(store.get(target(3)).unwrap(), Some(node(4)));
202        fs::remove_dir_all(path.parent().unwrap()).unwrap();
203    }
204
205    #[test]
206    fn identical_sets_still_append() {
207        let path = path();
208        let store = LaunchNodes::open(&path).unwrap();
209        store.set(target(1), node(2)).unwrap();
210        store.set(target(1), node(2)).unwrap();
211        assert_eq!(
212            fs::metadata(&path).unwrap().len(),
213            (RECORD_BYTES * 2) as u64
214        );
215        fs::remove_dir_all(path.parent().unwrap()).unwrap();
216    }
217
218    #[test]
219    fn partial_final_record_is_truncated() {
220        let path = path();
221        let store = LaunchNodes::open(&path).unwrap();
222        store.set(target(1), node(2)).unwrap();
223        drop(store);
224        OpenOptions::new()
225            .append(true)
226            .open(&path)
227            .unwrap()
228            .write_all(&[9, 9, 9])
229            .unwrap();
230
231        let store = LaunchNodes::open(&path).unwrap();
232        assert_eq!(store.get(target(1)).unwrap(), Some(node(2)));
233        assert_eq!(fs::metadata(&path).unwrap().len(), RECORD_BYTES as u64);
234        fs::remove_dir_all(path.parent().unwrap()).unwrap();
235    }
236
237    #[test]
238    fn corrupt_complete_record_fails_open() {
239        let path = path();
240        let store = LaunchNodes::open(&path).unwrap();
241        store.set(target(1), node(2)).unwrap();
242        drop(store);
243        let mut file = OpenOptions::new().write(true).open(&path).unwrap();
244        file.seek(SeekFrom::Start(3)).unwrap();
245        file.write_all(&[7]).unwrap();
246        file.sync_data().unwrap();
247
248        assert!(LaunchNodes::open(&path).is_err());
249        fs::remove_dir_all(path.parent().unwrap()).unwrap();
250    }
251}
kcode_k1_launch_nodes/lib.rs

kcode_k1_launch_nodes/
lib.rs
