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;

#[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>,
}

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)
            .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()
    }
}

/// Phase 2 adds `Hardlink`, `CharDevice`, `BlockDevice`, `Fifo` and `Socket`
/// for tar and cpio; `#[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,
    },
    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.
pub struct Entry<'a> {
    meta: EntryMeta,
    reader: Box<dyn Read + 'a>,
}

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

    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`].
    fn decoder(&self, src: Box<dyn Source>, o: &DecodeOpts) -> Result<Box<dyn Source>>;
    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 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");
    }
}