use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use procpilot::{Cmd, RetryPolicy, RunError, RunOutput};
pub fn run_jj(repo_path: &Path, args: &[&str]) -> Result<RunOutput, RunError> {
Cmd::new("jj").in_dir(repo_path).args(args).run()
}
pub fn run_git(repo_path: &Path, args: &[&str]) -> Result<RunOutput, RunError> {
Cmd::new("git").in_dir(repo_path).args(args).run()
}
pub fn run_jj_utf8(repo_path: &Path, args: &[&str]) -> Result<String, RunError> {
let out = run_jj(repo_path, args)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_jj_utf8_ignore_wc(repo_path: &Path, args: &[&str]) -> Result<String, RunError> {
let mut full = Vec::with_capacity(args.len() + 1);
full.push("--ignore-working-copy");
full.extend_from_slice(args);
run_jj_utf8(repo_path, &full)
}
pub fn run_git_utf8(repo_path: &Path, args: &[&str]) -> Result<String, RunError> {
let out = run_git(repo_path, args)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_jj_utf8_with_timeout(
repo_path: &Path,
args: &[&str],
timeout: Duration,
) -> Result<String, RunError> {
let out = run_jj_with_timeout(repo_path, args, timeout)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_git_utf8_with_timeout(
repo_path: &Path,
args: &[&str],
timeout: Duration,
) -> Result<String, RunError> {
let out = run_git_with_timeout(repo_path, args, timeout)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_jj_utf8_with_retry(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
) -> Result<String, RunError> {
let out = run_jj_with_retry(repo_path, args, is_transient)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_git_utf8_with_retry(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
) -> Result<String, RunError> {
let out = run_git_with_retry(repo_path, args, is_transient)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_jj_with_timeout(
repo_path: &Path,
args: &[&str],
timeout: Duration,
) -> Result<RunOutput, RunError> {
Cmd::new("jj")
.in_dir(repo_path)
.args(args)
.timeout(timeout)
.run()
}
pub fn run_git_with_timeout(
repo_path: &Path,
args: &[&str],
timeout: Duration,
) -> Result<RunOutput, RunError> {
Cmd::new("git")
.in_dir(repo_path)
.args(args)
.timeout(timeout)
.run()
}
pub fn run_jj_with_retry(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
) -> Result<RunOutput, RunError> {
Cmd::new("jj")
.in_dir(repo_path)
.args(args)
.retry(RetryPolicy::default().when(is_transient))
.run()
}
pub fn run_git_with_retry(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
) -> Result<RunOutput, RunError> {
Cmd::new("git")
.in_dir(repo_path)
.args(args)
.retry(RetryPolicy::default().when(is_transient))
.run()
}
pub fn run_jj_cancellable(
repo_path: &Path,
args: &[&str],
cancel: Arc<AtomicBool>,
) -> Result<RunOutput, RunError> {
Cmd::new("jj")
.in_dir(repo_path)
.args(args)
.cancel(cancel)
.run()
}
pub fn run_git_cancellable(
repo_path: &Path,
args: &[&str],
cancel: Arc<AtomicBool>,
) -> Result<RunOutput, RunError> {
Cmd::new("git")
.in_dir(repo_path)
.args(args)
.cancel(cancel)
.run()
}
pub fn run_jj_utf8_cancellable(
repo_path: &Path,
args: &[&str],
cancel: Arc<AtomicBool>,
) -> Result<String, RunError> {
let out = run_jj_cancellable(repo_path, args, cancel)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_git_utf8_cancellable(
repo_path: &Path,
args: &[&str],
cancel: Arc<AtomicBool>,
) -> Result<String, RunError> {
let out = run_git_cancellable(repo_path, args, cancel)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_jj_with_retry_cancellable(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
cancel: Arc<AtomicBool>,
) -> Result<RunOutput, RunError> {
Cmd::new("jj")
.in_dir(repo_path)
.args(args)
.retry(RetryPolicy::default().when(is_transient))
.cancel(cancel)
.run()
}
pub fn run_git_with_retry_cancellable(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
cancel: Arc<AtomicBool>,
) -> Result<RunOutput, RunError> {
Cmd::new("git")
.in_dir(repo_path)
.args(args)
.retry(RetryPolicy::default().when(is_transient))
.cancel(cancel)
.run()
}
pub fn run_jj_utf8_with_retry_cancellable(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
cancel: Arc<AtomicBool>,
) -> Result<String, RunError> {
let out = run_jj_with_retry_cancellable(repo_path, args, is_transient, cancel)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn run_git_utf8_with_retry_cancellable(
repo_path: &Path,
args: &[&str],
is_transient: impl Fn(&RunError) -> bool + Send + Sync + 'static,
cancel: Arc<AtomicBool>,
) -> Result<String, RunError> {
let out = run_git_with_retry_cancellable(repo_path, args, is_transient, cancel)?;
Ok(out.stdout_lossy().trim().to_string())
}
pub fn jj_merge_base(
repo_path: &Path,
a: &str,
b: &str,
) -> Result<Option<String>, RunError> {
let revset = format!("latest(::({a}) & ::({b}))");
let id = run_jj_utf8(
repo_path,
&[
"log", "-r", &revset, "--no-graph", "--limit", "1", "-T", "commit_id",
],
)?;
Ok(if id.is_empty() { None } else { Some(id) })
}
pub fn jj_current_operation_id(repo_path: &Path) -> Result<String, RunError> {
run_jj_utf8_ignore_wc(repo_path, &["op", "log", "-n1", "--no-graph", "-T", "id"])
}
pub fn jj_operation_log(
repo_path: &Path,
limit: usize,
) -> Result<Vec<crate::JjOperation>, RunError> {
const TEMPLATE: &str = r#"id ++ "\t" ++ description.first_line() ++ "\n""#;
let limit_str = limit.to_string();
let mut args: Vec<&str> = vec!["op", "log", "--no-graph", "-T", TEMPLATE];
if limit > 0 {
args.splice(2..2, ["-n", limit_str.as_str()]);
}
let out = run_jj_utf8_ignore_wc(repo_path, &args)?;
Ok(out
.lines()
.filter_map(|line| {
line.split_once('\t').map(|(id, desc)| crate::JjOperation {
id: id.to_string(),
description: desc.to_string(),
})
})
.collect())
}
pub fn jj_divergent_change_ids(repo_path: &Path) -> Result<Vec<String>, RunError> {
let out = run_jj_utf8_ignore_wc(
repo_path,
&["log", "-r", "divergent()", "--no-graph", "-T", r#"change_id ++ "\n""#],
)?;
let mut ids: Vec<String> = out.lines().filter(|l| !l.is_empty()).map(str::to_string).collect();
ids.sort();
ids.dedup();
Ok(ids)
}
pub fn jj_is_divergent_at_operation(repo_path: &Path, op_id: &str) -> Result<bool, RunError> {
let out = run_jj_utf8_ignore_wc(
repo_path,
&[
"--at-operation", op_id,
"log", "-r", "divergent()", "--no-graph", "-T", r#"change_id ++ "\n""#,
],
)?;
Ok(out.lines().any(|l| !l.is_empty()))
}
pub fn jj_op_restore(repo_path: &Path, op_id: &str) -> Result<(), RunError> {
run_jj(repo_path, &["op", "restore", op_id])?;
Ok(())
}
pub fn git_merge_base(
repo_path: &Path,
a: &str,
b: &str,
) -> Result<Option<String>, RunError> {
match run_git_utf8(repo_path, &["merge-base", a, b]) {
Ok(id) => Ok(if id.is_empty() { None } else { Some(id) }),
Err(RunError::NonZeroExit { status, .. }) if status.code() == Some(1) => Ok(None),
Err(e) => Err(e),
}
}
pub fn is_transient_error(err: &RunError) -> bool {
procpilot::default_transient(err)
}
#[cfg(test)]
mod tests {
use super::*;
fn jj_installed() -> bool {
procpilot::binary_available("jj")
}
fn git_installed() -> bool {
procpilot::binary_available("git")
}
#[test]
fn is_transient_matches_stale_and_lock() {
#[cfg(unix)]
let status = {
use std::os::unix::process::ExitStatusExt;
std::process::ExitStatus::from_raw(256)
};
#[cfg(windows)]
let status = {
use std::os::windows::process::ExitStatusExt;
std::process::ExitStatus::from_raw(1)
};
let err = RunError::NonZeroExit {
command: Cmd::new("jj").display(),
status,
stdout: vec![],
stderr: "The working copy is stale".into(),
attempts: 1,
};
assert!(is_transient_error(&err));
}
#[test]
fn run_jj_fails_gracefully_when_not_installed() {
if jj_installed() {
return;
}
let tmp = tempfile::tempdir().expect("tempdir");
let err = run_jj(tmp.path(), &["status"]).expect_err("jj not installed");
assert!(err.is_spawn_failure());
}
#[test]
fn run_jj_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_jj_cancellable(tmp.path(), &["status"], cancel)
.expect_err("preset cancel flag must return error");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_git_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_git_cancellable(tmp.path(), &["status"], cancel)
.expect_err("preset cancel flag must return error");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_jj_with_retry_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_jj_with_retry_cancellable(
tmp.path(),
&["status"],
is_transient_error,
cancel,
)
.expect_err("preset cancel flag must return error before retry");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_jj_utf8_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_jj_utf8_cancellable(tmp.path(), &["status"], cancel)
.expect_err("preset cancel flag must return error");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_git_utf8_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_git_utf8_cancellable(tmp.path(), &["status"], cancel)
.expect_err("preset cancel flag must return error");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_git_with_retry_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_git_with_retry_cancellable(
tmp.path(),
&["status"],
is_transient_error,
cancel,
)
.expect_err("preset cancel flag must return error before retry");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_jj_utf8_with_retry_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_jj_utf8_with_retry_cancellable(
tmp.path(),
&["status"],
is_transient_error,
cancel,
)
.expect_err("preset cancel flag must return error before retry");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn run_git_utf8_with_retry_cancellable_short_circuits_on_preset_flag() {
let tmp = tempfile::tempdir().expect("tempdir");
let cancel = Arc::new(AtomicBool::new(true));
let err = run_git_utf8_with_retry_cancellable(
tmp.path(),
&["status"],
is_transient_error,
cancel,
)
.expect_err("preset cancel flag must return error before retry");
assert!(err.is_cancelled(), "expected Cancelled, got: {err:?}");
}
#[test]
fn is_transient_does_not_retry_cancelled() {
let err = RunError::Cancelled {
command: Cmd::new("jj").display(),
stdout: vec![],
stderr: String::new(),
attempts: 1,
};
assert!(!is_transient_error(&err));
}
#[test]
fn git_merge_base_returns_none_for_unrelated() {
if !git_installed() {
return;
}
let tmp = tempfile::tempdir().expect("tempdir");
std::process::Command::new("git")
.args(["init", "--quiet"])
.current_dir(tmp.path())
.status()
.expect("git init");
let _ = git_merge_base(tmp.path(), "HEAD", "HEAD");
}
struct TestRepo {
dir: tempfile::TempDir,
}
impl TestRepo {
fn new(colocate: bool) -> Self {
let dir = tempfile::tempdir().expect("tempdir");
let mut args = vec!["git", "init"];
if colocate {
args.push("--colocate");
}
let ok = std::process::Command::new("jj")
.args(&args)
.current_dir(dir.path())
.output()
.expect("jj git init")
.status
.success();
assert!(ok, "jj git init failed");
Self { dir }
}
fn path(&self) -> &Path {
self.dir.path()
}
fn jj(&self, args: &[&str]) -> std::process::Output {
std::process::Command::new("jj")
.args(["--config=user.name=T", "--config=user.email=t@e.com"])
.args(args)
.current_dir(self.path())
.output()
.expect("jj command")
}
fn git(&self, args: &[&str]) -> std::process::Output {
std::process::Command::new("git")
.args(args)
.current_dir(self.path())
.output()
.expect("git command")
}
fn git_out(&self, args: &[&str]) -> String {
String::from_utf8_lossy(&self.git(args).stdout).trim().to_string()
}
fn seed_two_commits(&self) {
std::fs::write(self.path().join("f.txt"), "a\n").unwrap();
self.jj(&["describe", "-m", "A"]);
self.jj(&["new", "-m", "B"]);
}
fn force_divergence(&self) {
let out = self.jj(&["op", "log", "--no-graph", "-T", "id ++ \"\\n\""]);
let ops = String::from_utf8_lossy(&out.stdout);
let earlier = ops.lines().nth(2).expect("older op").to_string();
self.jj(&["--at-operation", &earlier, "describe", "-m", "FORKED"]);
self.jj(&["status"]); }
}
#[test]
fn operation_log_detection_and_op_restore_recovery() {
if !jj_installed() {
return;
}
let repo = TestRepo::new(false);
repo.seed_two_commits();
let path = repo.path();
let good = jj_current_operation_id(path).expect("current op id");
assert!(!good.is_empty());
assert!(jj_divergent_change_ids(path).unwrap().is_empty(), "clean to start");
repo.force_divergence();
assert!(!jj_divergent_change_ids(path).unwrap().is_empty(), "divergence detected");
let ops = jj_operation_log(path, 0).unwrap();
assert!(
ops.iter().any(|o| o.description.contains("reconcile divergent operations")),
"reconcile op should appear; got {ops:?}"
);
jj_op_restore(path, &good).unwrap();
assert!(jj_divergent_change_ids(path).unwrap().is_empty(), "restore should clear divergence");
}
#[test]
fn operation_log_respects_limit() {
if !jj_installed() {
return;
}
let repo = TestRepo::new(false);
repo.seed_two_commits();
let all = jj_operation_log(repo.path(), 0).unwrap();
let two = jj_operation_log(repo.path(), 2).unwrap();
assert!(all.len() > 2, "seeded repo should have several ops");
assert_eq!(two.len(), 2, "limit should cap the returned ops");
assert_eq!(two[0], all[0], "same newest-first ordering");
}
#[test]
fn is_divergent_at_operation_distinguishes_clean_from_divergent() {
if !jj_installed() {
return;
}
let repo = TestRepo::new(false);
repo.seed_two_commits();
let good = jj_current_operation_id(repo.path()).unwrap();
repo.force_divergence();
let head = &jj_operation_log(repo.path(), 1).unwrap()[0].id;
assert!(jj_is_divergent_at_operation(repo.path(), head).unwrap(), "head op is divergent");
assert!(!jj_is_divergent_at_operation(repo.path(), &good).unwrap(), "good op is clean");
}
#[test]
fn divergent_change_ids_dedups_to_distinct_changes() {
if !jj_installed() {
return;
}
let repo = TestRepo::new(false);
repo.seed_two_commits();
repo.force_divergence();
assert_eq!(jj_divergent_change_ids(repo.path()).unwrap().len(), 1);
}
#[test]
fn op_log_reads_do_not_snapshot_the_working_copy() {
if !jj_installed() {
return;
}
let repo = TestRepo::new(false);
repo.seed_two_commits();
let path = repo.path();
std::fs::write(path.join("f.txt"), "a\nwip\n").unwrap();
let wc_commit = |repo: &TestRepo| {
String::from_utf8_lossy(
&repo
.jj(&["--ignore-working-copy", "log", "-r", "@", "--no-graph", "-T", "commit_id"])
.stdout,
)
.trim()
.to_string()
};
let before = wc_commit(&repo);
let _ = jj_current_operation_id(path).unwrap();
let _ = jj_operation_log(path, 5).unwrap();
let _ = jj_divergent_change_ids(path).unwrap();
assert_eq!(before, wc_commit(&repo), "op-log/divergence reads must not snapshot the working copy");
}
fn assert_colocated_consistent(repo: &TestRepo, bookmark: &str) {
let jj_commit = String::from_utf8_lossy(
&repo.jj(&["log", "-r", bookmark, "--no-graph", "-T", "commit_id"]).stdout,
)
.trim()
.to_string();
assert_eq!(
jj_commit,
repo.git_out(&["rev-parse", bookmark]),
"git ref and jj bookmark for '{bookmark}' must agree"
);
}
fn assert_git_healthy(repo: &TestRepo) {
let fsck = repo.git(&["fsck", "--no-dangling"]);
assert!(fsck.status.success(), "git fsck: {}", String::from_utf8_lossy(&fsck.stderr));
assert!(repo.git(&["status"]).status.success(), "git status must work");
}
#[test]
fn colocation_consistent_for_a_plain_bookmark() {
if !jj_installed() || !git_installed() {
return;
}
let repo = TestRepo::new(true);
assert!(repo.path().join(".git").exists() && repo.path().join(".jj").exists());
std::fs::write(repo.path().join("f.txt"), "a\n").unwrap();
repo.jj(&["describe", "-m", "A"]);
repo.jj(&["bookmark", "create", "feat", "-r", "@"]);
repo.jj(&["new", "-m", "B"]);
let good = jj_current_operation_id(repo.path()).unwrap();
assert_colocated_consistent(&repo, "feat");
repo.force_divergence();
jj_op_restore(repo.path(), &good).unwrap();
assert!(jj_divergent_change_ids(repo.path()).unwrap().is_empty());
assert_colocated_consistent(&repo, "feat");
assert_git_healthy(&repo);
}
#[test]
fn colocation_consistent_when_bookmark_moved() {
if !jj_installed() || !git_installed() {
return;
}
let repo = TestRepo::new(true);
std::fs::write(repo.path().join("f.txt"), "a\n").unwrap();
repo.jj(&["describe", "-m", "A"]);
repo.jj(&["bookmark", "create", "feat", "-r", "@"]); repo.jj(&["new", "-m", "B"]);
repo.jj(&["new", "-m", "C"]); repo.jj(&["bookmark", "set", "feat", "-r", "@-"]);
assert_colocated_consistent(&repo, "feat");
let good = jj_current_operation_id(repo.path()).unwrap();
repo.force_divergence();
jj_op_restore(repo.path(), &good).unwrap();
assert!(jj_divergent_change_ids(repo.path()).unwrap().is_empty());
assert_colocated_consistent(&repo, "feat");
assert_git_healthy(&repo);
}
#[test]
fn colocation_consistent_for_a_stack() {
if !jj_installed() || !git_installed() {
return;
}
let repo = TestRepo::new(true);
std::fs::write(repo.path().join("f.txt"), "a\n").unwrap();
repo.jj(&["describe", "-m", "A"]);
repo.jj(&["bookmark", "create", "bottom", "-r", "@"]);
repo.jj(&["new", "-m", "B"]);
repo.jj(&["bookmark", "create", "top", "-r", "@"]);
repo.jj(&["new", "-m", "C"]);
let good = jj_current_operation_id(repo.path()).unwrap();
assert_colocated_consistent(&repo, "bottom");
assert_colocated_consistent(&repo, "top");
repo.force_divergence();
jj_op_restore(repo.path(), &good).unwrap();
assert!(jj_divergent_change_ids(repo.path()).unwrap().is_empty());
assert_colocated_consistent(&repo, "bottom");
assert_colocated_consistent(&repo, "top");
assert_git_healthy(&repo);
}
}