1#![allow(warnings)]
34
35use std::collections::HashMap;
36use std::path::{Path, PathBuf};
37
38use limnifs_core::codec;
39use limnifs_core::slab_store::SlabStore;
40use limnifs_core::{ContentHandle, ManifestCursor, MetadataBlob};
41
42use crate::sidecar_name;
43
44use crate::config::{ImageMode, WriteConfig};
45use crate::WriteError;
46
47pub struct RwImage {
49 manifest_path: PathBuf,
50 config: WriteConfig,
51 state: Option<OpenState>,
52 inode_map: HashMap<String, u64>,
55 pending_files: HashMap<String, Vec<u8>>,
57 pending_history: Vec<HistoryEntry>,
59 next_inode: u64,
61 base_manifest_hash: Option<[u8; 32]>,
68}
69
70struct OpenState {
73 blob: MetadataBlob,
74 root_inode: u64,
75 slab_store: Option<SlabStore>,
76}
77
78#[derive(Clone, Debug)]
80pub enum HistoryEntry {
81 Add {
82 path: String,
83 inode: u64,
84 size: u64,
85 },
86 Update {
87 path: String,
88 old_inode: u64,
89 new_inode: u64,
90 size: u64,
91 },
92 Delete {
93 path: String,
94 inode: u64,
95 },
96}
97
98impl RwImage {
99 pub fn open(path: &Path, config: WriteConfig) -> Result<Self, WriteError> {
106 cleanup_stale_swap_dir(path);
113
114 let manifest_bytes = std::fs::read(path).map_err(WriteError::Io)?;
115
116 let mut cursor = ManifestCursor::new(&manifest_bytes);
117 let _ = limnifs_core::parse_manifest_header(&mut cursor).map_err(core_to_io)?;
118 let _ = limnifs_core::parse_feature_flags_section(&mut cursor).map_err(core_to_io)?;
119 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).map_err(core_to_io)?;
120
121 let blob_bytes: Vec<u8> = if let Some(inline) = meta_ref.inline_metadata.as_ref() {
122 inline.clone()
123 } else {
124 let entry = meta_ref.locators.first().ok_or_else(|| {
125 WriteError::Io(std::io::Error::other(
126 "metadata_reference has neither inline data nor locators",
127 ))
128 })?;
129 let name = sidecar_name(&entry.uri)?;
130 let sidecar = path.parent().unwrap_or_else(|| Path::new(".")).join(name);
131 let wire_bytes = std::fs::read(&sidecar).map_err(WriteError::Io)?;
132 if meta_ref.codec == 0 {
133 wire_bytes
134 } else {
135 codec::decompress(meta_ref.codec, &wire_bytes, meta_ref.uncompressed_len)
136 .map_err(core_to_io)?
137 }
138 };
139
140 let mut blob_cursor = ManifestCursor::new(&blob_bytes);
141 let blob = limnifs_core::parse_metadata_blob(&mut blob_cursor).map_err(core_to_io)?;
142
143 let slab_index = limnifs_core::parse_slab_index(&mut cursor).map_err(core_to_io)?;
144 let slab_store = if slab_index.is_empty() {
145 None
146 } else {
147 Some(SlabStore::load_mmap(path, &slab_index).map_err(core_to_io)?)
148 };
149
150 let root_inode = blob.root_inode_number().ok_or_else(|| {
151 WriteError::Io(std::io::Error::other(
152 "metadata blob: could not identify a unique root directory inode",
153 ))
154 })?;
155
156 let path_index = blob.build_path_index();
157 let next_inode = blob.inodes.iter().map(|i| i.number).max().unwrap_or(0) + 1;
158
159 let mut image = Self {
160 manifest_path: path.to_path_buf(),
161 base_manifest_hash: Some(limnifs_core::hash_section(&manifest_bytes)),
162 config,
163 state: Some(OpenState {
164 blob,
165 root_inode,
166 slab_store,
167 }),
168 inode_map: path_index,
169 pending_files: HashMap::new(),
170 pending_history: Vec::new(),
171 next_inode,
172 };
173 let _ = image.replay_wal_if_present();
175 Ok(image)
176 }
177
178 #[must_use]
181 pub fn create_new(path: &Path, config: WriteConfig) -> Self {
182 Self {
183 manifest_path: path.to_path_buf(),
184 config,
185 state: None,
186 inode_map: HashMap::new(),
187 pending_files: HashMap::new(),
188 pending_history: Vec::new(),
189 next_inode: 1,
190 base_manifest_hash: None,
191 }
192 }
193
194 pub fn add_file(&mut self, path: &str, data: &[u8]) -> Result<u64, WriteError> {
200 let inode = self.next_inode;
201 self.next_inode += 1;
202 let key = normalize_path(path);
203 self.pending_files.insert(key.clone(), data.to_vec());
204 self.inode_map.insert(key.clone(), inode);
205 self.pending_history.push(HistoryEntry::Add {
206 path: key,
207 inode,
208 size: data.len() as u64,
209 });
210 Ok(inode)
211 }
212
213 pub fn update_file(&mut self, path: &str, data: &[u8]) -> Result<(), WriteError> {
219 let key = normalize_path(path);
220 let old_inode = *self.inode_map.get(&key).ok_or_else(|| {
221 WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
222 })?;
223 let new_inode = self.next_inode;
224 self.next_inode += 1;
225 self.pending_files.insert(key.clone(), data.to_vec());
226 self.inode_map.insert(key.clone(), new_inode);
227 self.pending_history.push(HistoryEntry::Update {
228 path: key,
229 old_inode,
230 new_inode,
231 size: data.len() as u64,
232 });
233 Ok(())
234 }
235
236 pub fn delete_file(&mut self, path: &str) -> Result<(), WriteError> {
242 let key = normalize_path(path);
243 let inode = self.inode_map.remove(&key).ok_or_else(|| {
244 WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
245 })?;
246 self.pending_files.remove(&key);
247 self.pending_history
248 .push(HistoryEntry::Delete { path: key, inode });
249 Ok(())
250 }
251
252 pub fn read_file(&self, path: &str) -> Result<Vec<u8>, WriteError> {
259 let state = self.state.as_ref().ok_or_else(|| {
260 WriteError::Io(std::io::Error::other("read_file: image was not opened"))
261 })?;
262 let key = normalize_path(path);
263 if let Some(data) = self.pending_files.get(&key) {
268 return Ok(data.clone());
269 }
270 let inode_num = *self.inode_map.get(&key).ok_or_else(|| {
271 WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
272 })?;
273 let inode = state.blob.inode_by_number(inode_num).ok_or_else(|| {
274 WriteError::Io(std::io::Error::other(format!("inode {inode_num} missing")))
275 })?;
276 match &inode.content_handle {
277 ContentHandle::InlineData(data) => Ok(data.clone()),
278 ContentHandle::SliceMap(slices) => {
279 let store = state.slab_store.as_ref().ok_or_else(|| {
280 WriteError::Io(std::io::Error::other(
281 "read_file: slice-backed file but no slab store",
282 ))
283 })?;
284 let mut out = Vec::new();
285 for slice in slices {
286 let plaintext = store
287 .plaintext_for(slice.drop_id.as_bytes())
288 .ok_or_else(|| {
289 WriteError::Io(std::io::Error::other("drop not in any slab"))
290 })?
291 .map_err(core_to_io)?;
292 out.extend_from_slice(&plaintext);
293 }
294 Ok(out)
295 }
296 _ => Err(WriteError::Io(std::io::Error::other(
297 "read_file: unsupported content handle",
298 ))),
299 }
300 }
301
302 #[must_use]
304 pub fn pending_changes(&self) -> usize {
305 self.pending_history.len()
306 }
307
308 #[must_use]
311 pub fn needs_turnover(&self) -> bool {
312 self.config.turnover_threshold > 0
313 && self.pending_history.len() >= self.config.turnover_threshold as usize
314 }
315
316 #[must_use]
318 pub fn mode(&self) -> &ImageMode {
319 &self.config.mode
320 }
321
322 pub fn commit(&mut self) -> Result<crate::WriteArtifact, WriteError> {
335 self.write_wal()?;
337 let staging = self.staging_dir();
338 self.write_staging_tree(&staging)?;
339 let artifact = crate::write_directory_with_config(&staging, &self.config)?;
340 let _ = std::fs::remove_dir_all(&staging);
341 self.write_artifact(&artifact)?;
342 let _ = std::fs::remove_file(self.wal_path());
346 self.base_manifest_hash = Some(limnifs_core::hash_section(&artifact.bytes));
347 Ok(artifact)
348 }
349
350 pub fn turnover(&self) -> Result<crate::WriteArtifact, WriteError> {
357 let staging = self.staging_dir();
358 if let Some(state) = &self.state {
359 self.write_live_tree_only(state, &staging)?;
360 } else {
361 let _ = std::fs::remove_dir_all(&staging);
362 std::fs::create_dir_all(&staging).map_err(WriteError::Io)?;
363 self.write_pending_only(&staging)?;
364 }
365 let artifact = crate::write_directory_with_config(&staging, &self.config)?;
366 let _ = std::fs::remove_dir_all(&staging);
367 self.write_artifact(&artifact)?;
368 Ok(artifact)
369 }
370
371 fn write_staging_tree(&self, staging: &Path) -> Result<(), WriteError> {
375 let _ = std::fs::remove_dir_all(staging);
376 std::fs::create_dir_all(staging).map_err(WriteError::Io)?;
377
378 if let Some(state) = &self.state {
379 self.write_live_tree(state, staging)?;
380 }
381
382 for entry in &self.pending_history {
385 if let HistoryEntry::Delete { path, .. } = entry {
386 if !self.pending_files.contains_key(path) {
387 let _ = std::fs::remove_file(staging.join(staging_relative(path)));
388 }
389 }
390 }
391
392 for (path, data) in &self.pending_files {
397 let file_path = staging.join(staging_relative(path));
398 if let Some(parent) = file_path.parent() {
399 std::fs::create_dir_all(parent).map_err(WriteError::Io)?;
400 }
401 let mode = preserved_mode(&file_path);
402 std::fs::write(&file_path, data).map_err(WriteError::Io)?;
403 if let Some(mode) = mode {
404 #[cfg(unix)]
405 {
406 use std::os::unix::fs::PermissionsExt as _;
407 let _ =
408 std::fs::set_permissions(&file_path, std::fs::Permissions::from_mode(mode));
409 }
410 #[cfg(not(unix))]
411 let _ = mode;
412 }
413 }
414 Ok(())
415 }
416
417 fn write_pending_only(&self, staging: &Path) -> Result<(), WriteError> {
418 for (path, data) in &self.pending_files {
419 let file_path = staging.join(staging_relative(path));
420 if let Some(parent) = file_path.parent() {
421 std::fs::create_dir_all(parent).map_err(WriteError::Io)?;
422 }
423 let mode = preserved_mode(&file_path);
424 std::fs::write(&file_path, data).map_err(WriteError::Io)?;
425 if let Some(mode) = mode {
426 #[cfg(unix)]
427 {
428 use std::os::unix::fs::PermissionsExt as _;
429 let _ =
430 std::fs::set_permissions(&file_path, std::fs::Permissions::from_mode(mode));
431 }
432 #[cfg(not(unix))]
433 let _ = mode;
434 }
435 }
436 Ok(())
437 }
438
439 fn write_live_tree_only(&self, state: &OpenState, staging: &Path) -> Result<(), WriteError> {
442 let _ = std::fs::remove_dir_all(staging);
443 std::fs::create_dir_all(staging).map_err(WriteError::Io)?;
444 self.write_live_tree(state, staging)?;
445 Ok(())
446 }
447
448 fn write_live_tree(&self, state: &OpenState, staging: &Path) -> Result<(), WriteError> {
453 let slab_ref: Option<&dyn limnifs_core::slab_source::SlabSource> = state
454 .slab_store
455 .as_ref()
456 .map(|s| s as &dyn limnifs_core::slab_source::SlabSource);
457 let mut sink = limnifs_core::live_tree::FilesystemSink::new(staging, slab_ref);
458 limnifs_core::live_tree::walk_live_tree(&state.blob, state.root_inode, &mut sink)
459 .map_err(core_to_io)?;
460 for dir in sink.finish() {
464 #[cfg(all(unix, feature = "xattr"))]
465 for x in &dir.xattrs {
466 let _ = xattr::set(&dir.path, &x.key, &x.value);
467 }
468 #[cfg(not(all(unix, feature = "xattr")))]
469 let _ = dir;
470 }
471 Ok(())
472 }
473
474 fn write_artifact(&self, artifact: &crate::WriteArtifact) -> Result<(), WriteError> {
489 let parent = self
490 .manifest_path
491 .parent()
492 .unwrap_or_else(|| Path::new("."))
493 .to_path_buf();
494 let staging = parent.join(format!(
500 "{}.new-{}-{}",
501 self.manifest_path
502 .file_name()
503 .and_then(std::ffi::OsStr::to_str)
504 .unwrap_or("image.lim"),
505 std::process::id(),
506 std::time::SystemTime::now()
507 .duration_since(std::time::UNIX_EPOCH)
508 .map(|d| d.as_nanos())
509 .unwrap_or(0),
510 ));
511 let _ = std::fs::remove_dir_all(&staging);
512 std::fs::create_dir_all(&staging).map_err(WriteError::Io)?;
513
514 let manifest_name = self
516 .manifest_path
517 .file_name()
518 .map(std::ffi::OsString::from)
519 .unwrap_or_else(|| std::ffi::OsString::from("image.lim"));
520 let manifest_staging = staging.join(&manifest_name);
521 std::fs::write(&manifest_staging, &artifact.bytes).map_err(WriteError::Io)?;
522
523 let mut slab_names: Vec<std::ffi::OsString> = Vec::new();
524 for slab in &artifact.slabs {
525 let name = sidecar_name(&slab.locator)?;
528 let os_name = std::ffi::OsString::from(name);
529 std::fs::write(staging.join(&os_name), &slab.bytes).map_err(WriteError::Io)?;
530 slab_names.push(os_name);
531 }
532 let sidecar_name: Option<std::ffi::OsString> =
533 if let Some(sidecar) = &artifact.metadata_sidecar {
534 let name = sidecar_name(&sidecar.locator)?;
535 let os_name = std::ffi::OsString::from(name);
536 std::fs::write(staging.join(&os_name), &sidecar.bytes).map_err(WriteError::Io)?;
537 Some(os_name)
538 } else {
539 None
540 };
541
542 if let Some(name) = &sidecar_name {
546 rename_or_fallback(staging.join(name), parent.join(name))?;
547 }
548 for name in &slab_names {
549 rename_or_fallback(staging.join(name), parent.join(name))?;
550 }
551 rename_or_fallback(manifest_staging, self.manifest_path.clone())?;
552
553 let _ = std::fs::remove_dir_all(&staging);
555 Ok(())
556 }
557
558 fn staging_dir(&self) -> PathBuf {
562 let nonce = format!("{}-{}", std::process::id(), self.next_inode);
563 let mut cur = self
564 .manifest_path
565 .parent()
566 .unwrap_or_else(|| Path::new("."))
567 .to_path_buf();
568 loop {
569 if cur.join("Cargo.toml").is_file() {
570 if std::fs::read_to_string(cur.join("Cargo.toml"))
571 .map(|s| s.contains("[workspace]"))
572 .unwrap_or(false)
573 {
574 return cur.join(".scratch").join(format!("limnifs-rw-{nonce}"));
575 }
576 }
577 if !cur.pop() {
578 break;
579 }
580 }
581 self.manifest_path
582 .parent()
583 .unwrap_or_else(|| Path::new("."))
584 .join(".scratch")
585 .join(format!("limnifs-rw-{nonce}"))
586 }
587
588 fn wal_path(&self) -> PathBuf {
590 let parent = self
591 .manifest_path
592 .parent()
593 .unwrap_or_else(|| Path::new("."));
594 let mut name = self
595 .manifest_path
596 .file_name()
597 .map(std::ffi::OsString::from)
598 .unwrap_or_else(|| std::ffi::OsString::from("image.lim"));
599 name.push(".wal");
600 parent.join(name)
601 }
602
603 fn write_wal(&self) -> Result<(), WriteError> {
606 let mut buf: Vec<u8> = Vec::new();
607 buf.extend_from_slice(b"LIMWAL\0\0");
609 buf.extend_from_slice(&self.base_manifest_hash.unwrap_or([0u8; 32]));
610 buf.extend_from_slice(&(self.pending_files.len() as u32).to_le_bytes());
612 for (path, data) in &self.pending_files {
613 write_path_str(&mut buf, path);
614 buf.extend_from_slice(&(data.len() as u64).to_le_bytes());
615 buf.extend_from_slice(data);
616 }
617 buf.extend_from_slice(&(self.pending_history.len() as u32).to_le_bytes());
619 for entry in &self.pending_history {
620 match entry {
621 HistoryEntry::Add { path, .. } => {
622 buf.push(1);
623 write_path_str(&mut buf, path);
624 }
625 HistoryEntry::Update { path, .. } => {
626 buf.push(2);
627 write_path_str(&mut buf, path);
628 }
629 HistoryEntry::Delete { path, .. } => {
630 buf.push(3);
631 write_path_str(&mut buf, path);
632 }
633 }
634 }
635 let wal_tmp = self.wal_path().with_extension("wal.tmp");
637 std::fs::write(&wal_tmp, &buf).map_err(WriteError::Io)?;
638 std::fs::rename(&wal_tmp, self.wal_path()).map_err(WriteError::Io)?;
639 Ok(())
640 }
641
642 fn replay_wal_if_present(&mut self) -> usize {
647 let wal_path = self.wal_path();
648 let Ok(bytes) = std::fs::read(&wal_path) else {
649 return 0;
650 };
651 if bytes.len() < 40 || &bytes[..8] != b"LIMWAL\0\0" {
652 let _ = std::fs::remove_file(&wal_path);
653 return 0;
654 }
655 let mut tag = [0u8; 32];
660 tag.copy_from_slice(&bytes[8..40]);
661 let current = std::fs::read(&self.manifest_path)
662 .map(|b| limnifs_core::hash_section(&b))
663 .ok();
664 if current != Some(tag) {
665 let _ = std::fs::remove_file(&wal_path);
666 return 0;
667 }
668 let mut cursor = WalCursor {
669 bytes: &bytes,
670 pos: 40,
671 };
672 let files_count = match cursor.read_u32_le() {
673 Ok(n) => n as usize,
674 Err(_) => {
675 let _ = std::fs::remove_file(&wal_path);
676 return 0;
677 }
678 };
679 for _ in 0..files_count {
680 let path = match cursor.read_path_str() {
681 Ok(p) => p,
682 Err(_) => break,
683 };
684 let len = match cursor.read_u64_le() {
685 Ok(n) => n as usize,
686 Err(_) => break,
687 };
688 let data = match cursor.read_bytes(len) {
689 Ok(d) => d.to_vec(),
690 Err(_) => break,
691 };
692 self.pending_files.insert(path, data);
693 }
694 let hist_count = match cursor.read_u32_le() {
695 Ok(n) => n as usize,
696 Err(_) => 0,
697 };
698 let mut replayed = 0;
699 for _ in 0..hist_count {
700 let op = match cursor.read_u8() {
701 Ok(b) => b,
702 Err(_) => break,
703 };
704 let path = match cursor.read_path_str() {
705 Ok(p) => p,
706 Err(_) => break,
707 };
708 match op {
709 1 => {
710 let inode = self.next_inode;
711 self.next_inode += 1;
712 let size = self
713 .pending_files
714 .get(&path)
715 .map(|v| v.len() as u64)
716 .unwrap_or(0);
717 self.inode_map.insert(path.clone(), inode);
718 self.pending_history
719 .push(HistoryEntry::Add { path, inode, size });
720 }
721 2 => {
722 let old_inode = self.inode_map.get(&path).copied().unwrap_or(0);
723 let new_inode = self.next_inode;
724 self.next_inode += 1;
725 let size = self
726 .pending_files
727 .get(&path)
728 .map(|v| v.len() as u64)
729 .unwrap_or(0);
730 self.inode_map.insert(path.clone(), new_inode);
731 self.pending_history.push(HistoryEntry::Update {
732 path,
733 old_inode,
734 new_inode,
735 size,
736 });
737 }
738 3 => {
739 let inode = self.inode_map.remove(&path).unwrap_or(0);
740 self.pending_files.remove(&path);
741 self.pending_history
742 .push(HistoryEntry::Delete { path, inode });
743 }
744 _ => break,
745 }
746 replayed += 1;
747 }
748 let _ = std::fs::remove_file(&wal_path);
750 replayed
751 }
752}
753
754fn write_path_str(out: &mut Vec<u8>, s: &str) {
755 let bytes = s.as_bytes();
756 out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
757 out.extend_from_slice(bytes);
758}
759
760struct WalCursor<'a> {
761 bytes: &'a [u8],
762 pos: usize,
763}
764
765impl<'a> WalCursor<'a> {
766 fn read_u8(&mut self) -> Result<u8, ()> {
767 let b = *self.bytes.get(self.pos).ok_or(())?;
768 self.pos += 1;
769 Ok(b)
770 }
771 fn read_u32_le(&mut self) -> Result<u32, ()> {
772 if self.pos + 4 > self.bytes.len() {
773 return Err(());
774 }
775 let mut arr = [0u8; 4];
776 arr.copy_from_slice(&self.bytes[self.pos..self.pos + 4]);
777 self.pos += 4;
778 Ok(u32::from_le_bytes(arr))
779 }
780 fn read_u64_le(&mut self) -> Result<u64, ()> {
781 if self.pos + 8 > self.bytes.len() {
782 return Err(());
783 }
784 let mut arr = [0u8; 8];
785 arr.copy_from_slice(&self.bytes[self.pos..self.pos + 8]);
786 self.pos += 8;
787 Ok(u64::from_le_bytes(arr))
788 }
789 fn read_bytes(&mut self, len: usize) -> Result<&'a [u8], ()> {
790 if self.pos + len > self.bytes.len() {
791 return Err(());
792 }
793 let slice = &self.bytes[self.pos..self.pos + len];
794 self.pos += len;
795 Ok(slice)
796 }
797 fn read_path_str(&mut self) -> Result<String, ()> {
798 let len = self.read_u32_le()? as usize;
799 let bytes = self.read_bytes(len)?;
800 std::str::from_utf8(bytes).map(String::from).map_err(|_| ())
801 }
802}
803
804fn normalize_path(path: &str) -> String {
807 let trimmed = path.trim_matches('/');
808 if trimmed.is_empty() {
809 "/".to_string()
810 } else {
811 format!("/{trimmed}")
812 }
813}
814
815#[cfg(unix)]
818fn preserved_mode(path: &std::path::Path) -> Option<u32> {
819 use std::os::unix::fs::MetadataExt as _;
820 std::fs::metadata(path).ok().map(|m| m.mode() & 0o7777)
821}
822
823#[cfg(not(unix))]
824fn preserved_mode(_path: &std::path::Path) -> Option<u32> {
825 None
826}
827
828fn staging_relative(path: &str) -> &str {
829 path.trim_start_matches('/')
830}
831
832fn rename_or_fallback(from: PathBuf, to: PathBuf) -> Result<(), WriteError> {
838 match std::fs::rename(&from, &to) {
839 Ok(()) => Ok(()),
840 Err(e) if e.raw_os_error() == Some(18) => {
841 let bytes = std::fs::read(&from).map_err(WriteError::Io)?;
843 std::fs::write(&to, &bytes).map_err(WriteError::Io)?;
844 let _ = std::fs::remove_file(&from);
845 Ok(())
846 }
847 Err(e) => Err(WriteError::Io(e)),
848 }
849}
850
851fn core_to_io(e: limnifs_core::CoreError) -> WriteError {
852 WriteError::Io(std::io::Error::other(format!("{e}")))
853}
854
855const STALE_SWAP_AGE: std::time::Duration = std::time::Duration::from_secs(30);
869
870fn cleanup_stale_swap_dir(path: &Path) {
871 let Some(name) = path.file_name().and_then(std::ffi::OsStr::to_str) else {
872 return;
873 };
874 let parent = path.parent().unwrap_or_else(|| Path::new("."));
875 let Ok(entries) = std::fs::read_dir(parent) else {
879 return;
880 };
881 for entry in entries.flatten() {
882 let file_name = entry.file_name();
883 let Some(fname) = file_name.to_str() else {
884 continue;
885 };
886 if fname == format!("{name}.new") {
887 let _ = std::fs::remove_dir_all(entry.path());
890 continue;
891 }
892 if let Some(rest) = fname.strip_prefix(&format!("{name}.new-")) {
893 let mtime_old = entry
895 .metadata()
896 .and_then(|m| m.modified())
897 .ok()
898 .and_then(|t| t.elapsed().ok())
899 .is_some_and(|age| age > STALE_SWAP_AGE);
900 if mtime_old {
901 let _ = std::fs::remove_dir_all(entry.path());
902 }
903 let _ = rest;
904 }
905 }
906}
907
908#[cfg(test)]
909mod tests {
910 use super::*;
911 use crate::profile;
912
913 #[cfg(unix)]
914 #[test]
915 fn commit_and_turnover_preserve_fidelity() {
916 use std::os::unix::fs::MetadataExt as _;
917 use std::os::unix::fs::PermissionsExt as _;
918
919 let workdir =
920 std::env::temp_dir().join(format!("limnifs-rw-fidelity-{}", std::process::id()));
921 let _ = std::fs::remove_dir_all(&workdir);
922 std::fs::create_dir_all(&workdir).expect("workdir");
923
924 let src = workdir.join("src");
926 std::fs::create_dir_all(&src).expect("src");
927 std::fs::write(src.join("run.sh"), b"#!/bin/sh\n").expect("write");
928 std::fs::set_permissions(src.join("run.sh"), std::fs::Permissions::from_mode(0o750))
929 .expect("chmod");
930 std::fs::write(src.join("data.bin"), vec![0x33u8; 300_000]).expect("write");
931 std::fs::hard_link(src.join("data.bin"), src.join("alias.bin")).expect("link");
932 let stamp = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::new(500_000, 42);
933 let f = std::fs::File::options()
934 .write(true)
935 .open(src.join("data.bin"))
936 .unwrap();
937 f.set_times(std::fs::FileTimes::new().set_modified(stamp))
938 .unwrap();
939 drop(f);
940
941 let manifest = workdir.join("image.lim");
942 let artifact =
943 crate::write_directory_with_config(&src, &profile::balanced_rw()).expect("pack");
944 std::fs::write(&manifest, &artifact.bytes).expect("manifest");
945 for slab in &artifact.slabs {
946 std::fs::write(
947 workdir.join(sidecar_name(&slab.locator).unwrap()),
948 &slab.bytes,
949 )
950 .expect("slab");
951 }
952
953 let mut image = RwImage::open(&manifest, profile::balanced_rw()).expect("open");
956 image
957 .update_file("/run.sh", b"#!/bin/sh\nupdated\n")
958 .expect("update");
959 let artifact = image.commit().expect("commit");
960
961 use limnifs_core::ManifestCursor;
963 let mut cursor = ManifestCursor::new(&artifact.bytes);
964 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
965 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
966 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta");
967 let inline = meta_ref.inline_metadata.as_ref().expect("inline");
968 let mut blob_cursor = ManifestCursor::new(inline);
969 let blob = limnifs_core::parse_metadata_blob(&mut blob_cursor).expect("blob");
970
971 let run = blob
972 .inodes
973 .iter()
974 .find(|i| i.mode == 0o100_750)
975 .expect("run.sh keeps its 0750 mode through commit");
976 let _ = run;
977 let data = blob
978 .inodes
979 .iter()
980 .find(|i| i.mode == 0o100_644 && i.nlink == 2)
981 .expect("hardlink keeps nlink 2 through commit");
982 assert_eq!(data.mtime_ns, 500_000_000_000_042, "mtime survives");
983
984 let mut image = RwImage::open(&manifest, profile::balanced_rw()).expect("reopen");
986 let artifact = image.turnover().expect("turnover");
987 let mut cursor = ManifestCursor::new(&artifact.bytes);
988 limnifs_core::parse_manifest_header(&mut cursor).expect("header");
989 limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
990 let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).expect("meta");
991 let inline = meta_ref.inline_metadata.as_ref().expect("inline");
992 let mut blob_cursor = ManifestCursor::new(inline);
993 let blob = limnifs_core::parse_metadata_blob(&mut blob_cursor).expect("blob");
994 assert!(
995 blob.inodes.iter().any(|i| i.mode == 0o100_750),
996 "turnover preserves the exec bit"
997 );
998 assert!(
999 blob.inodes.iter().any(|i| i.nlink == 2),
1000 "turnover preserves the hardlink"
1001 );
1002
1003 let _ = std::fs::remove_dir_all(&workdir);
1004 }
1005
1006 #[test]
1007 fn open_cleans_up_stale_new_directory() {
1008 let workdir = std::env::temp_dir().join(format!(
1012 "limnifs-crash-recovery-{}-{}",
1013 std::process::id(),
1014 std::time::SystemTime::now()
1015 .duration_since(std::time::UNIX_EPOCH)
1016 .map(|d| d.as_nanos() as u64)
1017 .unwrap_or(0),
1018 ));
1019 let _ = std::fs::remove_dir_all(&workdir);
1020 std::fs::create_dir_all(&workdir).expect("mkdir");
1021
1022 std::fs::write(workdir.join("data.txt"), b"alpha").expect("src");
1024 let manifest = workdir.join("image.lim");
1025 let artifact =
1026 crate::write_directory_with_config(&workdir, &profile::balanced()).expect("write");
1027 std::fs::write(&manifest, &artifact.bytes).expect("manifest");
1028 for slab in &artifact.slabs {
1029 let name = sidecar_name(&slab.locator).expect("slab locator");
1030 std::fs::write(workdir.join(name), &slab.bytes).expect("slab");
1031 }
1032 if let Some(sidecar) = &artifact.metadata_sidecar {
1033 let name = sidecar_name(&sidecar.locator).expect("sidecar locator");
1034 std::fs::write(workdir.join(name), &sidecar.bytes).expect("sidecar");
1035 }
1036
1037 let stale = workdir.join("image.lim.new");
1039 std::fs::create_dir_all(&stale).expect("mkdir stale");
1040 std::fs::write(stale.join("partial.lim"), b"garbage from crashed commit").expect("garbage");
1041 assert!(stale.is_dir(), "stale dir exists before open");
1042
1043 let _image = RwImage::open(&manifest, profile::balanced()).expect("open");
1045 assert!(!stale.exists(), "stale dir removed by open");
1046
1047 let _ = std::fs::remove_dir_all(&workdir);
1048 }
1049
1050 #[test]
1051 fn wal_round_trip_recovers_pending_state_after_simulated_crash() {
1052 let workdir = std::env::temp_dir().join(format!(
1066 "limnifs-wal-rt-{}-{}",
1067 std::process::id(),
1068 std::time::SystemTime::now()
1069 .duration_since(std::time::UNIX_EPOCH)
1070 .map(|d| d.as_nanos() as u64)
1071 .unwrap_or(0),
1072 ));
1073 let _ = std::fs::remove_dir_all(&workdir);
1074 std::fs::create_dir_all(&workdir).expect("mkdir");
1075 std::fs::write(workdir.join("a.txt"), b"alpha").expect("seed a");
1076 std::fs::write(workdir.join("b.txt"), b"beta").expect("seed b");
1077 let manifest = workdir.join("image.lim");
1078 let artifact =
1079 crate::write_directory_with_config(&workdir, &profile::balanced()).expect("write");
1080 std::fs::write(&manifest, &artifact.bytes).expect("manifest");
1081 for slab in &artifact.slabs {
1082 let name = sidecar_name(&slab.locator).expect("slab locator");
1083 std::fs::write(workdir.join(name), &slab.bytes).expect("slab");
1084 }
1085 if let Some(sidecar) = &artifact.metadata_sidecar {
1086 let name = sidecar_name(&sidecar.locator).expect("sidecar locator");
1087 std::fs::write(workdir.join(name), &sidecar.bytes).expect("sidecar");
1088 }
1089
1090 {
1092 let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
1093 image.add_file("c.txt", b"gamma").expect("add c");
1094 image.update_file("a.txt", b"alpha2").expect("update a");
1095 image.delete_file("b.txt").expect("delete b");
1096 assert_eq!(image.pending_changes(), 3);
1097 image.write_wal().expect("write WAL");
1100 assert!(
1101 manifest.with_extension("lim.wal").exists() || {
1102 let wal = image.wal_path();
1104 eprintln!("WAL path: {}", wal.display());
1105 wal.exists()
1106 }
1107 );
1108 }
1110
1111 let image = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1113 assert_eq!(
1114 image.pending_changes(),
1115 3,
1116 "WAL should have replayed 3 pending ops"
1117 );
1118 let _ = std::fs::remove_dir_all(&workdir);
1119 }
1120
1121 #[test]
1122 fn rw_image_create_and_add() {
1123 let config = profile::balanced();
1124 let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
1125 let inode = image.add_file("hello.txt", b"hello world").expect("add");
1126 assert_eq!(inode, 1);
1127 assert_eq!(image.pending_changes(), 1);
1128 }
1129
1130 #[test]
1131 fn rw_image_update_and_delete() {
1132 let config = profile::balanced();
1133 let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
1134 image.add_file("file.txt", b"original").expect("add");
1135 image.update_file("file.txt", b"updated").expect("update");
1136 assert_eq!(image.pending_changes(), 2);
1137 image.delete_file("file.txt").expect("delete");
1138 assert_eq!(image.pending_changes(), 3);
1139 }
1140
1141 #[test]
1142 fn rw_image_update_nonexistent_fails() {
1143 let config = profile::balanced();
1144 let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
1145 assert!(image.update_file("nope.txt", b"data").is_err());
1146 }
1147
1148 #[test]
1149 fn rw_image_needs_turnover() {
1150 let mut config = profile::balanced();
1151 config.turnover_threshold = 3;
1152 let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
1153 image.add_file("a", b"data").expect("add");
1154 image.add_file("b", b"data").expect("add");
1155 assert!(!image.needs_turnover());
1156 image.add_file("c", b"data").expect("add");
1157 assert!(image.needs_turnover());
1158 }
1159
1160 fn write_initial(files: &[(&str, &[u8])], config: &WriteConfig) -> PathBuf {
1163 let staging = std::env::temp_dir().join(format!(
1164 "limnifs-rw-test-init-{}-{}",
1165 std::process::id(),
1166 rand_u64()
1167 ));
1168 let _ = std::fs::remove_dir_all(&staging);
1169 std::fs::create_dir_all(&staging).expect("mkdir staging");
1170 for (name, data) in files {
1171 let path = staging.join(name);
1172 if let Some(parent) = path.parent() {
1173 std::fs::create_dir_all(parent).expect("mkdir parent");
1174 }
1175 std::fs::write(&path, data).expect("write file");
1176 }
1177 let manifest = staging.join("image.lim");
1178 let artifact = crate::write_directory_with_config(&staging, config).expect("write");
1179 std::fs::write(&manifest, &artifact.bytes).expect("write manifest");
1180 for slab in &artifact.slabs {
1181 let name = sidecar_name(&slab.locator).expect("locator");
1182 std::fs::write(staging.join(name), &slab.bytes).expect("write slab");
1183 }
1184 if let Some(sidecar) = &artifact.metadata_sidecar {
1185 let name = sidecar_name(&sidecar.locator).expect("locator");
1186 std::fs::write(staging.join(name), &sidecar.bytes).expect("write sidecar");
1187 }
1188 manifest
1189 }
1190
1191 fn rand_u64() -> u64 {
1194 use std::cell::Cell;
1195 use std::time::{SystemTime, UNIX_EPOCH};
1196 thread_local!(static SEED: Cell<u64> = {
1197 let nanos = SystemTime::now()
1198 .duration_since(UNIX_EPOCH)
1199 .map(|d| d.as_nanos() as u64)
1200 .unwrap_or(0);
1201 Cell::new(nanos ^ (std::process::id() as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15))
1202 });
1203 SEED.with(|s| {
1204 let mut x = s.get();
1205 x ^= x << 13;
1206 x ^= x >> 7;
1207 x ^= x << 17;
1208 s.set(x);
1209 x
1210 })
1211 }
1212
1213 #[test]
1214 fn rw_image_open_round_trip() {
1215 let config = profile::balanced();
1216 let manifest = write_initial(
1217 &[("hello.txt", b"hello world"), ("dir/note.txt", b"nested")],
1218 &config,
1219 );
1220 let image = RwImage::open(&manifest, profile::balanced()).expect("open");
1221 assert_eq!(
1222 image.read_file("hello.txt").expect("read hello"),
1223 b"hello world"
1224 );
1225 assert_eq!(
1226 image.read_file("dir/note.txt").expect("read note"),
1227 b"nested"
1228 );
1229 }
1230
1231 #[test]
1232 fn rw_image_commit_adds_file() {
1233 let config = profile::balanced();
1234 let manifest = write_initial(&[("a.txt", b"alpha")], &config);
1235
1236 let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
1237 image.add_file("b.txt", b"beta").expect("add");
1238 let _ = image.commit().expect("commit");
1239
1240 let reread = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1241 assert_eq!(reread.read_file("a.txt").expect("read a"), b"alpha");
1242 assert_eq!(reread.read_file("b.txt").expect("read b"), b"beta");
1243 }
1244
1245 #[test]
1246 fn concurrent_readers_never_observe_torn_state_during_commit() {
1247 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
1253 use std::sync::Arc;
1254
1255 let config = profile::balanced();
1256 let manifest = write_initial(&[("a.txt", b"gen0")], &config);
1257
1258 let stop = Arc::new(AtomicBool::new(false));
1259 let opens = Arc::new(AtomicUsize::new(0));
1260 let torn = Arc::new(AtomicUsize::new(0));
1261
1262 let mut handles = Vec::new();
1263 for _ in 0..4 {
1264 let m = manifest.clone();
1265 let stop = Arc::clone(&stop);
1266 let opens = Arc::clone(&opens);
1267 let torn = Arc::clone(&torn);
1268 handles.push(std::thread::spawn(move || {
1269 while !stop.load(Ordering::Relaxed) {
1270 match RwImage::open(&m, profile::balanced()) {
1271 Ok(image) => {
1272 opens.fetch_add(1, Ordering::Relaxed);
1273 match image.read_file("a.txt") {
1275 Ok(bytes) if bytes == b"gen0" => {}
1276 Ok(bytes) if bytes == b"genN" => {}
1277 Ok(bytes) => {
1278 eprintln!("torn read: {:?}", String::from_utf8_lossy(&bytes));
1279 torn.fetch_add(1, Ordering::Relaxed);
1280 }
1281 Err(e) => {
1282 eprintln!("torn open->read: {e}");
1283 torn.fetch_add(1, Ordering::Relaxed);
1284 }
1285 }
1286 }
1287 Err(e) => {
1288 eprintln!("torn open: {e}");
1289 torn.fetch_add(1, Ordering::Relaxed);
1290 }
1291 }
1292 }
1293 }));
1294 }
1295
1296 {
1298 let mut image = RwImage::open(&manifest, profile::balanced()).expect("open writer");
1299 image.update_file("a.txt", b"genN").expect("update");
1300 image.commit().expect("commit during readers");
1301 }
1302 std::thread::sleep(std::time::Duration::from_millis(50));
1303 {
1304 let mut image = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1305 image.update_file("a.txt", b"genN").expect("update 2");
1306 image.commit().expect("commit 2");
1307 }
1308
1309 stop.store(true, Ordering::Relaxed);
1310 for h in handles {
1311 h.join().expect("reader thread");
1312 }
1313 assert_eq!(
1314 torn.load(Ordering::Relaxed),
1315 0,
1316 "no reader saw a torn snapshot"
1317 );
1318 assert!(
1319 opens.load(Ordering::Relaxed) > 0,
1320 "readers actually opened the image"
1321 );
1322 }
1323
1324 #[test]
1325 fn rw_image_commit_updates_and_deletes() {
1326 let config = profile::balanced();
1327 let manifest = write_initial(&[("a.txt", b"alpha"), ("b.txt", b"beta")], &config);
1328
1329 let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
1330 image.update_file("a.txt", b"alpha2").expect("update");
1331 image.delete_file("b.txt").expect("delete");
1332 let _ = image.commit().expect("commit");
1333
1334 let reread = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1335 assert_eq!(reread.read_file("a.txt").expect("read a"), b"alpha2");
1336 assert!(reread.read_file("b.txt").is_err(), "b.txt must be gone");
1337 }
1338
1339 #[test]
1340 fn rw_image_turnover_preserves_tree() {
1341 let config = profile::max_write();
1342 let manifest = write_initial(
1343 &[("hello.txt", b"hello world"), ("dir/note.txt", b"nested")],
1344 &config,
1345 );
1346
1347 let image = RwImage::open(&manifest, profile::max_write()).expect("open");
1348 let _ = image.turnover().expect("turnover");
1349
1350 let reread = RwImage::open(&manifest, profile::max_write()).expect("reopen");
1351 assert_eq!(
1352 reread.read_file("hello.txt").expect("read hello"),
1353 b"hello world"
1354 );
1355 assert_eq!(
1356 reread.read_file("dir/note.txt").expect("read note"),
1357 b"nested"
1358 );
1359 }
1360}