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    /// BLAKE3 of the on-disk manifest bytes this handle's WAL entries
62    /// are based on. A WAL whose tag no longer matches the manifest
63    /// is stale (its commit completed) and is discarded instead of
64    /// replayed — replaying it would resurrect a pending inode number
65    /// that the committed blob never had (torn reads for concurrent
66    /// openers in the swap→WAL-discard window).
67    base_manifest_hash: Option<[u8; 32]>,
68}
69
70/// State populated by `RwImage::open` so subsequent `commit` /
71/// `turnover` calls can read the live tree without re-parsing.
72struct OpenState {
73    blob: MetadataBlob,
74    root_inode: u64,
75    slab_store: Option<SlabStore>,
76}
77
78/// A history operation recorded for incremental updates.
79#[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    /// Open an existing image for read-write access. Parses the
100    /// manifest, mmaps the slabs, and builds the path index.
101    ///
102    /// # Errors
103    /// Returns [`WriteError`] if the manifest cannot be parsed or
104    /// the slab files cannot be opened.
105    pub fn open(path: &Path, config: WriteConfig) -> Result<Self, WriteError> {
106        // Crash recovery: if a previous commit was interrupted
107        // mid-swap, `<path>.new/` may still exist. The previous
108        // manifest at `path` is intact (atomic swap was incomplete);
109        // the `.new/` directory is garbage. Clean it up before
110        // proceeding so the next commit's `write_artifact` doesn't
111        // trip over a stale directory.
112        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        // Replay WAL if present (crash recovery for pending state).
174        let _ = image.replay_wal_if_present();
175        Ok(image)
176    }
177
178    /// Create a new empty RW image. The first `commit` produces the
179    /// on-disk manifest + slabs.
180    #[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    /// Add a new file to the image. The plaintext is staged for the
195    /// next `commit`; no I/O happens until then.
196    ///
197    /// # Errors
198    /// Returns [`WriteError`] only if internal allocation fails.
199    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    /// Update an existing file. The old inode is marked superseded;
214    /// old drops remain in slabs until the next turnover.
215    ///
216    /// # Errors
217    /// Returns [`WriteError`] if the path doesn't exist.
218    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    /// Delete a file. The inode is removed from the path index; old
237    /// drops remain until the next turnover.
238    ///
239    /// # Errors
240    /// Returns [`WriteError`] if the path doesn't exist.
241    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    /// Read a file's plaintext from the in-memory state. Only
253    /// available for images that have been `open`ed.
254    ///
255    /// # Errors
256    /// Returns [`WriteError`] if the image was not opened, the path
257    /// is unknown, or the slab is missing/corrupt.
258    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        // Pending content wins: after a WAL replay (crash recovery, or
264        // a reader that opened while a commit was in flight) the
265        // inode map carries pending inode numbers the committed blob
266        // does not have — their bytes live in `pending_files`.
267        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    /// Number of pending (uncommitted) changes.
303    #[must_use]
304    pub fn pending_changes(&self) -> usize {
305        self.pending_history.len()
306    }
307
308    /// Check if auto-turnover should trigger based on the config's
309    /// `turnover_threshold`.
310    #[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    /// Get the image mode (RO vs RW sub-mode).
317    #[must_use]
318    pub fn mode(&self) -> &ImageMode {
319        &self.config.mode
320    }
321
322    /// Commit pending changes. Materializes the live tree (if any),
323    /// overlays pending writes, rebuilds the image with the
324    /// configured codecs, and writes the new manifest + slabs.
325    ///
326    /// **Crash safety**: writes the WAL with planned operations
327    /// *before* the manifest swap. If the swap is interrupted, the
328    /// WAL survives and is replayed on the next `open`, restoring
329    /// the user's pending writes. On successful swap, the WAL is
330    /// unlinked.
331    ///
332    /// # Errors
333    /// Returns [`WriteError`] on I/O or serialization failure.
334    pub fn commit(&mut self) -> Result<crate::WriteArtifact, WriteError> {
335        // Write the WAL first so a crash mid-swap preserves pending state.
336        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        // Successful swap — discard the WAL and re-tag this handle to
343        // the new manifest so a subsequent update's WAL carries the
344        // right base generation.
345        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    /// Turnover: rebuild the current live tree with the configured
351    /// codecs. Pending changes are dropped — this is a hygiene
352    /// operation, not a commit.
353    ///
354    /// # Errors
355    /// Returns [`WriteError`] on I/O or serialization failure.
356    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    /// Build the staging tree: live tree (if opened) + pending
372    /// changes (adds/updates overwriting live entries, deletes
373    /// removing them).
374    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        // Apply deletes for paths not covered by a subsequent
383        // pending write.
384        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        // Overlay pending writes. An UPDATE rewrites content on top
393        // of a materialized live file — `fs::write` would clobber
394        // its captured mode, so preserve it across the rewrite (the
395        // mtime refreshes honestly: the content changed).
396        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    /// Turnover helper: write the live tree verbatim (no pending
440    /// overlays) to `staging`.
441    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    /// Recursively walk the live tree and write each entry under
449    /// `staging`. Delegates to the shared
450    /// [`limnifs_core::live_tree::walk_live_tree`] with a
451    /// [`FilesystemSink`].
452    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        // Directory identities (and, under the xattr feature, their
461        // extended attributes) — without this, turnover strips what
462        // the pack captured.
463        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    /// Persist the produced manifest + slabs to disk, replacing the
475    /// previous files at the same paths.
476    /// Persist the produced manifest + slabs to disk atomically.
477    ///
478    /// Files are written to `<manifest_path>.new/` then renamed into
479    /// place. `rename(2)` is atomic for a single file on POSIX
480    /// filesystems (APFS, ext4, btrfs, xfs); ordering the renames
481    /// sidecar → slabs → manifest means a reader opening the manifest
482    /// always sees a consistent snapshot (referenced slabs already
483    /// exist).
484    ///
485    /// A crash mid-sequence leaves `<manifest_path>.new/` on disk;
486    /// the next `RwImage::open` could detect and clean it up (TODO:
487    /// `06-rw-crash-safety.md`).
488    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        // Unique staging name: a concurrent RwImage::open must never
495        // delete a LIVE writer's staging (that was a real race —
496        // readers ran cleanup_stale_swap_dir on `<name>.new` while a
497        // commit was mid-flight). Age-based stale cleanup in open()
498        // only touches directories older than STALE_SWAP_AGE.
499        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        // Write all files into staging first.
515        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            // Locators are content-derived in the writer — a commit
526            // never overwrites a slab file a live manifest references.
527            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        // Rename into place: sidecar → slabs → manifest. The manifest
543        // is last so a reader never sees a manifest that references
544        // missing slabs.
545        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        // Cleanup staging directory (now empty).
554        let _ = std::fs::remove_dir_all(&staging);
555        Ok(())
556    }
557
558    /// Pick a workspace-local scratch directory for staging. Walks up
559    /// from the manifest path to find a `Cargo.toml` with
560    /// `[workspace]`; falls back to the manifest's parent directory.
561    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    /// Path to the write-ahead log: `<manifest_path>.wal`.
589    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    /// Write the WAL atomically. Records every pending op so a crash
604    /// mid-swap can be recovered on next `open`.
605    fn write_wal(&self) -> Result<(), WriteError> {
606        let mut buf: Vec<u8> = Vec::new();
607        // Header: magic + base-manifest tag.
608        buf.extend_from_slice(b"LIMWAL\0\0");
609        buf.extend_from_slice(&self.base_manifest_hash.unwrap_or([0u8; 32]));
610        // pending_files.
611        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        // pending_history.
618        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        // Write to temp file, then rename (atomic on POSIX).
636        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    /// If `<manifest_path>.wal` exists, parse and replay pending
643    /// operations into the in-memory state. Returns the count of
644    /// replayed entries (0 if no WAL exists). Best-effort: corrupt
645    /// WAL is silently discarded with a stderr warning.
646    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        // Stale-WAL gate: the WAL belongs to the manifest generation
656        // recorded in its tag. If the on-disk manifest differs, the
657        // WAL's commit already completed — replaying would corrupt
658        // the inode map (torn reads). Discard.
659        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        // WAL replayed — discard so subsequent opens don't double-replay.
749        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
804/// Normalize a user-supplied path to a leading-`/` form so it
805/// matches `MetadataBlob::build_path_index` keys.
806fn 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/// Strip the leading `/` so a key can be safely joined onto a
816/// staging root.
817#[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
832/// `rename(2)` is atomic on POSIX filesystems when source and
833/// destination are on the same filesystem. If they're not (e.g.
834/// `/tmp` → `/`), `rename` fails with `EXDEV` — fall back to
835/// `write` + `remove` so we still get the final state, just without
836/// the cross-reader atomicity guarantee.
837fn 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            // EXDEV: cross-device rename. Fall back.
842            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
855/// Detect and remove a stale `<path>.new/` directory left behind by
856/// an interrupted commit. The previous manifest at `path` is intact
857/// (atomic swap is incomplete by construction — `write_artifact`
858/// renames the manifest last); the `.new/` is garbage.
859///
860/// Logs nothing on success; silently ignores missing directory. If
861/// the directory exists but cannot be removed (e.g. permissions),
862/// the next commit's `write_artifact` will fail with a clearer
863/// error when it tries to recreate the directory.
864/// Staging directories older than this are considered abandoned
865/// (crashed commits) and removed by `RwImage::open`. Live commits
866/// finish in milliseconds, so a live writer's staging is never
867/// old enough to be collected.
868const 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    // Legacy exact name plus unique-suffixed live scheme; only remove
876    // entries older than STALE_SWAP_AGE so a concurrent commit's
877    // staging is never touched.
878    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            // Legacy staging (pre unique-suffix): no live writer can
888            // be using this name anymore.
889            let _ = std::fs::remove_dir_all(entry.path());
890            continue;
891        }
892        if let Some(rest) = fname.strip_prefix(&format!("{name}.new-")) {
893            // Unique-suffix staging: only remove when abandoned.
894            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        // A fidelity-rich source: exec bit, hardlink, distinct mtime.
925        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        // RW update + commit: the untouched files' identity must
954        // survive the staging round-trip.
955        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        // Parse the committed blob and assert identity.
962        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        // Turnover on the same image also preserves identity.
985        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        // Simulate a crashed previous commit: image exists, plus a
1009        // stale <image>.new/ directory. RwImage::open must remove
1010        // the stale directory so the next commit doesn't trip.
1011        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        // Write a minimal valid image.
1023        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        // Simulate a crash: create <image>.new/ with garbage.
1038        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        // Open should clean it up.
1044        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        // 1. Build a base image.
1053        // 2. Open it, add a file, update another, delete a third
1054        //    (this populates pending_files/pending_history).
1055        // 3. Call commit() — but simulate a crash by manually
1056        //    keeping the WAL around after the swap (i.e., we don't
1057        //    unlink it). Actually, simpler: call commit() which
1058        //    writes the WAL and runs the swap; then re-create the
1059        //    WAL by writing it ourselves with the same pending state.
1060        // 4. Open again — WAL replay should restore pending state.
1061        //
1062        // The simplest faithful simulation: open → mutate → drop the
1063        // image without commit → manually call write_wal on a fresh
1064        // image pointing at the same manifest.
1065        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        // Mutate but don't commit (simulates crash before swap).
1091        {
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            // Write WAL without running swap (simulates crash between
1098            // WAL write and successful swap).
1099            image.write_wal().expect("write WAL");
1100            assert!(
1101                manifest.with_extension("lim.wal").exists() || {
1102                    // Some platforms the file_name handling differs; check via wal_path.
1103                    let wal = image.wal_path();
1104                    eprintln!("WAL path: {}", wal.display());
1105                    wal.exists()
1106                }
1107            );
1108            // Drop without calling commit. The image is unchanged on disk.
1109        }
1110
1111        // Reopen — WAL should replay and restore pending state.
1112        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    /// Helper: write `files` into a fresh staging dir, build an
1161    /// image with the given config, and return the manifest path.
1162    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    /// Tiny PRNG to avoid pulling the `rand` crate just for a
1192    /// non-colliding nonce in tests.
1193    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        // IMPL-4 (TODO.remaining): a second thread opening the image
1248        // during a commit must never observe an inconsistent snapshot
1249        // — either the old complete image or the new complete one.
1250        // The write_artifact ordering (sidecar → slabs → manifest)
1251        // plus per-file rename(2) atomicity guarantees this.
1252        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                            // Either generation is valid; anything else is torn.
1274                            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        // Commit several generations while readers loop.
1297        {
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}