#![cfg_attr(docsrs, feature(doc_cfg))]
#![deny(rustdoc::broken_intra_doc_links)]
use std::ffi::OsStr;
use std::fmt;
use std::future::Future;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use processkit::{
CancellationToken, CliClient, Command, Error, IntoCommand, JobRunner, OutputBufferPolicy,
OverflowMode, ProcessResult, ProcessRunner, Result,
};
pub mod credentials;
pub use credentials::{
Credential, CredentialProvider, CredentialRequest, CredentialService, EnvToken, FnProvider,
GitCredentialHelper, Secret, StaticCredential, git_credential_helper, https_host, provider_fn,
};
#[cfg(feature = "serde")]
#[cfg_attr(docsrs, doc(cfg(feature = "serde")))]
pub mod json {
use processkit::{Error, Result};
use serde::Deserialize;
use serde::de::DeserializeOwned;
pub fn null_to_empty<'de, D>(deserializer: D) -> ::core::result::Result<String, D::Error>
where
D: serde::Deserializer<'de>,
{
Ok(Option::<String>::deserialize(deserializer)?.unwrap_or_default())
}
pub fn from_json<T: DeserializeOwned>(program: &str, json: &str) -> Result<T> {
serde_json::from_str(json).map_err(|e| Error::parse(program, e.to_string()))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OutputBudget {
max_bytes: Option<usize>,
max_lines: Option<usize>,
}
impl OutputBudget {
pub const fn unlimited() -> Self {
Self {
max_bytes: None,
max_lines: None,
}
}
pub const fn bytes(max_bytes: usize) -> Self {
Self {
max_bytes: Some(max_bytes),
max_lines: None,
}
}
#[must_use]
pub const fn with_max_lines(mut self, max_lines: usize) -> Self {
self.max_lines = Some(max_lines);
self
}
pub const fn is_unlimited(&self) -> bool {
self.max_bytes.is_none() && self.max_lines.is_none()
}
pub const fn max_bytes(&self) -> Option<usize> {
self.max_bytes
}
pub const fn max_lines(&self) -> Option<usize> {
self.max_lines
}
pub fn content_policy(&self) -> Option<OutputBufferPolicy> {
if self.is_unlimited() {
return None;
}
let mut policy = match self.max_lines {
Some(lines) => OutputBufferPolicy::fail_loud(lines),
None => OutputBufferPolicy::unbounded().with_overflow(OverflowMode::Error),
};
if let Some(bytes) = self.max_bytes {
policy = policy.with_max_bytes(bytes);
}
Some(policy)
}
pub fn diagnostic_policy(&self) -> Option<OutputBufferPolicy> {
if self.is_unlimited() {
return None;
}
let mut policy = match self.max_lines {
Some(lines) => OutputBufferPolicy::bounded(lines),
None => OutputBufferPolicy::unbounded(),
};
if let Some(bytes) = self.max_bytes {
policy = policy.with_max_bytes(bytes);
}
Some(policy)
}
}
impl Default for OutputBudget {
fn default() -> Self {
Self::unlimited()
}
}
#[macro_export]
macro_rules! at_forwarders {
(
$view:ident, $field:ident, $client:literal,
bare { $( fn $bn:ident( $($ba:ident: $bt:ty),* $(,)? ) -> $br:ty; )* }
dir { $( fn $dn:ident( $($da:ident: $dt:ty),* $(,)? ) -> $dr:ty; )* }
$( raw { $( fn $rn:ident( $($ra:ident: $rt:ty),* $(,)? ) -> $rr:ty => $rtgt:ident; )* } )?
) => {
impl<'a, R: ::processkit::ProcessRunner> $view<'a, R> {
$(
#[doc = concat!("Bound form of [`", $client, "`]'s `", stringify!($bn), "`.")]
pub async fn $bn(&self, $($ba: $bt),*) -> $br {
self.$field.$bn($($ba),*).await
}
)*
$(
#[doc = concat!("Bound form of [`", $client, "`]'s `", stringify!($dn), "` (with `dir` pre-bound).")]
pub async fn $dn(&self, $($da: $dt),*) -> $dr {
self.$field.$dn(self.dir, $($da),*).await
}
)*
$($(
#[doc = concat!(
"Bound form of [`", $client, "`]'s `", stringify!($rn),
"` raw escape hatch — runs the given argv **in the bound `dir`** \
(forwards to the client's `", stringify!($rtgt), "`). For the \
process-cwd escape hatch, call `", stringify!($rn),
"` on [`", $client, "`] directly."
)]
pub async fn $rn(&self, $($ra: $rt),*) -> $rr {
self.$field.$rtgt(self.dir, $($ra),*).await
}
)*)?
}
};
}
#[macro_export]
macro_rules! managed_client {
(
$(#[$meta:meta])*
$vis:vis struct $name:ident => $binary:expr
$(, token_env = ($svc:expr, $var:expr) )?
$(, scrub_env = [ $($scrub:expr),* $(,)? ] )?
$(,)?
) => {
$(#[$meta])*
$vis struct $name<R: ::processkit::ProcessRunner = ::processkit::JobRunner> {
core: $crate::ManagedClient<R>,
}
impl<R: ::processkit::ProcessRunner> ::core::fmt::Debug for $name<R> {
fn fmt(&self, f: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
f.debug_struct(stringify!($name))
.field("core", &self.core)
.finish()
}
}
impl $name<::processkit::JobRunner> {
pub fn new() -> Self {
Self { core: $crate::ManagedClient::new($binary)
$(.with_token_env($svc, $var))?
$($(.default_env_remove($scrub))*)?
}
}
}
impl ::core::default::Default for $name<::processkit::JobRunner> {
fn default() -> Self {
Self::new()
}
}
impl<R: ::processkit::ProcessRunner> $name<R> {
pub fn with_runner(runner: R) -> Self {
Self {
core: $crate::ManagedClient::with_runner($binary, runner)
$(.with_token_env($svc, $var))?
$($(.default_env_remove($scrub))*)?,
}
}
pub fn default_timeout(mut self, timeout: ::core::time::Duration) -> Self {
self.core = self.core.default_timeout(timeout);
self
}
pub fn default_env(
mut self,
key: impl ::core::convert::AsRef<::std::ffi::OsStr>,
value: impl ::core::convert::AsRef<::std::ffi::OsStr>,
) -> Self {
self.core = self.core.default_env(key, value);
self
}
pub fn default_env_remove(
mut self,
key: impl ::core::convert::AsRef<::std::ffi::OsStr>,
) -> Self {
self.core = self.core.default_env_remove(key);
self
}
pub fn default_cancel_on(mut self, token: ::processkit::CancellationToken) -> Self {
self.core = self.core.default_cancel_on(token);
self
}
pub fn default_output_budget(mut self, budget: $crate::OutputBudget) -> Self {
self.core = self.core.default_output_budget(budget);
self
}
}
};
}
pub fn reject_flag_like(program: &str, what: &str, value: &str) -> Result<()> {
let trimmed = value.trim();
if trimmed.is_empty() || trimmed.starts_with('-') || value.contains('\0') {
return Err(Error::spawn(
program,
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!(
"{what} {value:?} would be parsed as a flag (or is empty / contains NUL) — \
refusing to pass it as a positional argument"
),
),
));
}
Ok(())
}
pub const FETCH_ATTEMPTS: u32 = 3;
pub const FETCH_BACKOFF: Duration = Duration::from_millis(500);
pub const FETCH_TIMEOUT_GRACE: Duration = Duration::from_secs(2);
const CONFLICT_MARKERS: &[&str] = &["conflict (", "automatic merge failed"];
const NOTHING_TO_COMMIT_MARKERS: &[&str] = &["nothing to commit", "nothing added to commit"];
const TRANSIENT_FETCH_MARKERS: &[&str] = &[
"could not resolve host",
"couldn't resolve host",
"temporary failure in name resolution",
"connection timed out",
"connection refused",
"operation timed out",
"network is unreachable",
"failed to connect",
"could not read from remote repository",
"the remote end hung up",
"early eof",
"rpc failed",
];
fn exit_output_matches(err: &Error, markers: &[&str]) -> bool {
let Error::Exit { stdout, stderr, .. } = err else {
return false;
};
let out = stdout.to_ascii_lowercase();
let errt = stderr.to_ascii_lowercase();
markers.iter().any(|m| out.contains(m) || errt.contains(m))
}
pub fn is_merge_conflict(err: &Error) -> bool {
exit_output_matches(err, CONFLICT_MARKERS)
}
pub fn is_nothing_to_commit(err: &Error) -> bool {
exit_output_matches(err, NOTHING_TO_COMMIT_MARKERS)
}
pub fn is_transient_fetch_error(err: &Error) -> bool {
err.is_transient() || exit_output_matches(err, TRANSIENT_FETCH_MARKERS)
}
const LOCK_CONTENTION_MARKERS: &[&str] = &[
"index.lock",
"failed to lock working copy",
"failed to lock operation heads store",
];
pub fn is_lock_contention(err: &Error) -> bool {
if exit_output_matches(err, &["refs/"]) {
return false;
}
exit_output_matches(err, LOCK_CONTENTION_MARKERS)
}
pub fn is_invalid_input(err: &Error) -> bool {
matches!(
err,
Error::Spawn { source, .. } if source.kind() == std::io::ErrorKind::InvalidInput
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct RetryPolicy {
pub attempts: u32,
pub base_backoff: Duration,
pub max_backoff: Duration,
pub jitter: bool,
}
impl RetryPolicy {
pub const fn none() -> Self {
Self {
attempts: 1,
base_backoff: Duration::ZERO,
max_backoff: Duration::ZERO,
jitter: false,
}
}
pub const fn lock_contention() -> Self {
Self {
attempts: 5,
base_backoff: Duration::from_millis(25),
max_backoff: Duration::from_millis(500),
jitter: true,
}
}
pub fn attempts(mut self, attempts: u32) -> Self {
self.attempts = attempts.max(1);
self
}
pub fn base_backoff(mut self, backoff: Duration) -> Self {
self.base_backoff = backoff;
self
}
pub fn max_backoff(mut self, max: Duration) -> Self {
self.max_backoff = max;
self
}
pub fn with_jitter(mut self, jitter: bool) -> Self {
self.jitter = jitter;
self
}
}
impl Default for RetryPolicy {
fn default() -> Self {
Self::none()
}
}
fn backoff_for(policy: &RetryPolicy, retry_index: u32) -> Duration {
if policy.base_backoff.is_zero() {
return Duration::ZERO;
}
let base = policy.base_backoff.as_nanos();
let scaled = base.saturating_mul(1u128 << retry_index.min(20));
let capped = if policy.max_backoff.is_zero() {
scaled
} else {
scaled.min(policy.max_backoff.as_nanos())
};
let delay = Duration::from_nanos(capped.min(u64::MAX as u128) as u64);
if policy.jitter {
full_jitter(delay)
} else {
delay
}
}
fn full_jitter(max: Duration) -> Duration {
use std::hash::{BuildHasher, Hasher};
let nanos = max.as_nanos();
if nanos == 0 {
return Duration::ZERO;
}
let mut hasher = std::collections::hash_map::RandomState::new().build_hasher();
hasher.write_u64(nanos as u64);
let r = hasher.finish() as u128;
Duration::from_nanos((r % (nanos + 1)).min(u64::MAX as u128) as u64)
}
fn cancelled_error(last_err: &Error) -> Error {
Error::Cancelled {
program: last_err.program().unwrap_or_default().to_owned(),
}
}
pub async fn retry_async<T, Fut>(
policy: &RetryPolicy,
cancel: Option<&CancellationToken>,
should_retry: impl Fn(&Error) -> bool,
mut op: impl FnMut() -> Fut,
) -> Result<T>
where
Fut: Future<Output = Result<T>>,
{
let attempts = policy.attempts.max(1);
for attempt in 1..=attempts {
match op().await {
Ok(value) => return Ok(value),
Err(err) => {
if attempt == attempts || !should_retry(&err) {
return Err(err);
}
let delay = backoff_for(policy, attempt - 1);
match cancel {
Some(token) => {
if !delay.is_zero() {
let _ = token.run_until_cancelled(tokio::time::sleep(delay)).await;
}
if token.is_cancelled() {
return Err(cancelled_error(&err));
}
}
None => {
if !delay.is_zero() {
tokio::time::sleep(delay).await;
}
}
}
}
}
}
unreachable!("the loop returns on the final attempt")
}
pub struct ManagedClient<R: ProcessRunner = JobRunner> {
inner: CliClient<R>,
retry: RetryPolicy,
credentials: Option<Arc<dyn CredentialProvider>>,
token_env: Option<(CredentialService, &'static str)>,
expected_host: Option<String>,
cancel: Option<CancellationToken>,
output_budget: OutputBudget,
}
impl<R: ProcessRunner> fmt::Debug for ManagedClient<R> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ManagedClient")
.field("inner", &self.inner)
.field("retry", &self.retry)
.field("credentials", &self.credentials.is_some())
.field("token_env", &self.token_env)
.field("expected_host", &self.expected_host)
.field("has_cancel", &self.cancel.is_some())
.field("output_budget", &self.output_budget)
.finish()
}
}
impl ManagedClient<JobRunner> {
pub fn new(program: impl AsRef<OsStr>) -> Self {
Self {
inner: CliClient::new(program),
retry: RetryPolicy::none(),
credentials: None,
token_env: None,
expected_host: None,
cancel: None,
output_budget: OutputBudget::unlimited(),
}
}
}
impl<R: ProcessRunner> ManagedClient<R> {
pub fn with_runner(program: impl AsRef<OsStr>, runner: R) -> Self {
Self {
inner: CliClient::with_runner(program, runner),
retry: RetryPolicy::none(),
credentials: None,
token_env: None,
expected_host: None,
cancel: None,
output_budget: OutputBudget::unlimited(),
}
}
pub fn with_retry(mut self, policy: RetryPolicy) -> Self {
self.retry = policy;
self
}
pub fn retry_policy(&self) -> RetryPolicy {
self.retry
}
#[must_use]
pub fn with_credentials(mut self, provider: Arc<dyn CredentialProvider>) -> Self {
self.credentials = Some(provider);
self
}
#[must_use]
pub fn with_token_env(mut self, service: CredentialService, var: &'static str) -> Self {
self.token_env = Some((service, var));
self
}
#[must_use]
pub fn with_expected_host(mut self, host: impl Into<String>) -> Self {
self.expected_host = Some(host.into());
self
}
#[must_use]
pub fn has_credentials(&self) -> bool {
self.credentials.is_some()
}
pub async fn resolve_credential(
&self,
service: CredentialService,
host: Option<&str>,
) -> Result<Option<Credential>> {
let Some(provider) = &self.credentials else {
return Ok(None);
};
let request = CredentialRequest { service, host };
Ok(provider
.credential(&request)
.await?
.filter(|cred| !cred.secret().expose().trim().is_empty()))
}
async fn prepare(&self, call: impl IntoCommand<R>) -> Result<Command> {
let cmd = call.into_command(&self.inner);
let Some((service, var)) = self.token_env else {
return Ok(cmd);
};
match self
.resolve_credential(service, self.expected_host.as_deref())
.await?
{
Some(cred) => Ok(cmd.env(var, cred.secret().expose())),
None => Ok(cmd),
}
}
pub fn default_timeout(mut self, timeout: Duration) -> Self {
self.inner = self.inner.default_timeout(timeout);
self
}
pub fn default_env(mut self, key: impl AsRef<OsStr>, value: impl AsRef<OsStr>) -> Self {
self.inner = self.inner.default_env(key, value);
self
}
pub fn default_env_remove(mut self, key: impl AsRef<OsStr>) -> Self {
self.inner = self.inner.default_env_remove(key);
self
}
pub fn default_cancel_on(mut self, token: CancellationToken) -> Self {
self.inner = self.inner.default_cancel_on(token.clone());
self.cancel = Some(token);
self
}
pub fn default_output_budget(mut self, budget: OutputBudget) -> Self {
self.output_budget = budget;
self
}
pub fn output_budget(&self) -> OutputBudget {
self.output_budget
}
pub fn budget_diagnostics(&self, cmd: Command) -> Command {
match self.output_budget.diagnostic_policy() {
Some(policy) => cmd.output_buffer(policy),
None => cmd,
}
}
pub fn command<I, S>(&self, args: I) -> Command
where
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
{
self.inner.command(args)
}
pub fn command_in<I, S>(&self, dir: &Path, args: I) -> Command
where
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
{
self.inner.command_in(dir, args)
}
pub fn runner(&self) -> &R {
self.inner.runner()
}
pub async fn run(&self, call: impl IntoCommand<R>) -> Result<String> {
let cmd = self.prepare(call).await?;
retry_async(
&self.retry,
self.cancel.as_ref(),
is_lock_contention,
|| self.inner.run(cmd.clone()),
)
.await
}
pub async fn run_unit(&self, call: impl IntoCommand<R>) -> Result<()> {
let cmd = self.prepare(call).await?;
retry_async(
&self.retry,
self.cancel.as_ref(),
is_lock_contention,
|| self.inner.run_unit(cmd.clone()),
)
.await
}
pub async fn output_string(&self, call: impl IntoCommand<R>) -> Result<ProcessResult<String>> {
let cmd = self.prepare(call).await?;
self.inner.output_string(cmd).await
}
pub async fn output_bytes(&self, call: impl IntoCommand<R>) -> Result<ProcessResult<Vec<u8>>> {
let cmd = self.prepare(call).await?;
self.inner.output_bytes(cmd).await
}
pub async fn run_untrimmed(&self, call: impl IntoCommand<R>) -> Result<String> {
self.run_untrimmed_within(call, self.output_budget).await
}
pub async fn run_untrimmed_within(
&self,
call: impl IntoCommand<R>,
budget: OutputBudget,
) -> Result<String> {
let cmd = self.prepare(call).await?;
let cmd = match budget.content_policy() {
Some(policy) => cmd.output_buffer(policy),
None => cmd,
};
let bytes = self
.inner
.output_bytes(cmd)
.await?
.ensure_success()?
.into_stdout();
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
pub async fn probe(&self, call: impl IntoCommand<R>) -> Result<bool> {
let cmd = self.prepare(call).await?;
retry_async(
&self.retry,
self.cancel.as_ref(),
is_lock_contention,
|| self.inner.probe(cmd.clone()),
)
.await
}
pub async fn exit_code(&self, call: impl IntoCommand<R>) -> Result<i32> {
let cmd = self.prepare(call).await?;
retry_async(
&self.retry,
self.cancel.as_ref(),
is_lock_contention,
|| self.inner.exit_code(cmd.clone()),
)
.await
}
pub async fn parse<T>(
&self,
call: impl IntoCommand<R>,
parser: impl FnOnce(&str) -> T + Send,
) -> Result<T>
where
T: Send,
{
let cmd = self.prepare(call).await?;
self.inner.parse(cmd, parser).await
}
pub async fn parse_bytes<T>(
&self,
call: impl IntoCommand<R>,
parser: impl FnOnce(&[u8]) -> T + Send,
) -> Result<T>
where
T: Send,
{
let cmd = self.prepare(call).await?;
let bytes = self
.inner
.output_bytes(cmd)
.await?
.ensure_success()?
.into_stdout();
Ok(parser(&bytes))
}
pub async fn try_parse<T>(
&self,
call: impl IntoCommand<R>,
parser: impl FnOnce(&str) -> Result<T> + Send,
) -> Result<T>
where
T: Send,
{
let cmd = self.prepare(call).await?;
self.inner.try_parse(cmd, parser).await
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rejects_empty_and_leading_dash() {
assert!(reject_flag_like("git", "branch name", "-evil").is_err());
assert!(reject_flag_like("git", "branch name", "").is_err());
assert!(reject_flag_like("git", "branch name", " ").is_err());
assert!(reject_flag_like("git", "branch name", "\t").is_err());
assert!(reject_flag_like("git", "branch name", "feature").is_ok());
assert!(reject_flag_like("git", "remote", " --upload-pack=evil").is_err());
assert!(reject_flag_like("git", "remote", "\t-x").is_err());
assert!(reject_flag_like("git", "path", "a\0b").is_err());
assert!(reject_flag_like("git", "branch name", " feature").is_ok());
let err = reject_flag_like("jj", "revset", "--remote").unwrap_err();
assert!(matches!(err, Error::Spawn { program, .. } if program == "jj"));
}
#[test]
fn classifies_merge_conflict() {
let on_stdout = Error::exit("git", 1, "CONFLICT (content): Merge conflict in a.rs", "");
let on_stderr = Error::exit(
"git",
1,
"",
"Automatic merge failed; fix conflicts and then commit",
);
let unrelated = Error::exit("git", 128, "", "fatal: not a git repository");
assert!(is_merge_conflict(&on_stdout));
assert!(is_merge_conflict(&on_stderr));
assert!(!is_merge_conflict(&unrelated));
assert!(!is_nothing_to_commit(&on_stdout));
}
#[test]
fn classifies_nothing_to_commit_and_transient_fetch() {
let nothing = Error::exit("git", 1, "nothing to commit, working tree clean", "");
assert!(is_nothing_to_commit(¬hing));
let dns = Error::exit(
"git",
128,
"",
"fatal: unable to access 'https://x/': Could not resolve host: x",
);
assert!(is_transient_fetch_error(&dns));
assert!(!is_transient_fetch_error(¬hing));
let timeout = Error::timeout("git", Duration::from_secs(10), "", "");
assert!(!is_transient_fetch_error(&timeout));
}
#[test]
fn classifies_io_transient_as_fetch_retryable() {
let interrupted =
Error::spawn("git", std::io::Error::from(std::io::ErrorKind::Interrupted));
assert!(
interrupted.is_transient(),
"processkit treats Interrupted as a transient io error"
);
assert!(is_transient_fetch_error(&interrupted));
let missing = Error::spawn("git", std::io::Error::from(std::io::ErrorKind::NotFound));
assert!(!is_transient_fetch_error(&missing));
}
#[test]
fn classifies_on_large_output_past_the_old_4kib_cap() {
let padding = "noise line that says nothing\n".repeat(500); let conflict = Error::exit(
"git",
1,
format!("{padding}CONFLICT (content): Merge conflict in late.rs"),
"",
);
assert!(
is_merge_conflict(&conflict),
"a conflict marker past 4 KiB must still classify"
);
let transient = Error::exit(
"git",
128,
"",
format!("{padding}fatal: unable to access: Could not resolve host: x"),
);
assert!(is_transient_fetch_error(&transient));
}
#[test]
fn unfamiliar_error_variants_are_not_classified() {
let not_ready = Error::NotReady {
program: "git".into(),
timeout: Duration::from_secs(5),
};
let unsupported = Error::Unsupported {
operation: "suspend".into(),
};
for err in [¬_ready, &unsupported] {
assert!(!is_merge_conflict(err));
assert!(!is_nothing_to_commit(err));
assert!(!is_transient_fetch_error(err));
}
}
#[test]
fn cancelled_is_not_transient_or_otherwise_classified() {
let cancelled = Error::Cancelled {
program: "git".into(),
};
assert!(!is_transient_fetch_error(&cancelled));
assert!(!is_merge_conflict(&cancelled));
assert!(!is_nothing_to_commit(&cancelled));
}
#[test]
fn signalled_is_terminal_not_transient() {
let signalled = Error::signalled(
"git",
Some(15),
"",
"fatal: unable to access: Could not resolve host: x",
);
assert!(!signalled.is_transient());
assert!(!is_transient_fetch_error(&signalled));
assert!(!is_merge_conflict(&signalled));
assert!(!is_nothing_to_commit(&signalled));
}
fn exit(program: &str, code: i32, stderr: &str) -> Error {
Error::exit(program, code, "", stderr)
}
#[test]
fn classifies_lock_contention() {
let lock_failures = [
exit(
"git",
128,
"fatal: Unable to create '/r/.git/index.lock': File exists.",
),
exit(
"git",
128,
"fatal: Konnte '/r/.git/index.lock' nicht erstellen: Datei existiert bereits",
),
exit("jj", 1, "Error: Failed to lock working copy"),
exit("jj", 1, "Error: Failed to lock operation heads store"),
];
for e in &lock_failures {
assert!(is_lock_contention(e), "should be lock contention: {e:?}");
assert!(!is_transient_fetch_error(e), "not a fetch error: {e:?}");
}
let not_locks = [
exit("git", 1, "CONFLICT (content): Merge conflict in a.rs"),
exit("git", 1, "error: pathspec 'x' did not match any file(s)"),
exit("git", 128, "fatal: not a git repository"),
exit(
"git",
1,
"error: cannot lock ref 'refs/heads/x': reference already exists",
),
exit(
"git",
128,
"Unable to create '/r/.git/packed-refs.lock': File exists.",
),
exit(
"git",
128,
"error: cannot lock ref 'refs/heads/index': Unable to create \
'/r/.git/refs/heads/index.lock': File exists.",
),
Error::timeout("git", Duration::from_secs(1), "", ""),
];
for e in ¬_locks {
assert!(
!is_lock_contention(e),
"should NOT be lock contention: {e:?}"
);
}
}
#[test]
fn classifies_invalid_input_from_the_guards() {
let rejected = reject_flag_like("git", "reference", "-x").unwrap_err();
assert!(
is_invalid_input(&rejected),
"guard rejection is invalid input"
);
assert!(is_invalid_input(
&reject_flag_like("git", "x", "").unwrap_err()
));
let not_input = [
Error::spawn("git", std::io::Error::from(std::io::ErrorKind::NotFound)),
exit("git", 1, "fatal: not a git repository"),
Error::timeout("git", Duration::from_secs(1), "", ""),
];
for e in ¬_input {
assert!(!is_invalid_input(e), "should NOT be invalid input: {e:?}");
}
}
#[test]
fn backoff_is_exponential_capped_and_zero_without_base() {
let p = RetryPolicy::none()
.attempts(6)
.base_backoff(Duration::from_millis(10))
.max_backoff(Duration::from_millis(80));
assert_eq!(backoff_for(&p, 0), Duration::from_millis(10));
assert_eq!(backoff_for(&p, 1), Duration::from_millis(20));
assert_eq!(backoff_for(&p, 2), Duration::from_millis(40));
assert_eq!(backoff_for(&p, 3), Duration::from_millis(80));
assert_eq!(
backoff_for(&p, 4),
Duration::from_millis(80),
"capped at max"
);
assert_eq!(
backoff_for(&RetryPolicy::none(), 3),
Duration::ZERO,
"no base → no wait"
);
}
#[test]
fn jitter_stays_within_cap_and_decorrelates() {
let p = RetryPolicy::none()
.attempts(8)
.base_backoff(Duration::from_millis(10))
.max_backoff(Duration::from_millis(80))
.with_jitter(true);
let cap = Duration::from_millis(80);
let mut seen = std::collections::HashSet::new();
for _ in 0..1000 {
let d = backoff_for(&p, 3);
assert!(
d <= cap,
"jittered backoff {d:?} must stay within the cap {cap:?}"
);
seen.insert(d.as_nanos());
}
assert!(
seen.len() > 1,
"full jitter must produce a spread of delays, not a constant"
);
assert_eq!(
backoff_for(&RetryPolicy::none().with_jitter(true), 2),
Duration::ZERO
);
}
#[tokio::test]
async fn retry_async_retries_then_succeeds_and_respects_the_predicate() {
use std::sync::atomic::{AtomicU32, Ordering};
let policy = RetryPolicy::none().attempts(4);
let lock = || {
exit(
"git",
128,
"Unable to create '/r/.git/index.lock': File exists.",
)
};
let calls = AtomicU32::new(0);
let out: Result<u32> = retry_async(&policy, None, is_lock_contention, || {
let n = calls.fetch_add(1, Ordering::SeqCst);
let lock = lock();
async move { if n < 2 { Err(lock) } else { Ok(n) } }
})
.await;
assert_eq!(out.unwrap(), 2);
assert_eq!(calls.load(Ordering::SeqCst), 3, "1 try + 2 retries");
let calls = AtomicU32::new(0);
let out: Result<u32> = retry_async(&policy, None, is_lock_contention, || {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(exit("git", 1, "real, deterministic failure")) }
})
.await;
assert!(out.is_err());
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"non-retryable → single attempt"
);
let calls = AtomicU32::new(0);
let out: Result<u32> = retry_async(&policy, None, is_lock_contention, || {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(exit("git", 128, "index.lock': File exists")) }
})
.await;
assert!(out.is_err());
assert_eq!(calls.load(Ordering::SeqCst), 4, "all attempts used");
}
fn lock_err() -> Error {
exit(
"git",
128,
"Unable to create '/r/.git/index.lock': File exists.",
)
}
#[tokio::test(start_paused = true)]
async fn cancel_before_backoff_aborts_without_waiting_or_retrying() {
use std::sync::atomic::{AtomicU32, Ordering};
let token = CancellationToken::new();
token.cancel(); let policy = RetryPolicy::none()
.attempts(5)
.base_backoff(Duration::from_secs(3600)); let calls = AtomicU32::new(0);
let start = tokio::time::Instant::now();
let out: Result<u32> = retry_async(&policy, Some(&token), is_lock_contention, || {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(lock_err()) }
})
.await;
assert!(
matches!(out, Err(Error::Cancelled { ref program }) if program == "git"),
"a fired token aborts with a program-named Cancelled, got {out:?}"
);
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"one attempt ran; the cancel launched no retry"
);
assert_eq!(
start.elapsed(),
Duration::ZERO,
"the backoff was cut short — no virtual time elapsed"
);
}
#[tokio::test(start_paused = true)]
async fn cancel_during_backoff_wakes_early_and_does_not_retry() {
use std::sync::atomic::{AtomicU32, Ordering};
let token = CancellationToken::new();
let policy = RetryPolicy::none()
.attempts(5)
.base_backoff(Duration::from_secs(3600)); let calls = AtomicU32::new(0);
let start = tokio::time::Instant::now();
let out: Result<u32> = retry_async(&policy, Some(&token), is_lock_contention, || {
let n = calls.fetch_add(1, Ordering::SeqCst);
let token = token.clone();
async move {
if n == 0 {
tokio::spawn(async move { token.cancel() });
}
Err(lock_err())
}
})
.await;
assert!(
matches!(out, Err(Error::Cancelled { ref program }) if program == "git"),
"a cancel during the sleep aborts with Cancelled, got {out:?}"
);
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"cancel woke the sleep early — no second attempt"
);
assert_eq!(
start.elapsed(),
Duration::ZERO,
"woke on the cancel, not after the 1 h delay"
);
}
#[tokio::test(start_paused = true)]
async fn cancel_right_before_next_attempt_aborts() {
use std::sync::atomic::{AtomicU32, Ordering};
let token = CancellationToken::new();
let policy = RetryPolicy::none().attempts(5); let calls = AtomicU32::new(0);
let out: Result<u32> = retry_async(&policy, Some(&token), is_lock_contention, || {
let n = calls.fetch_add(1, Ordering::SeqCst);
let token = token.clone();
async move {
if n == 0 {
token.cancel();
}
Err(lock_err())
}
})
.await;
assert!(
matches!(out, Err(Error::Cancelled { ref program }) if program == "git"),
"a cancel observed before the next attempt aborts with Cancelled, got {out:?}"
);
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"the guard stopped attempt #2 from launching"
);
}
#[tokio::test]
async fn no_token_backoff_is_unchanged() {
use std::sync::atomic::{AtomicU32, Ordering};
let policy = RetryPolicy::none().attempts(3); let calls = AtomicU32::new(0);
let out: Result<u32> = retry_async(&policy, None, is_lock_contention, || {
calls.fetch_add(1, Ordering::SeqCst);
async { Err(lock_err()) }
})
.await;
assert!(
matches!(out, Err(Error::Exit { .. })),
"last error is the lock exit, not Cancelled"
);
assert_eq!(
calls.load(Ordering::SeqCst),
3,
"all attempts used with no token"
);
}
#[tokio::test]
async fn retrying_client_resolves_credential_opt_in() {
let client = ManagedClient::new("git");
assert!(!client.has_credentials());
assert!(
client
.resolve_credential(CredentialService::Git, None)
.await
.unwrap()
.is_none(),
"no provider → ambient (None)"
);
let client = client.with_credentials(Arc::new(StaticCredential::token("t0k")));
assert!(client.has_credentials());
let got = client
.resolve_credential(CredentialService::Git, None)
.await
.unwrap()
.expect("provider yields a credential");
assert_eq!(got.secret().expose(), "t0k");
}
#[tokio::test]
async fn resolve_credential_treats_empty_secret_as_ambient() {
for blank in ["", " ", "\t\n"] {
let client = ManagedClient::new("git")
.with_credentials(Arc::new(StaticCredential::token(blank)));
for service in [CredentialService::GitHub, CredentialService::Git] {
assert!(
client
.resolve_credential(service, None)
.await
.unwrap()
.is_none(),
"blank secret {blank:?} → ambient (None) for {service:?}"
);
}
}
}
#[tokio::test]
async fn resolve_credential_routes_on_request_host() {
let provider = provider_fn(|r: &CredentialRequest<'_>| {
Ok(match r.host {
Some("github.com") => Some(Credential::token("saas")),
Some("ghe.example.com") => Some(Credential::token("ent")),
_ => None,
})
});
let client = ManagedClient::new("gh").with_credentials(Arc::new(provider));
let resolve =
|host: Option<&'static str>| client.resolve_credential(CredentialService::GitHub, host);
assert_eq!(
resolve(Some("github.com"))
.await
.unwrap()
.unwrap()
.secret()
.expose(),
"saas"
);
assert_eq!(
resolve(Some("ghe.example.com"))
.await
.unwrap()
.unwrap()
.secret()
.expose(),
"ent"
);
assert!(
resolve(Some("other.example")).await.unwrap().is_none(),
"a host the provider doesn't place → ambient (None), not a wrong secret"
);
assert!(
resolve(None).await.unwrap().is_none(),
"an absent host → ambient (None)"
);
}
#[tokio::test]
async fn resolve_credential_propagates_provider_error_fail_closed() {
let provider = provider_fn(|_r: &CredentialRequest<'_>| {
Err(Error::spawn(
"vault",
std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "vault unreachable"),
))
});
let client = ManagedClient::new("gh").with_credentials(Arc::new(provider));
for host in [Some("github.com"), None] {
assert!(
client
.resolve_credential(CredentialService::GitHub, host)
.await
.is_err(),
"provider error must propagate (fail-closed), host={host:?}"
);
}
}
#[test]
fn output_budget_default_is_unlimited() {
let b = OutputBudget::default();
assert!(b.is_unlimited());
assert_eq!(b, OutputBudget::unlimited());
assert_eq!(b.max_bytes(), None);
assert_eq!(b.max_lines(), None);
assert!(b.content_policy().is_none());
assert!(b.diagnostic_policy().is_none());
}
#[test]
fn output_budget_bytes_projects_to_both_policies() {
let b = OutputBudget::bytes(4096);
assert!(!b.is_unlimited());
assert_eq!(b.max_bytes(), Some(4096));
let content = b
.content_policy()
.expect("a byte budget yields a content policy");
assert_eq!(
content.overflow,
OverflowMode::Error,
"content is fail-loud"
);
assert_eq!(content.max_bytes, Some(4096));
assert_eq!(content.max_lines, None);
let diag = b
.diagnostic_policy()
.expect("a byte budget yields a diagnostic policy");
assert_eq!(
diag.overflow,
OverflowMode::DropOldest,
"diagnostics keep the tail, never OutputTooLarge"
);
assert_eq!(diag.max_bytes, Some(4096));
}
#[test]
fn output_budget_with_max_lines_composes() {
let b = OutputBudget::bytes(4096).with_max_lines(200);
assert_eq!(b.max_lines(), Some(200));
let content = b.content_policy().unwrap();
assert_eq!(content.max_lines, Some(200));
assert_eq!(content.max_bytes, Some(4096));
assert_eq!(content.overflow, OverflowMode::Error);
let diag = b.diagnostic_policy().unwrap();
assert_eq!(diag.max_lines, Some(200));
assert_eq!(diag.max_bytes, Some(4096));
assert_eq!(diag.overflow, OverflowMode::DropOldest);
}
#[test]
fn managed_client_default_output_budget_round_trips() {
let client = ManagedClient::new("git");
assert!(client.output_budget().is_unlimited());
let client = client.default_output_budget(OutputBudget::bytes(1 << 20));
assert_eq!(client.output_budget(), OutputBudget::bytes(1 << 20));
}
}