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`] and then the [`Builder`] APIs instead.
342///
343/// Hook implementations must return [`BuildFailure::recoverable`] only when the
344/// failed invocation wrote no output. Any failure after output may have begun
345/// must use [`BuildFailure::poisoned`].
346#[expect(
347    async_fn_in_trait,
348    reason = "archive writers may be !Send and run on a local executor"
349)]
350pub trait ArchiveBuilder: Sized {
351    /// The archive-format error returned while encoding entries.
352    type Error;
353
354    /// Wraps this format writer in a builder using default policy.
355    ///
356    /// Implementors should not override this default implementation.
357    fn builder(self) -> Builder<Self> {
358        Builder {
359            backend: self,
360            state: BuilderState::new(BuilderPolicy::default()),
361        }
362    }
363
364    /// Writes any format-specific archive terminator or index.
365    async fn finish_archive(&mut self) -> Result<(), BuildFailure<Self::Error>>;
366
367    /// Writes one regular-file member and its complete payload.
368    ///
369    /// Implementations must call [`FilePayload::next_chunk`] through
370    /// completion and classify failures using [`BuildFailure`].
371    async fn write_file_member(
372        &mut self,
373        path: &str,
374        payload: &mut FilePayload<'_>,
375        metadata: EntryMetadata,
376    ) -> Result<(), BuildFailure<Self::Error>>;
377
378    /// Writes one directory member.
379    async fn write_directory_member(&mut self, path: &str)
380    -> Result<(), BuildFailure<Self::Error>>;
381
382    /// Writes one symbolic-link member.
383    async fn write_symbolic_link_member(
384        &mut self,
385        path: &str,
386        target: &str,
387    ) -> Result<(), BuildFailure<Self::Error>>;
388}
389
390/// A stateful format-neutral archive construction engine.
391///
392/// Create this wrapper with [`ArchiveBuilder::builder`].
393pub struct Builder<B> {
394    backend: B,
395    state: BuilderState,
396}
397
398impl<B: ArchiveBuilder> Builder<B> {
399    /// Configures the policy used by this builder.
400    ///
401    /// Call before adding any entries.
402    pub fn with_policy(mut self, policy: BuilderPolicy) -> Self {
403        self.state.policy = policy;
404        self
405    }
406
407    /// Adds one regular file from a [`FilePayload`].
408    ///
409    /// If the payload ends before its declared size or returns an error, the
410    /// addition fails and the builder is poisoned if the archive member's
411    /// output may already have begun.
412    ///
413    /// The payload must not have been read through [`FilePayload::next_chunk`]
414    /// before this method is called.
415    pub async fn add_file<'a, P>(
416        &mut self,
417        path: P,
418        payload: impl Into<FilePayload<'a>>,
419        metadata: EntryMetadata,
420    ) -> Result<(), BuildError<B::Error>>
421    where
422        P: AsRef<Path>,
423    {
424        self.state.ensure_active()?;
425        let archive_path = path.as_ref();
426        let Some(path) = archive_path.to_str() else {
427            return Err(BuildError::InvalidArchivePath {
428                path: archive_path.to_path_buf(),
429                reason: "path is not valid UTF-8",
430            });
431        };
432        if !self.state.policy.name_validation.accepts(path) {
433            return Err(BuildError::NameRejected {
434                context: "member path",
435                value: path.to_owned(),
436            });
437        }
438        let path = path.to_owned();
439        let reservation = self
440            .state
441            .entries
442            .preflight_entry(&path, ArchivedEntry::NonDirectory)?;
443        let mut payload = payload.into();
444        if payload.started {
445            return Err(BuildError::FilePayloadAlreadyRead);
446        }
447        payload.swap_buffer(&mut self.state.source_buffer);
448        self.state.begin_write();
449        let result = self
450            .backend
451            .write_file_member(&path, &mut payload, metadata)
452            .await;
453        self.state.complete_write();
454        payload.swap_buffer(&mut self.state.source_buffer);
455        self.resolve_hook(result)?;
456        self.state.entries.commit_entry(&path, reservation);
457        Ok(())
458    }
459
460    /// Adds one directory member without reading from the filesystem.
461    ///
462    /// This creates only the named directory member. It does not add any child
463    /// members; use [`Self::add_directory_all`] to recursively add a filesystem
464    /// directory and its contents.
465    pub async fn add_directory<P: AsRef<Path>>(
466        &mut self,
467        path: P,
468    ) -> Result<(), BuildError<B::Error>> {
469        self.state.ensure_active()?;
470        let archive_path = path.as_ref();
471        let Some(path) = archive_path.to_str() else {
472            return Err(BuildError::InvalidArchivePath {
473                path: archive_path.to_path_buf(),
474                reason: "path is not valid UTF-8",
475            });
476        };
477        if !self.state.policy.name_validation.accepts(path) {
478            return Err(BuildError::NameRejected {
479                context: "member path",
480                value: path.to_owned(),
481            });
482        }
483        let path = path.to_owned();
484        let reservation = self
485            .state
486            .entries
487            .preflight_entry(&path, ArchivedEntry::Directory { explicit: true })?;
488        self.state.begin_write();
489        let result = self.backend.write_directory_member(&path).await;
490        self.state.complete_write();
491        self.resolve_hook(result)?;
492        self.state.entries.commit_entry(&path, reservation);
493        Ok(())
494    }
495
496    /// Recursively adds a filesystem directory beneath its UTF-8 basename.
497    ///
498    /// Entries are visited in deterministic sorted order and files are streamed
499    /// with bounded memory. Source symbolic links are rejected by default;
500    /// [`BuilderPolicy::symlink_policy`] can instead preserve them. A late
501    /// traversal or validation failure may leave partial output and poison
502    /// this builder.
503    pub async fn add_directory_all<P: AsRef<Path>>(
504        &mut self,
505        source: P,
506    ) -> Result<(), BuildError<B::Error>> {
507        self.state.ensure_active()?;
508        let source = source.as_ref().to_path_buf();
509        let mut entries = stream_directory_entries(
510            source,
511            self.state.policy.name_validation,
512            self.state.policy.symlink_policy,
513        )
514        .map_err(BuildError::Traversal)?;
515        self.state.begin_write();
516        let mut traversal = DirectoryBuild {
517            entries: &mut self.state.entries,
518            source_buffer: mem::take(&mut self.state.source_buffer),
519            emitted: false,
520        };
521        let write_result =
522            write_directory_entries(&mut self.backend, &mut entries, &mut traversal).await;
523        let traversal_result = entries
524            .finish()
525            .await
526            .map_err(BuildError::Traversal)
527            .map_err(BuildFailure::recoverable);
528        let result = write_result.and(traversal_result);
529        let DirectoryBuild {
530            entries: _,
531            source_buffer,
532            emitted,
533        } = traversal;
534        self.state.complete_write();
535        self.state.source_buffer = source_buffer;
536        match result {
537            Ok(()) => Ok(()),
538            Err(error) => {
539                let (error, hook_poisoned) = error.into_parts();
540                if emitted || hook_poisoned {
541                    self.state.poison();
542                }
543                Err(error)
544            }
545        }
546    }
547
548    /// Finalizes and consumes this archive builder.
549    pub async fn finish(self) -> Result<(), BuildError<B::Error>> {
550        self.finish_into_inner().await?;
551        Ok(())
552    }
553
554    /// Finalizes this archive builder and returns its format writer.
555    ///
556    /// This allows callers to recover and finalize an owned output sink after
557    /// the archive terminator has been written.
558    pub async fn finish_into_inner(mut self) -> Result<B, BuildError<B::Error>> {
559        self.state.ensure_active()?;
560        let result = self.backend.finish_archive().await;
561        self.resolve_hook(result)?;
562        Ok(self.backend)
563    }
564
565    fn resolve_hook<T>(
566        &mut self,
567        result: Result<T, BuildFailure<B::Error>>,
568    ) -> Result<T, BuildError<B::Error>> {
569        match result {
570            Ok(value) => Ok(value),
571            Err(error) => {
572                let (error, poisons_builder) = error.into_parts();
573                if poisons_builder {
574                    self.state.poison();
575                }
576                Err(error)
577            }
578        }
579    }
580}
581
582async fn write_directory_entries<B: ArchiveBuilder>(
583    builder: &mut B,
584    entries: &mut TraversalStream,
585    traversal: &mut DirectoryBuild<'_>,
586) -> Result<(), BuildFailure<B::Error>> {
587    while let Some(entries) = entries.recv().await {
588        let mut entries = VecDeque::from(entries);
589        while !entries.is_empty() {
590            let buffer = mem::take(&mut traversal.source_buffer);
591            let (prepared, remaining) = prepare_directory_entries(entries, buffer)
592                .await
593                .map_err(SourceError::into_build_error)
594                .map_err(BuildFailure::recoverable)?;
595            entries = remaining;
596            let PreparedDirectoryBatch {
597                entries: prepared_entries,
598                mut buffer,
599            } = prepared;
600            let result =
601                write_prepared_directory_entries(builder, prepared_entries, &mut buffer, traversal)
602                    .await;
603            traversal.source_buffer = buffer;
604            result?;
605        }
606    }
607    Ok(())
608}
609
610async fn write_prepared_directory_entries<B: ArchiveBuilder>(
611    builder: &mut B,
612    entries: Vec<PreparedTraversalEntry>,
613    buffer: &mut Vec<u8>,
614    traversal: &mut DirectoryBuild<'_>,
615) -> Result<(), BuildFailure<B::Error>> {
616    for entry in entries {
617        let reservation = traversal
618            .entries
619            .preflight_entry(
620                &entry.archive_path,
621                if matches!(&entry.kind, PreparedTraversalKind::Directory) {
622                    ArchivedEntry::Directory { explicit: true }
623                } else {
624                    ArchivedEntry::NonDirectory
625                },
626            )
627            .map_err(BuildFailure::recoverable)?;
628        match entry.kind {
629            PreparedTraversalKind::Directory => {
630                builder.write_directory_member(&entry.archive_path).await?;
631            }
632            PreparedTraversalKind::BufferedFile { range, executable } => {
633                let data = buffer.get(range).ok_or_else(|| {
634                    BuildFailure::recoverable(arithmetic_overflow(
635                        "prepared source file buffer range",
636                    ))
637                })?;
638                let mut payload = FilePayload::from(data);
639                builder
640                    .write_file_member(
641                        &entry.archive_path,
642                        &mut payload,
643                        EntryMetadata::default().executable(executable),
644                    )
645                    .await?;
646            }
647            PreparedTraversalKind::StreamingFile {
648                file,
649                path,
650                size,
651                executable,
652            } => {
653                let mut file = tokio::fs::File::from_std(file);
654                file.set_max_buf_size(FILE_PAYLOAD_CHUNK_BYTES);
655                let mut payload =
656                    FilePayload::streaming(size, file, mem::take(buffer), Some(path.as_path()));
657                let result = builder
658                    .write_file_member(
659                        &entry.archive_path,
660                        &mut payload,
661                        EntryMetadata::default().executable(executable),
662                    )
663                    .await;
664                payload.swap_buffer(buffer);
665                result?;
666            }
667            PreparedTraversalKind::SymbolicLink { target } => {
668                builder
669                    .write_symbolic_link_member(&entry.archive_path, &target)
670                    .await?;
671            }
672        }
673        traversal
674            .entries
675            .commit_entry(&entry.archive_path, reservation);
676        traversal.emitted = true;
677    }
678    Ok(())
679}
680
681struct DirectoryBuild<'entries> {
682    entries: &'entries mut BuildEntries,
683    source_buffer: Vec<u8>,
684    emitted: bool,
685}
686
687struct PreparedDirectoryBatch {
688    entries: Vec<PreparedTraversalEntry>,
689    buffer: Vec<u8>,
690}
691
692struct PreparedTraversalEntry {
693    archive_path: String,
694    kind: PreparedTraversalKind,
695}
696
697enum PreparedTraversalKind {
698    Directory,
699    BufferedFile {
700        range: Range<usize>,
701        executable: bool,
702    },
703    StreamingFile {
704        file: std::fs::File,
705        path: PathBuf,
706        size: u64,
707        executable: bool,
708    },
709    SymbolicLink {
710        target: String,
711    },
712}
713
714async fn prepare_directory_entries(
715    mut entries: VecDeque<TraversalEntry>,
716    mut buffer: Vec<u8>,
717) -> Result<(PreparedDirectoryBatch, VecDeque<TraversalEntry>), SourceError> {
718    tokio::task::spawn_blocking(move || {
719        buffer.clear();
720        let mut prepared = Vec::with_capacity(entries.len());
721        while let Some(entry) = entries.pop_front() {
722            let TraversalEntry {
723                source,
724                archive_path,
725                kind,
726            } = entry;
727            let (kind, batch_complete) = match kind {
728                TraversalKind::Directory => (PreparedTraversalKind::Directory, false),
729                TraversalKind::Regular => prepare_regular_file(source, &mut buffer)?,
730                TraversalKind::SymbolicLink { target } => {
731                    (PreparedTraversalKind::SymbolicLink { target }, false)
732                }
733            };
734            prepared.push(PreparedTraversalEntry { archive_path, kind });
735            if batch_complete {
736                break;
737            }
738        }
739        Ok((
740            PreparedDirectoryBatch {
741                entries: prepared,
742                buffer,
743            },
744            entries,
745        ))
746    })
747    .await
748    .map_err(SourceError::BlockingTask)?
749}
750
751fn prepare_regular_file(
752    path: PathBuf,
753    buffer: &mut Vec<u8>,
754) -> Result<(PreparedTraversalKind, bool), SourceError> {
755    let file = std::fs::File::open(&path)
756        .map_err(|source| SourceError::filesystem("open source file", &path, source))?;
757    let metadata = file
758        .metadata()
759        .map_err(|source| SourceError::filesystem("inspect source file", &path, source))?;
760    if !metadata.is_file() {
761        return Err(SourceError::filesystem(
762            "inspect source file",
763            &path,
764            io::Error::other("source is not a regular file"),
765        ));
766    }
767    let size = metadata.len();
768    let executable = is_executable(&metadata);
769    if size > BUFFERED_SOURCE_FILE_BYTES as u64 {
770        return Ok((
771            PreparedTraversalKind::StreamingFile {
772                file,
773                path,
774                size,
775                executable,
776            },
777            true,
778        ));
779    }
780    let payload_size = usize::try_from(size).map_err(|_| SourceError::ArithmeticOverflow {
781        context: "buffered source file size",
782    })?;
783    let start = buffer.len();
784    let end = start
785        .checked_add(payload_size)
786        .ok_or(SourceError::ArithmeticOverflow {
787            context: "buffered source batch size",
788        })?;
789    buffer.resize(end, 0);
790    (&file)
791        .read_exact(&mut buffer[start..end])
792        .map_err(|source| SourceError::filesystem("read source file", &path, source))?;
793    Ok((
794        PreparedTraversalKind::BufferedFile {
795            range: start..end,
796            executable,
797        },
798        buffer.len() >= SOURCE_FILE_PREPARATION_BATCH_BYTES,
799    ))
800}
801
802enum SourceError {
803    Filesystem {
804        operation: &'static str,
805        path: PathBuf,
806        source: io::Error,
807    },
808    BlockingTask(tokio::task::JoinError),
809    ArithmeticOverflow {
810        context: &'static str,
811    },
812}
813
814impl SourceError {
815    fn filesystem(operation: &'static str, path: &Path, source: io::Error) -> Self {
816        Self::Filesystem {
817            operation,
818            path: path.to_path_buf(),
819            source,
820        }
821    }
822
823    fn into_build_error<E>(self) -> BuildError<E> {
824        match self {
825            Self::Filesystem {
826                operation,
827                path,
828                source,
829            } => BuildError::Filesystem {
830                operation,
831                path,
832                source,
833            },
834            Self::BlockingTask(error) => BuildError::BlockingTask(error),
835            Self::ArithmeticOverflow { context } => BuildError::ArithmeticOverflow { context },
836        }
837    }
838}
839
840/// A failure while constructing an archive.
841#[derive(Debug, Error)]
842pub enum BuildError<E> {
843    /// The archive format encoder failed.
844    #[error(transparent)]
845    Encoder(E),
846    /// Traversing a recursive source failed.
847    #[error(transparent)]
848    Traversal(#[from] TraversalError),
849    /// A requested archive path cannot be represented by the UTF-8 builder.
850    #[error("invalid archive path {path:?}: {reason}")]
851    InvalidArchivePath {
852        /// The rejected archive path.
853        path: PathBuf,
854        /// The reason the path cannot be represented.
855        reason: &'static str,
856    },
857    /// An archive name was rejected by the configured [`BuilderPolicy`].
858    #[error("archive {context} rejected by builder policy: {value:?}")]
859    NameRejected {
860        /// The role of the rejected archive text.
861        context: &'static str,
862        /// The rejected UTF-8 value.
863        value: String,
864    },
865    /// An archive path collides with a previously reserved entry.
866    #[error("archive entry collides with existing path {path}")]
867    PathCollision {
868        /// The conflicting normalized archive path.
869        path: String,
870    },
871    /// A file payload was read before it was passed to [`Builder::add_file`].
872    #[error("file payload was already read before being added to the archive")]
873    FilePayloadAlreadyRead,
874    /// A source filesystem operation failed.
875    #[error("failed to {operation} {path}: {source}")]
876    Filesystem {
877        /// The operation that failed.
878        operation: &'static str,
879        /// The affected source filesystem path.
880        path: PathBuf,
881        /// The underlying I/O error.
882        #[source]
883        source: io::Error,
884    },
885    /// Reading an asynchronous file payload source failed.
886    #[error("failed to read archive file payload source")]
887    SourceRead {
888        /// The underlying I/O error.
889        #[source]
890        source: io::Error,
891    },
892    /// A blocking filesystem operation failed to complete.
893    #[error("failed to complete blocking archive filesystem operation: {0}")]
894    BlockingTask(#[from] tokio::task::JoinError),
895    /// The builder cannot continue because a prior failure may have written bytes.
896    #[error("archive builder is poisoned after a previous partial write")]
897    Poisoned,
898    /// A size computation exceeded this API's range.
899    #[error("arithmetic overflow while computing {context}")]
900    ArithmeticOverflow {
901        /// The failed computation.
902        context: &'static str,
903    },
904}
905
906#[derive(Clone, Copy, Debug)]
907enum ArchivedEntry {
908    Directory { explicit: bool },
909    NonDirectory,
910}
911
912/// Builder collision state keyed by literal `/`-separated archive components.
913#[derive(Debug)]
914struct BuildEntries(ComponentTree<Box<str>, ArchivedEntry>);
915
916/// Proof that an entry was checked against the current collision state.
917struct EntryReservation {
918    entry: ArchivedEntry,
919}
920
921impl BuildEntries {
922    fn new() -> Self {
923        Self(ComponentTree::new(None))
924    }
925
926    fn preflight_entry<E>(
927        &self,
928        path: &str,
929        entry: ArchivedEntry,
930    ) -> Result<EntryReservation, BuildError<E>> {
931        let mut parent = ROOT_NODE;
932        let mut components = archive_path_components(path).peekable();
933        while let Some((component, prefix)) = components.next() {
934            let Some(node) = self.0.child(parent, component) else {
935                return Ok(EntryReservation { entry });
936            };
937            if components.peek().is_some() {
938                match self.0.state(node) {
939                    Some(ArchivedEntry::Directory { .. }) => parent = node,
940                    Some(ArchivedEntry::NonDirectory) => return Err(path_collision(prefix)),
941                    None => return Ok(EntryReservation { entry }),
942                }
943            } else {
944                match (self.0.state(node), entry) {
945                    (
946                        Some(ArchivedEntry::Directory { explicit: false }),
947                        ArchivedEntry::Directory { .. },
948                    )
949                    | (None, _) => return Ok(EntryReservation { entry }),
950                    (Some(_), _) => return Err(path_collision(prefix)),
951                }
952            }
953        }
954        Ok(EntryReservation { entry })
955    }
956
957    fn commit_entry(&mut self, path: &str, reservation: EntryReservation) {
958        // The builder holds exclusive state access while the backend hook is
959        // awaited, so a successful reservation remains valid until this commit.
960        let mut parent = ROOT_NODE;
961        let mut components = archive_path_components(path).peekable();
962        while let Some((component, _)) = components.next() {
963            let node = self
964                .0
965                .ensure_child_with(parent, component, || component.into());
966            if components.peek().is_some() {
967                if self.0.state(node).is_none() {
968                    self.0
969                        .set_state(node, ArchivedEntry::Directory { explicit: false });
970                }
971            } else {
972                self.0.set_state(node, reservation.entry);
973            }
974            parent = node;
975        }
976    }
977
978    #[cfg(test)]
979    fn node_count(&self) -> usize {
980        self.0.node_count()
981    }
982
983    #[cfg(test)]
984    fn component_bytes(&self) -> usize {
985        self.0.components().map(|component| component.len()).sum()
986    }
987}
988
989/// Iterates the exact textual component and prefix at each `/` boundary.
990fn archive_path_components(path: &str) -> impl Iterator<Item = (&str, &str)> {
991    let mut component_start = 0;
992    path.split('/').map(move |component| {
993        let prefix_end = component_start + component.len();
994        let prefix = &path[..prefix_end];
995        component_start = if prefix_end < path.len() {
996            prefix_end + 1
997        } else {
998            prefix_end
999        };
1000        (component, prefix)
1001    })
1002}
1003
1004fn file_payload_read_error<E>(filesystem_path: Option<&Path>, source: io::Error) -> BuildError<E> {
1005    if let Some(path) = filesystem_path {
1006        BuildError::Filesystem {
1007            operation: "read source file",
1008            path: path.to_path_buf(),
1009            source,
1010        }
1011    } else {
1012        BuildError::SourceRead { source }
1013    }
1014}
1015
1016fn arithmetic_overflow<E>(context: &'static str) -> BuildError<E> {
1017    BuildError::ArithmeticOverflow { context }
1018}
1019
1020fn path_collision<E>(path: &str) -> BuildError<E> {
1021    BuildError::PathCollision {
1022        path: path.to_owned(),
1023    }
1024}
1025
1026#[cfg(unix)]
1027fn is_executable(metadata: &std::fs::Metadata) -> bool {
1028    use std::os::unix::fs::PermissionsExt;
1029
1030    metadata.permissions().mode() & 0o111 != 0
1031}
1032
1033#[cfg(not(unix))]
1034fn is_executable(_metadata: &std::fs::Metadata) -> bool {
1035    false
1036}
1037
1038#[cfg(test)]
1039mod tests {
1040    use std::fs;
1041
1042    use tempfile::tempdir;
1043
1044    use super::*;
1045
1046    #[derive(Debug)]
1047    struct TestError;
1048
1049    #[derive(Default)]
1050    struct NoopArchiveBuilder {
1051        fail_next_file: bool,
1052        fail_next_directory: bool,
1053    }
1054
1055    impl ArchiveBuilder for NoopArchiveBuilder {
1056        type Error = TestError;
1057
1058        async fn finish_archive(&mut self) -> Result<(), BuildFailure<Self::Error>> {
1059            Ok(())
1060        }
1061
1062        async fn write_file_member(
1063            &mut self,
1064            _path: &str,
1065            payload: &mut FilePayload<'_>,
1066            _metadata: EntryMetadata,
1067        ) -> Result<(), BuildFailure<Self::Error>> {
1068            if mem::take(&mut self.fail_next_file) {
1069                return Err(BuildFailure::recoverable(BuildError::Encoder(TestError)));
1070            }
1071            loop {
1072                match payload.next_chunk::<TestError>().await {
1073                    Ok(Some(_)) => {}
1074                    Ok(None) => return Ok(()),
1075                    Err(error) => return Err(BuildFailure::recoverable(error)),
1076                }
1077            }
1078        }
1079
1080        async fn write_directory_member(
1081            &mut self,
1082            _path: &str,
1083        ) -> Result<(), BuildFailure<Self::Error>> {
1084            if mem::take(&mut self.fail_next_directory) {
1085                return Err(BuildFailure::recoverable(BuildError::Encoder(TestError)));
1086            }
1087            Ok(())
1088        }
1089
1090        async fn write_symbolic_link_member(
1091            &mut self,
1092            _path: &str,
1093            _target: &str,
1094        ) -> Result<(), BuildFailure<Self::Error>> {
1095            Ok(())
1096        }
1097    }
1098
1099    #[tokio::test]
1100    async fn deep_manual_entry_uses_linear_component_storage() {
1101        const COMPONENT: &str = "segment";
1102        const DEPTH: usize = 4_096;
1103
1104        let mut path = format!("{COMPONENT}/").repeat(DEPTH);
1105        path.push_str("file");
1106        let mut builder = NoopArchiveBuilder::default().builder();
1107        builder
1108            .add_file(&path, b"".as_slice(), EntryMetadata::default())
1109            .await
1110            .expect("deep manual file should be added");
1111
1112        assert_eq!(builder.state.entries.node_count(), DEPTH + 2);
1113        assert_eq!(
1114            builder.state.entries.component_bytes(),
1115            DEPTH * COMPONENT.len() + "file".len()
1116        );
1117    }
1118
1119    #[tokio::test]
1120    async fn collision_state_preserves_literal_slash_components() {
1121        let mut builder = NoopArchiveBuilder::default().builder();
1122        for path in ["a//b", "a/b", "/absolute", "absolute", ".", ".."] {
1123            builder
1124                .add_file(path, b"".as_slice(), EntryMetadata::default())
1125                .await
1126                .expect("distinct textual path should be added");
1127        }
1128
1129        for (path, collision) in [("a//b", "a//b"), ("a/", "a/"), ("", ""), ("./child", ".")] {
1130            assert!(matches!(
1131                builder
1132                    .add_file(
1133                        path,
1134                        b"".as_slice(),
1135                        EntryMetadata::default(),
1136                    )
1137                    .await,
1138                Err(BuildError::PathCollision { path }) if path == collision
1139            ));
1140        }
1141    }
1142
1143    #[tokio::test]
1144    async fn recoverable_write_failure_does_not_commit_reservation() {
1145        let mut builder = NoopArchiveBuilder {
1146            fail_next_file: true,
1147            ..Default::default()
1148        }
1149        .builder();
1150        assert!(matches!(
1151            builder
1152                .add_file("parent/file", b"".as_slice(), EntryMetadata::default(),)
1153                .await,
1154            Err(BuildError::Encoder(TestError))
1155        ));
1156        builder
1157            .add_file("parent/file", b"".as_slice(), EntryMetadata::default())
1158            .await
1159            .expect("a recoverable failure should not reserve the path");
1160    }
1161
1162    #[tokio::test]
1163    async fn recoverable_directory_write_failure_does_not_commit_reservation() {
1164        let mut builder = NoopArchiveBuilder {
1165            fail_next_directory: true,
1166            ..Default::default()
1167        }
1168        .builder();
1169
1170        assert!(matches!(
1171            builder.add_directory("directory").await,
1172            Err(BuildError::Encoder(TestError))
1173        ));
1174        assert_eq!(builder.state.entries.node_count(), 1);
1175
1176        builder
1177            .add_directory("directory")
1178            .await
1179            .expect("a recoverable failure should not reserve the directory");
1180        assert_eq!(builder.state.entries.node_count(), 2);
1181    }
1182
1183    #[tokio::test]
1184    async fn repeated_directory_additions_use_linear_component_storage() {
1185        const DIRECTORIES: usize = 256;
1186
1187        let temp = tempdir().expect("temporary directory should be created");
1188        let mut builder = NoopArchiveBuilder::default().builder();
1189        for index in 0..DIRECTORIES {
1190            let source = temp.path().join(format!("directory-{index}"));
1191            fs::create_dir(&source).expect("source directory should be created");
1192            builder
1193                .add_directory_all(&source)
1194                .await
1195                .expect("empty source directory should be added");
1196        }
1197
1198        assert_eq!(builder.state.entries.node_count(), DIRECTORIES + 1);
1199    }
1200}