use crate::auto_push_single_flight::{PushAttempt, SingleFlightError};
use crate::diagnostic_log::{debug, warn};
use crate::push_deadline_advice::extend_clause;
use crate::push_progress::{Clock, WaitReport};
use crate::process_tree::{TreeMember, descendants_of, kill_members, member_of, still_running, terminate_members};
use crate::stopped_git_locks::LockLedger;
use crate::scratch_location::scratch_file;
use crate::start_retry::RetryStart;
use std::ffi::{OsStr, OsString};
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::time::{Duration, Instant};
const TAG: &str = "cli-git";
pub const PUSH_QUEUED_LINE: &str = "The push is queued: the commit service keeps it in its retry plan and tries it again.";
pub const PUSH_DEADLINE_VARIABLE: &str = "CLI_GIT_PUSH_DEADLINE_SECONDS";
pub const DEFAULT_PUSH_DEADLINE: Duration = Duration::from_secs(30);
pub const STOP_GRACE: Duration = Duration::from_secs(5);
pub const MAX_PUSH_OUTPUT: u64 = 4 * 1024 * 1024;
pub const STOPPED_EXIT_CODE: i64 = 124;
pub const WAIT_NOTE_PHRASE: &str = "is waiting for the server";
const POLL_PAUSE: Duration = Duration::from_millis(10);
pub fn positive_seconds(name: &str, value: &OsStr) -> Option<Duration> {
let parsed: Option<u64> = match value.to_str() {
Some(text) => text.trim().parse::<u64>().ok(),
None => None,
};
return match parsed {
Some(count) if count > 0 => Some(Duration::from_secs(count)),
Some(_) | None => {
debug(
TAG,
&format!("{name} is not a positive whole number; ignoring it"),
);
None
}
};
}
fn seconds_variable(environment: &[(OsString, OsString)], name: &str) -> Option<Duration> {
let mut found: Option<&OsString> = None;
for (key, value) in environment {
if key.to_str() == Some(name) {
found = Some(value);
}
}
let value: &OsString = found?;
return positive_seconds(name, value.as_os_str());
}
pub fn push_deadline(environment: &[(OsString, OsString)]) -> Duration {
return seconds_variable(environment, PUSH_DEADLINE_VARIABLE).unwrap_or(DEFAULT_PUSH_DEADLINE);
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Patience {
Fixed(Duration),
Progress {
stall: Duration,
},
}
pub fn push_patience(environment: &[(OsString, OsString)]) -> Patience {
return Patience::Progress { stall: push_deadline(environment) };
}
pub fn strip_one_line_break(text: &str) -> &str {
let without_lf = match text.strip_suffix('\n') {
Some(stripped) => stripped,
None => return text,
};
return without_lf.strip_suffix('\r').unwrap_or(without_lf);
}
#[cfg(unix)]
fn ask_to_stop(child: &Child) {
let Ok(pid) = libc::pid_t::try_from(child.id()) else {
return;
};
unsafe { libc::kill(pid, libc::SIGTERM) };
}
#[cfg(not(unix))]
fn ask_to_stop(_child: &Child) {}
fn grace_remains(asked: Duration, grace: Duration) -> bool {
return asked < grace;
}
fn is_flood(size: u64) -> bool {
return size > MAX_PUSH_OUTPUT;
}
fn wait_for_tree(tree: &[TreeMember], grace: Duration) {
let waited: Instant = Instant::now();
while grace_remains(waited.elapsed(), grace) && tree.iter().any(still_running) {
std::thread::sleep(POLL_PAUSE);
}
}
pub fn sweep_after_exit(mut child: Child, tree: Vec<TreeMember>, locks: LockLedger) -> Option<std::thread::JoinHandle<()>> {
return std::thread::Builder::new()
.name(String::from("cli-git-lock-sweep"))
.spawn(move || {
let _status = child.wait();
while tree.iter().any(still_running) {
std::thread::sleep(POLL_PAUSE);
}
locks.sweep(tree.as_slice());
})
.inspect_err(|error| {
debug(TAG, &format!("could not start the thread that sweeps a stopped push's locks: {error}"));
})
.ok();
}
pub fn stop_push(mut child: Child, grace: Duration, locks: LockLedger) {
let below: Vec<TreeMember> = i32::try_from(child.id()).map(descendants_of).unwrap_or_default();
let mut tree: Vec<TreeMember> = i32::try_from(child.id()).ok().and_then(member_of).into_iter().collect();
tree.extend(below.iter().cloned());
ask_to_stop(&child);
terminate_members(below.as_slice());
let asked: Instant = Instant::now();
let mut ended: bool = false;
while !ended && grace_remains(asked.elapsed(), grace) {
match child.try_wait() {
Ok(Some(_)) | Err(_) => ended = true,
Ok(None) => std::thread::sleep(POLL_PAUSE),
}
}
let mut collected: bool = ended;
if !ended {
let _killed = child.kill();
let killed_at: Instant = Instant::now();
while grace_remains(killed_at.elapsed(), grace) {
match child.try_wait() {
Ok(Some(_)) | Err(_) => {
collected = true;
break;
}
Ok(None) => std::thread::sleep(POLL_PAUSE),
}
}
}
while grace_remains(asked.elapsed(), grace) && below.iter().any(still_running) {
std::thread::sleep(POLL_PAUSE);
}
kill_members(below.as_slice());
let _sweeper = settle_locks(child, collected, tree, locks, grace);
}
pub fn settle_locks(child: Child, collected: bool, tree: Vec<TreeMember>, locks: LockLedger, grace: Duration) -> Option<std::thread::JoinHandle<()>> {
if collected {
wait_for_tree(tree.as_slice(), grace);
if tree.iter().all(|member| { return !still_running(member); }) {
locks.sweep(tree.as_slice());
return None;
}
}
return sweep_after_exit(child, tree, locks);
}
fn read_output(file: &mut File) -> Result<String, SingleFlightError> {
file.seek(SeekFrom::Start(0))
.map_err(SingleFlightError::Start)?;
let mut bytes: Vec<u8> = Vec::new();
file.take(MAX_PUSH_OUTPUT)
.read_to_end(&mut bytes)
.map_err(SingleFlightError::Start)?;
return Ok(String::from_utf8_lossy(&bytes).into_owned());
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct StopReport {
pub limit: Duration,
pub moved: u64,
pub elapsed: Duration,
}
pub enum StopCause {
Deadline,
Stalled(StopReport),
Flood,
}
pub fn wait_until_done(
child: &mut Child,
outputs: &[&File],
patience: Patience,
quiet: &mut Option<Duration>,
) -> Result<Result<ExitStatus, StopCause>, SingleFlightError> {
let mut waits: Vec<WaitReport> = Vec::new();
let clock: Clock = Clock::start(child.id(), patience).without_wall_fallback();
return wait_with_clock(child, outputs, clock, "git", &mut waits, quiet);
}
pub fn wait_until_done_noting(
child: &mut Child,
outputs: &[&File],
patience: Patience,
label: &str,
waits: &mut Vec<WaitReport>,
) -> Result<Result<ExitStatus, StopCause>, SingleFlightError> {
let clock: Clock = Clock::start(child.id(), patience);
let mut unused: Option<Duration> = None;
return wait_with_clock(child, outputs, clock, label, waits, &mut unused);
}
fn wait_with_clock(
child: &mut Child,
outputs: &[&File],
mut clock: Clock,
label: &str,
waits: &mut Vec<WaitReport>,
quiet: &mut Option<Duration>,
) -> Result<Result<ExitStatus, StopCause>, SingleFlightError> {
let kept = |watched: &Clock| -> Option<Duration> { return watched.measures().then(|| { return watched.longest_quiet(); }); };
loop {
if let Some(status) = child.try_wait().map_err(SingleFlightError::Start)? {
*quiet = kept(&clock);
return Ok(Ok(status));
}
for output in outputs {
let size: u64 = match output.metadata() {
Ok(metadata) => metadata.len(),
Err(_) => u64::MAX,
};
if is_flood(size) {
*quiet = kept(&clock);
return Ok(Err(StopCause::Flood));
}
}
if let Some(cause) = clock.check() {
*quiet = kept(&clock);
return Ok(Err(cause));
}
if let Some(report) = clock.take_wait_note() {
warn(TAG, &wait_line(label, &report));
waits.push(report);
}
std::thread::sleep(POLL_PAUSE);
}
}
pub fn command_label(command: &Command) -> String {
let program: String = command.get_program().to_string_lossy().rsplit('/').next().unwrap_or_default().to_owned();
let mut arguments = command.get_args();
while let Some(argument) = arguments.next() {
let word: String = argument.to_string_lossy().into_owned();
if word == "-c" || word == "-C" {
let _value = arguments.next();
continue;
}
if word.starts_with('-') {
continue;
}
return format!("{program} {word}");
}
return program;
}
fn wait_line(label: &str, report: &WaitReport) -> String {
return format!(
"cli-git: {label} {WAIT_NOTE_PHRASE}: the connection to the remote is open, and nothing has moved for {} seconds \
(elapsed_seconds={}); the remote may be checking the push or preparing its answer. It is waited for until it answers or \
the connection closes; no time limit applies while the connection is open.",
report.silent.as_secs(),
report.elapsed.as_secs(),
);
}
pub fn run_bounded_push(
command: Command,
patience: Patience,
grace: Duration,
environment: &[(OsString, OsString)],
) -> Result<PushAttempt, SingleFlightError> {
return crate::push_lock_clearing::run_clearing_locks(command, patience, grace, environment, &crate::stale_ref_lock_judge::Judge::here());
}
pub fn run_bounded_push_once(
mut command: Command,
patience: Patience,
grace: Duration,
environment: &[(OsString, OsString)],
) -> Result<PushAttempt, SingleFlightError> {
let mut output: File = scratch_file(environment).map_err(SingleFlightError::Start)?;
let for_stdout: File = output.try_clone().map_err(SingleFlightError::Start)?;
let for_stderr: File = output.try_clone().map_err(SingleFlightError::Start)?;
command
.stdin(Stdio::null())
.stdout(Stdio::from(for_stdout))
.stderr(Stdio::from(for_stderr));
let locks: LockLedger = LockLedger::before_start(&command);
let label: String = command_label(&command);
let mut child: Child = command.spawn_retrying().map_err(SingleFlightError::Start)?;
drop(command);
let mut waits: Vec<WaitReport> = Vec::new();
let ended: Result<ExitStatus, StopCause> = wait_until_done_noting(&mut child, &[&output], patience, label.as_str(), &mut waits)?;
if ended.is_err() {
stop_push(child, grace, locks);
}
let captured: String = read_output(&mut output)?;
let mut shown_text: String = strip_one_line_break(captured.as_str()).to_owned();
for report in &waits {
let line: String = wait_line(label.as_str(), report);
if !shown_text.is_empty() {
shown_text.push('\n');
}
shown_text.push_str(line.as_str());
}
let shown: &str = shown_text.as_str();
return Ok(match ended {
Ok(status) => PushAttempt {
exit_code: i64::from(status.code().unwrap_or(1)),
output: shown.to_owned(),
},
Err(cause) => {
let reason: String = match cause {
StopCause::Deadline => format!(
"cli-git: git push did not finish within {} seconds and was stopped (limit=deadline); \
{}. {PUSH_QUEUED_LINE} Hint: if you were pushing over a slow or \
offline connection, your commit is already saved; push again when the connection works.",
match patience {
Patience::Fixed(limit) => limit.as_secs(),
Patience::Progress { stall } => stall.as_secs(),
},
extend_clause()
),
StopCause::Stalled(report) => format!(
"cli-git: git push made no progress for {} seconds and was stopped (limit=stall, bytes_moved={}, \
elapsed_seconds={}); {}. {PUSH_QUEUED_LINE} Hint: if you were pushing over a slow or \
offline connection, your commit is already saved; push again when the connection works.",
report.limit.as_secs(),
report.moved,
report.elapsed.as_secs(),
extend_clause()
),
StopCause::Flood => format!(
"cli-git: git push wrote more than {MAX_PUSH_OUTPUT} bytes of output and was stopped. Hint: if a hook or the \
remote prints this much, run `git ls-remote` to see whether your commits arrived, and read the \
hook's own output by running it directly before pushing again."
),
};
debug(TAG, &reason);
let joined: String = if shown.is_empty() {
reason
} else {
format!("{shown}\n{reason}")
};
PushAttempt {
exit_code: STOPPED_EXIT_CODE,
output: joined,
}
}
});
}
#[cfg(test)]
#[path = "bounded_push_tests.rs"]
mod tests;