stuffr_core/

archive.rs

//! The two orthogonal traits, and the entry model they share.

use std::io::{Read, Write};
use std::time::SystemTime;

use crate::error::Result;
use crate::fidelity::FidelityReport;
use crate::format::{CodecCaps, ContainerCaps, FormatId};
use crate::source::Source;

/// What a codec's decoder is opened with.
///
/// **Clones share one [`FidelitySink`](crate::FidelitySink).** `Clone`
/// copies the [`fidelity`](Self::fidelity) handle, not the warnings behind
/// it, so every clone of one value — and every decoder opened with any of
/// them — records into the same sink. Reusing one `DecodeOpts` (or clones of
/// it) across independent decodes therefore accumulates their warnings
/// together until the sink is drained; give each decode whose fidelity is
/// reported on its own a fresh value (`DecodeOpts::default()`, or a new
/// sink in `fidelity`).
#[derive(Clone, Default)]
pub struct DecodeOpts {
    /// Worker hint. The governor has final say; this is only a request.
    pub threads: Option<usize>,
    /// The shared worker/memory budget. `None` means single-threaded — a codec
    /// parallelises only if handed a governor, never from an ambient global.
    pub governor: Option<std::sync::Arc<crate::Governor>>,
    /// Cap on memory a decoder may ask the allocator for. `None` means
    /// unbounded, which is the library default; the CLI always sets a value.
    ///
    /// This exists because three pure codecs size a dictionary buffer from a
    /// value the *file declares in its own header*, before producing any
    /// output — measured at 69.35 MB peak RSS for a 60-byte `.xz`, and 538 MB
    /// for a crafted 114-byte `.lz` that decoded correctly, so the cost came
    /// purely from the declaration. `--max-ratio` cannot see it: that guard
    /// counts decoded output bytes and the allocation precedes any output.
    ///
    /// Exceeding it is [`crate::Error::ResourceLimit`] (exit 6), never
    /// `Corrupt` (exit 5) — the file is not damaged, this build simply will
    /// not allocate that much.
    pub memory_limit: Option<u64>,
    /// Where the decoder records what it approximated — today, bytes after a
    /// gzip or bzip2 file's last verified member that are neither another
    /// member nor padding ([`crate::Fidelity::TrailingDataIgnored`]). The
    /// default is a fresh, empty sink; a driver that reports fidelity keeps a
    /// clone and drains it into its report once the stream has been read.
    /// Shared by every clone of this value (see the type's doc). See
    /// [`crate::FidelitySink`].
    pub fidelity: crate::fidelity::FidelitySink,
}

impl std::fmt::Debug for DecodeOpts {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DecodeOpts")
            .field("threads", &self.threads)
            .field("governor", &self.governor.as_ref().map(|_| "Governor"))
            .field("memory_limit", &self.memory_limit)
            .field("fidelity", &self.fidelity)
            .finish()
    }
}

#[derive(Clone, Default)]
pub struct EncodeOpts {
    /// Format-relative compression level. `None` means the format's default.
    pub level: Option<i32>,
    /// Worker hint. The governor has final say; this is only a request.
    pub threads: Option<usize>,
    /// The shared worker/memory budget. `None` means single-threaded — a codec
    /// parallelises only if handed a governor, never from an ambient global.
    pub governor: Option<std::sync::Arc<crate::Governor>>,
}

impl std::fmt::Debug for EncodeOpts {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EncodeOpts")
            .field("level", &self.level)
            .field("threads", &self.threads)
            .field("governor", &self.governor.as_ref().map(|_| "Governor"))
            .finish()
    }
}

#[derive(Clone, Default)]
pub struct OpenOpts {
    pub password: Option<String>,
    /// The shared worker/memory budget. `None` means single-threaded — a
    /// container parallelises only if handed a governor, never from an
    /// ambient global. Needed so a container with `per_entry_codec` can build
    /// a `DecodeOpts` per entry, carrying the same governor forward.
    pub governor: Option<std::sync::Arc<crate::Governor>>,
}

impl std::fmt::Debug for OpenOpts {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OpenOpts")
            .field("password", &self.password)
            .field("governor", &self.governor.as_ref().map(|_| "Governor"))
            .finish()
    }
}

#[derive(Clone, Default)]
pub struct CreateOpts {
    pub level: Option<i32>,
    /// Codec for entry payloads, for containers with `per_entry_codec`.
    pub entry_codec: Option<FormatId>,
    /// The shared worker/memory budget. `None` means single-threaded — a
    /// container parallelises only if handed a governor, never from an
    /// ambient global. Needed so a container with `per_entry_codec` can build
    /// an `EncodeOpts` per entry, carrying the same governor forward.
    pub governor: Option<std::sync::Arc<crate::Governor>>,
}

impl std::fmt::Debug for CreateOpts {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CreateOpts")
            .field("level", &self.level)
            .field("entry_codec", &self.entry_codec)
            .field("governor", &self.governor.as_ref().map(|_| "Governor"))
            .finish()
    }
}

/// What an entry is. `Hardlink` is the same content as an EARLIER entry; a
/// later phase adds `CharDevice`, `BlockDevice`, `Fifo` and `Socket` for tar
/// and cpio, and `#[non_exhaustive]` keeps that from breaking downstream
/// matches. Struct types deliberately do NOT carry this attribute — see the
/// `#[non_exhaustive]` section of CONTRIBUTING.md.
#[non_exhaustive]
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub enum EntryKind {
    #[default]
    File,
    Dir,
    Symlink {
        target: String,
    },
    /// The same content as the EARLIER entry named `target`, in the
    /// archive's own name space (exactly the `EntryMeta::name` that entry
    /// carried, before any destination join).
    ///
    /// A `Hardlink` entry's own reader yields no bytes, and `meta.size` is
    /// the shared content's size when the reader knows it, else `None`.
    /// Consumers never read a link's payload; they take it from the target.
    Hardlink {
        target: String,
    },
    Other,
}

#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct EntryMeta {
    pub name: String,
    pub size: Option<u64>,
    pub compressed_size: Option<u64>,
    pub mtime: Option<SystemTime>,
    pub mode: Option<u32>,
    pub uid: Option<u32>,
    pub gid: Option<u32>,
    pub kind: EntryKind,
    /// Codec of this entry's payload, for `per_entry_codec` containers.
    pub codec: Option<FormatId>,
}

impl EntryMeta {
    pub fn file(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            ..Default::default()
        }
    }
}

/// One entry, borrowed from the archive it came from.
///
/// The reader is deliberately **not** `Send`. It was until Phase 2's first
/// real container, and the bound was unsound to require: `tar::Entry` holds
/// `&Archive<dyn Read>` over a `RefCell`, so it is `!Send` for any reader,
/// and every container built on the `tar` crate would have had to assert
/// `unsafe impl Send` for a property that is false — a lie that becomes UB
/// the day anything actually sends one.
///
/// Nothing needs it. An entry borrows the archive it came from
/// (`ArchiveRead::next_entry(&mut self) -> Entry<'_>`), so sending an entry
/// to another thread would mean sending the archive with it, and no caller in
/// this tree does either: entries are read on the thread that asked for
/// them. Dropping the bound is strictly more permissive, so a container whose
/// reader IS `Send` is unaffected.
///
/// Since 0.23.0 a codec decodes straight from it: `Source` is no longer
/// `Send` and [`Codec::decoder`] takes a borrowed `Box<dyn Source + 'a>`, so
/// `decode_reader(&codec, entry.reader(), &opts)` decodes a `.Z` or `.gz`
/// member in place, without copying it out first.
pub struct Entry<'a> {
    meta: EntryMeta,
    reader: Box<dyn Read + 'a>,
    announces_links: Option<u32>,
}

impl<'a> Entry<'a> {
    pub fn new(meta: EntryMeta, reader: Box<dyn Read + 'a>) -> Self {
        Self {
            meta,
            reader,
            announces_links: None,
        }
    }

    /// Marks this entry as one the reader KNOWS later entries in the same
    /// archive are hard links to, and how many: `links_to_come` names besides
    /// this one share it (cpio reads `nlink`; tar cannot know, so it never
    /// sets this). A hint for consumers that keep payloads for later links,
    /// so they can drop a payload once its last link has passed; never a
    /// promise that a link follows, nor that no more will (an archive can
    /// hold fewer names than `nlink` says, or a crafted larger count).
    #[must_use]
    pub fn announce_links(mut self, links_to_come: u32) -> Self {
        self.announces_links = Some(links_to_come);
        self
    }

    /// How many later hard links the reader announced to this entry; see
    /// [`Self::announce_links`]. `None` unless a reader set it.
    pub fn announces_links(&self) -> Option<u32> {
        self.announces_links
    }

    pub fn meta(&self) -> &EntryMeta {
        &self.meta
    }

    pub fn reader(&mut self) -> &mut (dyn Read + 'a) {
        &mut *self.reader
    }

    /// Takes the payload reader out, leaving the metadata behind — the one
    /// way to rebuild an `Entry` with different metadata over the SAME,
    /// still-streaming reader.
    ///
    /// `#[cfg(test)]` and `pub(crate)` on purpose: no container needs it,
    /// and nothing outside this crate should be rewriting an entry's
    /// metadata between the container that read it and the caller that acts
    /// on it. It exists for `container_conformance`'s property 11 doubles,
    /// which have to report a mode the framed mock's wire format does not
    /// carry while still satisfying property 8 — buffering the payload to
    /// fake one would trip the incrementality property first and the double
    /// would then prove the wrong thing. Not `pub`: widening the visibility
    /// to silence the dead-code lint would ship a metadata-rewriting hook to
    /// every caller for a test's convenience.
    #[cfg(test)]
    pub(crate) fn into_reader(self) -> Box<dyn Read + 'a> {
        self.reader
    }
}

/// Manual impl: `reader` is `Box<dyn Read>`, which cannot derive `Debug`.
/// Needed so `Result<Entry<'_>, Error>::unwrap_err()` type-checks in tests
/// (`unwrap_err` requires the `Ok` side to be `Debug`).
impl std::fmt::Debug for Entry<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Entry")
            .field("meta", &self.meta)
            .finish_non_exhaustive()
    }
}

/// A writer that needs an explicit, fallible completion step (trailers, CRCs).
///
/// `finish` must flush the underlying writer before returning, once every
/// trailer byte has been written to it. `Codec::encoder` takes the
/// destination `Box<dyn Write + Send>` by value, so once a caller has handed
/// it over, `finish` is the only code left holding it — the caller has no
/// remaining handle to flush afterwards. Skipping this is silent on an
/// unbuffered destination (a plain `File`) but loses the tail of the stream
/// on a buffered one: notably stdout's `LineWriter`, which only auto-flushes
/// on `\n` and may see none in a long run of binary output.
///
/// `#[must_use]` on the trait, so a `Box<dyn Sink>` that is produced and then
/// dropped without `finish` is a warning rather than a silent truncation.
/// `ArchiveWrite::finish` returns one precisely so the layer beneath can be
/// finished next; `archive.finish()?;` on its own compiles perfectly well and
/// loses every trailer below the container — the exact defect Phase 2c
/// exists to fix, reintroduced by a line that looks finished.
#[must_use = "a Sink must be completed with `finish`; dropping it skips the \
              trailer and the flush, truncating the output"]
pub trait Sink: Write + Send {
    fn finish(self: Box<Self>) -> Result<()>;
}

/// A [`Sink`] over a destination that needs no trailer — an uncompressed file.
///
/// Exists so `Container::create` can take a `Box<dyn Sink>` uniformly. A
/// composed write never constructs one: there the codec's own `Sink` IS the
/// container's destination. `finish` flushes, which is exactly what all four
/// containers used to do by hand at the end of their own `finish`.
pub struct PlainSink(Box<dyn Write + Send>);

impl PlainSink {
    /// Returns the boxed trait object rather than `Self` on purpose: every
    /// call site wants a `Box<dyn Sink>`, and handing back `Self` would make
    /// each of them re-box it.
    #[allow(clippy::new_ret_no_self)]
    pub fn new(w: Box<dyn Write + Send>) -> Box<dyn Sink> {
        Box::new(Self(w))
    }
}

impl Write for PlainSink {
    fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
        self.0.write(b)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.0.flush()
    }
}

impl Sink for PlainSink {
    fn finish(mut self: Box<Self>) -> Result<()> {
        self.0.flush()?;
        Ok(())
    }
}

/// A byte-stream transform. Knows nothing about files, names, or entries.
pub trait Codec: Send + Sync {
    fn id(&self) -> FormatId;
    fn caps(&self) -> CodecCaps;
    /// Validates encode options without touching the filesystem.
    ///
    /// `ops` calls this before opening any destination, so a rejected option
    /// costs nothing — no temp file created, no existing file disturbed. The
    /// default accepts everything, for codecs with no options to reject.
    ///
    /// With one codec this looks like ceremony. With nine, each carrying its
    /// own level range, "validate before touching the filesystem" has to be a
    /// contract rather than a habit: in Phase 1b an invalid level created and
    /// then deleted a file for nothing.
    fn check_encode_opts(&self, _o: &EncodeOpts) -> Result<()> {
        Ok(())
    }
    /// Decodes `src`. Returns a [`Source`] rather than a bare reader so a codec
    /// carrying a frame index can advertise seekability to the layer above;
    /// forward-only codecs wrap their reader in [`crate::StreamOnly`].
    ///
    /// `src` may borrow (`'a`) and need not be `Send`: an archive member's
    /// reader is both, and the decoder over it lives no longer than it. A
    /// caller holding a plain reader uses [`crate::decode_reader`].
    fn decoder<'a>(
        &self,
        src: Box<dyn Source + 'a>,
        o: &DecodeOpts,
    ) -> Result<Box<dyn Source + 'a>>;
    fn encoder(&self, dst: Box<dyn Write + Send>, o: &EncodeOpts) -> Result<Box<dyn Sink>>;
}

/// A structure of entries. May invoke codecs internally.
pub trait Container: Send + Sync {
    fn id(&self) -> FormatId;
    fn caps(&self) -> ContainerCaps;
    /// Opens an archive from a ladder-resolved source.
    ///
    /// Takes the whole [`crate::Resolved`] rather than just its source so the
    /// container knows which rung it was given — which is what makes
    /// `Rung::Degraded` implementable — and so it can seed its fidelity report
    /// from the ladder's rather than re-deriving the same facts.
    fn open(&self, resolved: crate::ladder::Resolved, o: &OpenOpts)
    -> Result<Box<dyn ArchiveRead>>;
    /// Creates an archive writing into `dst`.
    ///
    /// Takes a [`Sink`] rather than a bare `Write` so a container over a codec
    /// composes: the codec's own `Sink` is handed straight in, and
    /// [`ArchiveWrite::finish`] gives it back so the caller can finish it.
    /// Passing a bare file means wrapping it in [`PlainSink`].
    fn create(&self, dst: Box<dyn Sink>, o: &CreateOpts) -> Result<Box<dyn ArchiveWrite>>;
}

pub trait ArchiveRead {
    /// Forward iteration. Available for every container on every input — this
    /// is the streaming path, and the reason `Entry` borrows.
    fn next_entry(&mut self) -> Result<Option<Entry<'_>>>;

    /// Random access. `Err(NotSeekable)` when the ladder could not supply seek.
    fn by_index(&mut self, index: usize) -> Result<Entry<'_>>;

    fn fidelity(&self) -> &FidelityReport;
}

pub trait ArchiveWrite {
    fn add(&mut self, meta: &EntryMeta, data: &mut dyn Read) -> Result<()>;
    /// Writes the container's trailer and returns the destination.
    ///
    /// Returning it is what makes container-over-codec possible: the caller
    /// still holds a `Sink` afterwards and can finish the codec layer, which
    /// writes ITS trailer. Dropping the destination here — what this used to
    /// do — silently truncated every composed archive.
    ///
    /// Implementors must NOT call `finish` on the returned sink. Finishing is
    /// the caller's, once, at the outermost layer.
    fn finish(self: Box<Self>) -> Result<Box<dyn Sink>>;
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn an_entry_announces_links_only_when_its_reader_says_so() {
        let plain = Entry::new(EntryMeta::default(), Box::new(std::io::empty()));
        assert_eq!(plain.announces_links(), None);
        let marked = Entry::new(EntryMeta::default(), Box::new(std::io::empty())).announce_links(2);
        assert_eq!(marked.announces_links(), Some(2));
    }

    #[test]
    fn a_plain_sink_flushes_its_destination_on_finish() {
        use std::sync::{Arc, Mutex};

        #[derive(Default)]
        struct Rec {
            flushed: bool,
            bytes: Vec<u8>,
        }
        struct W(Arc<Mutex<Rec>>);
        impl Write for W {
            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
                self.0.lock().unwrap().bytes.extend_from_slice(b);
                Ok(b.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                self.0.lock().unwrap().flushed = true;
                Ok(())
            }
        }

        let rec = Arc::new(Mutex::new(Rec::default()));
        let mut sink = PlainSink::new(Box::new(W(Arc::clone(&rec))));
        sink.write_all(b"payload").unwrap();
        assert!(
            !rec.lock().unwrap().flushed,
            "finish has not been called yet"
        );
        sink.finish().unwrap();

        let r = rec.lock().unwrap();
        assert!(
            r.flushed,
            "PlainSink::finish must flush — a buffered destination loses its tail otherwise"
        );
        assert_eq!(r.bytes, b"payload");
    }
}