1use super::capture_invocation::InvocationCapture;
16use super::commit_facts::PrivateIndexFacts;
17use super::commit_landing_lock::{acquire_reserved_lock, landing_is_nested};
18use super::commit_preparation::PreparedCommit;
19use super::commit_replay_options::ReplayOptions;
20use super::commit_revalidation::{Revalidation, RevalidationContext, revalidate_replay};
21use super::commit_transaction::{CommitInvocation, TransactionStop, failed};
22use super::commit_workspace::CommitWorkspace;
23use super::commit_worktree_records::{HookChanges, settled_entry};
24use super::diagnostic_log::debug;
25use super::event_path::EventPath;
26use super::landing_attempt::{LandingOutcome, LandingPayload, LandingRequest, land_transaction};
27use super::landing_findings::{LANDING_FINDING_EXIT_CODE, landing_finding};
28use super::landing_objects::migrate_shadow_objects;
29use super::owner_lock::OwnerLock;
30use super::policy_checks::ShippedChecks;
31use super::policy_engine::StageRequest;
32use super::policy_events::PolicyEvent;
33use super::replay_commit::{MergeStep, ReplayError, ReplayRequest, merge_replay, rebuild_replay};
34use super::replay_merge::ReplayedCommit;
35use super::replay_reservation::{LandingReservation, ReservationError};
36use super::repository_facts::RepositoryFacts;
37use super::transaction_journal::{AddedPath, Base, TransactionMode};
38use super::transaction_phase::{Phase, reach_transaction_phase};
39use std::path::{Path, PathBuf};
40
41const TAG: &str = "landWithReplay";
43
44struct Candidate {
54 new_oid: String,
56 tree: String,
58 expected_old: Base,
60 index: PathBuf,
62 exact_private_index: bool,
64 approved_tree: String,
66 committed_paths: Vec<Vec<u8>>,
68 added_paths: Vec<AddedPath>,
70 worktree_records: Vec<AddedPath>,
72 hook_changes: HookChanges,
74 events: Vec<PolicyEvent>,
76 attempt: i64,
78 lost_races: u64,
80}
81
82pub struct LoopInput<'a, F: RepositoryFacts> {
92 pub invocation: &'a CommitInvocation<'a>,
94 pub workspace: &'a mut CommitWorkspace,
96 pub capture: &'a InvocationCapture,
98 pub mode: TransactionMode,
100 pub request: &'a StageRequest,
102 pub checks: &'a mut ShippedChecks<PrivateIndexFacts<F>>,
104 pub prepared: &'a PreparedCommit,
106 pub settled_events: Vec<PolicyEvent>,
108 pub options: &'a ReplayOptions,
110 pub global_args: &'a [String],
112 pub read_only_selection: bool,
114 pub named_hooks_possible: bool,
116 pub committed_paths: Vec<Vec<u8>>,
118 pub added_paths: Vec<AddedPath>,
120 pub worktree_records: Vec<AddedPath>,
122 pub hook_changes: HookChanges,
124 pub index_lock_timeout_ms: u64,
126 pub reserve_after_lost_races: u64,
128}
129
130pub enum LoopOutcome {
140 Landed {
142 oid: String,
144 events: Vec<PolicyEvent>,
146 worktree_records: Vec<AddedPath>,
148 added_paths: Vec<AddedPath>,
150 },
151 Failed {
153 events: Vec<PolicyEvent>,
155 exit_code: i32,
157 },
158}
159
160fn retarget_records(
170 input: &CommitInvocation<'_>,
171 index: &Path,
172 records: &[AddedPath],
173) -> Result<Vec<AddedPath>, TransactionStop> {
174 let mut retargeted: Vec<AddedPath> = Vec::new();
175 for record in records {
176 let Some((mode, oid)) = settled_entry(input.context, input.cwd, index, record.path.as_slice())? else {
177 continue;
178 };
179 if mode != record.git_mode {
180 continue;
181 }
182 retargeted.push(AddedPath {
183 intended_oid: oid,
184 ..record.clone()
185 });
186 }
187 return Ok(retargeted);
188}
189
190fn replay_request<'a, F: RepositoryFacts>(
200 input: &'a LoopInput<'a, F>,
201 replay: u64,
202 onto: &'a str,
203) -> ReplayRequest<'a> {
204 return ReplayRequest {
205 context: input.invocation.context,
206 shadow: input.workspace.landing.shadow.as_path(),
207 object_directory: input.workspace.object_directory.as_path(),
208 worktree_root: input.capture.layout.repository_root.as_path(),
209 git_dir: input.capture.layout.git_dir.as_path(),
210 transaction_directory: input.workspace.landing.directory.as_path(),
211 replay,
212 base: match &input.capture.base {
213 Base::Commit(oid) => Some(oid.as_str()),
214 Base::Unborn => None,
215 },
216 empty_tree: input.capture.layout.empty_tree_oid.as_str(),
217 current: onto,
218 prepared: input.prepared.oid.as_str(),
219 signing_key: input.options.signing_key.as_deref(),
220 global_arguments: input.global_args,
221 };
222}
223
224fn replay_failure(capture: &InvocationCapture, onto: &str, cause: String) -> LoopOutcome {
235 return LoopOutcome::Failed {
236 events: vec![PolicyEvent::EngineFailure {
237 code: super::diagnostics::EngineFailureCode::TransactionFailed,
238 message: format!(
239 "{} moved while this commit was prepared, and replaying it onto {onto} failed; nothing landed: {cause}",
240 capture.layout.target_ref
241 ),
242 trigger: Some(super::policy_trigger::Trigger::PreForward),
243 policy: None,
244 path: None,
245 }],
246 exit_code: 2,
247 };
248}
249
250fn failed_events(settled: &[PolicyEvent], raced: &[PolicyEvent], last: Vec<PolicyEvent>) -> Vec<PolicyEvent> {
260 let mut events: Vec<PolicyEvent> = settled.to_vec();
261 events.extend(raced.iter().cloned());
262 events.extend(last);
263 return events;
264}
265
266fn replay_candidate<F: RepositoryFacts>(
276 input: &mut LoopInput<'_, F>,
277 candidate: &Candidate,
278 raced: Vec<PolicyEvent>,
279 onto: &str,
280) -> Result<Result<Candidate, LoopOutcome>, TransactionStop> {
281 let replay: u64 = lost_races_after(candidate.lost_races);
282 let merge: Result<MergeStep, ReplayError> = merge_replay(&replay_request(input, replay, onto));
283 let merged_tree: String = match merge {
284 Err(error) => return Ok(Err(replay_failure(input.capture, onto, replay_error_text(&error)))),
285 Ok(MergeStep::Conflict(conflict)) => {
286 migrate_shadow_objects(
288 input.invocation.context,
289 input.invocation.cwd,
290 input.workspace.landing.shadow.as_path(),
291 input.prepared.oid.as_str(),
292 &input.capture.base,
293 None,
294 )?;
295 let last: Vec<PolicyEvent> = vec![PolicyEvent::ReplayConflict {
296 target_ref: input.capture.layout.target_ref.clone(),
297 paths: conflict
298 .paths
299 .iter()
300 .map(|path| { return EventPath::from_git_bytes(path.as_slice()); })
301 .collect(),
302 winning_oid: conflict.winning_oid,
303 prepared_oid: conflict.prepared_oid,
304 }];
305 return Ok(Err(LoopOutcome::Failed {
306 events: failed_events(input.settled_events.as_slice(), raced.as_slice(), last),
307 exit_code: 1,
308 }));
309 }
310 Ok(MergeStep::Merged { tree, subsumed_paths, kept_landed_paths }) => {
311 debug(
312 TAG,
313 format!("replay onto {onto} keeps the prepared entries of {subsumed_paths:?} and the landed entries of {kept_landed_paths:?}").as_str(),
314 );
315 tree
316 }
317 };
318 let revalidation: Revalidation = {
319 let context: RevalidationContext<'_> = RevalidationContext {
320 invocation: input.invocation,
321 capture: input.capture,
322 read_only_selection: input.read_only_selection,
323 global_args: input.global_args,
324 no_verify: input.options.no_verify,
325 named_hooks_possible: input.named_hooks_possible,
326 };
327 revalidate_replay(
328 &context,
329 input.workspace,
330 input.request,
331 input.checks,
332 replay,
333 merged_tree.as_str(),
334 onto,
335 candidate.approved_tree.as_str(),
336 )?
337 };
338 let (tree, index, revalidation_events, revalidated_paths, revalidated_hooks): (String, PathBuf, Vec<PolicyEvent>, Vec<Vec<u8>>, HookChanges) =
339 match revalidation {
340 Revalidation::Blocked { events, exit_code } => {
341 return Ok(Err(LoopOutcome::Failed {
342 events: failed_events(input.settled_events.as_slice(), raced.as_slice(), events),
343 exit_code,
344 }));
345 }
346 Revalidation::Revalidated {
347 tree,
348 index,
349 events,
350 committed_paths,
351 hook_changes,
352 } => (tree, index, events, committed_paths, hook_changes),
353 };
354 let rebuilt = match rebuild_replay(&replay_request(input, replay, onto), tree.as_str()) {
355 Ok(found) => found,
356 Err(error) => return Ok(Err(replay_failure(input.capture, onto, replay_error_text(&error)))),
357 };
358 let committed_paths: Vec<Vec<u8>> = union_paths(input.committed_paths.as_slice(), revalidated_paths);
359 let mut worktree_records: Vec<AddedPath> = retarget_records(input.invocation, index.as_path(), input.worktree_records.as_slice())?;
360 worktree_records.extend(revalidated_hooks.worktree_records.iter().cloned());
361 let mut events: Vec<PolicyEvent> = replayed_events(
362 raced,
363 input.prepared.oid.as_str(),
364 &input.capture.base,
365 onto,
366 &rebuilt,
367 );
368 events.extend(revalidation_events);
369 return Ok(Ok(Candidate {
370 new_oid: rebuilt.oid,
371 tree: tree.clone(),
372 expected_old: Base::Commit(String::from(onto)),
373 index: index.clone(),
374 exact_private_index: false,
375 approved_tree: tree,
376 committed_paths,
377 added_paths: retarget_records(input.invocation, index.as_path(), input.added_paths.as_slice())?,
378 worktree_records,
379 hook_changes: HookChanges::combine(&input.hook_changes, &revalidated_hooks),
380 events,
381 attempt: next_attempt_number(candidate.attempt),
382 lost_races: lost_races_after(candidate.lost_races),
383 }));
384}
385
386pub fn union_paths(committed: &[Vec<u8>], revalidated: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
396 let mut paths: Vec<Vec<u8>> = committed.to_vec();
397 let known: std::collections::HashSet<&[u8]> = committed.iter().map(Vec::as_slice).collect();
399 let mut added: std::collections::HashSet<Vec<u8>> = std::collections::HashSet::new();
400 for path in revalidated {
401 if known.contains(path.as_slice()) || !added.insert(path.clone()) {
402 continue;
403 }
404 paths.push(path);
405 }
406 return paths;
407}
408
409pub fn replayed_events(
419 raced: Vec<PolicyEvent>,
420 prepared_oid: &str,
421 base: &Base,
422 onto: &str,
423 rebuilt: &ReplayedCommit,
424) -> Vec<PolicyEvent> {
425 let mut events: Vec<PolicyEvent> = raced;
426 events.push(PolicyEvent::CommitReplayed {
427 prepared_oid: String::from(prepared_oid),
428 from_base: match base {
429 Base::Commit(oid) => Some(oid.clone()),
430 Base::Unborn => None,
431 },
432 onto: String::from(onto),
433 oid: rebuilt.oid.clone(),
434 });
435 if !rebuilt.dropped_headers.is_empty() {
436 events.push(PolicyEvent::ReplayHeadersDropped {
437 prepared_oid: String::from(prepared_oid),
438 oid: rebuilt.oid.clone(),
439 headers: rebuilt.dropped_headers.clone(),
440 });
441 }
442 return events;
443}
444
445pub fn next_attempt_number(attempt: i64) -> i64 {
455 return attempt + 1;
456}
457
458pub fn lost_races_after(lost_races: u64) -> u64 {
468 return lost_races + 1;
469}
470
471fn replay_error_text(error: &ReplayError) -> String {
481 return match error {
482 ReplayError::Shared(inner) => format!("{inner:?}"),
483 ReplayError::Capture(inner) => format!("{inner:?}"),
484 ReplayError::Subsumption(inner) => format!("{inner:?}"),
485 ReplayError::Merge(inner) => format!("{inner:?}"),
486 ReplayError::Git(inner) => inner.0.clone(),
487 ReplayError::Object(text) => text.clone(),
488 };
489}
490
491fn reservation_stop(error: ReservationError) -> TransactionStop {
501 return failed(format!("the landing reservation failed: {error}"));
502}
503
504pub fn land_with_replay<F: RepositoryFacts>(mut input: LoopInput<'_, F>) -> Result<LoopOutcome, TransactionStop> {
515 let layout = &input.capture.layout;
516 let nested: bool = landing_is_nested(layout, input.invocation.environment);
517 let mut reservation: LandingReservation = LandingReservation::open(
518 layout.registry_root.as_path(),
519 input.workspace.landing.directory.as_path(),
520 input.workspace.landing.transaction_id.as_str(),
521 nested,
522 );
523 let result: Result<LoopOutcome, TransactionStop> = run_loop(&mut input, &mut reservation);
524 if let Err(error) = reservation.release() {
526 debug(TAG, format!("releasing the landing reservation failed: {error:?}").as_str());
527 }
528 return result;
529}
530
531fn run_loop<F: RepositoryFacts>(
541 input: &mut LoopInput<'_, F>,
542 reservation: &mut LandingReservation,
543) -> Result<LoopOutcome, TransactionStop> {
544 let mut candidate: Candidate = Candidate {
545 new_oid: input.prepared.oid.clone(),
546 tree: input.prepared.tree_oid.clone(),
547 expected_old: input.capture.base.clone(),
548 index: input.workspace.landing.commit_index.clone(),
549 exact_private_index: true,
550 approved_tree: input.prepared.tree_oid.clone(),
551 committed_paths: input.committed_paths.clone(),
552 added_paths: input.added_paths.clone(),
553 worktree_records: input.worktree_records.clone(),
554 hook_changes: input.hook_changes.clone(),
555 events: Vec::new(),
556 attempt: 1,
557 lost_races: 0,
558 };
559 loop {
560 let payload: LandingPayload = LandingPayload::Commit {
561 new_oid: candidate.new_oid.clone(),
562 landed_tree_oid: candidate.tree.clone(),
563 expected_old: candidate.expected_old.clone(),
564 landed_index: candidate.index.clone(),
565 exact_private_index: candidate.exact_private_index,
566 guarded_paths: candidate.hook_changes.paths.clone(),
567 };
568 let request: LandingRequest<'_> = LandingRequest {
569 capture: input.capture,
570 mode: input.mode,
571 payload: &payload,
572 committed_paths: candidate.committed_paths.as_slice(),
573 added_paths: candidate.added_paths.as_slice(),
574 selected_worktree_paths: candidate.worktree_records.as_slice(),
575 index_lock_timeout_ms: input.index_lock_timeout_ms,
576 attempt: candidate.attempt,
577 };
578 let outcome: LandingOutcome = {
579 let lock: OwnerLock = acquire_reserved_lock(
581 input.invocation.context,
582 input.invocation.cwd,
583 &input.capture.layout,
584 input.invocation.environment,
585 reservation,
586 )
587 .map_err(reservation_stop)?;
588 land_transaction(
589 input.invocation.context,
590 input.invocation.cwd,
591 &lock,
592 &mut input.workspace.landing,
593 &request,
594 )?
595 };
596 let onto: String = match &outcome {
597 LandingOutcome::Landed { oid } => {
598 return Ok(LoopOutcome::Landed {
599 oid: oid.clone(),
600 events: candidate.events,
601 worktree_records: candidate.worktree_records,
602 added_paths: candidate.added_paths,
603 });
604 }
605 LandingOutcome::HeadMoved {
606 current: Base::Commit(oid),
607 replayable: true,
608 } => oid.clone(),
609 _ => {
610 let mut last: Vec<PolicyEvent> = Vec::new();
611 if let Some(event) = landing_finding(&outcome, input.capture, Some(input.prepared.oid.as_str()), false) {
612 last.push(event);
613 }
614 return Ok(LoopOutcome::Failed {
615 events: failed_events(input.settled_events.as_slice(), candidate.events.as_slice(), last),
616 exit_code: LANDING_FINDING_EXIT_CODE,
617 });
618 }
619 };
620 let lost_races: u64 = lost_races_after(candidate.lost_races);
621 debug(TAG, format!("lost landing race {lost_races} to {onto}").as_str());
622 let reserved: bool = lost_races >= input.reserve_after_lost_races
623 && reservation.reserve(lost_races).map_err(reservation_stop)?;
624 reach_transaction_phase(Phase::RaceLost, Some(lost_races))?;
625 let mut raced: Vec<PolicyEvent> = candidate.events.clone();
626 raced.push(PolicyEvent::LandingRaceLost {
627 attempt: lost_races,
628 winning_oid: onto.clone(),
629 });
630 if reserved {
631 raced.push(PolicyEvent::LandingReserved { lost_races });
632 }
633 match replay_candidate(input, &candidate, raced, onto.as_str())? {
634 Ok(next) => candidate = next,
635 Err(ended) => return Ok(ended),
636 }
637 }
638}
639
640#[cfg(test)]
642#[path = "commit_landing_loop_tests.rs"]
643mod tests;