stuffr-core 0.12.0

Core traits, stream ladder and resource governor for the stuffr toolkit
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
//! 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()
    }
}

/// 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.
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`].
    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 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");
    }
}