oci-image 0.1.3

//! One-pass layer unpack.
//!
//! The tar is read once, in order. Whiteouts (`.wh.<name>`) and opaque
//! markers (`.wh..wh..opq`) remove what LOWER layers put there; anything this
//! layer itself writes, before or after the marker, stays. That is the OCI
//! rule, and it makes the result independent of where the markers sit in the
//! tar, so no separate whiteout pass is needed.
//!
//! Nothing is written outside the rootfs, whatever the layer holds and
//! whatever another process changes in the rootfs meanwhile: every entry's
//! parent is resolved beneath the rootfs's descriptor (`openat2` with
//! `RESOLVE_BENEATH`, or an `O_NOFOLLOW` walk), and every write is an `*at`
//! call on that descriptor that never follows the last component. A symlink
//! is resolved inside the rootfs; one that leads out (absolute, or `..` above
//! the root) is an [`Error::Escape`] for anything written or removed below it.
//!
//! Directories, links, special files and the creation of every regular file
//! happen in order on the calling thread. Regular files up to [`SMALL_FILE`]
//! are read into memory and filled by a pool of worker threads (write, chown,
//! chmod, mtime, all on the descriptor opened in place), with at most
//! [`IN_FLIGHT`] bytes and [`OPEN_FILES`] files waiting.

use std::collections::HashSet;
use std::ffi::{CStr, OsStr};
use std::fs::{self, File};
use std::io::{self, Read, Write};
use std::os::fd::{AsRawFd, OwnedFd, RawFd};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::MetadataExt;
use std::path::{Component, Path, PathBuf};
use std::sync::{mpsc, Arc, Condvar, Mutex};
use std::thread;

use crate::beneath::{
    cstr, cvt, escape, fstat, fstatat, is, is_escape, list, openat, remove_at, Root,
};
use crate::decode::{open_layer, Decoder};
use crate::error::Context;
use crate::{Error, Result};

/// Regular files up to this size go to the writer pool; larger ones are
/// streamed to disk on the unpacking thread.
pub const SMALL_FILE: u64 = 1 << 20;
/// The most file bytes held in memory for the writer pool.
pub const IN_FLIGHT: usize = 16 << 20;
/// The most files open and waiting for the writer pool.
pub const OPEN_FILES: usize = 256;
/// Read-ahead chunk between the decoder thread and the tar reader.
const CHUNK: usize = 1 << 20;

/// How to unpack layers.
#[derive(Debug, Clone)]
pub struct UnpackOptions {
    /// Threads for file writes and parallel gzip decode; 0 means all cores
    /// this process may use. 1 unpacks on the calling thread only.
    pub threads: usize,
    /// Decode gzip layers with `lgz` on `threads` workers (false: one-core
    /// zlib-rs).
    pub parallel_gzip: bool,
    /// Restore the uid/gid of every entry (needs root or CAP_CHOWN).
    pub same_owner: bool,
    /// Apply whiteouts onto `rootfs` (flatten), or keep them in overlayfs
    /// form for a per-layer directory.
    pub layout: Layout,
}

/// What an unpack does with whiteouts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Layout {
    /// Apply them: the layer lands on top of the layers already in `rootfs`.
    #[default]
    Flatten,
    /// Keep them for overlayfs: `.wh.<name>` becomes a 0/0 character device,
    /// `.wh..wh..opq` an `overlay.opaque` xattr on its directory. For one
    /// layer into its own empty directory.
    Overlay(OverlayXattr),
}

/// The xattr namespace overlayfs reads its markers from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OverlayXattr {
    /// `trusted.overlay.*`: a mount by real root.
    Trusted,
    /// `user.overlay.*`, mounted with `userxattr` (in a user namespace).
    User,
}

impl Default for UnpackOptions {
    fn default() -> Self {
        Self {
            threads: 0,
            parallel_gzip: true,
            same_owner: fs::metadata("/proc/self").is_ok_and(|m| m.uid() == 0),
            layout: Layout::Flatten,
        }
    }
}

impl UnpackOptions {
    /// The thread count to use.
    pub fn threads(&self) -> usize {
        if self.threads > 0 {
            self.threads
        } else {
            thread::available_parallelism().map_or(1, |n| n.get())
        }
    }
}

/// What one layer unpack did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnpackStats {
    pub decoder: Decoder,
    /// Tar entries read.
    pub entries: u64,
    /// Regular files written.
    pub files: u64,
    /// Bytes of regular file content written.
    pub bytes: u64,
    /// Whiteout and opaque markers applied.
    pub whiteouts: u64,
    /// Device nodes that could not be created (in a user namespace) and were skipped.
    pub skipped_devices: u64,
}

/// Apply the layer blob at `blob` (plain, gzip or zstd) onto `rootfs`.
pub fn apply_layer_file(blob: &Path, rootfs: &Path, opts: &UnpackOptions) -> Result<UnpackStats> {
    let (reader, decoder, _) = open_layer(blob, opts, false)?;
    let mut stats = apply_layer(reader, rootfs, opts)?;
    stats.decoder = decoder;
    Ok(stats)
}

/// [`apply_layer_file`], also returning the sha256 of the uncompressed tar
/// (the layer's diff_id). On the lgz path it is computed on one more thread
/// from the decoder's in-order output, beside the tar reader: no extra pass.
pub fn apply_layer_file_digest(
    blob: &Path,
    rootfs: &Path,
    opts: &UnpackOptions,
) -> Result<(UnpackStats, crate::Digest)> {
    let (reader, decoder, digest) = open_layer(blob, opts, true)?;
    let mut stats = apply_layer(reader, rootfs, opts)?;
    stats.decoder = decoder;
    let digest = digest
        .finish()
        .ok_or_else(|| Error::Invalid(format!("hashing layer {} failed", blob.display())))?;
    Ok((stats, digest))
}

/// Apply one uncompressed layer tar onto `rootfs`.
///
/// A refused layer leaves what it wrote so far (inside the rootfs): unpack
/// into a fresh directory and remove it on error, as [`crate::LayerStore`]
/// does.
pub fn apply_layer<R: Read + Send>(
    tar: R,
    rootfs: &Path,
    opts: &UnpackOptions,
) -> Result<UnpackStats> {
    let threads = opts.threads();
    let root = Root::open(rootfs).ctx(|| format!("rootfs {}", rootfs.display()))?;
    let mut w = Writer {
        root: &root,
        same_owner: opts.same_owner,
        layout: opts.layout,
        ours: HashSet::new(),
        dirs: Vec::new(),
        stats: UnpackStats {
            decoder: Decoder::Plain,
            entries: 0,
            files: 0,
            bytes: 0,
            whiteouts: 0,
            skipped_devices: 0,
        },
    };

    thread::scope(|s| -> Result<()> {
        let reader: Box<dyn Read> = if threads > 1 {
            Box::new(read_ahead(s, tar))
        } else {
            Box::new(tar)
        };
        let mut pool = Pool::new(s, threads.saturating_sub(1), opts.same_owner);
        let mut archive = tar::Archive::new(reader);
        let r = (|| -> Result<()> {
            for entry in archive.entries().ctx(|| "reading layer".into())? {
                let mut entry = entry.ctx(|| "reading layer".into())?;
                w.entry(&mut entry, &mut pool)?;
            }
            Ok(())
        })();
        // The pool is joined either way; the first error wins.
        let done = pool.finish();
        r?;
        done?;
        // Read to the end so the gzip/zstd trailer is checked.
        let mut rest = archive.into_inner();
        io::copy(&mut rest, &mut io::sink()).ctx(|| "reading the layer to its end".into())?;
        Ok(())
    })?;

    // Directory mtimes last, deepest first: writing into a directory moves it.
    for (rel, mtime) in w.dirs.iter().rev() {
        if let Ok(Some(d)) = w.open_dir(rel, false) {
            let t = times(*mtime);
            unsafe { libc::futimens(d.as_raw_fd(), t.as_ptr()) };
        }
    }
    Ok(w.stats)
}

/// Owner, mode and mtime of an entry.
struct FileMeta {
    mode: u32,
    uid: u64,
    gid: u64,
    mtime: u64,
}

fn times(mtime: u64) -> [libc::timespec; 2] {
    let t = libc::timespec {
        tv_sec: mtime as libc::time_t,
        tv_nsec: 0,
    };
    [t, t]
}

/// Owner, mode and mtime of a file just filled, on its descriptor.
fn finish_file(f: &File, meta: &FileMeta, same_owner: bool) -> io::Result<()> {
    let fd = f.as_raw_fd();
    if same_owner {
        cvt(unsafe { libc::fchown(fd, meta.uid as u32, meta.gid as u32) })?;
    }
    cvt(unsafe { libc::fchmod(fd, meta.mode) })?;
    let t = times(meta.mtime);
    cvt(unsafe { libc::futimens(fd, t.as_ptr()) })?;
    Ok(())
}

/// `EXDEV` (the path leaves the rootfs) is an escape; anything else an io
/// error about `rel`.
fn err(e: io::Error, rel: &Path) -> Error {
    if is_escape(&e) {
        Error::Escape(rel.to_path_buf())
    } else {
        Error::Io {
            context: format!("unpacking {}", rel.display()),
            source: e,
        }
    }
}

fn split(rel: &Path) -> (&Path, &OsStr) {
    (
        rel.parent().unwrap_or(Path::new("")),
        rel.file_name().unwrap_or_default(),
    )
}

struct Writer<'r> {
    root: &'r Root,
    same_owner: bool,
    layout: Layout,
    /// Paths this layer has put down (and their parents): whiteouts only
    /// hide lower layers.
    ours: HashSet<PathBuf>,
    dirs: Vec<(PathBuf, u64)>,
    stats: UnpackStats,
}

impl Writer<'_> {
    /// The parent directory of `rel`, resolved beneath the root. Without
    /// `create`, `None` if it is not there.
    fn parent(&self, rel: &Path, create: bool) -> Result<Option<OwnedFd>> {
        match self.root.dir(split(rel).0, create) {
            Ok(fd) => Ok(Some(fd)),
            Err(e) if !create && matches!(e.raw_os_error(), Some(libc::ENOENT | libc::ENOTDIR)) => {
                Ok(None)
            }
            Err(e) => Err(err(e, rel)),
        }
    }

    /// The directory `rel` itself, opened readable (for listing, xattrs and
    /// times).
    fn open_dir(&self, rel: &Path, create: bool) -> Result<Option<OwnedFd>> {
        let p = match self.root.dir(rel, create) {
            Ok(p) => p,
            Err(e) if !create && matches!(e.raw_os_error(), Some(libc::ENOENT | libc::ENOTDIR)) => {
                return Ok(None)
            }
            Err(e) => return Err(err(e, rel)),
        };
        openat(p.as_raw_fd(), c".", libc::O_RDONLY | libc::O_DIRECTORY, 0)
            .map(Some)
            .map_err(|e| err(e, rel))
    }

    /// Hide `rel` from lower layers: remove it, unless this layer wrote it;
    /// in a directory this layer wrote into, hide what is inside instead.
    fn hide(&self, rel: &Path) -> Result<()> {
        let Some(pfd) = self.parent(rel, false)? else {
            return Ok(());
        };
        let c = cstr(split(rel).1).map_err(|e| err(e, rel))?;
        if !self.ours.contains(rel) {
            return remove_at(pfd.as_raw_fd(), &c).map_err(|e| err(e, rel));
        }
        let st = fstatat(pfd.as_raw_fd(), &c).map_err(|e| err(e, rel))?;
        if st.is_some_and(|st| is(&st, libc::S_IFDIR)) {
            let d = openat(
                pfd.as_raw_fd(),
                &c,
                libc::O_RDONLY | libc::O_DIRECTORY | libc::O_NOFOLLOW,
                0,
            )
            .map_err(|e| err(e, rel))?;
            for child in list(d.as_raw_fd()).map_err(|e| err(e, rel))? {
                self.hide(&rel.join(OsStr::from_bytes(child.to_bytes())))?;
            }
        }
        Ok(())
    }

    fn whiteout(&mut self, parent: &Path, hidden: Option<&str>) -> Result<()> {
        self.stats.whiteouts += 1;
        match (self.layout, hidden) {
            (Layout::Flatten, None) => {
                if let Some(d) = self.open_dir(parent, false)? {
                    for child in list(d.as_raw_fd()).map_err(|e| err(e, parent))? {
                        self.hide(&parent.join(OsStr::from_bytes(child.to_bytes())))?;
                    }
                }
            }
            (Layout::Flatten, Some(h)) => self.hide(&parent.join(h))?,
            (Layout::Overlay(xattr), None) => {
                let d = self.open_dir(parent, true)?.expect("created");
                let name: &CStr = match xattr {
                    OverlayXattr::Trusted => c"trusted.overlay.opaque",
                    OverlayXattr::User => c"user.overlay.opaque",
                };
                cvt(unsafe {
                    libc::fsetxattr(d.as_raw_fd(), name.as_ptr(), b"y".as_ptr().cast(), 1, 0)
                })
                .map_err(|e| err(e, parent))?;
                add_with_parents(&mut self.ours, parent);
            }
            (Layout::Overlay(_), Some(h)) => {
                let rel = parent.join(h);
                let pfd = self.parent(&rel, true)?.expect("created");
                let c = cstr(OsStr::new(h)).map_err(|e| err(e, &rel))?;
                remove_at(pfd.as_raw_fd(), &c).map_err(|e| err(e, &rel))?;
                // An overlayfs whiteout: a character device 0/0.
                cvt(unsafe { libc::mknodat(pfd.as_raw_fd(), c.as_ptr(), libc::S_IFCHR, 0) })
                    .map_err(|e| err(e, &rel))?;
                add_with_parents(&mut self.ours, &rel);
            }
        }
        Ok(())
    }

    /// Owner, mode and times of a fifo or device just made, through an
    /// `O_PATH` descriptor of it (`/proc/self/fd`), so a swap cannot redirect
    /// them.
    fn special_meta(&self, pfd: RawFd, name: &CStr, m: &FileMeta) -> io::Result<()> {
        let fd = openat(pfd, name, libc::O_PATH | libc::O_NOFOLLOW, 0)?;
        if is(&fstat(fd.as_raw_fd())?, libc::S_IFLNK) {
            return Err(escape());
        }
        if self.same_owner {
            cvt(unsafe {
                libc::fchownat(
                    fd.as_raw_fd(),
                    c"".as_ptr(),
                    m.uid as u32,
                    m.gid as u32,
                    libc::AT_EMPTY_PATH,
                )
            })?;
        }
        let proc = cstr(OsStr::new(&format!("/proc/self/fd/{}", fd.as_raw_fd())))?;
        cvt(unsafe { libc::chmod(proc.as_ptr(), m.mode) })?;
        let t = times(m.mtime);
        cvt(unsafe { libc::utimensat(libc::AT_FDCWD, proc.as_ptr(), t.as_ptr(), 0) })?;
        Ok(())
    }

    fn entry<R: Read>(&mut self, entry: &mut tar::Entry<'_, R>, pool: &mut Pool) -> Result<()> {
        self.stats.entries += 1;
        let raw = entry.path().ctx(|| "entry path".into())?.into_owned();
        let Some(rel) = entry_path(&raw) else {
            // The root itself is skipped; a `..` is a hostile layer.
            if raw
                .components()
                .any(|c| matches!(c, Component::ParentDir | Component::Prefix(_)))
            {
                return Err(Error::Escape(raw));
            }
            return Ok(());
        };
        let (parent, name) = split(&rel);
        let name_s = name.to_str().unwrap_or("");
        if name_s == ".wh..wh..opq" {
            return self.whiteout(parent, None);
        }
        if let Some(hidden) = name_s.strip_prefix(".wh.") {
            // `.wh..` and `.wh...` would hide the directory itself or the one
            // above it (for a top-level entry: the rootfs, or its parent).
            if matches!(hidden, "" | "." | "..") {
                return Err(Error::Escape(rel));
            }
            return self.whiteout(parent, Some(hidden));
        }

        let h = entry.header();
        let kind = h.entry_type();
        let known = kind.is_dir()
            || kind.is_file()
            || kind == tar::EntryType::Continuous
            || kind.is_symlink()
            || kind.is_hard_link()
            || kind.is_fifo()
            || kind.is_character_special()
            || kind.is_block_special();
        if !known {
            return Ok(());
        }
        let m = FileMeta {
            mode: h.mode().unwrap_or(0o644) & 0o7777,
            uid: h.uid().unwrap_or(0),
            gid: h.gid().unwrap_or(0),
            mtime: h.mtime().unwrap_or(0),
        };
        let pfd = self.parent(&rel, true)?.expect("created");
        let p = pfd.as_raw_fd();
        let c = cstr(name).map_err(|e| err(e, &rel))?;
        let io = |e: io::Error| err(e, &rel);

        // A file replacing a directory (or the reverse) needs the old one gone.
        if let Some(st) = fstatat(p, &c).map_err(io)? {
            if !(is(&st, libc::S_IFDIR) && kind.is_dir()) {
                remove_at(p, &c).map_err(io)?;
            }
        }

        if kind.is_dir() {
            match cvt(unsafe { libc::mkdirat(p, c.as_ptr(), 0o700) }) {
                Ok(_) => {}
                Err(e) if e.raw_os_error() == Some(libc::EEXIST) => {}
                Err(e) => return Err(io(e)),
            }
            let d = openat(
                p,
                &c,
                libc::O_RDONLY | libc::O_DIRECTORY | libc::O_NOFOLLOW,
                0,
            )
            .map_err(io)?;
            if self.same_owner {
                cvt(unsafe { libc::fchown(d.as_raw_fd(), m.uid as u32, m.gid as u32) })
                    .map_err(io)?;
            }
            cvt(unsafe { libc::fchmod(d.as_raw_fd(), m.mode) }).map_err(io)?;
            self.dirs.push((rel.clone(), m.mtime));
        } else if kind.is_file() || kind == tar::EntryType::Continuous {
            let fd = openat(
                p,
                &c,
                libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_NOFOLLOW,
                0o600,
            )
            .map_err(io)?;
            let mut f = File::from(fd);
            let size = entry.size();
            if pool.active() && size > 0 && size <= SMALL_FILE {
                let mut data = Vec::with_capacity(size as usize);
                entry
                    .read_to_end(&mut data)
                    .ctx(|| format!("reading {}", rel.display()))?;
                pool.submit(Job {
                    file: f,
                    data,
                    meta: m,
                    rel: rel.clone(),
                })?;
            } else {
                io::copy(entry, &mut f).ctx(|| format!("writing {}", rel.display()))?;
                finish_file(&f, &m, self.same_owner).map_err(io)?;
            }
            self.stats.files += 1;
            self.stats.bytes += size;
        } else if kind.is_symlink() {
            let target = entry
                .link_name()
                .ctx(|| format!("link of {}", rel.display()))?
                .ok_or_else(|| {
                    Error::Invalid(format!("symlink {} has no target", rel.display()))
                })?;
            let t = cstr(target.as_os_str()).map_err(io)?;
            cvt(unsafe { libc::symlinkat(t.as_ptr(), p, c.as_ptr()) }).map_err(io)?;
            if self.same_owner {
                cvt(unsafe {
                    libc::fchownat(
                        p,
                        c.as_ptr(),
                        m.uid as u32,
                        m.gid as u32,
                        libc::AT_SYMLINK_NOFOLLOW,
                    )
                })
                .map_err(io)?;
            }
            let ts = times(m.mtime);
            cvt(unsafe { libc::utimensat(p, c.as_ptr(), ts.as_ptr(), libc::AT_SYMLINK_NOFOLLOW) })
                .map_err(io)?;
        } else if kind.is_hard_link() {
            let target = entry
                .link_name()
                .ctx(|| format!("link of {}", rel.display()))?
                .ok_or_else(|| {
                    Error::Invalid(format!("hard link {} has no target", rel.display()))
                })?;
            let src = entry_path(&target).ok_or_else(|| Error::Escape(target.to_path_buf()))?;
            let spfd = self
                .root
                .dir(split(&src).0, false)
                .map_err(|e| err(e, &src))?;
            let sc = cstr(split(&src).1).map_err(io)?;
            // Flags 0: a symlink source is linked itself, never followed.
            cvt(unsafe { libc::linkat(spfd.as_raw_fd(), sc.as_ptr(), p, c.as_ptr(), 0) })
                .map_err(io)?;
        } else {
            let (t, dev) = if kind.is_fifo() {
                (libc::S_IFIFO, 0)
            } else {
                let maj = h.device_major().ok().flatten().unwrap_or(0);
                let min = h.device_minor().ok().flatten().unwrap_or(0);
                let t = if kind.is_character_special() {
                    libc::S_IFCHR
                } else {
                    libc::S_IFBLK
                };
                (t, libc::makedev(maj, min))
            };
            match cvt(unsafe { libc::mknodat(p, c.as_ptr(), t | 0o600, dev) }) {
                Ok(_) => self.special_meta(p, &c, &m).map_err(io)?,
                // Device nodes cannot be created in a user namespace; a
                // container gets its own /dev anyway.
                Err(_) if !kind.is_fifo() => {
                    self.stats.skipped_devices += 1;
                    return Ok(());
                }
                Err(e) => return Err(io(e)),
            }
        }
        add_with_parents(&mut self.ours, &rel);
        Ok(())
    }
}

/// One step of a layer, as [`changes`] reports it.
pub enum Change<'a, 'b, R: Read> {
    /// `.wh..wh..opq` in this directory: lower layers' contents are hidden.
    Opaque(PathBuf),
    /// `.wh.<name>`: this path from lower layers is hidden.
    Whiteout(PathBuf),
    /// Any other entry, with its clean relative path.
    Entry(PathBuf, &'a mut tar::Entry<'b, R>),
}

/// Walk an uncompressed layer tar in order, classifying whiteouts, for
/// readers that keep their own view of the filesystem (scanners, catalogers).
/// Entries whose path would escape the root, and whiteouts of `.` or `..`,
/// are skipped.
pub fn changes<R: Read>(
    tar: R,
    mut f: impl FnMut(Change<'_, '_, R>) -> io::Result<()>,
) -> Result<()> {
    let mut archive = tar::Archive::new(tar);
    for entry in archive.entries().ctx(|| "reading layer".into())? {
        let mut entry = entry.ctx(|| "reading layer".into())?;
        let raw = entry.path().ctx(|| "entry path".into())?.into_owned();
        let Some(rel) = entry_path(&raw) else {
            continue;
        };
        let name = rel.file_name().and_then(|n| n.to_str()).unwrap_or("");
        let parent = rel.parent().unwrap_or(Path::new("")).to_path_buf();
        if name
            .strip_prefix(".wh.")
            .is_some_and(|h| matches!(h, "" | "." | ".."))
        {
            continue;
        }
        let change = if name == ".wh..wh..opq" {
            Change::Opaque(parent)
        } else if let Some(hidden) = name.strip_prefix(".wh.") {
            Change::Whiteout(parent.join(hidden))
        } else {
            Change::Entry(rel, &mut entry)
        };
        f(change).ctx(|| "layer visitor".into())?;
    }
    Ok(())
}

/// Clean relative path of a tar entry, or `None` if it would escape (or is
/// the root itself).
pub fn entry_path(path: &Path) -> Option<PathBuf> {
    let mut out = PathBuf::new();
    for c in path.components() {
        match c {
            Component::Normal(p) => out.push(p),
            Component::CurDir | Component::RootDir => {}
            Component::ParentDir | Component::Prefix(_) => return None,
        }
    }
    (!out.as_os_str().is_empty()).then_some(out)
}

fn add_with_parents(set: &mut HashSet<PathBuf>, rel: &Path) {
    let mut p = Some(rel);
    while let Some(q) = p.filter(|q| !q.as_os_str().is_empty()) {
        if !set.insert(q.to_path_buf()) {
            break;
        }
        p = q.parent();
    }
}

/// A small file created in place, to be filled by the pool.
struct Job {
    file: File,
    data: Vec<u8>,
    meta: FileMeta,
    rel: PathBuf,
}

#[derive(Default)]
struct State {
    pending: usize,
    bytes: usize,
    error: Option<Error>,
}

#[derive(Default)]
struct Shared {
    state: Mutex<State>,
    cv: Condvar,
}

/// File writer threads.
struct Pool {
    tx: Option<mpsc::Sender<Job>>,
    shared: Arc<Shared>,
}

impl Pool {
    fn new<'s>(s: &'s thread::Scope<'s, '_>, workers: usize, same_owner: bool) -> Self {
        let shared = Arc::new(Shared::default());
        if workers == 0 {
            return Self { tx: None, shared };
        }
        let (tx, rx) = mpsc::channel::<Job>();
        let rx = Arc::new(Mutex::new(rx));
        for _ in 0..workers {
            let (rx, shared) = (rx.clone(), shared.clone());
            // A32-14, stays a thread: a scoped file writer of one layer's unpack, joined when
            // the unpack's thread::scope ends.
            s.spawn(move || loop {
                let job = rx.lock().unwrap().recv();
                let Ok(mut job) = job else { return };
                let len = job.data.len();
                let r = job
                    .file
                    .write_all(&job.data)
                    .and_then(|()| finish_file(&job.file, &job.meta, same_owner))
                    .ctx(|| format!("writing {}", job.rel.display()));
                drop(job);
                let mut st = shared.state.lock().unwrap();
                st.pending -= 1;
                st.bytes -= len;
                if let Err(e) = r {
                    st.error.get_or_insert(e);
                }
                shared.cv.notify_all();
            });
        }
        Self {
            tx: Some(tx),
            shared,
        }
    }

    fn active(&self) -> bool {
        self.tx.is_some()
    }

    fn submit(&mut self, job: Job) -> Result<()> {
        let len = job.data.len();
        let mut st = self.shared.state.lock().unwrap();
        while st.pending > 0
            && (st.bytes + len > IN_FLIGHT || st.pending >= OPEN_FILES)
            && st.error.is_none()
        {
            st = self.shared.cv.wait(st).unwrap();
        }
        if let Some(e) = st.error.take() {
            return Err(e);
        }
        st.pending += 1;
        st.bytes += len;
        drop(st);
        self.tx
            .as_ref()
            .unwrap()
            .send(job)
            .expect("writer threads live");
        Ok(())
    }

    /// Wait for every submitted write, then stop the workers.
    fn finish(mut self) -> Result<()> {
        let mut st = self.shared.state.lock().unwrap();
        while st.pending > 0 {
            st = self.shared.cv.wait(st).unwrap();
        }
        let r = st.error.take().map_or(Ok(()), Err);
        drop(st);
        self.tx = None;
        r
    }
}

/// Decode on its own thread, a few chunks ahead of the tar reader.
fn read_ahead<'s, R: Read + Send + 's>(
    s: &'s thread::Scope<'s, '_>,
    mut inner: R,
) -> ChannelReader {
    let (tx, rx) = mpsc::sync_channel::<io::Result<Vec<u8>>>(4);
    // A32-14, stays a thread: the scoped read-ahead of one layer's tar, joined when the unpack's
    // thread::scope ends.
    s.spawn(move || loop {
        let mut buf = vec![0u8; CHUNK];
        let mut n = 0;
        while n < CHUNK {
            match inner.read(&mut buf[n..]) {
                Ok(0) => break,
                Ok(k) => n += k,
                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
                Err(e) => {
                    let _ = tx.send(Err(e));
                    return;
                }
            }
        }
        buf.truncate(n);
        let done = n < CHUNK;
        if (n > 0 && tx.send(Ok(buf)).is_err()) || done {
            return;
        }
    });
    ChannelReader {
        rx,
        buf: Vec::new(),
        pos: 0,
    }
}

struct ChannelReader {
    rx: mpsc::Receiver<io::Result<Vec<u8>>>,
    buf: Vec<u8>,
    pos: usize,
}

impl Read for ChannelReader {
    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
        while self.pos == self.buf.len() {
            match self.rx.recv() {
                Ok(chunk) => {
                    self.buf = chunk?;
                    self.pos = 0;
                }
                Err(_) => return Ok(0),
            }
        }
        let n = out.len().min(self.buf.len() - self.pos);
        out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]);
        self.pos += n;
        Ok(n)
    }
}