Skip to main content

limnifs_write/
rw.rs

1//! Read-write image API — LimniFS's key differentiator.
2//!
3//! Unlike SquashFS/DwarFS (read-only), LimniFS images support
4//! incremental updates: add, modify, and delete files without
5//! rebuilding the entire image.
6//!
7//! ## Architecture
8//!
9//! ```text
10//! RwImage
11//!   ├── manifest_path: where the .lim lives
12//!   ├── config: WriteConfig with write_codec + turnover defaults
13//!   ├── state: parsed metadata blob + slab store (populated on open)
14//!   ├── inode_map: path → inode map
15//!   ├── pending_files: path → plaintext for staged writes
16//!   ├── pending_history: operation log (add/update/delete)
17//!   └── next_inode: allocator
18//! ```
19//!
20//! ## Lifecycle
21//!
22//! 1. **Open** (`RwImage::open`): parse manifest, mmap slabs, build
23//!    path index. Pending changes start empty.
24//! 2. **Mutate**: `add_file` / `update_file` / `delete_file` stage
25//!    changes. Files are kept as plaintext in memory.
26//! 3. **Commit**: materialize the live tree (if opened) into a
27//!    workspace `.scratch/` directory, overlay pending changes, and
28//!    rebuild the image with the configured codecs.
29//! 4. **Turnover**: re-build the current live tree with the
30//!    configured (turnover) codecs, discarding pending changes. This
31//!    is the hygiene operation that reclaims unreferenced drops.
32
33#![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
47/// An open read-write LimniFS image.
48pub struct RwImage {
49    manifest_path: PathBuf,
50    config: WriteConfig,
51    state: Option<OpenState>,
52    /// Path → inode number map for fast lookups. Keys are absolute
53    /// POSIX-style paths (with leading `/`).
54    inode_map: HashMap<String, u64>,
55    /// Files staged for write/update. Keys mirror `inode_map`.
56    pending_files: HashMap<String, Vec<u8>>,
57    /// Pending operations since open.
58    pending_history: Vec<HistoryEntry>,
59    /// Next available inode number.
60    next_inode: u64,
61}
62
63/// State populated by `RwImage::open` so subsequent `commit` /
64/// `turnover` calls can read the live tree without re-parsing.
65struct OpenState {
66    blob: MetadataBlob,
67    root_inode: u64,
68    slab_store: Option<SlabStore>,
69}
70
71/// A history operation recorded for incremental updates.
72#[derive(Clone, Debug)]
73pub enum HistoryEntry {
74    Add {
75        path: String,
76        inode: u64,
77        size: u64,
78    },
79    Update {
80        path: String,
81        old_inode: u64,
82        new_inode: u64,
83        size: u64,
84    },
85    Delete {
86        path: String,
87        inode: u64,
88    },
89}
90
91impl RwImage {
92    /// Open an existing image for read-write access. Parses the
93    /// manifest, mmaps the slabs, and builds the path index.
94    ///
95    /// # Errors
96    /// Returns [`WriteError`] if the manifest cannot be parsed or
97    /// the slab files cannot be opened.
98    pub fn open(path: &Path, config: WriteConfig) -> Result<Self, WriteError> {
99        // Crash recovery: if a previous commit was interrupted
100        // mid-swap, `<path>.new/` may still exist. The previous
101        // manifest at `path` is intact (atomic swap was incomplete);
102        // the `.new/` directory is garbage. Clean it up before
103        // proceeding so the next commit's `write_artifact` doesn't
104        // trip over a stale directory.
105        cleanup_stale_swap_dir(path);
106
107        let manifest_bytes = std::fs::read(path).map_err(WriteError::Io)?;
108
109        let mut cursor = ManifestCursor::new(&manifest_bytes);
110        let _ = limnifs_core::parse_manifest_header(&mut cursor).map_err(core_to_io)?;
111        let _ = limnifs_core::parse_feature_flags_section(&mut cursor).map_err(core_to_io)?;
112        let meta_ref = limnifs_core::parse_metadata_reference(&mut cursor).map_err(core_to_io)?;
113
114        let blob_bytes: Vec<u8> = if let Some(inline) = meta_ref.inline_metadata.as_ref() {
115            inline.clone()
116        } else {
117            let entry = meta_ref.locators.first().ok_or_else(|| {
118                WriteError::Io(std::io::Error::other(
119                    "metadata_reference has neither inline data nor locators",
120                ))
121            })?;
122            let name = sidecar_name(&entry.uri)?;
123            let sidecar = path.parent().unwrap_or_else(|| Path::new(".")).join(name);
124            let wire_bytes = std::fs::read(&sidecar).map_err(WriteError::Io)?;
125            if meta_ref.codec == 0 {
126                wire_bytes
127            } else {
128                codec::decompress(meta_ref.codec, &wire_bytes, meta_ref.uncompressed_len)
129                    .map_err(core_to_io)?
130            }
131        };
132
133        let mut blob_cursor = ManifestCursor::new(&blob_bytes);
134        let blob = limnifs_core::parse_metadata_blob(&mut blob_cursor).map_err(core_to_io)?;
135
136        let slab_index = limnifs_core::parse_slab_index(&mut cursor).map_err(core_to_io)?;
137        let slab_store = if slab_index.is_empty() {
138            None
139        } else {
140            Some(SlabStore::load_mmap(path, &slab_index).map_err(core_to_io)?)
141        };
142
143        let root_inode = blob.root_inode_number().ok_or_else(|| {
144            WriteError::Io(std::io::Error::other(
145                "metadata blob: could not identify a unique root directory inode",
146            ))
147        })?;
148
149        let path_index = blob.build_path_index();
150        let next_inode = blob.inodes.iter().map(|i| i.number).max().unwrap_or(0) + 1;
151
152        let mut image = Self {
153            manifest_path: path.to_path_buf(),
154            config,
155            state: Some(OpenState {
156                blob,
157                root_inode,
158                slab_store,
159            }),
160            inode_map: path_index,
161            pending_files: HashMap::new(),
162            pending_history: Vec::new(),
163            next_inode,
164        };
165        // Replay WAL if present (crash recovery for pending state).
166        let _ = image.replay_wal_if_present();
167        Ok(image)
168    }
169
170    /// Create a new empty RW image. The first `commit` produces the
171    /// on-disk manifest + slabs.
172    #[must_use]
173    pub fn create_new(path: &Path, config: WriteConfig) -> Self {
174        Self {
175            manifest_path: path.to_path_buf(),
176            config,
177            state: None,
178            inode_map: HashMap::new(),
179            pending_files: HashMap::new(),
180            pending_history: Vec::new(),
181            next_inode: 1,
182        }
183    }
184
185    /// Add a new file to the image. The plaintext is staged for the
186    /// next `commit`; no I/O happens until then.
187    ///
188    /// # Errors
189    /// Returns [`WriteError`] only if internal allocation fails.
190    pub fn add_file(&mut self, path: &str, data: &[u8]) -> Result<u64, WriteError> {
191        let inode = self.next_inode;
192        self.next_inode += 1;
193        let key = normalize_path(path);
194        self.pending_files.insert(key.clone(), data.to_vec());
195        self.inode_map.insert(key.clone(), inode);
196        self.pending_history.push(HistoryEntry::Add {
197            path: key,
198            inode,
199            size: data.len() as u64,
200        });
201        Ok(inode)
202    }
203
204    /// Update an existing file. The old inode is marked superseded;
205    /// old drops remain in slabs until the next turnover.
206    ///
207    /// # Errors
208    /// Returns [`WriteError`] if the path doesn't exist.
209    pub fn update_file(&mut self, path: &str, data: &[u8]) -> Result<(), WriteError> {
210        let key = normalize_path(path);
211        let old_inode = *self.inode_map.get(&key).ok_or_else(|| {
212            WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
213        })?;
214        let new_inode = self.next_inode;
215        self.next_inode += 1;
216        self.pending_files.insert(key.clone(), data.to_vec());
217        self.inode_map.insert(key.clone(), new_inode);
218        self.pending_history.push(HistoryEntry::Update {
219            path: key,
220            old_inode,
221            new_inode,
222            size: data.len() as u64,
223        });
224        Ok(())
225    }
226
227    /// Delete a file. The inode is removed from the path index; old
228    /// drops remain until the next turnover.
229    ///
230    /// # Errors
231    /// Returns [`WriteError`] if the path doesn't exist.
232    pub fn delete_file(&mut self, path: &str) -> Result<(), WriteError> {
233        let key = normalize_path(path);
234        let inode = self.inode_map.remove(&key).ok_or_else(|| {
235            WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
236        })?;
237        self.pending_files.remove(&key);
238        self.pending_history
239            .push(HistoryEntry::Delete { path: key, inode });
240        Ok(())
241    }
242
243    /// Read a file's plaintext from the in-memory state. Only
244    /// available for images that have been `open`ed.
245    ///
246    /// # Errors
247    /// Returns [`WriteError`] if the image was not opened, the path
248    /// is unknown, or the slab is missing/corrupt.
249    pub fn read_file(&self, path: &str) -> Result<Vec<u8>, WriteError> {
250        let state = self.state.as_ref().ok_or_else(|| {
251            WriteError::Io(std::io::Error::other("read_file: image was not opened"))
252        })?;
253        let key = normalize_path(path);
254        let inode_num = *self.inode_map.get(&key).ok_or_else(|| {
255            WriteError::Io(std::io::Error::other(format!("path not found: {path}")))
256        })?;
257        let inode = state.blob.inode_by_number(inode_num).ok_or_else(|| {
258            WriteError::Io(std::io::Error::other(format!("inode {inode_num} missing")))
259        })?;
260        match &inode.content_handle {
261            ContentHandle::InlineData(data) => Ok(data.clone()),
262            ContentHandle::SliceMap(slices) => {
263                let store = state.slab_store.as_ref().ok_or_else(|| {
264                    WriteError::Io(std::io::Error::other(
265                        "read_file: slice-backed file but no slab store",
266                    ))
267                })?;
268                let mut out = Vec::new();
269                for slice in slices {
270                    let plaintext = store
271                        .plaintext_for(slice.drop_id.as_bytes())
272                        .ok_or_else(|| {
273                            WriteError::Io(std::io::Error::other("drop not in any slab"))
274                        })?
275                        .map_err(core_to_io)?;
276                    out.extend_from_slice(&plaintext);
277                }
278                Ok(out)
279            }
280            _ => Err(WriteError::Io(std::io::Error::other(
281                "read_file: unsupported content handle",
282            ))),
283        }
284    }
285
286    /// Number of pending (uncommitted) changes.
287    #[must_use]
288    pub fn pending_changes(&self) -> usize {
289        self.pending_history.len()
290    }
291
292    /// Check if auto-turnover should trigger based on the config's
293    /// `turnover_threshold`.
294    #[must_use]
295    pub fn needs_turnover(&self) -> bool {
296        self.config.turnover_threshold > 0
297            && self.pending_history.len() >= self.config.turnover_threshold as usize
298    }
299
300    /// Get the image mode (RO vs RW sub-mode).
301    #[must_use]
302    pub fn mode(&self) -> &ImageMode {
303        &self.config.mode
304    }
305
306    /// Commit pending changes. Materializes the live tree (if any),
307    /// overlays pending writes, rebuilds the image with the
308    /// configured codecs, and writes the new manifest + slabs.
309    ///
310    /// **Crash safety**: writes the WAL with planned operations
311    /// *before* the manifest swap. If the swap is interrupted, the
312    /// WAL survives and is replayed on the next `open`, restoring
313    /// the user's pending writes. On successful swap, the WAL is
314    /// unlinked.
315    ///
316    /// # Errors
317    /// Returns [`WriteError`] on I/O or serialization failure.
318    pub fn commit(&self) -> Result<crate::WriteArtifact, WriteError> {
319        // Write the WAL first so a crash mid-swap preserves pending state.
320        self.write_wal()?;
321        let staging = self.staging_dir();
322        self.write_staging_tree(&staging)?;
323        let artifact = crate::write_directory_with_config(&staging, &self.config)?;
324        let _ = std::fs::remove_dir_all(&staging);
325        self.write_artifact(&artifact)?;
326        // Successful swap — discard the WAL.
327        let _ = std::fs::remove_file(self.wal_path());
328        Ok(artifact)
329    }
330
331    /// Turnover: rebuild the current live tree with the configured
332    /// codecs. Pending changes are dropped — this is a hygiene
333    /// operation, not a commit.
334    ///
335    /// # Errors
336    /// Returns [`WriteError`] on I/O or serialization failure.
337    pub fn turnover(&self) -> Result<crate::WriteArtifact, WriteError> {
338        let staging = self.staging_dir();
339        if let Some(state) = &self.state {
340            self.write_live_tree_only(state, &staging)?;
341        } else {
342            let _ = std::fs::remove_dir_all(&staging);
343            std::fs::create_dir_all(&staging).map_err(WriteError::Io)?;
344            self.write_pending_only(&staging)?;
345        }
346        let artifact = crate::write_directory_with_config(&staging, &self.config)?;
347        let _ = std::fs::remove_dir_all(&staging);
348        self.write_artifact(&artifact)?;
349        Ok(artifact)
350    }
351
352    /// Build the staging tree: live tree (if opened) + pending
353    /// changes (adds/updates overwriting live entries, deletes
354    /// removing them).
355    fn write_staging_tree(&self, staging: &Path) -> Result<(), WriteError> {
356        let _ = std::fs::remove_dir_all(staging);
357        std::fs::create_dir_all(staging).map_err(WriteError::Io)?;
358
359        if let Some(state) = &self.state {
360            self.write_live_tree(state, staging)?;
361        }
362
363        // Apply deletes for paths not covered by a subsequent
364        // pending write.
365        for entry in &self.pending_history {
366            if let HistoryEntry::Delete { path, .. } = entry {
367                if !self.pending_files.contains_key(path) {
368                    let _ = std::fs::remove_file(staging.join(staging_relative(path)));
369                }
370            }
371        }
372
373        // Overlay pending writes.
374        for (path, data) in &self.pending_files {
375            let file_path = staging.join(staging_relative(path));
376            if let Some(parent) = file_path.parent() {
377                std::fs::create_dir_all(parent).map_err(WriteError::Io)?;
378            }
379            std::fs::write(&file_path, data).map_err(WriteError::Io)?;
380        }
381        Ok(())
382    }
383
384    fn write_pending_only(&self, staging: &Path) -> Result<(), WriteError> {
385        for (path, data) in &self.pending_files {
386            let file_path = staging.join(staging_relative(path));
387            if let Some(parent) = file_path.parent() {
388                std::fs::create_dir_all(parent).map_err(WriteError::Io)?;
389            }
390            std::fs::write(&file_path, data).map_err(WriteError::Io)?;
391        }
392        Ok(())
393    }
394
395    /// Turnover helper: write the live tree verbatim (no pending
396    /// overlays) to `staging`.
397    fn write_live_tree_only(&self, state: &OpenState, staging: &Path) -> Result<(), WriteError> {
398        let _ = std::fs::remove_dir_all(staging);
399        std::fs::create_dir_all(staging).map_err(WriteError::Io)?;
400        self.write_live_tree(state, staging)?;
401        Ok(())
402    }
403
404    /// Recursively walk the live tree and write each entry under
405    /// `staging`. Delegates to the shared
406    /// [`limnifs_core::live_tree::walk_live_tree`] with a
407    /// [`FilesystemSink`].
408    fn write_live_tree(&self, state: &OpenState, staging: &Path) -> Result<(), WriteError> {
409        let slab_ref: Option<&dyn limnifs_core::slab_source::SlabSource> = state
410            .slab_store
411            .as_ref()
412            .map(|s| s as &dyn limnifs_core::slab_source::SlabSource);
413        let mut sink = limnifs_core::live_tree::FilesystemSink::new(staging, slab_ref);
414        limnifs_core::live_tree::walk_live_tree(&state.blob, state.root_inode, &mut sink)
415            .map_err(core_to_io)
416    }
417
418    /// Persist the produced manifest + slabs to disk, replacing the
419    /// previous files at the same paths.
420    /// Persist the produced manifest + slabs to disk atomically.
421    ///
422    /// Files are written to `<manifest_path>.new/` then renamed into
423    /// place. `rename(2)` is atomic for a single file on POSIX
424    /// filesystems (APFS, ext4, btrfs, xfs); ordering the renames
425    /// sidecar → slabs → manifest means a reader opening the manifest
426    /// always sees a consistent snapshot (referenced slabs already
427    /// exist).
428    ///
429    /// A crash mid-sequence leaves `<manifest_path>.new/` on disk;
430    /// the next `RwImage::open` could detect and clean it up (TODO:
431    /// `06-rw-crash-safety.md`).
432    fn write_artifact(&self, artifact: &crate::WriteArtifact) -> Result<(), WriteError> {
433        let parent = self
434            .manifest_path
435            .parent()
436            .unwrap_or_else(|| Path::new("."))
437            .to_path_buf();
438        let staging = parent.join(format!(
439            "{}.new",
440            self.manifest_path
441                .file_name()
442                .and_then(std::ffi::OsStr::to_str)
443                .unwrap_or("image.lim"),
444        ));
445        let _ = std::fs::remove_dir_all(&staging);
446        std::fs::create_dir_all(&staging).map_err(WriteError::Io)?;
447
448        // Write all files into staging first.
449        let manifest_name = self
450            .manifest_path
451            .file_name()
452            .map(std::ffi::OsString::from)
453            .unwrap_or_else(|| std::ffi::OsString::from("image.lim"));
454        let manifest_staging = staging.join(&manifest_name);
455        std::fs::write(&manifest_staging, &artifact.bytes).map_err(WriteError::Io)?;
456
457        let mut slab_names: Vec<std::ffi::OsString> = Vec::new();
458        for slab in &artifact.slabs {
459            let name = sidecar_name(&slab.locator)?;
460            let os_name = std::ffi::OsString::from(name);
461            std::fs::write(staging.join(&os_name), &slab.bytes).map_err(WriteError::Io)?;
462            slab_names.push(os_name);
463        }
464        let sidecar_name: Option<std::ffi::OsString> =
465            if let Some(sidecar) = &artifact.metadata_sidecar {
466                let name = sidecar_name(&sidecar.locator)?;
467                let os_name = std::ffi::OsString::from(name);
468                std::fs::write(staging.join(&os_name), &sidecar.bytes).map_err(WriteError::Io)?;
469                Some(os_name)
470            } else {
471                None
472            };
473
474        // Rename into place: sidecar → slabs → manifest. The manifest
475        // is last so a reader never sees a manifest that references
476        // missing slabs.
477        if let Some(name) = &sidecar_name {
478            rename_or_fallback(staging.join(name), parent.join(name))?;
479        }
480        for name in &slab_names {
481            rename_or_fallback(staging.join(name), parent.join(name))?;
482        }
483        rename_or_fallback(manifest_staging, self.manifest_path.clone())?;
484
485        // Cleanup staging directory (now empty).
486        let _ = std::fs::remove_dir_all(&staging);
487        Ok(())
488    }
489
490    /// Pick a workspace-local scratch directory for staging. Walks up
491    /// from the manifest path to find a `Cargo.toml` with
492    /// `[workspace]`; falls back to the manifest's parent directory.
493    fn staging_dir(&self) -> PathBuf {
494        let nonce = format!("{}-{}", std::process::id(), self.next_inode);
495        let mut cur = self
496            .manifest_path
497            .parent()
498            .unwrap_or_else(|| Path::new("."))
499            .to_path_buf();
500        loop {
501            if cur.join("Cargo.toml").is_file() {
502                if std::fs::read_to_string(cur.join("Cargo.toml"))
503                    .map(|s| s.contains("[workspace]"))
504                    .unwrap_or(false)
505                {
506                    return cur.join(".scratch").join(format!("limnifs-rw-{nonce}"));
507                }
508            }
509            if !cur.pop() {
510                break;
511            }
512        }
513        self.manifest_path
514            .parent()
515            .unwrap_or_else(|| Path::new("."))
516            .join(".scratch")
517            .join(format!("limnifs-rw-{nonce}"))
518    }
519
520    /// Path to the write-ahead log: `<manifest_path>.wal`.
521    fn wal_path(&self) -> PathBuf {
522        let parent = self
523            .manifest_path
524            .parent()
525            .unwrap_or_else(|| Path::new("."));
526        let mut name = self
527            .manifest_path
528            .file_name()
529            .map(std::ffi::OsString::from)
530            .unwrap_or_else(|| std::ffi::OsString::from("image.lim"));
531        name.push(".wal");
532        parent.join(name)
533    }
534
535    /// Write the WAL atomically. Records every pending op so a crash
536    /// mid-swap can be recovered on next `open`.
537    fn write_wal(&self) -> Result<(), WriteError> {
538        let mut buf: Vec<u8> = Vec::new();
539        // Header: magic + version.
540        buf.extend_from_slice(b"LIMWAL\0\0");
541        // pending_files.
542        buf.extend_from_slice(&(self.pending_files.len() as u32).to_le_bytes());
543        for (path, data) in &self.pending_files {
544            write_path_str(&mut buf, path);
545            buf.extend_from_slice(&(data.len() as u64).to_le_bytes());
546            buf.extend_from_slice(data);
547        }
548        // pending_history.
549        buf.extend_from_slice(&(self.pending_history.len() as u32).to_le_bytes());
550        for entry in &self.pending_history {
551            match entry {
552                HistoryEntry::Add { path, .. } => {
553                    buf.push(1);
554                    write_path_str(&mut buf, path);
555                }
556                HistoryEntry::Update { path, .. } => {
557                    buf.push(2);
558                    write_path_str(&mut buf, path);
559                }
560                HistoryEntry::Delete { path, .. } => {
561                    buf.push(3);
562                    write_path_str(&mut buf, path);
563                }
564            }
565        }
566        // Write to temp file, then rename (atomic on POSIX).
567        let wal_tmp = self.wal_path().with_extension("wal.tmp");
568        std::fs::write(&wal_tmp, &buf).map_err(WriteError::Io)?;
569        std::fs::rename(&wal_tmp, self.wal_path()).map_err(WriteError::Io)?;
570        Ok(())
571    }
572
573    /// If `<manifest_path>.wal` exists, parse and replay pending
574    /// operations into the in-memory state. Returns the count of
575    /// replayed entries (0 if no WAL exists). Best-effort: corrupt
576    /// WAL is silently discarded with a stderr warning.
577    fn replay_wal_if_present(&mut self) -> usize {
578        let wal_path = self.wal_path();
579        let Ok(bytes) = std::fs::read(&wal_path) else {
580            return 0;
581        };
582        if bytes.len() < 8 || &bytes[..8] != b"LIMWAL\0\0" {
583            let _ = std::fs::remove_file(&wal_path);
584            return 0;
585        }
586        let mut cursor = WalCursor {
587            bytes: &bytes,
588            pos: 8,
589        };
590        let files_count = match cursor.read_u32_le() {
591            Ok(n) => n as usize,
592            Err(_) => {
593                let _ = std::fs::remove_file(&wal_path);
594                return 0;
595            }
596        };
597        for _ in 0..files_count {
598            let path = match cursor.read_path_str() {
599                Ok(p) => p,
600                Err(_) => break,
601            };
602            let len = match cursor.read_u64_le() {
603                Ok(n) => n as usize,
604                Err(_) => break,
605            };
606            let data = match cursor.read_bytes(len) {
607                Ok(d) => d.to_vec(),
608                Err(_) => break,
609            };
610            self.pending_files.insert(path, data);
611        }
612        let hist_count = match cursor.read_u32_le() {
613            Ok(n) => n as usize,
614            Err(_) => 0,
615        };
616        let mut replayed = 0;
617        for _ in 0..hist_count {
618            let op = match cursor.read_u8() {
619                Ok(b) => b,
620                Err(_) => break,
621            };
622            let path = match cursor.read_path_str() {
623                Ok(p) => p,
624                Err(_) => break,
625            };
626            match op {
627                1 => {
628                    let inode = self.next_inode;
629                    self.next_inode += 1;
630                    let size = self
631                        .pending_files
632                        .get(&path)
633                        .map(|v| v.len() as u64)
634                        .unwrap_or(0);
635                    self.inode_map.insert(path.clone(), inode);
636                    self.pending_history
637                        .push(HistoryEntry::Add { path, inode, size });
638                }
639                2 => {
640                    let old_inode = self.inode_map.get(&path).copied().unwrap_or(0);
641                    let new_inode = self.next_inode;
642                    self.next_inode += 1;
643                    let size = self
644                        .pending_files
645                        .get(&path)
646                        .map(|v| v.len() as u64)
647                        .unwrap_or(0);
648                    self.inode_map.insert(path.clone(), new_inode);
649                    self.pending_history.push(HistoryEntry::Update {
650                        path,
651                        old_inode,
652                        new_inode,
653                        size,
654                    });
655                }
656                3 => {
657                    let inode = self.inode_map.remove(&path).unwrap_or(0);
658                    self.pending_files.remove(&path);
659                    self.pending_history
660                        .push(HistoryEntry::Delete { path, inode });
661                }
662                _ => break,
663            }
664            replayed += 1;
665        }
666        // WAL replayed — discard so subsequent opens don't double-replay.
667        let _ = std::fs::remove_file(&wal_path);
668        replayed
669    }
670}
671
672fn write_path_str(out: &mut Vec<u8>, s: &str) {
673    let bytes = s.as_bytes();
674    out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
675    out.extend_from_slice(bytes);
676}
677
678struct WalCursor<'a> {
679    bytes: &'a [u8],
680    pos: usize,
681}
682
683impl<'a> WalCursor<'a> {
684    fn read_u8(&mut self) -> Result<u8, ()> {
685        let b = *self.bytes.get(self.pos).ok_or(())?;
686        self.pos += 1;
687        Ok(b)
688    }
689    fn read_u32_le(&mut self) -> Result<u32, ()> {
690        if self.pos + 4 > self.bytes.len() {
691            return Err(());
692        }
693        let mut arr = [0u8; 4];
694        arr.copy_from_slice(&self.bytes[self.pos..self.pos + 4]);
695        self.pos += 4;
696        Ok(u32::from_le_bytes(arr))
697    }
698    fn read_u64_le(&mut self) -> Result<u64, ()> {
699        if self.pos + 8 > self.bytes.len() {
700            return Err(());
701        }
702        let mut arr = [0u8; 8];
703        arr.copy_from_slice(&self.bytes[self.pos..self.pos + 8]);
704        self.pos += 8;
705        Ok(u64::from_le_bytes(arr))
706    }
707    fn read_bytes(&mut self, len: usize) -> Result<&'a [u8], ()> {
708        if self.pos + len > self.bytes.len() {
709            return Err(());
710        }
711        let slice = &self.bytes[self.pos..self.pos + len];
712        self.pos += len;
713        Ok(slice)
714    }
715    fn read_path_str(&mut self) -> Result<String, ()> {
716        let len = self.read_u32_le()? as usize;
717        let bytes = self.read_bytes(len)?;
718        std::str::from_utf8(bytes).map(String::from).map_err(|_| ())
719    }
720}
721
722/// Normalize a user-supplied path to a leading-`/` form so it
723/// matches `MetadataBlob::build_path_index` keys.
724fn normalize_path(path: &str) -> String {
725    let trimmed = path.trim_matches('/');
726    if trimmed.is_empty() {
727        "/".to_string()
728    } else {
729        format!("/{trimmed}")
730    }
731}
732
733/// Strip the leading `/` so a key can be safely joined onto a
734/// staging root.
735fn staging_relative(path: &str) -> &str {
736    path.trim_start_matches('/')
737}
738
739/// `rename(2)` is atomic on POSIX filesystems when source and
740/// destination are on the same filesystem. If they're not (e.g.
741/// `/tmp` → `/`), `rename` fails with `EXDEV` — fall back to
742/// `write` + `remove` so we still get the final state, just without
743/// the cross-reader atomicity guarantee.
744fn rename_or_fallback(from: PathBuf, to: PathBuf) -> Result<(), WriteError> {
745    match std::fs::rename(&from, &to) {
746        Ok(()) => Ok(()),
747        Err(e) if e.raw_os_error() == Some(18) => {
748            // EXDEV: cross-device rename. Fall back.
749            let bytes = std::fs::read(&from).map_err(WriteError::Io)?;
750            std::fs::write(&to, &bytes).map_err(WriteError::Io)?;
751            let _ = std::fs::remove_file(&from);
752            Ok(())
753        }
754        Err(e) => Err(WriteError::Io(e)),
755    }
756}
757
758fn core_to_io(e: limnifs_core::CoreError) -> WriteError {
759    WriteError::Io(std::io::Error::other(format!("{e}")))
760}
761
762/// Detect and remove a stale `<path>.new/` directory left behind by
763/// an interrupted commit. The previous manifest at `path` is intact
764/// (atomic swap is incomplete by construction — `write_artifact`
765/// renames the manifest last); the `.new/` is garbage.
766///
767/// Logs nothing on success; silently ignores missing directory. If
768/// the directory exists but cannot be removed (e.g. permissions),
769/// the next commit's `write_artifact` will fail with a clearer
770/// error when it tries to recreate the directory.
771fn cleanup_stale_swap_dir(path: &Path) {
772    let Some(name) = path.file_name().and_then(std::ffi::OsStr::to_str) else {
773        return;
774    };
775    let parent = path.parent().unwrap_or_else(|| Path::new("."));
776    let stale = parent.join(format!("{name}.new"));
777    if stale.is_dir() {
778        let _ = std::fs::remove_dir_all(&stale);
779    }
780}
781
782#[cfg(test)]
783mod tests {
784    use super::*;
785    use crate::profile;
786
787    #[test]
788    fn open_cleans_up_stale_new_directory() {
789        // Simulate a crashed previous commit: image exists, plus a
790        // stale <image>.new/ directory. RwImage::open must remove
791        // the stale directory so the next commit doesn't trip.
792        let workdir = std::env::temp_dir().join(format!(
793            "limnifs-crash-recovery-{}-{}",
794            std::process::id(),
795            std::time::SystemTime::now()
796                .duration_since(std::time::UNIX_EPOCH)
797                .map(|d| d.as_nanos() as u64)
798                .unwrap_or(0),
799        ));
800        let _ = std::fs::remove_dir_all(&workdir);
801        std::fs::create_dir_all(&workdir).expect("mkdir");
802
803        // Write a minimal valid image.
804        std::fs::write(workdir.join("data.txt"), b"alpha").expect("src");
805        let manifest = workdir.join("image.lim");
806        let artifact =
807            crate::write_directory_with_config(&workdir, &profile::balanced()).expect("write");
808        std::fs::write(&manifest, &artifact.bytes).expect("manifest");
809        for slab in &artifact.slabs {
810            let name = sidecar_name(&slab.locator).expect("slab locator");
811            std::fs::write(workdir.join(name), &slab.bytes).expect("slab");
812        }
813        if let Some(sidecar) = &artifact.metadata_sidecar {
814            let name = sidecar_name(&sidecar.locator).expect("sidecar locator");
815            std::fs::write(workdir.join(name), &sidecar.bytes).expect("sidecar");
816        }
817
818        // Simulate a crash: create <image>.new/ with garbage.
819        let stale = workdir.join("image.lim.new");
820        std::fs::create_dir_all(&stale).expect("mkdir stale");
821        std::fs::write(stale.join("partial.lim"), b"garbage from crashed commit").expect("garbage");
822        assert!(stale.is_dir(), "stale dir exists before open");
823
824        // Open should clean it up.
825        let _image = RwImage::open(&manifest, profile::balanced()).expect("open");
826        assert!(!stale.exists(), "stale dir removed by open");
827
828        let _ = std::fs::remove_dir_all(&workdir);
829    }
830
831    #[test]
832    fn wal_round_trip_recovers_pending_state_after_simulated_crash() {
833        // 1. Build a base image.
834        // 2. Open it, add a file, update another, delete a third
835        //    (this populates pending_files/pending_history).
836        // 3. Call commit() — but simulate a crash by manually
837        //    keeping the WAL around after the swap (i.e., we don't
838        //    unlink it). Actually, simpler: call commit() which
839        //    writes the WAL and runs the swap; then re-create the
840        //    WAL by writing it ourselves with the same pending state.
841        // 4. Open again — WAL replay should restore pending state.
842        //
843        // The simplest faithful simulation: open → mutate → drop the
844        // image without commit → manually call write_wal on a fresh
845        // image pointing at the same manifest.
846        let workdir = std::env::temp_dir().join(format!(
847            "limnifs-wal-rt-{}-{}",
848            std::process::id(),
849            std::time::SystemTime::now()
850                .duration_since(std::time::UNIX_EPOCH)
851                .map(|d| d.as_nanos() as u64)
852                .unwrap_or(0),
853        ));
854        let _ = std::fs::remove_dir_all(&workdir);
855        std::fs::create_dir_all(&workdir).expect("mkdir");
856        std::fs::write(workdir.join("a.txt"), b"alpha").expect("seed a");
857        std::fs::write(workdir.join("b.txt"), b"beta").expect("seed b");
858        let manifest = workdir.join("image.lim");
859        let artifact =
860            crate::write_directory_with_config(&workdir, &profile::balanced()).expect("write");
861        std::fs::write(&manifest, &artifact.bytes).expect("manifest");
862        for slab in &artifact.slabs {
863            let name = sidecar_name(&slab.locator).expect("slab locator");
864            std::fs::write(workdir.join(name), &slab.bytes).expect("slab");
865        }
866        if let Some(sidecar) = &artifact.metadata_sidecar {
867            let name = sidecar_name(&sidecar.locator).expect("sidecar locator");
868            std::fs::write(workdir.join(name), &sidecar.bytes).expect("sidecar");
869        }
870
871        // Mutate but don't commit (simulates crash before swap).
872        {
873            let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
874            image.add_file("c.txt", b"gamma").expect("add c");
875            image.update_file("a.txt", b"alpha2").expect("update a");
876            image.delete_file("b.txt").expect("delete b");
877            assert_eq!(image.pending_changes(), 3);
878            // Write WAL without running swap (simulates crash between
879            // WAL write and successful swap).
880            image.write_wal().expect("write WAL");
881            assert!(
882                manifest.with_extension("lim.wal").exists() || {
883                    // Some platforms the file_name handling differs; check via wal_path.
884                    let wal = image.wal_path();
885                    eprintln!("WAL path: {}", wal.display());
886                    wal.exists()
887                }
888            );
889            // Drop without calling commit. The image is unchanged on disk.
890        }
891
892        // Reopen — WAL should replay and restore pending state.
893        let image = RwImage::open(&manifest, profile::balanced()).expect("reopen");
894        assert_eq!(
895            image.pending_changes(),
896            3,
897            "WAL should have replayed 3 pending ops"
898        );
899        let _ = std::fs::remove_dir_all(&workdir);
900    }
901
902    #[test]
903    fn rw_image_create_and_add() {
904        let config = profile::balanced();
905        let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
906        let inode = image.add_file("hello.txt", b"hello world").expect("add");
907        assert_eq!(inode, 1);
908        assert_eq!(image.pending_changes(), 1);
909    }
910
911    #[test]
912    fn rw_image_update_and_delete() {
913        let config = profile::balanced();
914        let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
915        image.add_file("file.txt", b"original").expect("add");
916        image.update_file("file.txt", b"updated").expect("update");
917        assert_eq!(image.pending_changes(), 2);
918        image.delete_file("file.txt").expect("delete");
919        assert_eq!(image.pending_changes(), 3);
920    }
921
922    #[test]
923    fn rw_image_update_nonexistent_fails() {
924        let config = profile::balanced();
925        let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
926        assert!(image.update_file("nope.txt", b"data").is_err());
927    }
928
929    #[test]
930    fn rw_image_needs_turnover() {
931        let mut config = profile::balanced();
932        config.turnover_threshold = 3;
933        let mut image = RwImage::create_new(Path::new("/tmp/test.lim"), config);
934        image.add_file("a", b"data").expect("add");
935        image.add_file("b", b"data").expect("add");
936        assert!(!image.needs_turnover());
937        image.add_file("c", b"data").expect("add");
938        assert!(image.needs_turnover());
939    }
940
941    /// Helper: write `files` into a fresh staging dir, build an
942    /// image with the given config, and return the manifest path.
943    fn write_initial(files: &[(&str, &[u8])], config: &WriteConfig) -> PathBuf {
944        let staging = std::env::temp_dir().join(format!(
945            "limnifs-rw-test-init-{}-{}",
946            std::process::id(),
947            rand_u64()
948        ));
949        let _ = std::fs::remove_dir_all(&staging);
950        std::fs::create_dir_all(&staging).expect("mkdir staging");
951        for (name, data) in files {
952            let path = staging.join(name);
953            if let Some(parent) = path.parent() {
954                std::fs::create_dir_all(parent).expect("mkdir parent");
955            }
956            std::fs::write(&path, data).expect("write file");
957        }
958        let manifest = staging.join("image.lim");
959        let artifact = crate::write_directory_with_config(&staging, config).expect("write");
960        std::fs::write(&manifest, &artifact.bytes).expect("write manifest");
961        for slab in &artifact.slabs {
962            let name = sidecar_name(&slab.locator).expect("locator");
963            std::fs::write(staging.join(name), &slab.bytes).expect("write slab");
964        }
965        if let Some(sidecar) = &artifact.metadata_sidecar {
966            let name = sidecar_name(&sidecar.locator).expect("locator");
967            std::fs::write(staging.join(name), &sidecar.bytes).expect("write sidecar");
968        }
969        manifest
970    }
971
972    /// Tiny PRNG to avoid pulling the `rand` crate just for a
973    /// non-colliding nonce in tests.
974    fn rand_u64() -> u64 {
975        use std::cell::Cell;
976        use std::time::{SystemTime, UNIX_EPOCH};
977        thread_local!(static SEED: Cell<u64> = {
978            let nanos = SystemTime::now()
979                .duration_since(UNIX_EPOCH)
980                .map(|d| d.as_nanos() as u64)
981                .unwrap_or(0);
982            Cell::new(nanos ^ (std::process::id() as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15))
983        });
984        SEED.with(|s| {
985            let mut x = s.get();
986            x ^= x << 13;
987            x ^= x >> 7;
988            x ^= x << 17;
989            s.set(x);
990            x
991        })
992    }
993
994    #[test]
995    fn rw_image_open_round_trip() {
996        let config = profile::balanced();
997        let manifest = write_initial(
998            &[("hello.txt", b"hello world"), ("dir/note.txt", b"nested")],
999            &config,
1000        );
1001        let image = RwImage::open(&manifest, profile::balanced()).expect("open");
1002        assert_eq!(
1003            image.read_file("hello.txt").expect("read hello"),
1004            b"hello world"
1005        );
1006        assert_eq!(
1007            image.read_file("dir/note.txt").expect("read note"),
1008            b"nested"
1009        );
1010    }
1011
1012    #[test]
1013    fn rw_image_commit_adds_file() {
1014        let config = profile::balanced();
1015        let manifest = write_initial(&[("a.txt", b"alpha")], &config);
1016
1017        let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
1018        image.add_file("b.txt", b"beta").expect("add");
1019        let _ = image.commit().expect("commit");
1020
1021        let reread = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1022        assert_eq!(reread.read_file("a.txt").expect("read a"), b"alpha");
1023        assert_eq!(reread.read_file("b.txt").expect("read b"), b"beta");
1024    }
1025
1026    #[test]
1027    fn rw_image_commit_updates_and_deletes() {
1028        let config = profile::balanced();
1029        let manifest = write_initial(&[("a.txt", b"alpha"), ("b.txt", b"beta")], &config);
1030
1031        let mut image = RwImage::open(&manifest, profile::balanced()).expect("open");
1032        image.update_file("a.txt", b"alpha2").expect("update");
1033        image.delete_file("b.txt").expect("delete");
1034        let _ = image.commit().expect("commit");
1035
1036        let reread = RwImage::open(&manifest, profile::balanced()).expect("reopen");
1037        assert_eq!(reread.read_file("a.txt").expect("read a"), b"alpha2");
1038        assert!(reread.read_file("b.txt").is_err(), "b.txt must be gone");
1039    }
1040
1041    #[test]
1042    fn rw_image_turnover_preserves_tree() {
1043        let config = profile::max_write();
1044        let manifest = write_initial(
1045            &[("hello.txt", b"hello world"), ("dir/note.txt", b"nested")],
1046            &config,
1047        );
1048
1049        let image = RwImage::open(&manifest, profile::max_write()).expect("open");
1050        let _ = image.turnover().expect("turnover");
1051
1052        let reread = RwImage::open(&manifest, profile::max_write()).expect("reopen");
1053        assert_eq!(
1054            reread.read_file("hello.txt").expect("read hello"),
1055            b"hello world"
1056        );
1057        assert_eq!(
1058            reread.read_file("dir/note.txt").expect("read note"),
1059            b"nested"
1060        );
1061    }
1062}