use super::journald_native::Priority;
use super::service_fault::reach;
use super::service_head::{follows_move, still_checked_out};
use super::service_pruned::batch_content_missing;
use super::service_journal::{Intent, OPEN_EXTENSION, remove_start, write_intent, write_start};
pub use super::service_land_finish::{RecoveryReport, recover, recover_instance};
#[cfg(test)]
use super::service_land_finish::follow_up;
#[cfg(test)]
use super::service_journal::{Recovery, list_intents};
#[cfg(test)]
use super::service_index_lock::try_acquire;
use super::service_land_finish::finish_after_move;
use super::service_land_start::witness_of;
use super::service_index_lock::{Announce, IndexLock, WaitNotes, acquire_noting_wait_within};
use super::service_timing::{LandTiming, wait_line};
use super::config_snapshot::read_config_snapshot;
use super::service_land_follow::{classify, storage_failure};
use super::service_protocol::{CommitRequest, FailureCode, is_zero_object};
use super::transaction_git::{GitContext, GitOutput, GitRequest, run_git};
use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::time::Instant;
pub const LANDING_ATTEMPTS: usize = 3;
pub struct LandContext {
pub git: GitContext,
pub journal: PathBuf,
pub on_objects_written: Option<Box<dyn Fn() + Send + Sync>>,
pub lock_ceiling: std::time::Duration,
}
fn keep_unlanded(git: &GitContext, repository: &Path, seq: u64, tip: &str) -> String {
let name: String = format!("refs/cli-git/kept/{seq}");
let made = run(git, repository, &["update-ref", name.as_str(), tip], None);
if !made.is_ok_and(|output| { return output.succeeded(); }) {
return String::from("cli-git: the commit built for you could not be kept on a side ref; your staged changes and your worktree are as they were, so run the commit again when the program above is done.\n");
}
return format!(
"cli-git: the commit built for you is kept at {name} (see it with `git log {name}`); nothing lands later by itself, and your staged changes and your worktree are as they were. Run the commit again when the program above is done.\n"
);
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Outcome {
Landed { commit: String },
RetryPrepare { base: String },
BranchMoved,
}
fn run(git: &GitContext, repository: &Path, arguments: &[&str], input: Option<Vec<u8>>) -> Result<GitOutput, FailureCode> {
let mut request: GitRequest = GitRequest::new(repository, arguments);
request.input = input;
let output: GitOutput = run_git(git, &request).map_err(|error| {
if super::child_watch::is_ceiling(&error) {
CEILING_STEP.with(|step| { *step.borrow_mut() = Some(error.to_string()); });
return FailureCode::StepTooSlow;
}
return FailureCode::Unavailable;
})?;
return Ok(output);
}
thread_local! {
static CEILING_STEP: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
}
fn take_ceiling_step() -> Option<String> {
return CEILING_STEP.with(|step| { return step.borrow_mut().take(); });
}
fn checked(git: &GitContext, repository: &Path, arguments: &[&str], input: Option<Vec<u8>>) -> Result<String, FailureCode> {
let output: GitOutput = run(git, repository, arguments, input)?;
if !output.succeeded() {
return Err(classify(&output));
}
return Ok(String::from_utf8_lossy(output.stdout.as_slice()).trim().to_string());
}
fn quoted(path: &str) -> String {
let mut text: String = String::from("\"");
for character in path.chars() {
match character {
'"' => text.push_str("\\\""),
'\\' => text.push_str("\\\\"),
other => text.push(other),
}
}
text.push('"');
return text;
}
fn overlaps(path: &str, changed: &str) -> bool {
return path == changed || path.starts_with(&format!("{changed}/")) || changed.starts_with(&format!("{path}/"));
}
fn import_stream(scratch: &str, old: &str, requests: &[&CommitRequest], amended_parents: &[String]) -> Vec<u8> {
let mut bytes: Vec<u8> = Vec::new();
for (index, request) in requests.iter().enumerate() {
bytes.extend(format!("commit {scratch}\nmark :{}\n", index + 1).as_bytes());
if let Some(author) = &request.author {
bytes.extend(format!("author {author}\n").as_bytes());
}
bytes.extend(format!("committer {}\ndata {}\n", request.committer, request.message.len()).as_bytes());
bytes.extend(request.message.as_bytes());
bytes.push(b'\n');
if let Some(tree) = request.tree.as_deref().filter(|_| { return request.amend; }) {
for (position, parent) in amended_parents.iter().enumerate() {
bytes.extend(format!("{} {parent}\n", if position == 0 { "from" } else { "merge" }).as_bytes());
}
bytes.extend(format!("M 040000 {tree} \"\"\n").as_bytes());
bytes.push(b'\n');
continue;
}
if index == 0 {
bytes.extend(format!("from {old}\n").as_bytes());
}
for change in &request.changes {
match &change.blob {
Some(blob) => bytes.extend(format!("M {:o} {blob} {}\n", change.mode, quoted(change.path.as_str())).as_bytes()),
None => bytes.extend(format!("D {}\n", quoted(change.path.as_str())).as_bytes()),
}
}
bytes.push(b'\n');
}
bytes.extend(b"done\n");
return bytes;
}
fn commits_from_marks(text: &str, count: usize) -> Option<Vec<String>> {
let mut commits: Vec<Option<String>> = vec![None; count];
for line in text.lines() {
let mut parts = line.split(' ');
let (Some(mark), Some(name)) = (parts.next(), parts.next()) else { continue };
let slot: usize = mark.trim_start_matches(':').parse::<usize>().ok()?;
if slot >= 1 && slot <= count {
commits[slot - 1] = Some(name.to_string());
}
}
return commits.into_iter().collect();
}
pub fn land_batch(context: &LandContext, seq: u64, requests: &[CommitRequest]) -> Result<Vec<Outcome>, FailureCode> {
return land_batch_announcing(context, seq, requests, &|_| {});
}
pub fn land_batch_announcing(context: &LandContext, seq: u64, requests: &[CommitRequest], announce: &dyn Fn(Announce)) -> Result<Vec<Outcome>, FailureCode> {
return land_batch_timed(context, seq, requests, announce, &mut LandTiming::default());
}
pub fn land_batch_timed(
context: &LandContext,
seq: u64,
requests: &[CommitRequest],
announce: &dyn Fn(Announce),
timing: &mut LandTiming,
) -> Result<Vec<Outcome>, FailureCode> {
let hardens_index: bool = requests
.first()
.and_then(|first| { return read_config_snapshot(&context.git, first.repository.as_path()); })
.and_then(|snapshot| { return snapshot.hardens_index(); })
.unwrap_or(true);
let deferral = super::service_durable::begin(hardens_index);
let result: Result<Vec<Outcome>, FailureCode> = land_batch_body(context, seq, requests, announce, timing);
if result == Err(FailureCode::StepTooSlow)
&& let Some(step) = take_ceiling_step()
&& let Some(first) = requests.first()
{
announce(Announce::Waiting(format!("cli-git: {step}, so this commit was not made.\n")));
let git: GitContext = GitContext { common_dir: Some(first.common_dir.clone()), ..context.git.clone() };
let scratch: String = format!("refs/worktree/cli-git-service/{seq}");
if let Ok(tip) = checked(&git, first.repository.as_path(), &["rev-parse", "--verify", "--quiet", scratch.as_str()], None) {
announce(Announce::Waiting(keep_unlanded(&git, first.repository.as_path(), seq, tip.as_str())));
let _deleted = run(&git, first.repository.as_path(), &["update-ref", "-d", scratch.as_str()], None);
}
}
let published: bool = result.as_ref().is_ok_and(|outcomes| { return outcomes.iter().any(|outcome| { return matches!(outcome, Outcome::Landed { .. }); }); });
deferral.settle(published);
return result;
}
fn land_batch_body(
context: &LandContext,
seq: u64,
requests: &[CommitRequest],
announce: &dyn Fn(Announce),
timing: &mut LandTiming,
) -> Result<Vec<Outcome>, FailureCode> {
let first: &CommitRequest = requests.first().ok_or(FailureCode::Rejected)?;
if requests.iter().any(|request| {
return request.repository != first.repository || request.common_dir != first.common_dir || request.reference != first.reference;
}) {
return Err(FailureCode::Rejected);
}
let alone: bool = requests.iter().any(CommitRequest::lands_alone);
if alone && requests.len() != 1 {
return Err(FailureCode::Rejected);
}
let repository: &Path = first.repository.as_path();
let git: GitContext = GitContext { common_dir: Some(first.common_dir.clone()), ..context.git.clone() };
let scratch: String = format!("refs/worktree/cli-git-service/{seq}");
let mut outcomes: Vec<Outcome>;
let mut attempts_left: usize = LANDING_ATTEMPTS;
let unborn: bool = is_zero_object(first.base.as_str());
loop {
timing.attempts += 1;
let branch_commit: String = format!("{}^{{commit}}", first.reference);
let old: String = match checked(&git, repository, &["rev-parse", "--verify", "--quiet", branch_commit.as_str()], None) {
Ok(found) => found,
Err(_) if unborn => first.base.clone(),
Err(code) => return Err(code),
};
if alone && first.base != old {
return Ok(vec![Outcome::BranchMoved]);
}
if !still_checked_out(&git, repository, first.git_dir.as_path(), first.common_dir.as_path(), first.reference.as_str())? {
return Ok(vec![Outcome::BranchMoved; requests.len()]);
}
let amended_parents: Vec<String> = if alone && !unborn {
let listing: String = checked(&git, repository, &["rev-list", "--parents", "--max-count=1", first.base.as_str()], None)?;
listing.split_whitespace().skip(1).map(str::to_string).collect()
} else {
Vec::new()
};
let mut changed_since: HashMap<&str, BTreeSet<String>> = HashMap::new();
let mut landing: Vec<(usize, &CommitRequest)> = Vec::new();
outcomes = vec![Outcome::RetryPrepare { base: old.clone() }; requests.len()];
for (index, request) in requests.iter().enumerate() {
if request.base != old && !changed_since.contains_key(request.base.as_str()) {
let listing: String = checked(&git, repository, &["diff-tree", "-r", "--name-only", "-z", request.base.as_str(), old.as_str()], None)?;
changed_since.insert(request.base.as_str(), listing.split('\0').filter(|name| { return !name.is_empty(); }).map(str::to_string).collect());
}
let conflicts: bool = request.base != old
&& request.changes.iter().any(|change| {
return changed_since.get(request.base.as_str()).is_some_and(|names| { return names.iter().any(|name| { return overlaps(change.path.as_str(), name.as_str()); }); });
});
let shadowed: bool = landing.iter().any(|(_, earlier)| {
let later_capture: bool = earlier.captured != 0 && request.captured > earlier.captured;
return !later_capture && request.changes.iter().any(|change| { return earlier.changes.iter().any(|member| { return overlaps(change.path.as_str(), member.path.as_str()); }); });
});
if !conflicts && !shadowed {
landing.push((index, request));
}
}
if landing.is_empty() {
return Ok(outcomes);
}
let batch: Vec<&CommitRequest> = landing.iter().map(|(_, request)| { return *request; }).collect();
if batch_content_missing(first.common_dir.as_path(), batch.as_slice()) {
return Err(FailureCode::Unavailable);
}
let objects_began: Instant = Instant::now();
let (commits, new): (Vec<String>, String) = if let Some(built) = first.commit.as_deref() {
let listing: GitOutput = run(&git, repository, &["rev-list", "--parents", "--max-count=1", built], None)?;
let text: String = String::from_utf8_lossy(listing.stdout.as_slice()).into_owned();
let mut words = text.split_whitespace();
let named_itself: bool = listing.succeeded() && words.next() == Some(built);
let parents: Vec<String> = words.map(str::to_string).collect();
let acceptable: bool = if first.amend {
parents == amended_parents
} else if unborn {
parents.is_empty()
} else {
parents.first() == Some(&first.base)
};
if !named_itself || !acceptable {
return Err(FailureCode::Rejected);
}
let connected: Vec<&str> = if unborn { vec!["rev-list", "--objects", built] } else { vec!["rev-list", "--objects", built, "--not", old.as_str()] };
if !run(&git, repository, connected.as_slice(), None)?.succeeded() {
return Err(FailureCode::Unavailable);
}
(vec![built.to_string()], built.to_string())
} else {
let marks: PathBuf = context.journal.join(format!("marks-{seq}"));
let marks_argument: String = format!("--export-marks={}", marks.display());
let imported: GitOutput = run(&git, repository, &["fast-import", "--done", marks_argument.as_str()], Some(import_stream(scratch.as_str(), old.as_str(), &batch, amended_parents.as_slice())))?;
if !imported.succeeded() {
return Err(classify(&imported));
}
crate::inner_git_words::record_words(&["fast-import", "--done"], &imported);
let marks_text: String = std::fs::read_to_string(&marks).map_err(|_| { return FailureCode::Unavailable; })?;
let _marks_removed = std::fs::remove_file(&marks);
let made: Vec<String> = commits_from_marks(marks_text.as_str(), batch.len()).ok_or(FailureCode::Unavailable)?;
let last: String = made.last().cloned().ok_or(FailureCode::Unavailable)?;
(made, last)
};
timing.objects += objects_began.elapsed();
reach("after-objects");
if let Some(seam) = &context.on_objects_written {
seam();
}
let derived: bool = first.commit.is_some();
let paths: Vec<String> = if derived {
Vec::new()
} else {
batch.iter().flat_map(|request| { return request.changes.iter().map(|change| { return change.path.clone(); }); }).collect()
};
let intent: Intent = Intent {
seq,
repository: first.repository.clone(),
common_dir: first.common_dir.clone(),
reference: first.reference.clone(),
old: old.clone(),
new: new.clone(),
ids: batch.iter().map(|request| { return request.id.clone(); }).collect(),
commits: commits.clone(),
paths,
derived,
cleanup: first.cleanup.clone(),
index: first.index.clone(),
git_dir: first.git_dir.clone(),
};
let journal_began: Instant = Instant::now();
let written = write_intent(context.journal.as_path(), &intent);
timing.journal += journal_began.elapsed();
written.map_err(|error| { return storage_failure(&error); })?;
reach("after-intent");
let waiting_since: Instant = Instant::now();
let mut notes: WaitNotes = WaitNotes::default();
let taken = acquire_noting_wait_within(first.index.as_path(), seq, context.lock_ceiling, announce, &mut notes);
let waited: std::time::Duration = waiting_since.elapsed();
timing.lock_wait += waited;
if notes.looks > 0 || timing.attempts == 1 {
timing.lock_holder.clone_from(¬es.holder);
timing.lock_notes = notes.clone();
}
if let Some(line) = wait_line(seq, waited, ¬es, taken.is_ok()) {
let priority: Priority = if taken.is_ok() { Priority::Info } else { Priority::Error };
crate::service_timing::log_entry(priority, line.as_str(), &[("CLI_GIT_BATCH", seq.to_string())]);
}
let held: IndexLock = match taken {
Ok(lock) => lock,
Err(failure) => {
let _intent_removed = std::fs::remove_file(super::service_journal::batch_path(context.journal.as_path(), seq, OPEN_EXTENSION));
if failure.code == FailureCode::IndexInUse {
announce(Announce::Waiting(keep_unlanded(&git, repository, seq, new.as_str())));
}
let _deleted = run(&git, repository, &["update-ref", "-d", scratch.as_str()], None);
return Err(failure.code);
}
};
let hold_began: Instant = Instant::now();
let still: Result<bool, FailureCode> = still_checked_out(&git, repository, first.git_dir.as_path(), first.common_dir.as_path(), first.reference.as_str());
if !matches!(still, Ok(true)) {
let _intent_removed = std::fs::remove_file(super::service_journal::batch_path(context.journal.as_path(), seq, OPEN_EXTENSION));
let _deleted = run(&git, repository, &["update-ref", "-d", scratch.as_str()], None);
return match still {
Ok(_) => Ok(vec![Outcome::BranchMoved; requests.len()]),
Err(code) => Err(code),
};
}
match witness_of(first.index.as_path()) {
Ok(witness) => {
if let Err(error) = write_start(context.journal.as_path(), seq, &witness) {
crate::service_error!("batch {seq}: the index file the landing saw could not be recorded ({error}); a restart will not replay the index update");
}
}
Err(error) => crate::service_error!("batch {seq}: the index file cannot be looked at ({error}); a restart will not replay the index update"),
}
reach("before-ref");
let removal: String = if first.commit.is_some() { String::new() } else { format!("delete {scratch}\n") };
let transaction: String = format!("start\nupdate {} {new} {old}\n{removal}prepare\ncommit\n", first.reference);
let ref_began: Instant = Instant::now();
let ran = run(&git, repository, &["update-ref", "--stdin"], Some(transaction.into_bytes()));
timing.ref_update += ref_began.elapsed();
let moved: GitOutput = match ran {
Ok(output) => output,
Err(FailureCode::StepTooSlow) => {
let step: String = take_ceiling_step().unwrap_or_else(|| { return String::from("the reference update ran too long and was ended"); });
let tip: Option<String> = checked(&git, repository, &["rev-parse", "--verify", "--quiet", first.reference.as_str()], None).ok();
if tip.as_deref() == Some(new.as_str()) {
let _deleted = run(&git, repository, &["update-ref", "-d", scratch.as_str()], None);
GitOutput { code: Some(0), stdout: Vec::new(), stderr: Vec::new() }
} else {
let _intent_removed = std::fs::remove_file(super::service_journal::batch_path(context.journal.as_path(), seq, OPEN_EXTENSION));
remove_start(context.journal.as_path(), seq);
announce(Announce::Waiting(format!("cli-git: {step}, so this commit was not made.\n")));
announce(Announce::Waiting(keep_unlanded(&git, repository, seq, new.as_str())));
let _deleted = run(&git, repository, &["update-ref", "-d", scratch.as_str()], None);
return Err(FailureCode::StepTooSlow);
}
}
Err(code) => return Err(code),
};
if !moved.succeeded() {
let _intent_removed = std::fs::remove_file(super::service_journal::batch_path(context.journal.as_path(), seq, OPEN_EXTENSION));
remove_start(context.journal.as_path(), seq);
let _deleted = run(&git, repository, &["update-ref", "-d", scratch.as_str()], None);
attempts_left -= 1;
timing.hold += hold_began.elapsed();
if attempts_left == 0 {
return Err(FailureCode::BranchKeepsChanging);
}
continue;
}
reach("after-ref");
let followed: Result<bool, FailureCode> = follows_move(&git, repository, first.git_dir.as_path(), first.common_dir.as_path(), first.reference.as_str(), new.as_str());
let sync_began: Instant = Instant::now();
let settled: bool = finish_after_move(&git, &intent, held, context.journal.as_path(), followed);
timing.sync += sync_began.elapsed();
timing.hold += hold_began.elapsed();
if settled {
reach("after-index");
let _marked = super::service_journal::mark_done(context.journal.as_path(), seq);
}
for (slot, (index, _request)) in landing.iter().enumerate() {
outcomes[*index] = Outcome::Landed { commit: commits[slot].clone() };
}
return Ok(outcomes);
}
}
#[cfg(test)]
#[path = "service_land_tests.rs"]
mod tests;
#[cfg(test)]
#[path = "service_land_witness_tests.rs"]
mod witness_tests;
#[cfg(test)]
#[path = "service_land_head_tests.rs"]
mod head_tests;