nornir 0.4.54

//! Publish-order walker. Per-repo TOML defines `publish_order` as
//! `Vec<Vec<String>>`: outer entries are sequential phases, inner
//! entries are independent and may be published in parallel.
//!
//! Tag creation is done in-process via the pure-Rust `gix` crate (no
//! `git` shellout). Tag *pushing* over the network is deferred to a
//! follow-up that wires `gix` network features — for now we print the
//! exact `git push` invocation so a human or CI can complete the step.
//!
//! `cargo publish` itself remains a `cargo` subprocess because cargo
//! owns crate packaging (sources → `.crate` tarball → upload). That is
//! cargo's domain, not a script we can replace with a few lines of
//! HTTP. Isolated to [`run_cargo_publish`] and clearly annotated.

use std::path::Path;
use std::process::Command;

use anyhow::{anyhow, Context, Result};

/// Outcome reported per crate so the orchestrator (and progress
/// writer) can distinguish "really uploaded" from "already on the
/// registry" — re-running a release of a tree that's partially-shipped
/// is a routine case, not an error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublishOutcome {
    Published,
    AlreadyPublished,
    DryRun,
}

/// Walk the publish order, invoking cargo to publish each crate.
/// `dry_run` adds `--dry-run`. Returns the per-crate outcomes in the
/// same order they were published (outer phase order, inner declared
/// order). Errors only on real cargo failures — "crate already on the
/// registry" downgrades to [`PublishOutcome::AlreadyPublished`].
pub fn publish_all(
    repo_root: &Path,
    order: &[Vec<String>],
    dry_run: bool,
) -> Result<Vec<(String, PublishOutcome)>> {
    // No `publish_order` configured → DERIVE it from the workspace dependency
    // graph (topological, deps-first) instead of silently publishing nothing
    // (the bug that made `nornir release publish` exit 0 having uploaded nothing).
    // The manual `publish_order` is now an OVERRIDE, not a requirement.
    let derived;
    let order: &[Vec<String>] = if order.iter().all(|phase| phase.is_empty()) {
        derived = derive_publish_order(repo_root).with_context(|| {
            "no `publish_order` configured and could not derive one from `cargo metadata`"
        })?;
        if derived.iter().all(|p| p.is_empty()) {
            anyhow::bail!(
                "nothing to publish: no publishable crates in the workspace at {} \
                 (all `publish = false`?). Set `publish_order` in nornir.toml to override.",
                repo_root.display()
            );
        }
        &derived
    } else {
        order
    };
    let mut out: Vec<(String, PublishOutcome)> = Vec::new();
    for phase in order {
        for krate in phase {
            let outcome = run_cargo_publish(repo_root, krate, dry_run)
                .with_context(|| format!("cargo publish -p {krate}"))?;
            out.push((krate.clone(), outcome));
        }
    }
    Ok(out)
}

/// Deterministically derive the publish order from the workspace dependency
/// graph via `cargo metadata`: a topological sort (deps first) of the
/// **publishable** workspace members, grouped into phases (crates with no
/// remaining intra-workspace dependency publish together). Always possible for a
/// valid workspace — a publishable dep graph is a DAG (cargo forbids cycles); the
/// only filter is dropping `publish = false` crates. Fully deterministic: members
/// and phase contents are alphabetically ordered.
pub fn derive_publish_order(repo_root: &Path) -> Result<Vec<Vec<String>>> {
    let meta = cargo_metadata::MetadataCommand::new()
        .current_dir(repo_root)
        .exec()
        .with_context(|| format!("cargo metadata in {}", repo_root.display()))?;
    let ws: std::collections::BTreeSet<String> =
        meta.workspace_packages().iter().map(|p| p.name.to_string()).collect();
    let mut members: Vec<(String, Vec<String>)> = Vec::new();
    for p in meta.workspace_packages() {
        // `publish = false` serializes as `Some([])` → unpublishable, skip it.
        if matches!(&p.publish, Some(v) if v.is_empty()) {
            continue;
        }
        let deps: Vec<String> = p
            .dependencies
            .iter()
            .map(|d| d.name.clone())
            .filter(|n| ws.contains(n))
            .collect();
        members.push((p.name.to_string(), deps));
    }
    Ok(topo_phases(members))
}

/// Pure, deterministic Kahn topological sort into phases (deps-first). Each
/// `(name, deps)` lists the crate's intra-workspace dependencies. A crate is
/// "ready" when none of its workspace deps remain; ready crates form one phase
/// (alphabetical). Robust to a cycle (shouldn't occur for a publishable
/// workspace): the unresolved remainder is appended as a final phase so nothing
/// is silently dropped.
pub(crate) fn topo_phases(members: Vec<(String, Vec<String>)>) -> Vec<Vec<String>> {
    use std::collections::{BTreeMap, BTreeSet};
    let names: BTreeSet<String> = members.iter().map(|(n, _)| n.clone()).collect();
    let mut indeg: BTreeMap<String, usize> = BTreeMap::new();
    let mut dependents: BTreeMap<String, Vec<String>> = BTreeMap::new(); // dep → its dependents
    for (n, deps) in &members {
        let real: Vec<&String> =
            deps.iter().filter(|d| names.contains(*d) && d.as_str() != n).collect();
        indeg.insert(n.clone(), real.len());
        for d in real {
            dependents.entry(d.clone()).or_default().push(n.clone());
        }
    }
    let mut phases: Vec<Vec<String>> = Vec::new();
    let mut remaining: BTreeSet<String> = names;
    while !remaining.is_empty() {
        let ready: Vec<String> = remaining
            .iter()
            .filter(|n| indeg.get(*n).copied().unwrap_or(0) == 0)
            .cloned()
            .collect();
        if ready.is_empty() {
            phases.push(remaining.iter().cloned().collect()); // cycle guard — drop nothing
            break;
        }
        for n in &ready {
            remaining.remove(n);
            if let Some(deps) = dependents.get(n) {
                for dep in deps {
                    if let Some(e) = indeg.get_mut(dep) {
                        *e = e.saturating_sub(1);
                    }
                }
            }
        }
        phases.push(ready); // already alphabetical (from the BTreeSet iteration)
    }
    phases
}

/// Default cap on crates.io 429 rate-limit retries for one crate. crates.io's
/// publish limit replenishes one slot per ~60 s after a burst, so a handful of
/// retries clears any realistic backlog; beyond that we bail rather than loop
/// forever (mirrors [`wait_for_index`]'s bounded poll).
pub const DEFAULT_RATE_LIMIT_RETRIES: u32 = 6;

/// Hard ceiling on a single parsed 429 wait so a server-suggested
/// multi-hour delay can't hang the pipeline. crates.io waits are seconds-to-
/// low-minutes in practice.
pub const MAX_RATE_LIMIT_WAIT: std::time::Duration = std::time::Duration::from_secs(20 * 60);

/// Parse the wait duration out of crates.io's publish rate-limit (HTTP 429)
/// diagnostic. cargo surfaces the body verbatim, e.g.:
///   "You have published too many crates in a short period of time,
///    please try again after Wed, 18 Jun 2026 12:34:56 GMT, or email ..."
/// or the shorter relative form:
///   "... please try again after 56 seconds ..."
///   "... please try again in 4 minutes ..."
///
/// Returns `Some(duration)` if the message is recognisably a publish rate
/// limit, parsing a relative "<n> seconds/minutes" if present and otherwise
/// falling back to `default` (we can't reliably parse an absolute RFC-2822
/// timestamp without a date lib, and a fixed fallback is safer than guessing).
/// Returns `None` when the message is not a rate-limit error at all.
pub fn parse_rate_limit_wait(
    stderr: &str,
    default: std::time::Duration,
) -> Option<std::time::Duration> {
    let lower = stderr.to_ascii_lowercase();
    let is_rate_limit = lower.contains("too many crates")
        || (lower.contains("rate limit") && lower.contains("publish"))
        || (lower.contains("429") && lower.contains("publish"));
    if !is_rate_limit {
        return None;
    }
    // Look for a relative "<n> second(s)" / "<n> minute(s)" hint anywhere in
    // the message (handles both "after N seconds" and "in N minutes").
    let parse_after = |unit: &str, scale: u64| -> Option<std::time::Duration> {
        let idx = lower.find(unit)?;
        // Walk back over the unit's leading whitespace to the trailing digits.
        let prefix = &lower[..idx];
        let digits: String = prefix
            .trim_end()
            .chars()
            .rev()
            .take_while(|c| c.is_ascii_digit())
            .collect::<String>()
            .chars()
            .rev()
            .collect();
        let n: u64 = digits.parse().ok()?;
        Some(std::time::Duration::from_secs(n.saturating_mul(scale)))
    };
    let parsed = parse_after(" second", 1)
        .or_else(|| parse_after(" minute", 60))
        .unwrap_or(default);
    Some(parsed.min(MAX_RATE_LIMIT_WAIT))
}

/// The single intentional `cargo` subprocess. Cargo owns crate
/// packaging+upload; replacing it would require reimplementing
/// `cargo package` (thousands of lines covering manifest validation,
/// `.cargo_vcs_info.json`, exclude/include globs, tarball
/// determinism, license-file handling, etc.). Out of scope; not
/// every wheel needs reinventing.
///
/// On a crates.io publish rate-limit (HTTP 429 "You have published too many
/// crates in a short period of time, please try again after <t>") the wait is
/// parsed, slept off, and the same crate retried — up to
/// [`DEFAULT_RATE_LIMIT_RETRIES`] times — mirroring [`wait_for_index`]'s
/// bounded poll. A dry-run is never rate-limited (no upload), so the retry
/// loop is a no-op there.
pub fn run_cargo_publish(repo_root: &Path, krate: &str, dry_run: bool) -> Result<PublishOutcome> {
    run_cargo_publish_with_retry(
        repo_root,
        krate,
        dry_run,
        DEFAULT_RATE_LIMIT_RETRIES,
        std::time::Duration::from_secs(60),
        |d| std::thread::sleep(d),
    )
}

/// Testable core of [`run_cargo_publish`]: the `sleep` is injected so the
/// inject-assert test can drive the retry loop without real time or network.
/// `default_wait` is used when the 429 body carries no parseable relative wait.
pub fn run_cargo_publish_with_retry(
    repo_root: &Path,
    krate: &str,
    dry_run: bool,
    max_retries: u32,
    default_wait: std::time::Duration,
    mut sleep: impl FnMut(std::time::Duration),
) -> Result<PublishOutcome> {
    let mut attempt = 0u32;
    loop {
        let mut cmd = Command::new("cargo");
        cmd.arg("publish").arg("-p").arg(krate);
        if dry_run {
            cmd.arg("--dry-run");
        }
        cmd.current_dir(repo_root);
        let output = cmd.output().context("spawn cargo publish")?;
        if output.status.success() {
            return Ok(if dry_run {
                PublishOutcome::DryRun
            } else {
                PublishOutcome::Published
            });
        }
        let stderr = String::from_utf8_lossy(&output.stderr);
        // cargo's "already uploaded" diagnostic. Treat as success so a
        // re-run of a partially-shipped tree is idempotent.
        if stderr.contains("already uploaded")
            || stderr.contains("already exists on crates.io")
            || stderr.contains("crate version") && stderr.contains("is already uploaded")
        {
            eprintln!("    (skip: {krate} already on registry)");
            return Ok(PublishOutcome::AlreadyPublished);
        }
        // crates.io publish rate-limit (HTTP 429): parse the wait, sleep, retry.
        if let Some(wait) = parse_rate_limit_wait(&stderr, default_wait) {
            if attempt < max_retries {
                attempt += 1;
                eprintln!(
                    "    (rate-limited publishing {krate}: waiting {}s then retrying \
                     [attempt {attempt}/{max_retries}])",
                    wait.as_secs()
                );
                sleep(wait);
                continue;
            }
            eprintln!("{stderr}");
            return Err(anyhow!(
                "cargo publish -p {krate}: crates.io rate limit not cleared after \
                 {max_retries} retries"
            ));
        }
        // Re-emit cargo's stderr so the operator sees the real reason.
        eprintln!("{stderr}");
        return Err(anyhow!("cargo publish -p {krate} exited {}", output.status));
    }
}

/// Stage every worktree change (modifications, new files, deletions)
/// and create a commit on `HEAD`, returning the new commit SHA — or
/// `Ok(None)` when the working tree had nothing to commit (the common
/// case after a release where the README/CHANGELOG rendered
/// byte-identical to what's on disk).
///
/// 100% pure-Rust via `gix` — no `git` subprocess. The flow mirrors
/// `git add -A && git commit`:
///   1. enumerate `HEAD`→index and index→worktree status in one pass;
///   2. refuse to run if the index carries *staged* changes that differ
///      from `HEAD`. The release pipeline always regenerates docs on a
///      clean checkout (index == HEAD), so a dirty index signals an
///      unexpected state; committing anyway could silently diverge from
///      `git add -A` semantics, so we bail loudly instead;
///   3. apply every worktree delta onto a tree editor seeded from the
///      `HEAD` tree, write the tree, and commit it with `HEAD` as parent
///      (updates the `HEAD` ref + reflog);
///   4. refresh the on-disk index from the new tree so `git status`
///      reads clean afterwards.
///
/// Note: blobs are written from raw worktree bytes. nornir only commits
/// its own generated docs (UTF-8 markdown/TOML) in repos without
/// CRLF/clean `.gitattributes` filters, so no smudge/clean normalisation
/// is performed; repos relying on content filters are out of scope.
pub fn commit_release(repo_root: &Path, message: &str) -> Result<Option<String>> {
    use gix::bstr::BString;
    use gix::dir::entry::{Kind as DiskKind, Status as DiskStatus};
    use gix::status::index_worktree::Item as IwItem;
    use gix::status::plumbing::index_as_worktree::{Change, EntryStatus};

    let repo =
        gix::open(repo_root).with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo
        .head_commit()
        .context("resolve HEAD commit (unborn branch is unsupported here)")?;
    let head_id = head_commit.id;
    let head_tree_id = head_commit.tree().context("resolve HEAD tree")?.id;

    let iter = repo
        .status(gix::progress::Discard)
        .context("init status")?
        .head_tree(head_tree_id)
        .untracked_files(gix::status::UntrackedFiles::Files)
        .into_iter(Vec::<BString>::new())
        .context("start status walk")?;

    let mut editor = repo
        .edit_tree(head_tree_id)
        .context("seed tree editor from HEAD")?;
    let mut changed = false;

    for item in iter {
        match item.context("status item")? {
            // A tree→index difference means the index is dirty (staged
            // content differs from HEAD). Out of contract for releases.
            gix::status::Item::TreeIndex(change) => {
                return Err(anyhow!(
                    "refusing to commit: index has a staged change at `{}`; \
                     commit or reset it before releasing",
                    change.location()
                ));
            }
            gix::status::Item::IndexWorktree(iw) => match iw {
                IwItem::Modification {
                    rela_path, status, ..
                } => match status {
                    EntryStatus::Change(Change::Removed) => {
                        let rp: &gix::bstr::BStr = rela_path.as_ref();
                        editor
                            .remove(rp)
                            .with_context(|| format!("tree remove {rela_path}"))?;
                        changed = true;
                    }
                    EntryStatus::Conflict { .. } => {
                        return Err(anyhow!(
                            "refusing to commit: unresolved merge conflict at `{rela_path}`"
                        ));
                    }
                    // Modification / Type change / NeedsUpdate / IntentToAdd /
                    // SubmoduleModification: re-stage current worktree content.
                    _ => {
                        upsert_from_worktree(&repo, &mut editor, repo_root, rela_path.as_ref())?;
                        changed = true;
                    }
                },
                IwItem::DirectoryContents { entry, .. } => {
                    if matches!(entry.status, DiskStatus::Untracked)
                        && matches!(
                            entry.disk_kind,
                            Some(DiskKind::File) | Some(DiskKind::Symlink)
                        )
                    {
                        upsert_from_worktree(
                            &repo,
                            &mut editor,
                            repo_root,
                            entry.rela_path.as_ref(),
                        )?;
                        changed = true;
                    }
                }
                IwItem::Rewrite { .. } => {}
            },
        }
    }

    if !changed {
        return Ok(None);
    }
    let new_tree = editor.write().context("write release tree")?.detach();
    if new_tree == head_tree_id {
        return Ok(None);
    }
    let commit = repo
        .commit("HEAD", message, new_tree, Some(head_id))
        .context("create release commit")?;

    // Refresh the on-disk index from the new tree so the working tree
    // reads clean after the commit (as `git commit` would leave it).
    let mut index = repo
        .index_from_tree(&new_tree)
        .context("rebuild index from release tree")?;
    index
        .write(gix::index::write::Options::default())
        .context("persist refreshed index")?;

    Ok(Some(commit.to_string()))
}

/// Read `rela_path` from the working tree and upsert it into `editor`
/// with the correct blob kind (regular / executable / symlink).
fn upsert_from_worktree(
    repo: &gix::Repository,
    editor: &mut gix::object::tree::Editor<'_>,
    repo_root: &Path,
    rela_path: &gix::bstr::BStr,
) -> Result<()> {
    use gix::object::tree::EntryKind;

    let disk_path = repo_root.join(gix::path::from_bstr(rela_path).as_ref());
    let meta = std::fs::symlink_metadata(&disk_path)
        .with_context(|| format!("stat {}", disk_path.display()))?;

    let (bytes, kind): (Vec<u8>, EntryKind) = if meta.file_type().is_symlink() {
        let target = std::fs::read_link(&disk_path)
            .with_context(|| format!("readlink {}", disk_path.display()))?;
        let target = gix::path::into_bstr(target).into_owned();
        (target.into(), EntryKind::Link)
    } else {
        let bytes = std::fs::read(&disk_path)
            .with_context(|| format!("read {}", disk_path.display()))?;
        #[cfg(unix)]
        let kind = {
            use std::os::unix::fs::PermissionsExt;
            if meta.permissions().mode() & 0o111 != 0 {
                EntryKind::BlobExecutable
            } else {
                EntryKind::Blob
            }
        };
        #[cfg(not(unix))]
        let kind = EntryKind::Blob;
        (bytes, kind)
    };

    let blob = repo.write_blob(&bytes).context("write blob")?;
    editor
        .upsert(rela_path, kind, blob.detach())
        .with_context(|| format!("tree upsert {rela_path}"))?;
    Ok(())
}

/// Release commits are created locally; pushing is intentionally left to
/// the operator/CI. gix (0.84) implements fetch but not push/send-pack,
/// and nornir forbids both `git` subprocesses and C dependencies
/// (libgit2), so there is no in-process push path. This mirrors
/// [`tag()`], which likewise creates the ref locally and prints the push
/// command. Returns `false` to signal "not pushed" so callers don't
/// report a push that didn't happen.
pub fn push(repo_root: &Path, push_tags: bool) -> Result<bool> {
    let branch = crate::gitio::head_branch(repo_root).unwrap_or_else(|_| "HEAD".to_string());
    eprintln!("  ⏭  push skipped (pure-Rust build has no in-process push).");
    eprintln!("     run: git -C {} push origin {branch}", repo_root.display());
    if push_tags {
        eprintln!("     run: git -C {} push origin --tags", repo_root.display());
    }
    Ok(false)
}

/// Create an annotated `vX.Y.Z` tag pointing at HEAD using the
/// pure-Rust `gix` crate (no `git` shellout). Does *not* push — prints
/// the exact push command the operator/CI should run.
pub fn tag(repo_root: &Path, version: &str) -> Result<()> {
    let tag = format!("v{version}");
    let repo = gix::open(repo_root)
        .with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo.head_commit().context("resolve HEAD commit")?;

    let message = format!("Release {tag}\n");
    let signature = repo
        .author()
        .ok_or_else(|| anyhow!("git author not configured (user.name / user.email)"))?
        .map_err(|e| anyhow!("read git author: {e}"))?;

    repo.tag(
        &tag,
        head_commit.id,
        gix::objs::Kind::Commit,
        Some(signature),
        &message,
        gix::refs::transaction::PreviousValue::MustNotExist,
    )
    .with_context(|| format!("create tag {tag}"))?;

    eprintln!(
        "created local tag {tag} at {}. Push with: git push origin {tag}",
        head_commit.id
    );
    Ok(())
}

// ─── Feature 3: crates.io index propagation poll ────────────────────

/// Block until `crates.io/api/v1/crates/<name>` advertises `version`
/// as `max_version`, or `timeout` elapses. Returns the wait time in
/// milliseconds (0 if the version was already there) so callers can
/// log it / persist it into `publish_attempts.wait_ms`.
///
/// Replaces "just `sleep 20` between phases" — crates.io's index
/// propagates in 5-90 s depending on load, so a fixed sleep is either
/// too long (slowing the release) or too short (next phase tries to
/// resolve a dep that's not yet visible and fails).
///
/// Uses `curl` rather than a Rust HTTP client to avoid pulling
/// reqwest/tokio into the library footprint just for one poll loop —
/// matches the existing crates.io-probe pattern in the publish flow.
/// crates.io returns HTTP 403 without a non-default User-Agent.
pub fn wait_for_index(
    crate_name: &str,
    version: &str,
    timeout: std::time::Duration,
) -> Result<u64> {
    use std::time::{Duration, Instant};
    let url = format!("https://crates.io/api/v1/crates/{crate_name}");
    let agent = "nornir-release-bot/0.1 (cargo-pipeline; +https://crates.io)";
    let started = Instant::now();
    loop {
        let out = Command::new("curl")
            .args(["-sSL", "-A", agent, "--max-time", "10", &url])
            .output()
            .context("spawn curl for crates.io poll")?;
        if out.status.success() {
            if let Ok(json) = serde_json::from_slice::<serde_json::Value>(&out.stdout) {
                let max = json.get("crate")
                    .and_then(|c| c.get("max_version"))
                    .and_then(|v| v.as_str());
                if max == Some(version) {
                    return Ok(started.elapsed().as_millis() as u64);
                }
            }
        }
        if started.elapsed() >= timeout {
            return Err(anyhow!(
                "timed out waiting for crates.io index: {crate_name}@{version} after {:?}",
                timeout
            ));
        }
        std::thread::sleep(Duration::from_secs(3));
    }
}

// ─── Feature 5: .crate tarball stats ────────────────────────────────

/// Stats from a `cargo package --list` run. Used by the bloat gate
/// and persisted into the `crate_tarball_stats` iceberg table.
#[derive(Debug, Clone)]
pub struct TarballStats {
    pub crate_name: String,
    pub version: String,
    pub tarball_bytes: u64,
    pub file_count: usize,
    pub largest_file: Option<String>,
    pub largest_file_bytes: Option<u64>,
}

/// Run `cargo package -p <krate> --list --allow-dirty` and inspect
/// the resulting `.crate` tarball in `target/package/`. Does *not*
/// upload anything. Reports file count + size of largest entry so the
/// caller can warn on bloat (typical libraries are <100 kB; >5 MB is
/// almost always test fixtures or planet-scale data leaking in via a
/// missing `exclude = [...]`).
pub fn tarball_stats(repo_root: &Path, krate: &str) -> Result<TarballStats> {
    use std::process::Command;
    let pkg = Command::new("cargo")
        .args(["package", "-p", krate, "--allow-dirty", "--no-verify"])
        .current_dir(repo_root)
        .output()
        .context("spawn cargo package")?;
    if !pkg.status.success() {
        return Err(anyhow!(
            "cargo package -p {krate} exited {}: {}",
            pkg.status,
            String::from_utf8_lossy(&pkg.stderr)
        ));
    }
    let target = repo_root.join("target").join("package");
    // Find the .crate file we just produced; pick the newest matching.
    let mut newest: Option<(std::path::PathBuf, std::time::SystemTime)> = None;
    if let Ok(rd) = std::fs::read_dir(&target) {
        for e in rd.flatten() {
            let p = e.path();
            if p.extension().and_then(|s| s.to_str()) != Some("crate") { continue }
            let fname = p.file_name().and_then(|s| s.to_str()).unwrap_or("");
            if !fname.starts_with(&format!("{krate}-")) { continue }
            let mtime = e.metadata().and_then(|m| m.modified()).unwrap_or(std::time::UNIX_EPOCH);
            if newest.as_ref().map(|(_, t)| mtime > *t).unwrap_or(true) {
                newest = Some((p, mtime));
            }
        }
    }
    let (path, _) = newest.ok_or_else(|| anyhow!("no .crate tarball found in {}", target.display()))?;
    let tarball_bytes = std::fs::metadata(&path)?.len();
    let fname = path.file_name().and_then(|s| s.to_str()).unwrap_or("");
    let version = fname
        .strip_prefix(&format!("{krate}-"))
        .and_then(|s| s.strip_suffix(".crate"))
        .unwrap_or("?")
        .to_string();
    // Walk the tarball to count files + find the largest.
    use flate2::read::GzDecoder;
    use tar::Archive;
    let f = std::fs::File::open(&path)?;
    let dec = GzDecoder::new(f);
    let mut ar = Archive::new(dec);
    let mut file_count = 0usize;
    let mut largest: Option<(String, u64)> = None;
    for entry in ar.entries()? {
        let entry = entry?;
        let size = entry.header().size().unwrap_or(0);
        let p = entry.path()?.into_owned();
        file_count += 1;
        if largest.as_ref().map(|(_, s)| size > *s).unwrap_or(true) {
            largest = Some((p.display().to_string(), size));
        }
    }
    Ok(TarballStats {
        crate_name: krate.to_string(),
        version,
        tarball_bytes,
        file_count,
        largest_file: largest.as_ref().map(|(n, _)| n.clone()),
        largest_file_bytes: largest.map(|(_, s)| s),
    })
}

/// Default bloat threshold: 5 MB. Tweakable per-call.
pub const DEFAULT_TARBALL_BYTES_THRESHOLD: u64 = 5 * 1024 * 1024;

#[cfg(test)]
mod tests {
    use super::*;

    fn m(n: &str, deps: &[&str]) -> (String, Vec<String>) {
        (n.to_string(), deps.iter().map(|s| s.to_string()).collect())
    }

    #[test]
    fn topo_phases_orders_deps_before_dependents_deterministically() {
        // testmatrix ← robotui ← nornir ; airgap ← nornir ; testmatrix ← nornir.
        let members = vec![
            m("nornir", &["nornir-testmatrix", "nornir-robotui", "nornir-airgap"]),
            m("nornir-robotui", &["nornir-testmatrix"]),
            m("nornir-testmatrix", &[]),
            m("nornir-airgap", &[]),
        ];
        let phases = topo_phases(members);
        let flat: Vec<&str> = phases.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c).unwrap();
        // deps strictly before dependents
        assert!(pos("nornir-testmatrix") < pos("nornir-robotui"));
        assert!(pos("nornir-robotui") < pos("nornir"));
        assert!(pos("nornir-airgap") < pos("nornir"));
        assert!(pos("nornir-testmatrix") < pos("nornir"));
        // deterministic: leaves (testmatrix, airgap) share phase 0, alphabetical
        assert_eq!(phases[0], vec!["nornir-airgap", "nornir-testmatrix"]);
        // nornir is last (its own phase)
        assert_eq!(phases.last().unwrap(), &vec!["nornir".to_string()]);
    }

    #[test]
    fn topo_phases_cycle_drops_nothing() {
        let members = vec![m("a", &["b"]), m("b", &["a"])]; // pathological cycle
        let phases = topo_phases(members);
        let flat: std::collections::BTreeSet<&str> =
            phases.iter().flatten().map(String::as_str).collect();
        assert!(flat.contains("a") && flat.contains("b"), "cycle members not dropped");
    }

    #[test]
    fn parse_rate_limit_wait_parses_relative_seconds() {
        // crates.io's real 429 body, relative form.
        let msg = "error: failed to publish to registry at https://crates.io\n\
                   Caused by: the remote server responded with an error (status 429 Too \
                   Many Requests): You have published too many crates in a short period \
                   of time, please try again after 56 seconds, or email help@crates.io";
        let got = parse_rate_limit_wait(msg, std::time::Duration::from_secs(60))
            .expect("recognised as rate limit");
        assert_eq!(got, std::time::Duration::from_secs(56), "parsed the 56s wait");
    }

    #[test]
    fn parse_rate_limit_wait_parses_relative_minutes_and_falls_back() {
        let mins = "status 429 ... too many crates ... please try again in 4 minutes";
        assert_eq!(
            parse_rate_limit_wait(mins, std::time::Duration::from_secs(60)),
            Some(std::time::Duration::from_secs(240)),
            "4 minutes → 240s"
        );
        // Absolute RFC-2822 timestamp form → no relative number → fallback.
        let abs = "too many crates in a short period of time, please try again after \
                   Wed, 18 Jun 2026 12:34:56 GMT";
        assert_eq!(
            parse_rate_limit_wait(abs, std::time::Duration::from_secs(42)),
            Some(std::time::Duration::from_secs(42)),
            "unparseable absolute date falls back to the default wait"
        );
    }

    #[test]
    fn parse_rate_limit_wait_ignores_non_rate_limit_errors() {
        assert_eq!(
            parse_rate_limit_wait(
                "error: failed to verify package tarball\nexpected `foo`",
                std::time::Duration::from_secs(60)
            ),
            None,
            "a normal cargo failure is NOT a rate limit"
        );
    }

    #[test]
    fn parse_rate_limit_wait_caps_at_max() {
        let huge = "429 too many crates, please try again in 600 minutes"; // 10h
        let got = parse_rate_limit_wait(huge, std::time::Duration::from_secs(60)).unwrap();
        assert_eq!(got, MAX_RATE_LIMIT_WAIT, "absurd wait is clamped to the ceiling");
    }

    #[test]
    fn run_cargo_publish_retries_after_rate_limit_then_succeeds() {
        // Inject-assert: feed the real 429 string the FIRST two cargo invocations,
        // then a success, through a fake `cargo` on PATH. Assert the loop slept the
        // PARSED waits (3s, then 3s) and that the final outcome is Published.
        use std::io::Write;
        let dir = std::env::temp_dir().join(format!("nornir-rl-test-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let counter = dir.join("count");
        std::fs::write(&counter, "0").unwrap();
        // Fake cargo: 1st+2nd call print the 429 body + exit 1; 3rd call exits 0.
        let fake = dir.join("cargo");
        let script = format!(
            "#!/bin/sh\n\
             c=$(cat '{cnt}')\n\
             c=$((c+1)); echo $c > '{cnt}'\n\
             if [ \"$c\" -lt 3 ]; then\n\
               echo 'status 429 Too Many Requests: You have published too many crates in a short period of time, please try again after 3 seconds' 1>&2\n\
               exit 1\n\
             fi\n\
             exit 0\n",
            cnt = counter.display()
        );
        {
            let mut f = std::fs::File::create(&fake).unwrap();
            f.write_all(script.as_bytes()).unwrap();
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&fake, std::fs::Permissions::from_mode(0o755)).unwrap();
        }
        let old_path = std::env::var("PATH").unwrap_or_default();
        // SAFETY: single-threaded test; PATH is restored before return.
        unsafe {
            std::env::set_var("PATH", format!("{}:{}", dir.display(), old_path));
        }

        let mut slept: Vec<std::time::Duration> = Vec::new();
        let outcome = run_cargo_publish_with_retry(
            &dir,
            "demo-crate",
            false,
            5,
            std::time::Duration::from_secs(60),
            |d| slept.push(d),
        );

        unsafe {
            std::env::set_var("PATH", old_path);
        }
        let calls: u32 = std::fs::read_to_string(&counter)
            .ok()
            .and_then(|s| s.trim().parse().ok())
            .unwrap_or(0);
        let _ = std::fs::remove_dir_all(&dir);

        let outcome = outcome.expect("publish should succeed after retries");
        assert_eq!(outcome, PublishOutcome::Published, "succeeded on the 3rd attempt");
        assert_eq!(
            slept,
            vec![
                std::time::Duration::from_secs(3),
                std::time::Duration::from_secs(3)
            ],
            "slept the PARSED 3s wait before each of the 2 retries"
        );
        assert_eq!(calls, 3, "cargo was invoked 3× (2 rate-limited + 1 success)");
    }

    #[test]
    fn derive_publish_order_from_this_real_workspace() {
        // Deterministic derivation against nornir's OWN workspace: testmatrix must
        // come before nornir (nornir depends on it), and nornir is the last crate.
        let order = derive_publish_order(std::path::Path::new(".")).expect("derive");
        let flat: Vec<&str> = order.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c);
        assert!(flat.contains(&"nornir"), "nornir is publishable: {flat:?}");
        assert!(flat.contains(&"nornir-testmatrix"), "testmatrix is publishable: {flat:?}");
        assert!(
            pos("nornir-testmatrix") < pos("nornir"),
            "testmatrix publishes before nornir: {flat:?}"
        );
    }
}