Skip to main content

limnifs_write/
stream.rs

1//! Streaming multi-file writer — the no-materialisation seam.
2//!
3//! `write_stream` packs ONE named stream; [`StreamWriter`]
4//! generalises it to a whole tree of streams (tar archives, pipe
5//! bundles, network feeds) without ever touching the filesystem:
6//! entries are chunked straight off their readers via
7//! [`Chunker::chunk_reader`], whose internal buffering is bounded by
8//! the chunker's max chunk size plus one read buffer.
9//!
10//! ## Tree construction
11//!
12//! Entries arrive in arbitrary order; the tree is a nested
13//! `BTreeMap` so directory entries materialise name-sorted at
14//! `finish`. File and symlink inodes are allocated (and pushed) in
15//! arrival order; directory inodes are allocated parent-first
16//! during `finish`. The numbering therefore differs from a
17//! directory pack of the same tree (where the DFS orders
18//! allocation) — the format only requires unique numbers — but the
19//! same entry sequence always produces byte-identical images.
20//!
21//! [`Chunker::chunk_reader`]: crate::chunker::Chunker::chunk_reader
22
23use std::collections::BTreeMap;
24use std::io::Read;
25use std::path::PathBuf;
26
27use crate::chunker::Chunker;
28use crate::classifier;
29use crate::config::WriteConfig;
30use crate::{
31    encode_dir_node, hash_section, PendingContent, PendingFile, PendingInode, TournamentSpec,
32    WriteArtifact, WriteContext, WriteError,
33};
34
35/// Codec setup shared by every entry of one stream write. Built
36/// once at construction so per-entry cost is pure chunk + compress.
37struct StreamCodecs {
38    chunker: crate::chunker::ParallelFastCDC,
39    classifier: classifier::Classifier,
40    text_codec: u8,
41    binary_codec: u8,
42    tunables: limnifs_core::codec::CodecTunables,
43    tournament: TournamentSpec,
44}
45
46impl StreamCodecs {
47    fn from_config(
48        chunker: crate::chunker::ParallelFastCDC,
49        classifier: classifier::Classifier,
50        config: &WriteConfig,
51    ) -> Result<Self, WriteError> {
52        let registry = config
53            .codec_registry()
54            .map_err(|e| WriteError::Io(std::io::Error::other(format!("codec registry: {e}"))))?;
55        let tournament_codec_ids: Vec<u8> = config
56            .tournament
57            .codecs
58            .iter()
59            .filter_map(|n| registry.lookup_by_name(n))
60            .collect();
61        Ok(Self {
62            chunker,
63            classifier,
64            text_codec: config.text_codec_id().unwrap_or(0x04),
65            binary_codec: config.binary_codec_id().unwrap_or(0x01),
66            tunables: config.to_core_tunables(),
67            tournament: TournamentSpec {
68                codec_ids: tournament_codec_ids,
69                min_size: config.tournament.min_size_threshold as usize,
70                skip_for_binary: config.tournament.skip_for_binary,
71                short_circuit_permille: config.tournament.short_circuit_threshold,
72            },
73        })
74    }
75}
76
77/// One entry's unix identity, applied to its emitted inode:
78/// nanosecond mtime, permission bits, and owner. Replaces the
79/// scattered `(mtime_ns, perms)` pairs — ownership is identity,
80/// not an afterthought (the tar headers carry it first-class and
81/// the directory writer captures it since v0.3.37).
82#[derive(Clone, Copy, Debug)]
83pub struct EntryMeta {
84    pub mtime_ns: u64,
85    pub perms: u32,
86    pub uid: u32,
87    pub gid: u32,
88}
89
90impl EntryMeta {
91    /// Identity with root ownership; chain `.owner` for real uid/gid.
92    #[must_use]
93    pub const fn new(mtime_ns: u64, perms: u32) -> Self {
94        Self {
95            mtime_ns,
96            perms,
97            uid: 0,
98            gid: 0,
99        }
100    }
101
102    /// Set the owner.
103    #[must_use]
104    pub const fn owner(mut self, uid: u32, gid: u32) -> Self {
105        self.uid = uid;
106        self.gid = gid;
107        self
108    }
109}
110
111/// A directory level under construction: children in name order,
112/// plus the identity carried by an explicit `add_dir` (implicit
113/// directories created by a nested file path keep mtime 0 / root).
114struct StreamDir {
115    meta: EntryMeta,
116    children: BTreeMap<String, StreamNode>,
117}
118
119impl Default for StreamDir {
120    fn default() -> Self {
121        Self {
122            meta: EntryMeta::new(0, 0o755),
123            children: BTreeMap::new(),
124        }
125    }
126}
127
128enum StreamNode {
129    Dir(StreamDir),
130    /// Inode already pushed; the number wires the tree at `finish`.
131    File {
132        inode_number: u64,
133    },
134    Symlink {
135        inode_number: u64,
136    },
137}
138
139/// Build one `.lim` image from a sequence of named streams.
140///
141/// Create with [`StreamWriter::new`], add entries in any order
142/// ([`add_file`], [`add_dir`], [`add_symlink`]), then [`finish`] to
143/// assemble the artifact. Names are `/`-separated image-relative
144/// paths; parent directories are materialised implicitly, or
145/// explicitly via [`add_dir`] to control mtimes and empty
146/// directories.
147///
148/// [`add_file`]: Self::add_file
149/// [`add_dir`]: Self::add_dir
150/// [`add_symlink`]: Self::add_symlink
151/// [`finish`]: Self::finish
152///
153/// # Errors
154///
155/// [`WriteError::Io`] on reader failure, invalid names, conflicting
156/// paths, or any writer-pipeline error.
157pub struct StreamWriter<'a> {
158    ctx: WriteContext,
159    config: &'a WriteConfig,
160    codecs: StreamCodecs,
161    inline_threshold: u64,
162    tree: StreamDir,
163    /// Random-access entries awaiting the parallel flush at
164    /// `finish` (see [`stage_file`]). Flushed in stage order, after
165    /// every immediate `add_*` call.
166    ///
167    /// [`stage_file`]: Self::stage_file
168    staged: Vec<StagedEntry<'a>>,
169}
170
171/// One deferred stream entry: the tree slot and inode exist; only
172/// the chunk/hash/compress work is pending.
173struct StagedEntry<'a> {
174    name: String,
175    meta: EntryMeta,
176    xattrs: Vec<limnifs_core::inode::XAttr>,
177    inode_number: u64,
178    data: &'a [u8],
179}
180
181impl<'a> StreamWriter<'a> {
182    /// Start a stream write under `config`.
183    ///
184    /// # Errors
185    ///
186    /// [`WriteError::Io`] if the config's codec registry or chunking
187    /// parameters are invalid.
188    pub fn new(config: &'a WriteConfig) -> Result<Self, WriteError> {
189        let mut ctx = WriteContext::new();
190        ctx.chunker = crate::chunker_from_config(config)?;
191        ctx.categorizers_disabled = config.categorizers.is_empty();
192        ctx.rw_mode = matches!(config.mode, crate::config::ImageMode::ReadWrite(_));
193        ctx.auto_turnover = config.turnover_threshold > 0;
194        ctx.collect_dict_samples = config.dictionaries.enabled;
195        ctx.inline_threshold = config.defaults.inline_threshold as usize;
196        ctx.metadata_externalize_threshold = config.defaults.metadata_externalize_threshold;
197        ctx.emit_shared_inline = config.defaults.shared_inline;
198        let classifier = ctx.classifier;
199        let chunker = ctx.chunker.clone();
200        Ok(Self {
201            codecs: StreamCodecs::from_config(chunker, classifier, config)?,
202            inline_threshold: u64::try_from(ctx.inline_threshold).unwrap_or(u64::MAX),
203            ctx,
204            config,
205            tree: StreamDir::default(),
206            staged: Vec::new(),
207        })
208    }
209
210    /// Add a regular file at `name`, streaming `reader` through the
211    /// chunker. Small entries (within the config's inline
212    /// threshold) are stored inline, matching the directory writer.
213    ///
214    /// # Errors
215    ///
216    /// [`WriteError::Io`] if the name is invalid or conflicts with
217    /// an existing entry, or the reader fails.
218    pub fn add_file(
219        &mut self,
220        name: &str,
221        meta: EntryMeta,
222        xattrs: &[(String, Vec<u8>)],
223        reader: &mut dyn Read,
224    ) -> Result<(), WriteError> {
225        let (parent, leaf) = descend(&mut self.tree, name)?;
226        if parent.children.contains_key(leaf) {
227            return Err(name_conflict(name));
228        }
229        let inode_number = self.ctx.alloc_inode();
230        let pf = PendingFile {
231            path: PathBuf::from(name),
232            inode_number,
233            file_len: 0,
234            mtime_ns: meta.mtime_ns,
235            mode: limnifs_core::inode::S_IFREG | (meta.perms & 0o7777),
236            uid: meta.uid,
237            gid: meta.gid,
238        };
239        let wire_xattrs = to_wire_xattrs(xattrs)?;
240        self.ctx.pending_files.push(pf.clone());
241
242        let chunks = self.codecs.chunker.chunk_reader(reader)?;
243        let total_len: u64 = chunks.iter().map(|c| c.len() as u64).sum();
244        self.ctx.file_count += 1;
245
246        if total_len <= self.inline_threshold {
247            let mut data = Vec::with_capacity(total_len as usize);
248            for chunk in &chunks {
249                data.extend_from_slice(chunk);
250            }
251            self.ctx.inodes.push(PendingInode {
252                number: inode_number,
253                mode: limnifs_core::inode::S_IFREG | (meta.perms & 0o7777),
254                uid: meta.uid,
255                gid: meta.gid,
256                mtime_ns: meta.mtime_ns,
257                xattrs: wire_xattrs,
258                content: PendingContent::Inline(data),
259            });
260        } else {
261            let mut drops = Vec::with_capacity(chunks.len());
262            let mut slices = Vec::with_capacity(chunks.len());
263            let mut offset: u64 = 0;
264            for chunk in &chunks {
265                let drop_id = hash_section(chunk);
266                slices.push(crate::PendingSlice {
267                    drop_id,
268                    file_byte_start: offset,
269                    file_byte_end: offset + chunk.len() as u64,
270                });
271                offset += chunk.len() as u64;
272                let class = self.codecs.classifier.classify(chunk);
273                let (codec_id, compressed) = crate::compress_chunk_with_tournament(
274                    chunk,
275                    class,
276                    self.codecs.text_codec,
277                    self.codecs.binary_codec,
278                    &self.codecs.tunables,
279                    &self.codecs.tournament,
280                );
281                drops.push((drop_id, chunk.clone(), compressed, codec_id, 0));
282            }
283            self.ctx
284                .merge_chunked_file(&pf, crate::ChunkedFileResult { drops, slices });
285            // merge_chunked_file read file_len from the placeholder;
286            // correct the just-pushed inode now that it is known.
287            if let Some(inode) = self.ctx.inodes.last_mut() {
288                if let PendingContent::DropBacked { file_len, .. } = &mut inode.content {
289                    *file_len = total_len;
290                }
291            }
292            self.ctx
293                .pending_files
294                .last_mut()
295                .expect("pushed above")
296                .file_len = total_len;
297        }
298        parent
299            .children
300            .insert(leaf.to_owned(), StreamNode::File { inode_number });
301        Ok(())
302    }
303
304    /// Stage a random-access entry for parallel packing: like
305    /// [`add_file`], but the data is an in-memory slice (e.g. an
306    /// mmap'd archive entry) whose byte range is already known, so
307    /// chunk/hash/compress work is deferred to [`finish`], which
308    /// fans the staged set across rayon workers and merges the
309    /// results serially in stage order. Same entries, same order →
310    /// byte-identical image to the serial `add_file` path.
311    ///
312    /// The borrow of `data` must outlive the writer.
313    ///
314    /// # Errors
315    ///
316    /// [`WriteError::Io`] if the name is invalid or conflicts with
317    /// an existing entry.
318    pub fn stage_file(
319        &mut self,
320        name: &str,
321        meta: EntryMeta,
322        xattrs: &[(String, Vec<u8>)],
323        data: &'a [u8],
324    ) -> Result<(), WriteError> {
325        let (parent, leaf) = descend(&mut self.tree, name)?;
326        if parent.children.contains_key(leaf) {
327            return Err(name_conflict(name));
328        }
329        let inode_number = self.ctx.alloc_inode();
330        self.ctx.file_count += 1;
331        crate::progress::emit_file(std::path::Path::new(name), data.len() as u64);
332        parent
333            .children
334            .insert(leaf.to_owned(), StreamNode::File { inode_number });
335        self.staged.push(StagedEntry {
336            name: name.to_owned(),
337            meta,
338            xattrs: to_wire_xattrs(xattrs)?,
339            inode_number,
340            data,
341        });
342        Ok(())
343    }
344
345    /// Add (or declare) a directory at `name` with the given
346    /// identity. Implicit parents created by nested entries keep
347    /// mtime 0 / root; calling this on an existing implicit
348    /// directory stamps it.
349    ///
350    /// # Errors
351    ///
352    /// [`WriteError::Io`] if the name is invalid or conflicts with
353    /// a non-directory entry.
354    pub fn add_dir(&mut self, name: &str, meta: EntryMeta) -> Result<(), WriteError> {
355        if name == "/" {
356            return Ok(()); // the root is materialised at finish
357        }
358        let (parent, leaf) = descend(&mut self.tree, name)?;
359        match parent.children.get_mut(leaf) {
360            None => {
361                parent.children.insert(
362                    leaf.to_owned(),
363                    StreamNode::Dir(StreamDir {
364                        meta,
365                        children: BTreeMap::new(),
366                    }),
367                );
368                Ok(())
369            }
370            Some(StreamNode::Dir(dir)) => {
371                dir.meta.mtime_ns = meta.mtime_ns;
372                Ok(())
373            }
374            Some(_) => Err(name_conflict(name)),
375        }
376    }
377
378    /// Add a hardlink at `name` referencing the file already added
379    /// (by any path method) at `target` — both names share one
380    /// inode, and the emitted inode carries the real nlink. The
381    /// target must exist in the tree and be a regular file.
382    ///
383    /// # Errors
384    ///
385    /// [`WriteError::Io`] if `name` is invalid or conflicting, the
386    /// target is missing, or the target is not a regular file.
387    pub fn add_hardlink(&mut self, name: &str, target: &str) -> Result<(), WriteError> {
388        // Resolve the target before mutably borrowing the tree for
389        // the new name.
390        let inode_number = resolve_file_inode(&self.tree, target)?;
391        let (parent, leaf) = descend(&mut self.tree, name)?;
392        if parent.children.contains_key(leaf) {
393            return Err(name_conflict(name));
394        }
395        *self.ctx.nlink_counts.entry(inode_number).or_insert(1) += 1;
396        parent
397            .children
398            .insert(leaf.to_owned(), StreamNode::File { inode_number });
399        Ok(())
400    }
401
402    /// Add a symbolic link at `name` pointing at `target` (stored
403    /// raw, exactly as given).
404    ///
405    /// # Errors
406    ///
407    /// [`WriteError::Io`] if the name is invalid or conflicts with
408    /// an existing entry.
409    pub fn add_symlink(
410        &mut self,
411        name: &str,
412        target: &str,
413        meta: EntryMeta,
414    ) -> Result<(), WriteError> {
415        let (parent, leaf) = descend(&mut self.tree, name)?;
416        if parent.children.contains_key(leaf) {
417            return Err(name_conflict(name));
418        }
419        let inode_number = self.ctx.alloc_inode();
420        self.ctx.inodes.push(PendingInode {
421            number: inode_number,
422            mode: limnifs_core::inode::S_IFLNK | (meta.perms & 0o7777),
423            uid: meta.uid,
424            gid: meta.gid,
425            mtime_ns: meta.mtime_ns,
426            xattrs: Vec::new(),
427            content: PendingContent::Symlink(target.to_owned()),
428        });
429        parent
430            .children
431            .insert(leaf.to_owned(), StreamNode::Symlink { inode_number });
432        Ok(())
433    }
434
435    /// Materialise the tree and assemble the image.
436    ///
437    /// # Errors
438    ///
439    /// [`WriteError::Io`] on any writer-pipeline error.
440    pub fn finish(mut self) -> Result<WriteArtifact, WriteError> {
441        self.flush_staged()?;
442        let tree = std::mem::take(&mut self.tree);
443        self.ctx.root_inode_number = self.materialize_dir(tree);
444        self.ctx
445            .train_and_apply_dictionary(&self.config.dictionaries);
446        Ok(self.ctx.assemble())
447    }
448
449    /// Chunk/hash/compress every staged entry across rayon workers,
450    /// then merge serially in stage order. The parallel map is
451    /// order-preserving and the merge replays the exact same
452    /// per-entry steps as [`add_file`], so output is identical to
453    /// the serial path. Large entries additionally hit the
454    /// boundary-identical parallel FastCDC inside their slice —
455    /// nested rayon, the same work-stealing shape the write
456    /// pipeline already uses.
457    fn flush_staged(&mut self) -> Result<(), WriteError> {
458        if self.staged.is_empty() {
459            return Ok(());
460        }
461        let staged = std::mem::take(&mut self.staged);
462        let codecs = &self.codecs;
463        use rayon::prelude::*;
464        let results: Vec<crate::ChunkedFileResult> = staged
465            .par_iter()
466            .map(|entry| {
467                let chunks: Vec<&[u8]> = codecs.chunker.chunk_slice(entry.data);
468                let mut drops = Vec::with_capacity(chunks.len());
469                let mut slices = Vec::with_capacity(chunks.len());
470                let mut offset: u64 = 0;
471                for chunk in &chunks {
472                    let drop_id = crate::hash_section(chunk);
473                    slices.push(crate::PendingSlice {
474                        drop_id,
475                        file_byte_start: offset,
476                        file_byte_end: offset + chunk.len() as u64,
477                    });
478                    offset += chunk.len() as u64;
479                    let class = codecs.classifier.classify(chunk);
480                    let (codec_id, compressed) = crate::compress_chunk_with_tournament(
481                        chunk,
482                        class,
483                        codecs.text_codec,
484                        codecs.binary_codec,
485                        &codecs.tunables,
486                        &codecs.tournament,
487                    );
488                    drops.push((drop_id, (*chunk).to_vec(), compressed, codec_id, 0));
489                }
490                crate::ChunkedFileResult { drops, slices }
491            })
492            .collect();
493        for (entry, result) in staged.iter().zip(results) {
494            // Unlike the streaming path, the length is known upfront,
495            // so no post-merge inode patching is needed.
496            let total_len = entry.data.len() as u64;
497            let pf = PendingFile {
498                path: std::path::PathBuf::from(&entry.name),
499                inode_number: entry.inode_number,
500                file_len: total_len,
501                mtime_ns: entry.meta.mtime_ns,
502                mode: limnifs_core::inode::S_IFREG | (entry.meta.perms & 0o7777),
503                uid: entry.meta.uid,
504                gid: entry.meta.gid,
505            };
506            self.ctx.pending_files.push(pf.clone());
507            if total_len <= self.inline_threshold {
508                // Below the inline threshold chunk_slice yields the
509                // whole entry as one chunk, so this equals the
510                // serial path's chunk concatenation.
511                let mut data = Vec::with_capacity(entry.data.len());
512                data.extend_from_slice(entry.data);
513                self.ctx.inodes.push(PendingInode {
514                    number: entry.inode_number,
515                    mode: limnifs_core::inode::S_IFREG | (entry.meta.perms & 0o7777),
516                    uid: entry.meta.uid,
517                    gid: entry.meta.gid,
518                    mtime_ns: entry.meta.mtime_ns,
519                    xattrs: entry.xattrs.clone(),
520                    content: PendingContent::Inline(data),
521                });
522            } else {
523                if !entry.xattrs.is_empty() {
524                    self.ctx
525                        .inode_xattrs
526                        .insert(entry.inode_number, entry.xattrs.clone());
527                }
528                self.ctx.merge_chunked_file(&pf, result);
529            }
530        }
531        Ok(())
532    }
533
534    /// Allocate this directory's inode, then recurse into children
535    /// in name order — parent-first, mirroring the directory walk.
536    fn materialize_dir(&mut self, dir: StreamDir) -> u64 {
537        let inode_number = self.ctx.alloc_inode();
538        self.ctx.dir_count += 1;
539        let mut entries = Vec::with_capacity(dir.children.len());
540        for (name, node) in dir.children {
541            let (child_inode, entry_type) = match node {
542                StreamNode::Dir(child) => (self.materialize_dir(child), 0x02),
543                StreamNode::File { inode_number } => (inode_number, 0x01),
544                StreamNode::Symlink { inode_number } => (inode_number, 0x03),
545            };
546            entries.push((name, child_inode, entry_type));
547        }
548        // BTreeMap iterates name-sorted; the explicit sort keeps the
549        // invariant local, exactly like fold_survey.
550        entries.sort_by(|a, b| a.0.cmp(&b.0));
551        self.ctx.dir_nodes.push(encode_dir_node(&entries));
552        self.ctx.inodes.push(PendingInode {
553            number: inode_number,
554            mode: limnifs_core::inode::S_IFDIR | (dir.meta.perms & 0o7777),
555            uid: dir.meta.uid,
556            gid: dir.meta.gid,
557            mtime_ns: dir.meta.mtime_ns,
558            xattrs: Vec::new(),
559            content: PendingContent::Directory(entries),
560        });
561        inode_number
562    }
563}
564
565/// Walk (creating implicit directories) to `name`'s parent and
566/// return it plus the leaf component.
567fn descend<'a, 'b>(
568    root: &'a mut StreamDir,
569    name: &'b str,
570) -> Result<(&'a mut StreamDir, &'b str), WriteError> {
571    if name.is_empty() || name.starts_with('/') || name.ends_with('/') {
572        return Err(bad_name(name));
573    }
574    // `\` is a path separator on Windows and NUL is an io error:
575    // both must fail loudly at pack time, or the image they
576    // produce escapes the extract root on Windows binaries.
577    if name.contains(['\\', '\0']) {
578        return Err(bad_name(name));
579    }
580    let mut dir = root;
581    let mut components = name.split('/').peekable();
582    let leaf = components.next_back().expect("non-empty name has a leaf");
583    for component in components {
584        if component.is_empty() || component == "." || component == ".." {
585            return Err(bad_name(name));
586        }
587        dir = match dir
588            .children
589            .entry(component.to_owned())
590            .or_insert_with(|| StreamNode::Dir(StreamDir::default()))
591        {
592            StreamNode::Dir(child) => child,
593            StreamNode::File { .. } | StreamNode::Symlink { .. } => {
594                return Err(name_conflict(name))
595            }
596        };
597    }
598    if leaf.is_empty() || leaf == "." || leaf == ".." {
599        return Err(bad_name(name));
600    }
601    Ok((dir, leaf))
602}
603
604/// Resolve `target` to a file entry's inode number inside the
605/// stream tree. Errors when any component is missing, is a
606/// symlink, or the final component is a directory.
607fn resolve_file_inode(root: &StreamDir, target: &str) -> Result<u64, WriteError> {
608    let bad = || {
609        WriteError::Io(std::io::Error::other(format!(
610            "hardlink target {target:?} is not a file in the tree"
611        )))
612    };
613    let mut dir = root;
614    let mut components = target.split('/').filter(|c| !c.is_empty());
615    loop {
616        let Some(component) = components.next() else {
617            return Err(bad());
618        };
619        match dir.children.get(component) {
620            Some(StreamNode::File { inode_number }) if components.next().is_none() => {
621                return Ok(*inode_number);
622            }
623            Some(StreamNode::Dir(child)) => dir = child,
624            _ => return Err(bad()),
625        }
626    }
627}
628
629/// Validate and convert caller-supplied xattrs to wire form:
630/// namespace 0, 64 KiB total cap (metadata DoS guard), and the pax
631/// transport's hard limits — keys and values must be NUL-free and
632/// newline-free (a pax record is a length-prefixed text line).
633/// Returns an error naming the offending attribute.
634fn to_wire_xattrs(
635    raw: &[(String, Vec<u8>)],
636) -> Result<Vec<limnifs_core::inode::XAttr>, WriteError> {
637    const TOTAL_CAP: usize = 64 * 1024;
638    let mut out = Vec::with_capacity(raw.len());
639    let mut total = 0usize;
640    for (key, value) in raw {
641        if key.contains('\0') || key.contains('\n') || value.contains(&0) || value.contains(&b'\n')
642        {
643            return Err(WriteError::Io(std::io::Error::other(format!(
644                "xattr {key:?} carries NUL or newline bytes the pax record format cannot represent"
645            ))));
646        }
647        total += key.len() + value.len();
648        if total > TOTAL_CAP {
649            break;
650        }
651        out.push(limnifs_core::inode::XAttr {
652            namespace: 0,
653            key: key.clone(),
654            value: value.clone(),
655        });
656    }
657    Ok(out)
658}
659
660fn bad_name(name: &str) -> WriteError {
661    WriteError::Io(std::io::Error::other(format!(
662        "invalid stream entry name {name:?}: must be a non-empty relative path without '.', '..', '\\', or NUL components"
663    )))
664}
665
666fn name_conflict(name: &str) -> WriteError {
667    WriteError::Io(std::io::Error::other(format!(
668        "stream entry conflict: {name:?} already exists with a different type"
669    )))
670}
671
672#[cfg(test)]
673mod tests {
674    use super::*;
675
676    fn writer() -> StreamWriter<'static> {
677        // Leak is test-only and the config has no drop significance.
678        let config: &'static WriteConfig = Box::leak(Box::new(WriteConfig::default_v0_1()));
679        StreamWriter::new(config).expect("default config is valid")
680    }
681
682    fn pseudo_random_bytes(seed: u64, count: usize) -> Vec<u8> {
683        let mut state = seed;
684        let mut out = Vec::with_capacity(count);
685        for _ in 0..count {
686            state = state
687                .wrapping_mul(6_364_136_223_846_793_005)
688                .wrapping_add(1_442_695_040_888_963_407);
689            out.push(u8::try_from(state >> 56).expect("fits u8"));
690        }
691        out
692    }
693
694    fn add_all(w: &mut StreamWriter<'_>) {
695        w.add_dir("docs", EntryMeta::new(7_000_000_000_000, 0o755))
696            .expect("dir");
697        w.add_file(
698            "docs/readme.txt",
699            EntryMeta::new(1_000_000_000, 0o644),
700            &[],
701            &mut b"hello stream writer\n".as_slice(),
702        )
703        .expect("file 1");
704        let big = pseudo_random_bytes(9, 600 * 1024);
705        w.add_file(
706            "data/big.bin",
707            EntryMeta::new(2_000_000_000, 0o755),
708            &[],
709            &mut big.as_slice(),
710        )
711        .expect("file 2");
712        w.add_symlink(
713            "latest",
714            "docs/readme.txt",
715            EntryMeta::new(3_000_000_000, 0o777),
716        )
717        .expect("symlink");
718    }
719
720    #[test]
721    fn same_entry_sequence_packs_identically() {
722        let a = {
723            let mut w = writer();
724            add_all(&mut w);
725            w.finish().expect("finish a").bytes
726        };
727        let b = {
728            let mut w = writer();
729            add_all(&mut w);
730            w.finish().expect("finish b").bytes
731        };
732        assert_eq!(a, b);
733    }
734
735    #[test]
736    fn empty_stream_writes_root_only() {
737        let artifact = writer().finish().expect("finish");
738        assert_eq!(artifact.dir_count, 1);
739        assert_eq!(artifact.file_count, 0);
740        assert!(artifact.slabs.is_empty());
741    }
742
743    /// v0.3.44: the stream seam carries ownership, not just
744    /// mtime/perms — EntryMeta lands verbatim on the emitted inode
745    /// for every entry kind (file, staged file, dir, symlink).
746    #[test]
747    fn entry_meta_lands_on_inodes() {
748        let artifact = {
749            let mut w = writer();
750            w.add_dir("d", EntryMeta::new(1_111_111_111_111, 0o755).owner(12, 34))
751                .expect("dir");
752            w.add_file(
753                "d/f.txt",
754                EntryMeta::new(1_234_567_891_234, 0o640).owner(1000, 20),
755                &[],
756                &mut b"body\n".as_slice(),
757            )
758            .expect("file");
759            w.stage_file(
760                "d/s.bin",
761                EntryMeta::new(2_222_222_222_222, 0o600).owner(1001, 21),
762                &[],
763                b"staged",
764            )
765            .expect("staged");
766            w.add_symlink(
767                "d/l",
768                "d/f.txt",
769                EntryMeta::new(3_333_333_333_333, 0o777).owner(1002, 22),
770            )
771            .expect("symlink");
772            w.finish().expect("finish")
773        };
774        use limnifs_core::{parse_metadata_blob, parse_metadata_reference, ManifestCursor};
775        let mut cursor = ManifestCursor::new(&artifact.bytes);
776        limnifs_core::parse_manifest_header(&mut cursor).expect("header");
777        limnifs_core::parse_feature_flags_section(&mut cursor).expect("flags");
778        let meta_ref = parse_metadata_reference(&mut cursor).expect("meta");
779        let inline = meta_ref.inline_metadata.as_ref().expect("inline");
780        let mut blob_cursor = ManifestCursor::new(inline);
781        let blob = parse_metadata_blob(&mut blob_cursor).expect("blob");
782
783        let file = blob
784            .inodes
785            .iter()
786            .find(|i| i.uid == 1000 && i.gid == 20)
787            .expect("owned file inode");
788        assert_eq!(file.mtime_ns, 1_234_567_891_234);
789        assert_eq!(file.mode & 0o7777, 0o640);
790        assert!(blob
791            .inodes
792            .iter()
793            .any(|i| i.uid == 1001 && i.gid == 21 && i.mtime_ns == 2_222_222_222_222));
794        assert!(blob
795            .inodes
796            .iter()
797            .any(|i| i.uid == 1002 && i.gid == 22 && i.mtime_ns == 3_333_333_333_333));
798        assert!(blob.inodes.iter().any(|i| i.is_directory()
799            && i.uid == 12
800            && i.gid == 34
801            && i.mtime_ns == 1_111_111_111_111));
802    }
803
804    #[test]
805    fn small_files_inline_and_big_files_slab() {
806        let artifact = {
807            let mut w = writer();
808            add_all(&mut w);
809            w.finish().expect("finish")
810        };
811        assert_eq!(artifact.file_count, 2);
812        assert_eq!(artifact.dir_count, 3); // root + docs + data (implicit)
813        assert_eq!(artifact.slabs.len(), 1);
814    }
815
816    #[test]
817    fn staged_path_is_byte_identical_to_serial() {
818        let big_a = pseudo_random_bytes(31, 600 * 1024);
819        let big_b = pseudo_random_bytes(32, 900 * 1024);
820        let staged = {
821            let mut w = writer();
822            w.add_dir("docs", EntryMeta::new(7_000_000_000_000, 0o755))
823                .expect("dir");
824            w.add_file(
825                "tiny.txt",
826                EntryMeta::new(1, 0o644),
827                &[],
828                &mut b"small inline entry\n".as_slice(),
829            )
830            .expect("immediate file");
831            w.stage_file("docs/a.bin", EntryMeta::new(2, 0o644), &[], &big_a)
832                .expect("staged a");
833            w.stage_file("docs/b.bin", EntryMeta::new(3, 0o755), &[], &big_b)
834                .expect("staged b");
835            w.stage_file(
836                "docs/tiny2.txt",
837                EntryMeta::new(4, 0o600),
838                &[],
839                b"also inline\n",
840            )
841            .expect("staged tiny");
842            w.finish().expect("finish staged").bytes
843        };
844        let serial = {
845            let mut w = writer();
846            w.add_dir("docs", EntryMeta::new(7_000_000_000_000, 0o755))
847                .expect("dir");
848            w.add_file(
849                "tiny.txt",
850                EntryMeta::new(1, 0o644),
851                &[],
852                &mut b"small inline entry\n".as_slice(),
853            )
854            .expect("immediate file");
855            w.add_file(
856                "docs/a.bin",
857                EntryMeta::new(2, 0o644),
858                &[],
859                &mut big_a.as_slice(),
860            )
861            .expect("serial a");
862            w.add_file(
863                "docs/b.bin",
864                EntryMeta::new(3, 0o755),
865                &[],
866                &mut big_b.as_slice(),
867            )
868            .expect("serial b");
869            w.add_file(
870                "docs/tiny2.txt",
871                EntryMeta::new(4, 0o600),
872                &[],
873                &mut b"also inline\n".as_slice(),
874            )
875            .expect("serial tiny");
876            w.finish().expect("finish serial").bytes
877        };
878        assert_eq!(staged, serial, "staged flush must equal the serial path");
879    }
880
881    #[test]
882    fn staged_detects_conflicts_and_bad_names() {
883        let mut w = writer();
884        w.stage_file("a.txt", EntryMeta::new(0, 0o644), &[], b"x")
885            .expect("stage");
886        assert!(w
887            .stage_file("a.txt", EntryMeta::new(0, 0o644), &[], b"y")
888            .is_err());
889        assert!(w
890            .stage_file("", EntryMeta::new(0, 0o644), &[], b"y")
891            .is_err());
892        assert!(w
893            .stage_file("/abs", EntryMeta::new(0, 0o644), &[], b"y")
894            .is_err());
895        assert!(w
896            .stage_file("a.txt/child", EntryMeta::new(0, 0o644), &[], b"y")
897            .is_err());
898    }
899
900    #[test]
901    fn rejects_bad_and_conflicting_names() {
902        let mut w = writer();
903        assert!(w
904            .add_file("", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
905            .is_err());
906        assert!(w
907            .add_file("/abs", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
908            .is_err());
909        assert!(w
910            .add_file("a/../b", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
911            .is_err());
912        assert!(w
913            .add_file("ok.txt", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
914            .is_ok());
915        // Same leaf again, even with identical type: conflict.
916        assert!(w
917            .add_file("ok.txt", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
918            .is_err());
919        // File where a directory must pass through.
920        assert!(w
921            .add_file(
922                "ok.txt/child",
923                EntryMeta::new(0, 0o644),
924                &[],
925                &mut [].as_slice()
926            )
927            .is_err());
928        // Symlink over a file.
929        assert!(w
930            .add_symlink("ok.txt", "x", EntryMeta::new(0, 0o777))
931            .is_err());
932        // Windows-separator and NUL names escape extraction on
933        // Windows binaries (or crash the io layer); reject at pack.
934        assert!(w
935            .add_file(r"a\b", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
936            .is_err());
937        assert!(w
938            .add_dir(r"\\server\share", EntryMeta::new(0, 0o755))
939            .is_err());
940        assert!(w
941            .add_file("nul\0x", EntryMeta::new(0, 0o644), &[], &mut [].as_slice())
942            .is_err());
943    }
944}