Skip to main content

archive_trait/
builder.rs

1//! Format-neutral archive construction.
2//!
3//! Archive formats implement [`ArchiveBuilder`] and wrap the resulting writer
4//! in a stateful [`Builder`] to use the format-neutral construction APIs.
5
6mod traversal;
7
8use std::{
9    collections::VecDeque,
10    io::{self, Read},
11    mem,
12    ops::Range,
13    path::{Path, PathBuf},
14    pin::Pin,
15};
16
17use thiserror::Error;
18use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt};
19
20pub use self::traversal::TraversalError;
21use self::traversal::{TraversalEntry, TraversalKind, TraversalStream, stream_directory_entries};
22use crate::{
23    NameValidator,
24    component_tree::{ComponentTree, ROOT_NODE},
25    name::NameValidation,
26};
27
28const BUFFERED_SOURCE_FILE_BYTES: usize = 1024 * 1024;
29const FILE_PAYLOAD_CHUNK_BYTES: usize = 2 * 1024 * 1024;
30// A preparation batch may exceed this target by one buffered file, so its
31// payload storage remains below twice this value.
32const SOURCE_FILE_PREPARATION_BATCH_BYTES: usize = BUFFERED_SOURCE_FILE_BYTES;
33
34/// Minimal regular-file metadata accepted by [`Builder::add_file`].
35#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
36pub struct EntryMetadata {
37    executable: bool,
38}
39
40impl EntryMetadata {
41    /// Configures whether the regular file carries executable intent.
42    pub fn executable(mut self, executable: bool) -> Self {
43        self.executable = executable;
44        self
45    }
46
47    /// Returns whether this entry carries executable intent.
48    pub fn is_executable(self) -> bool {
49        self.executable
50    }
51}
52
53/// Controls format-neutral archive construction behavior.
54#[derive(Clone, Copy, Debug, Default)]
55pub struct BuilderPolicy {
56    name_validation: NameValidation,
57    symlink_policy: SymlinkPolicy,
58}
59
60/// Controls how source symbolic links are handled during recursive builds.
61#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
62pub enum SymlinkPolicy {
63    /// Reject recursive sources containing symbolic links.
64    #[default]
65    Reject,
66    /// Preserve symbolic links as link members in the resulting archive.
67    Preserve,
68    // TODO: Consider adding some kind of "Dereference" policy in the future,
69    // where symlinks get followed and replaced with their normal file/directory
70    // contents.
71}
72
73impl BuilderPolicy {
74    /// Configures validation for member names and preserved symbolic-link targets.
75    ///
76    /// Passing [`None`] disables configurable name validation. UTF-8 and
77    /// archive-format requirements still apply.
78    pub fn name_validator(mut self, validator: Option<NameValidator>) -> Self {
79        self.name_validation = NameValidation::from_validator(validator);
80        self
81    }
82
83    /// Configures how recursive builds handle source symbolic links.
84    ///
85    /// Symbolic links are **rejected by default**. Use
86    /// [`SymlinkPolicy::Preserve`] to write link members instead.
87    pub fn symlink_policy(mut self, policy: SymlinkPolicy) -> Self {
88        self.symlink_policy = policy;
89        self
90    }
91}
92
93struct BuilderState {
94    policy: BuilderPolicy,
95    entries: BuildEntries,
96    source_buffer: Vec<u8>,
97    poisoned: bool,
98}
99
100impl BuilderState {
101    fn new(policy: BuilderPolicy) -> Self {
102        Self {
103            policy,
104            entries: BuildEntries::new(),
105            source_buffer: Vec::new(),
106            poisoned: false,
107        }
108    }
109
110    fn ensure_active<E>(&self) -> Result<(), BuildError<E>> {
111        if self.poisoned {
112            return Err(BuildError::Poisoned);
113        }
114        Ok(())
115    }
116
117    // A backend write is provisionally poisoning. Completion clears this flag
118    // before the returned failure is classified; cancellation leaves it set.
119    fn begin_write(&mut self) {
120        self.poisoned = true;
121    }
122
123    fn complete_write(&mut self) {
124        self.poisoned = false;
125    }
126
127    fn poison(&mut self) {
128        self.poisoned = true;
129    }
130}
131
132/// A format-neutral, uncompressed file payload supplied to an [`ArchiveBuilder`]
133/// implementation.
134pub struct FilePayload<'a> {
135    size: u64,
136    started: bool,
137    inner: FilePayloadInner<'a>,
138}
139
140enum FilePayloadInner<'a> {
141    /// A pre-buffered payload.
142    ///
143    /// Observe that this is stored as an `Option` so that we can use the `None`
144    /// state to communicate that the payload has been consumed.
145    Buffered(Option<&'a [u8]>),
146    Reader {
147        source: Pin<Box<dyn AsyncRead + 'a>>,
148        filesystem_path: Option<&'a Path>,
149        buffer: Vec<u8>,
150        remaining: u64,
151        filled: usize,
152    },
153}
154
155impl<'a> FilePayload<'a> {
156    /// Creates a payload with a declared logical size and [`AsyncRead`] source.
157    ///
158    /// The builder reads exactly `size` bytes from `source`. Additional bytes
159    /// remain unread, while a source that ends early causes the addition to
160    /// fail.
161    pub fn new<R>(size: u64, source: R) -> Self
162    where
163        R: AsyncRead + 'a,
164    {
165        Self::streaming(size, source, Vec::new(), None)
166    }
167
168    /// Returns the logical, uncompressed source size in bytes.
169    ///
170    /// This is the total number of bytes yielded by [`Self::next_chunk`], not
171    /// necessarily the size ultimately stored by the archive format.
172    pub fn size(&self) -> u64 {
173        self.size
174    }
175
176    /// Returns the next chunk of logical, uncompressed source bytes.
177    ///
178    /// Once this method has been called, the payload cannot be passed to
179    /// [`Builder::add_file`].
180    pub async fn next_chunk<E>(&mut self) -> Result<Option<&[u8]>, BuildError<E>> {
181        self.started = true;
182        match &mut self.inner {
183            FilePayloadInner::Buffered(bytes) => Ok(bytes.take().filter(|bytes| !bytes.is_empty())),
184            FilePayloadInner::Reader {
185                source,
186                filesystem_path,
187                buffer,
188                remaining,
189                filled,
190            } => read_streaming_chunk(source, buffer, remaining, filled, *filesystem_path).await,
191        }
192    }
193
194    fn streaming<R>(
195        size: u64,
196        source: R,
197        buffer: Vec<u8>,
198        filesystem_path: Option<&'a Path>,
199    ) -> Self
200    where
201        R: AsyncRead + 'a,
202    {
203        Self {
204            size,
205            started: false,
206            inner: FilePayloadInner::Reader {
207                source: Box::pin(source),
208                filesystem_path,
209                buffer,
210                remaining: size,
211                filled: 0,
212            },
213        }
214    }
215
216    fn swap_buffer(&mut self, buffer: &mut Vec<u8>) {
217        if let FilePayloadInner::Reader {
218            buffer: payload_buffer,
219            ..
220        } = &mut self.inner
221        {
222            mem::swap(payload_buffer, buffer);
223        }
224    }
225}
226
227impl FilePayload<'static> {
228    /// Opens `path` and creates a payload from the complete regular file.
229    pub async fn from_path<P: AsRef<Path>>(path: P) -> io::Result<Self> {
230        let file = tokio::fs::File::open(path).await?;
231        Self::from_file(file).await
232    }
233
234    /// Creates a payload from the unread contents of a [`tokio::fs::File`].
235    ///
236    /// The declared size is the file length minus its current stream position.
237    /// This constructor rejects filesystem objects that are not regular files.
238    pub async fn from_file(mut file: tokio::fs::File) -> io::Result<Self> {
239        let metadata = file.metadata().await?;
240        if !metadata.is_file() {
241            return Err(io::Error::other(
242                "file payload source is not a regular file",
243            ));
244        }
245        let position = file.stream_position().await?;
246        Ok(Self::new(metadata.len().saturating_sub(position), file))
247    }
248}
249
250impl<'a> From<&'a [u8]> for FilePayload<'a> {
251    fn from(bytes: &'a [u8]) -> Self {
252        let size = bytes.len() as u64;
253        Self {
254            size,
255            started: false,
256            inner: FilePayloadInner::Buffered(Some(bytes)),
257        }
258    }
259}
260
261async fn read_streaming_chunk<'a, E, R>(
262    source: &mut R,
263    buffer: &'a mut Vec<u8>,
264    remaining: &mut u64,
265    filled: &mut usize,
266    filesystem_path: Option<&Path>,
267) -> Result<Option<&'a [u8]>, BuildError<E>>
268where
269    R: AsyncRead + Unpin + ?Sized,
270{
271    if *remaining == 0 {
272        return Ok(None);
273    }
274
275    let chunk_size = (*remaining).min(FILE_PAYLOAD_CHUNK_BYTES as u64);
276    let chunk_len = usize::try_from(chunk_size)
277        .map_err(|_| arithmetic_overflow("file payload read buffer size"))?;
278    buffer.resize(chunk_len, 0);
279    // Progress lives in the payload rather than this future, so cancelling and
280    // retrying `FilePayload::next_chunk` cannot discard completed reads.
281    while *filled < chunk_len {
282        let read = source
283            .read(&mut buffer[*filled..])
284            .await
285            .map_err(|source| file_payload_read_error(filesystem_path, source))?;
286        if read == 0 {
287            return Err(file_payload_read_error(
288                filesystem_path,
289                io::Error::new(
290                    io::ErrorKind::UnexpectedEof,
291                    "file payload source ended before its declared size",
292                ),
293            ));
294        }
295        *filled += read;
296    }
297    *remaining -= chunk_size;
298    *filled = 0;
299    Ok(Some(buffer))
300}
301
302/// A failure returned by an [`ArchiveBuilder`] format hook.
303///
304/// This distinguishes errors known to precede output from errors that may have
305/// left a partial member in the output archive.
306#[derive(Debug)]
307pub struct BuildFailure<E> {
308    error: BuildError<E>,
309    // TODO: Maybe make all failures poisoning?
310    // I'm not sure we really need the distinction here.
311    poisons_builder: bool,
312}
313
314impl<E> BuildFailure<E> {
315    /// Reports a failure that occurred before the hook wrote any output.
316    pub fn recoverable(error: BuildError<E>) -> Self {
317        Self {
318            error,
319            poisons_builder: false,
320        }
321    }
322
323    /// Reports a failure that may have left partial output.
324    pub fn poisoned(error: BuildError<E>) -> Self {
325        Self {
326            error,
327            poisons_builder: true,
328        }
329    }
330
331    fn into_parts(self) -> (BuildError<E>, bool) {
332        (self.error, self.poisons_builder)
333    }
334}
335
336/// A format-specific archive writer that can create a stateful [`Builder`].
337///
338/// The asynchronous methods on this trait are implementation hooks for
339/// [`Builder`]. Archive construction callers must not invoke them directly;
340/// doing so bypasses builder policy, collision tracking, and cancellation
341/// poisoning. Use [`Self::builder`] or [`Self::builder_with_policy`] and then
342/// the [`Builder`] APIs instead.
343///
344/// Hook implementations must return [`BuildFailure::recoverable`] only when the
345/// failed invocation wrote no output. Any failure after output may have begun
346/// must use [`BuildFailure::poisoned`].
347#[expect(
348    async_fn_in_trait,
349    reason = "archive writers may be !Send and run on a local executor"
350)]
351pub trait ArchiveBuilder: Sized {
352    /// The archive-format error returned while encoding entries.
353    type Error;
354
355    /// Wraps this format writer in a builder using default policy.
356    ///
357    /// Implementors should not override this default implementation.
358    fn builder(self) -> Builder<Self> {
359        Builder {
360            backend: self,
361            state: BuilderState::new(BuilderPolicy::default()),
362        }
363    }
364
365    /// Wraps this format writer in a builder using `policy`.
366    ///
367    /// Implementors should not override this default implementation.
368    fn builder_with_policy(self, policy: BuilderPolicy) -> Builder<Self> {
369        Builder {
370            backend: self,
371            state: BuilderState::new(policy),
372        }
373    }
374
375    /// Writes any format-specific archive terminator or index.
376    async fn finish_archive(&mut self) -> Result<(), BuildFailure<Self::Error>>;
377
378    /// Writes one regular-file member and its complete payload.
379    ///
380    /// Implementations must call [`FilePayload::next_chunk`] through
381    /// completion and classify failures using [`BuildFailure`].
382    async fn write_file_member(
383        &mut self,
384        path: &str,
385        payload: &mut FilePayload<'_>,
386        metadata: EntryMetadata,
387    ) -> Result<(), BuildFailure<Self::Error>>;
388
389    /// Writes one directory member.
390    async fn write_directory_member(&mut self, path: &str)
391    -> Result<(), BuildFailure<Self::Error>>;
392
393    /// Writes one symbolic-link member.
394    async fn write_symbolic_link_member(
395        &mut self,
396        path: &str,
397        target: &str,
398    ) -> Result<(), BuildFailure<Self::Error>>;
399}
400
401/// A stateful format-neutral archive construction engine.
402///
403/// Create this wrapper with [`ArchiveBuilder::builder`] or
404/// [`ArchiveBuilder::builder_with_policy`].
405pub struct Builder<B> {
406    backend: B,
407    state: BuilderState,
408}
409
410impl<B: ArchiveBuilder> Builder<B> {
411    /// Adds one regular file from a [`FilePayload`].
412    ///
413    /// If the payload ends before its declared size or returns an error, the
414    /// addition fails and the builder is poisoned if the archive member's
415    /// output may already have begun.
416    ///
417    /// The payload must not have been read through [`FilePayload::next_chunk`]
418    /// before this method is called.
419    pub async fn add_file<'a, P>(
420        &mut self,
421        path: P,
422        payload: impl Into<FilePayload<'a>>,
423        metadata: EntryMetadata,
424    ) -> Result<(), BuildError<B::Error>>
425    where
426        P: AsRef<Path>,
427    {
428        self.state.ensure_active()?;
429        let archive_path = path.as_ref();
430        let Some(path) = archive_path.to_str() else {
431            return Err(BuildError::InvalidArchivePath {
432                path: archive_path.to_path_buf(),
433                reason: "path is not valid UTF-8",
434            });
435        };
436        if !self.state.policy.name_validation.accepts(path) {
437            return Err(BuildError::NameRejected {
438                context: "member path",
439                value: path.to_owned(),
440            });
441        }
442        let path = path.to_owned();
443        let reservation = self
444            .state
445            .entries
446            .preflight_entry(&path, ArchivedEntry::NonDirectory)?;
447        let mut payload = payload.into();
448        if payload.started {
449            return Err(BuildError::FilePayloadAlreadyRead);
450        }
451        payload.swap_buffer(&mut self.state.source_buffer);
452        self.state.begin_write();
453        let result = self
454            .backend
455            .write_file_member(&path, &mut payload, metadata)
456            .await;
457        self.state.complete_write();
458        payload.swap_buffer(&mut self.state.source_buffer);
459        self.resolve_hook(result)?;
460        self.state.entries.commit_entry(&path, reservation);
461        Ok(())
462    }
463
464    /// Adds one directory member without reading from the filesystem.
465    ///
466    /// This creates only the named directory member. It does not add any child
467    /// members; use [`Self::add_directory_all`] to recursively add a filesystem
468    /// directory and its contents.
469    pub async fn add_directory<P: AsRef<Path>>(
470        &mut self,
471        path: P,
472    ) -> Result<(), BuildError<B::Error>> {
473        self.state.ensure_active()?;
474        let archive_path = path.as_ref();
475        let Some(path) = archive_path.to_str() else {
476            return Err(BuildError::InvalidArchivePath {
477                path: archive_path.to_path_buf(),
478                reason: "path is not valid UTF-8",
479            });
480        };
481        if !self.state.policy.name_validation.accepts(path) {
482            return Err(BuildError::NameRejected {
483                context: "member path",
484                value: path.to_owned(),
485            });
486        }
487        let path = path.to_owned();
488        let reservation = self
489            .state
490            .entries
491            .preflight_entry(&path, ArchivedEntry::Directory { explicit: true })?;
492        self.state.begin_write();
493        let result = self.backend.write_directory_member(&path).await;
494        self.state.complete_write();
495        self.resolve_hook(result)?;
496        self.state.entries.commit_entry(&path, reservation);
497        Ok(())
498    }
499
500    /// Recursively adds a filesystem directory beneath its UTF-8 basename.
501    ///
502    /// Entries are visited in deterministic sorted order and files are streamed
503    /// with bounded memory. Source symbolic links are rejected by default;
504    /// [`BuilderPolicy::symlink_policy`] can instead preserve them. A late
505    /// traversal or validation failure may leave partial output and poison
506    /// this builder.
507    pub async fn add_directory_all<P: AsRef<Path>>(
508        &mut self,
509        source: P,
510    ) -> Result<(), BuildError<B::Error>> {
511        self.state.ensure_active()?;
512        let source = source.as_ref().to_path_buf();
513        let mut entries = stream_directory_entries(
514            source,
515            self.state.policy.name_validation,
516            self.state.policy.symlink_policy,
517        )
518        .map_err(BuildError::Traversal)?;
519        self.state.begin_write();
520        let mut traversal = DirectoryBuild {
521            entries: &mut self.state.entries,
522            source_buffer: mem::take(&mut self.state.source_buffer),
523            emitted: false,
524        };
525        let write_result =
526            write_directory_entries(&mut self.backend, &mut entries, &mut traversal).await;
527        let traversal_result = entries
528            .finish()
529            .await
530            .map_err(BuildError::Traversal)
531            .map_err(BuildFailure::recoverable);
532        let result = write_result.and(traversal_result);
533        let DirectoryBuild {
534            entries: _,
535            source_buffer,
536            emitted,
537        } = traversal;
538        self.state.complete_write();
539        self.state.source_buffer = source_buffer;
540        match result {
541            Ok(()) => Ok(()),
542            Err(error) => {
543                let (error, hook_poisoned) = error.into_parts();
544                if emitted || hook_poisoned {
545                    self.state.poison();
546                }
547                Err(error)
548            }
549        }
550    }
551
552    /// Finalizes and consumes this archive builder.
553    pub async fn finish(self) -> Result<(), BuildError<B::Error>> {
554        self.finish_into_inner().await?;
555        Ok(())
556    }
557
558    /// Finalizes this archive builder and returns its format writer.
559    ///
560    /// This allows callers to recover and finalize an owned output sink after
561    /// the archive terminator has been written.
562    pub async fn finish_into_inner(mut self) -> Result<B, BuildError<B::Error>> {
563        self.state.ensure_active()?;
564        let result = self.backend.finish_archive().await;
565        self.resolve_hook(result)?;
566        Ok(self.backend)
567    }
568
569    fn resolve_hook<T>(
570        &mut self,
571        result: Result<T, BuildFailure<B::Error>>,
572    ) -> Result<T, BuildError<B::Error>> {
573        match result {
574            Ok(value) => Ok(value),
575            Err(error) => {
576                let (error, poisons_builder) = error.into_parts();
577                if poisons_builder {
578                    self.state.poison();
579                }
580                Err(error)
581            }
582        }
583    }
584}
585
586async fn write_directory_entries<B: ArchiveBuilder>(
587    builder: &mut B,
588    entries: &mut TraversalStream,
589    traversal: &mut DirectoryBuild<'_>,
590) -> Result<(), BuildFailure<B::Error>> {
591    while let Some(entries) = entries.recv().await {
592        let mut entries = VecDeque::from(entries);
593        while !entries.is_empty() {
594            let buffer = mem::take(&mut traversal.source_buffer);
595            let (prepared, remaining) = prepare_directory_entries(entries, buffer)
596                .await
597                .map_err(SourceError::into_build_error)
598                .map_err(BuildFailure::recoverable)?;
599            entries = remaining;
600            let PreparedDirectoryBatch {
601                entries: prepared_entries,
602                mut buffer,
603            } = prepared;
604            let result =
605                write_prepared_directory_entries(builder, prepared_entries, &mut buffer, traversal)
606                    .await;
607            traversal.source_buffer = buffer;
608            result?;
609        }
610    }
611    Ok(())
612}
613
614async fn write_prepared_directory_entries<B: ArchiveBuilder>(
615    builder: &mut B,
616    entries: Vec<PreparedTraversalEntry>,
617    buffer: &mut Vec<u8>,
618    traversal: &mut DirectoryBuild<'_>,
619) -> Result<(), BuildFailure<B::Error>> {
620    for entry in entries {
621        let reservation = traversal
622            .entries
623            .preflight_entry(
624                &entry.archive_path,
625                if matches!(&entry.kind, PreparedTraversalKind::Directory) {
626                    ArchivedEntry::Directory { explicit: true }
627                } else {
628                    ArchivedEntry::NonDirectory
629                },
630            )
631            .map_err(BuildFailure::recoverable)?;
632        match entry.kind {
633            PreparedTraversalKind::Directory => {
634                builder.write_directory_member(&entry.archive_path).await?;
635            }
636            PreparedTraversalKind::BufferedFile { range, executable } => {
637                let data = buffer.get(range).ok_or_else(|| {
638                    BuildFailure::recoverable(arithmetic_overflow(
639                        "prepared source file buffer range",
640                    ))
641                })?;
642                let mut payload = FilePayload::from(data);
643                builder
644                    .write_file_member(
645                        &entry.archive_path,
646                        &mut payload,
647                        EntryMetadata::default().executable(executable),
648                    )
649                    .await?;
650            }
651            PreparedTraversalKind::StreamingFile {
652                file,
653                path,
654                size,
655                executable,
656            } => {
657                let mut file = tokio::fs::File::from_std(file);
658                file.set_max_buf_size(FILE_PAYLOAD_CHUNK_BYTES);
659                let mut payload =
660                    FilePayload::streaming(size, file, mem::take(buffer), Some(path.as_path()));
661                let result = builder
662                    .write_file_member(
663                        &entry.archive_path,
664                        &mut payload,
665                        EntryMetadata::default().executable(executable),
666                    )
667                    .await;
668                payload.swap_buffer(buffer);
669                result?;
670            }
671            PreparedTraversalKind::SymbolicLink { target } => {
672                builder
673                    .write_symbolic_link_member(&entry.archive_path, &target)
674                    .await?;
675            }
676        }
677        traversal
678            .entries
679            .commit_entry(&entry.archive_path, reservation);
680        traversal.emitted = true;
681    }
682    Ok(())
683}
684
685struct DirectoryBuild<'entries> {
686    entries: &'entries mut BuildEntries,
687    source_buffer: Vec<u8>,
688    emitted: bool,
689}
690
691struct PreparedDirectoryBatch {
692    entries: Vec<PreparedTraversalEntry>,
693    buffer: Vec<u8>,
694}
695
696struct PreparedTraversalEntry {
697    archive_path: String,
698    kind: PreparedTraversalKind,
699}
700
701enum PreparedTraversalKind {
702    Directory,
703    BufferedFile {
704        range: Range<usize>,
705        executable: bool,
706    },
707    StreamingFile {
708        file: std::fs::File,
709        path: PathBuf,
710        size: u64,
711        executable: bool,
712    },
713    SymbolicLink {
714        target: String,
715    },
716}
717
718async fn prepare_directory_entries(
719    mut entries: VecDeque<TraversalEntry>,
720    mut buffer: Vec<u8>,
721) -> Result<(PreparedDirectoryBatch, VecDeque<TraversalEntry>), SourceError> {
722    tokio::task::spawn_blocking(move || {
723        buffer.clear();
724        let mut prepared = Vec::with_capacity(entries.len());
725        while let Some(entry) = entries.pop_front() {
726            let TraversalEntry {
727                source,
728                archive_path,
729                kind,
730            } = entry;
731            let (kind, batch_complete) = match kind {
732                TraversalKind::Directory => (PreparedTraversalKind::Directory, false),
733                TraversalKind::Regular => prepare_regular_file(source, &mut buffer)?,
734                TraversalKind::SymbolicLink { target } => {
735                    (PreparedTraversalKind::SymbolicLink { target }, false)
736                }
737            };
738            prepared.push(PreparedTraversalEntry { archive_path, kind });
739            if batch_complete {
740                break;
741            }
742        }
743        Ok((
744            PreparedDirectoryBatch {
745                entries: prepared,
746                buffer,
747            },
748            entries,
749        ))
750    })
751    .await
752    .map_err(SourceError::BlockingTask)?
753}
754
755fn prepare_regular_file(
756    path: PathBuf,
757    buffer: &mut Vec<u8>,
758) -> Result<(PreparedTraversalKind, bool), SourceError> {
759    let file = std::fs::File::open(&path)
760        .map_err(|source| SourceError::filesystem("open source file", &path, source))?;
761    let metadata = file
762        .metadata()
763        .map_err(|source| SourceError::filesystem("inspect source file", &path, source))?;
764    if !metadata.is_file() {
765        return Err(SourceError::filesystem(
766            "inspect source file",
767            &path,
768            io::Error::other("source is not a regular file"),
769        ));
770    }
771    let size = metadata.len();
772    let executable = is_executable(&metadata);
773    if size > BUFFERED_SOURCE_FILE_BYTES as u64 {
774        return Ok((
775            PreparedTraversalKind::StreamingFile {
776                file,
777                path,
778                size,
779                executable,
780            },
781            true,
782        ));
783    }
784    let payload_size = usize::try_from(size).map_err(|_| SourceError::ArithmeticOverflow {
785        context: "buffered source file size",
786    })?;
787    let start = buffer.len();
788    let end = start
789        .checked_add(payload_size)
790        .ok_or(SourceError::ArithmeticOverflow {
791            context: "buffered source batch size",
792        })?;
793    buffer.resize(end, 0);
794    (&file)
795        .read_exact(&mut buffer[start..end])
796        .map_err(|source| SourceError::filesystem("read source file", &path, source))?;
797    Ok((
798        PreparedTraversalKind::BufferedFile {
799            range: start..end,
800            executable,
801        },
802        buffer.len() >= SOURCE_FILE_PREPARATION_BATCH_BYTES,
803    ))
804}
805
806enum SourceError {
807    Filesystem {
808        operation: &'static str,
809        path: PathBuf,
810        source: io::Error,
811    },
812    BlockingTask(tokio::task::JoinError),
813    ArithmeticOverflow {
814        context: &'static str,
815    },
816}
817
818impl SourceError {
819    fn filesystem(operation: &'static str, path: &Path, source: io::Error) -> Self {
820        Self::Filesystem {
821            operation,
822            path: path.to_path_buf(),
823            source,
824        }
825    }
826
827    fn into_build_error<E>(self) -> BuildError<E> {
828        match self {
829            Self::Filesystem {
830                operation,
831                path,
832                source,
833            } => BuildError::Filesystem {
834                operation,
835                path,
836                source,
837            },
838            Self::BlockingTask(error) => BuildError::BlockingTask(error),
839            Self::ArithmeticOverflow { context } => BuildError::ArithmeticOverflow { context },
840        }
841    }
842}
843
844/// A failure while constructing an archive.
845#[derive(Debug, Error)]
846pub enum BuildError<E> {
847    /// The archive format encoder failed.
848    #[error(transparent)]
849    Encoder(E),
850    /// Traversing a recursive source failed.
851    #[error(transparent)]
852    Traversal(#[from] TraversalError),
853    /// A requested archive path cannot be represented by the UTF-8 builder.
854    #[error("invalid archive path {path:?}: {reason}")]
855    InvalidArchivePath {
856        /// The rejected archive path.
857        path: PathBuf,
858        /// The reason the path cannot be represented.
859        reason: &'static str,
860    },
861    /// An archive name was rejected by the configured [`BuilderPolicy`].
862    #[error("archive {context} rejected by builder policy: {value:?}")]
863    NameRejected {
864        /// The role of the rejected archive text.
865        context: &'static str,
866        /// The rejected UTF-8 value.
867        value: String,
868    },
869    /// An archive path collides with a previously reserved entry.
870    #[error("archive entry collides with existing path {path}")]
871    PathCollision {
872        /// The conflicting normalized archive path.
873        path: String,
874    },
875    /// A file payload was read before it was passed to [`Builder::add_file`].
876    #[error("file payload was already read before being added to the archive")]
877    FilePayloadAlreadyRead,
878    /// A source filesystem operation failed.
879    #[error("failed to {operation} {path}: {source}")]
880    Filesystem {
881        /// The operation that failed.
882        operation: &'static str,
883        /// The affected source filesystem path.
884        path: PathBuf,
885        /// The underlying I/O error.
886        #[source]
887        source: io::Error,
888    },
889    /// Reading an asynchronous file payload source failed.
890    #[error("failed to read archive file payload source")]
891    SourceRead {
892        /// The underlying I/O error.
893        #[source]
894        source: io::Error,
895    },
896    /// A blocking filesystem operation failed to complete.
897    #[error("failed to complete blocking archive filesystem operation: {0}")]
898    BlockingTask(#[from] tokio::task::JoinError),
899    /// The builder cannot continue because a prior failure may have written bytes.
900    #[error("archive builder is poisoned after a previous partial write")]
901    Poisoned,
902    /// A size computation exceeded this API's range.
903    #[error("arithmetic overflow while computing {context}")]
904    ArithmeticOverflow {
905        /// The failed computation.
906        context: &'static str,
907    },
908}
909
910#[derive(Clone, Copy, Debug)]
911enum ArchivedEntry {
912    Directory { explicit: bool },
913    NonDirectory,
914}
915
916/// Builder collision state keyed by literal `/`-separated archive components.
917#[derive(Debug)]
918struct BuildEntries(ComponentTree<Box<str>, ArchivedEntry>);
919
920/// Proof that an entry was checked against the current collision state.
921struct EntryReservation {
922    entry: ArchivedEntry,
923}
924
925impl BuildEntries {
926    fn new() -> Self {
927        Self(ComponentTree::new(None))
928    }
929
930    fn preflight_entry<E>(
931        &self,
932        path: &str,
933        entry: ArchivedEntry,
934    ) -> Result<EntryReservation, BuildError<E>> {
935        let mut parent = ROOT_NODE;
936        let mut components = archive_path_components(path).peekable();
937        while let Some((component, prefix)) = components.next() {
938            let Some(node) = self.0.child(parent, component) else {
939                return Ok(EntryReservation { entry });
940            };
941            if components.peek().is_some() {
942                match self.0.state(node) {
943                    Some(ArchivedEntry::Directory { .. }) => parent = node,
944                    Some(ArchivedEntry::NonDirectory) => return Err(path_collision(prefix)),
945                    None => return Ok(EntryReservation { entry }),
946                }
947            } else {
948                match (self.0.state(node), entry) {
949                    (
950                        Some(ArchivedEntry::Directory { explicit: false }),
951                        ArchivedEntry::Directory { .. },
952                    )
953                    | (None, _) => return Ok(EntryReservation { entry }),
954                    (Some(_), _) => return Err(path_collision(prefix)),
955                }
956            }
957        }
958        Ok(EntryReservation { entry })
959    }
960
961    fn commit_entry(&mut self, path: &str, reservation: EntryReservation) {
962        // The builder holds exclusive state access while the backend hook is
963        // awaited, so a successful reservation remains valid until this commit.
964        let mut parent = ROOT_NODE;
965        let mut components = archive_path_components(path).peekable();
966        while let Some((component, _)) = components.next() {
967            let node = self
968                .0
969                .ensure_child_with(parent, component, || component.into());
970            if components.peek().is_some() {
971                if self.0.state(node).is_none() {
972                    self.0
973                        .set_state(node, ArchivedEntry::Directory { explicit: false });
974                }
975            } else {
976                self.0.set_state(node, reservation.entry);
977            }
978            parent = node;
979        }
980    }
981
982    #[cfg(test)]
983    fn node_count(&self) -> usize {
984        self.0.node_count()
985    }
986
987    #[cfg(test)]
988    fn component_bytes(&self) -> usize {
989        self.0.components().map(|component| component.len()).sum()
990    }
991}
992
993/// Iterates the exact textual component and prefix at each `/` boundary.
994fn archive_path_components(path: &str) -> impl Iterator<Item = (&str, &str)> {
995    let mut component_start = 0;
996    path.split('/').map(move |component| {
997        let prefix_end = component_start + component.len();
998        let prefix = &path[..prefix_end];
999        component_start = if prefix_end < path.len() {
1000            prefix_end + 1
1001        } else {
1002            prefix_end
1003        };
1004        (component, prefix)
1005    })
1006}
1007
1008fn file_payload_read_error<E>(filesystem_path: Option<&Path>, source: io::Error) -> BuildError<E> {
1009    if let Some(path) = filesystem_path {
1010        BuildError::Filesystem {
1011            operation: "read source file",
1012            path: path.to_path_buf(),
1013            source,
1014        }
1015    } else {
1016        BuildError::SourceRead { source }
1017    }
1018}
1019
1020fn arithmetic_overflow<E>(context: &'static str) -> BuildError<E> {
1021    BuildError::ArithmeticOverflow { context }
1022}
1023
1024fn path_collision<E>(path: &str) -> BuildError<E> {
1025    BuildError::PathCollision {
1026        path: path.to_owned(),
1027    }
1028}
1029
1030#[cfg(unix)]
1031fn is_executable(metadata: &std::fs::Metadata) -> bool {
1032    use std::os::unix::fs::PermissionsExt;
1033
1034    metadata.permissions().mode() & 0o111 != 0
1035}
1036
1037#[cfg(not(unix))]
1038fn is_executable(_metadata: &std::fs::Metadata) -> bool {
1039    false
1040}
1041
1042#[cfg(test)]
1043mod tests {
1044    use std::fs;
1045
1046    use tempfile::tempdir;
1047
1048    use super::*;
1049
1050    #[derive(Debug)]
1051    struct TestError;
1052
1053    #[derive(Default)]
1054    struct NoopArchiveBuilder {
1055        fail_next_file: bool,
1056        fail_next_directory: bool,
1057    }
1058
1059    impl ArchiveBuilder for NoopArchiveBuilder {
1060        type Error = TestError;
1061
1062        async fn finish_archive(&mut self) -> Result<(), BuildFailure<Self::Error>> {
1063            Ok(())
1064        }
1065
1066        async fn write_file_member(
1067            &mut self,
1068            _path: &str,
1069            payload: &mut FilePayload<'_>,
1070            _metadata: EntryMetadata,
1071        ) -> Result<(), BuildFailure<Self::Error>> {
1072            if mem::take(&mut self.fail_next_file) {
1073                return Err(BuildFailure::recoverable(BuildError::Encoder(TestError)));
1074            }
1075            loop {
1076                match payload.next_chunk::<TestError>().await {
1077                    Ok(Some(_)) => {}
1078                    Ok(None) => return Ok(()),
1079                    Err(error) => return Err(BuildFailure::recoverable(error)),
1080                }
1081            }
1082        }
1083
1084        async fn write_directory_member(
1085            &mut self,
1086            _path: &str,
1087        ) -> Result<(), BuildFailure<Self::Error>> {
1088            if mem::take(&mut self.fail_next_directory) {
1089                return Err(BuildFailure::recoverable(BuildError::Encoder(TestError)));
1090            }
1091            Ok(())
1092        }
1093
1094        async fn write_symbolic_link_member(
1095            &mut self,
1096            _path: &str,
1097            _target: &str,
1098        ) -> Result<(), BuildFailure<Self::Error>> {
1099            Ok(())
1100        }
1101    }
1102
1103    #[tokio::test]
1104    async fn deep_manual_entry_uses_linear_component_storage() {
1105        const COMPONENT: &str = "segment";
1106        const DEPTH: usize = 4_096;
1107
1108        let mut path = format!("{COMPONENT}/").repeat(DEPTH);
1109        path.push_str("file");
1110        let mut builder = NoopArchiveBuilder::default().builder();
1111        builder
1112            .add_file(&path, b"".as_slice(), EntryMetadata::default())
1113            .await
1114            .expect("deep manual file should be added");
1115
1116        assert_eq!(builder.state.entries.node_count(), DEPTH + 2);
1117        assert_eq!(
1118            builder.state.entries.component_bytes(),
1119            DEPTH * COMPONENT.len() + "file".len()
1120        );
1121    }
1122
1123    #[tokio::test]
1124    async fn collision_state_preserves_literal_slash_components() {
1125        let mut builder = NoopArchiveBuilder::default().builder();
1126        for path in ["a//b", "a/b", "/absolute", "absolute", ".", ".."] {
1127            builder
1128                .add_file(path, b"".as_slice(), EntryMetadata::default())
1129                .await
1130                .expect("distinct textual path should be added");
1131        }
1132
1133        for (path, collision) in [("a//b", "a//b"), ("a/", "a/"), ("", ""), ("./child", ".")] {
1134            assert!(matches!(
1135                builder
1136                    .add_file(
1137                        path,
1138                        b"".as_slice(),
1139                        EntryMetadata::default(),
1140                    )
1141                    .await,
1142                Err(BuildError::PathCollision { path }) if path == collision
1143            ));
1144        }
1145    }
1146
1147    #[tokio::test]
1148    async fn recoverable_write_failure_does_not_commit_reservation() {
1149        let mut builder = NoopArchiveBuilder {
1150            fail_next_file: true,
1151            ..Default::default()
1152        }
1153        .builder();
1154        assert!(matches!(
1155            builder
1156                .add_file("parent/file", b"".as_slice(), EntryMetadata::default(),)
1157                .await,
1158            Err(BuildError::Encoder(TestError))
1159        ));
1160        builder
1161            .add_file("parent/file", b"".as_slice(), EntryMetadata::default())
1162            .await
1163            .expect("a recoverable failure should not reserve the path");
1164    }
1165
1166    #[tokio::test]
1167    async fn recoverable_directory_write_failure_does_not_commit_reservation() {
1168        let mut builder = NoopArchiveBuilder {
1169            fail_next_directory: true,
1170            ..Default::default()
1171        }
1172        .builder();
1173
1174        assert!(matches!(
1175            builder.add_directory("directory").await,
1176            Err(BuildError::Encoder(TestError))
1177        ));
1178        assert_eq!(builder.state.entries.node_count(), 1);
1179
1180        builder
1181            .add_directory("directory")
1182            .await
1183            .expect("a recoverable failure should not reserve the directory");
1184        assert_eq!(builder.state.entries.node_count(), 2);
1185    }
1186
1187    #[tokio::test]
1188    async fn repeated_directory_additions_use_linear_component_storage() {
1189        const DIRECTORIES: usize = 256;
1190
1191        let temp = tempdir().expect("temporary directory should be created");
1192        let mut builder = NoopArchiveBuilder::default().builder();
1193        for index in 0..DIRECTORIES {
1194            let source = temp.path().join(format!("directory-{index}"));
1195            fs::create_dir(&source).expect("source directory should be created");
1196            builder
1197                .add_directory_all(&source)
1198                .await
1199                .expect("empty source directory should be added");
1200        }
1201
1202        assert_eq!(builder.state.entries.node_count(), DIRECTORIES + 1);
1203    }
1204}