use std::{
collections::HashMap,
fs::{File, OpenOptions},
io::{self, Read, Write},
path::Path,
sync::Mutex,
};
pub use kcode_k1_kmap_format::NodeId;
pub const TARGET_ID_BYTES: usize = 12;
pub type TargetId = [u8; TARGET_ID_BYTES];
const RECORD_BYTES: usize = TARGET_ID_BYTES + 12 + 4;
type Bindings = HashMap<TargetId, NodeId>;
struct State {
file: File,
bindings: Bindings,
poisoned: bool,
}
pub struct LaunchNodes {
state: Mutex<State>,
}
impl LaunchNodes {
pub fn open(path: &Path) -> Result<Self, String> {
let (mut file, created) = open_file(path).map_err(io_error)?;
if created {
sync_parent(path).map_err(io_error)?;
}
let bindings = replay(&mut file).map_err(io_error)?;
Ok(Self {
state: Mutex::new(State {
file,
bindings,
poisoned: false,
}),
})
}
pub fn get(&self, target_id: TargetId) -> Result<Option<NodeId>, String> {
Ok(self.lock()?.bindings.get(&target_id).copied())
}
pub fn set(&self, target_id: TargetId, node_id: NodeId) -> Result<(), String> {
let mut state = self.lock()?;
let result = state
.file
.write_all(&record(target_id, node_id))
.and_then(|_| state.file.sync_data())
.map_err(io_error);
if let Err(error) = result {
state.poisoned = true;
return Err(error);
}
state.bindings.insert(target_id, node_id);
Ok(())
}
fn lock(&self) -> Result<std::sync::MutexGuard<'_, State>, String> {
let state = self
.state
.lock()
.map_err(|_| "launch-node store lock poisoned".to_owned())?;
if state.poisoned {
Err("launch-node store is unavailable".to_owned())
} else {
Ok(state)
}
}
}
fn open_file(path: &Path) -> io::Result<(File, bool)> {
match OpenOptions::new()
.read(true)
.append(true)
.create_new(true)
.open(path)
{
Ok(file) => Ok((file, true)),
Err(error) if error.kind() == io::ErrorKind::AlreadyExists => OpenOptions::new()
.read(true)
.append(true)
.open(path)
.map(|file| (file, false)),
Err(error) => Err(error),
}
}
fn replay(file: &mut File) -> io::Result<Bindings> {
let mut bytes = Vec::new();
file.read_to_end(&mut bytes)?;
let complete = bytes.len() / RECORD_BYTES * RECORD_BYTES;
if complete != bytes.len() {
file.set_len(complete as u64)?;
file.sync_data()?;
}
let mut bindings = HashMap::new();
for bytes in bytes[..complete].chunks_exact(RECORD_BYTES) {
let target_id = bytes[..TARGET_ID_BYTES]
.try_into()
.expect("record target has fixed width");
let node_bytes: [u8; 12] = bytes[TARGET_ID_BYTES..24]
.try_into()
.expect("record node has fixed width");
let expected = checksum(&bytes[..24]).to_le_bytes();
if bytes[24..] != expected {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"corrupt launch-node record",
));
}
bindings.insert(target_id, NodeId(node_bytes));
}
Ok(bindings)
}
fn record(target_id: TargetId, node_id: NodeId) -> [u8; RECORD_BYTES] {
let mut record = [0; RECORD_BYTES];
record[..TARGET_ID_BYTES].copy_from_slice(&target_id);
record[TARGET_ID_BYTES..24].copy_from_slice(&node_id.0);
let checksum = checksum(&record[..24]).to_le_bytes();
record[24..].copy_from_slice(&checksum);
record
}
fn checksum(bytes: &[u8]) -> u32 {
bytes.iter().fold(2_166_136_261, |sum, byte| {
(sum ^ u32::from(*byte)).wrapping_mul(16_777_619)
})
}
fn sync_parent(path: &Path) -> io::Result<()> {
File::open(
path.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new(".")),
)?
.sync_all()
}
fn io_error(error: io::Error) -> String {
format!("launch-node store I/O: {error}")
}
#[cfg(test)]
mod tests {
use super::*;
use std::{
fs,
io::{Seek, SeekFrom},
path::PathBuf,
sync::atomic::{AtomicU64, Ordering},
};
static NEXT: AtomicU64 = AtomicU64::new(0);
fn path() -> PathBuf {
let root = std::env::temp_dir().join(format!(
"launch-nodes-{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::Relaxed)
));
fs::create_dir(&root).unwrap();
root.join("bindings")
}
fn target(byte: u8) -> TargetId {
[byte; TARGET_ID_BYTES]
}
fn node(byte: u8) -> NodeId {
NodeId([byte; 12])
}
#[test]
fn empty_store_has_no_bindings() {
let path = path();
let store = LaunchNodes::open(&path).unwrap();
assert_eq!(store.get(target(1)).unwrap(), None);
fs::remove_dir_all(path.parent().unwrap()).unwrap();
}
#[test]
fn set_is_last_write_wins_and_replays() {
let path = path();
let store = LaunchNodes::open(&path).unwrap();
store.set(target(1), node(2)).unwrap();
store.set(target(3), node(4)).unwrap();
store.set(target(1), node(5)).unwrap();
assert_eq!(store.get(target(1)).unwrap(), Some(node(5)));
assert_eq!(store.get(target(3)).unwrap(), Some(node(4)));
drop(store);
let store = LaunchNodes::open(&path).unwrap();
assert_eq!(store.get(target(1)).unwrap(), Some(node(5)));
assert_eq!(store.get(target(3)).unwrap(), Some(node(4)));
fs::remove_dir_all(path.parent().unwrap()).unwrap();
}
#[test]
fn identical_sets_still_append() {
let path = path();
let store = LaunchNodes::open(&path).unwrap();
store.set(target(1), node(2)).unwrap();
store.set(target(1), node(2)).unwrap();
assert_eq!(
fs::metadata(&path).unwrap().len(),
(RECORD_BYTES * 2) as u64
);
fs::remove_dir_all(path.parent().unwrap()).unwrap();
}
#[test]
fn partial_final_record_is_truncated() {
let path = path();
let store = LaunchNodes::open(&path).unwrap();
store.set(target(1), node(2)).unwrap();
drop(store);
OpenOptions::new()
.append(true)
.open(&path)
.unwrap()
.write_all(&[9, 9, 9])
.unwrap();
let store = LaunchNodes::open(&path).unwrap();
assert_eq!(store.get(target(1)).unwrap(), Some(node(2)));
assert_eq!(fs::metadata(&path).unwrap().len(), RECORD_BYTES as u64);
fs::remove_dir_all(path.parent().unwrap()).unwrap();
}
#[test]
fn corrupt_complete_record_fails_open() {
let path = path();
let store = LaunchNodes::open(&path).unwrap();
store.set(target(1), node(2)).unwrap();
drop(store);
let mut file = OpenOptions::new().write(true).open(&path).unwrap();
file.seek(SeekFrom::Start(3)).unwrap();
file.write_all(&[7]).unwrap();
file.sync_data().unwrap();
assert!(LaunchNodes::open(&path).is_err());
fs::remove_dir_all(path.parent().unwrap()).unwrap();
}
}