Skip to main content

git_policy_cli/landing/
service_work.rs

1//! What:
2//!  The service's consumer of after-commit work: pending `push` and `maintenance` work items of one repository are
3//!  claimed, coalesced, run, settled, and a push that failed leaves a record that the person's next wrapper command shows.
4//! Why:
5//!  A commit must not wait for the network or for upkeep, so the wrapper leaves a record and the resident service does
6//!  the work on its own thread (`doc/planning/cli-git-resident-service.md`, "After-commit work"). The service is the
7//!  only pusher of the repositories it serves, so no lock is needed: several pending pushes of one branch become one
8//!  push, and a service that died mid-push is noticed by the lease of its claims running out, never by a process id: the
9//!  holder renews the lease while it works, so a successor that starts beside a holder that is still running leaves its claims
10//!  alone, and one that starts after the holder died takes them back as soon as the lease is over. Other kinds of work
11//!  (the standards review) are never claimed here.
12//!
13//! In TS you'd write (pseudocode):
14//! ```ts
15//! for (const group of coalesce(await claimAll(workDir, ['push', 'maintenance']))) settle(group, await handler(group));
16//! ```
17
18use crate::auto_push::push_once;
19use crate::journald_native::Priority;
20use crate::auto_push_single_flight::PushGit;
21use crate::child_environment::{environment_with_overlay, lockfile_pid_overlay};
22use crate::config_snapshot::read_config_snapshot;
23use crate::lfs_pre_push::uses_lfs;
24use crate::maintenance_run::{AutoMaintenanceOutcome, run_auto_maintenance_with};
25use crate::auto_push::AutoPushOutcome;
26use crate::claim_time::ClaimTime;
27use crate::push_failure_notice::{clear_push_failures_before, record_push_failure, record_push_skip};
28use crate::push_retry::{Due, PushRetries, RetryChange, Retrying, final_failure, pause_before, plain_words, unrecognised_rejection};
29use crate::repo_read_config_text::{last_entry, parse_bool, parse_config_text};
30use crate::service_claim_renewal::{ClaimRenewal, HeldClaims, Renewal, held_claims, renewal_interval};
31use crate::service_review::{options_for, review_item, review_variables};
32use crate::parallel_budget::HostResources;
33use crate::service_review_key::ReviewKey;
34use crate::service_round::{Assignment, RoundAnswer, RoundDrain, drain_directories};
35use crate::service_timing::log_entry;
36use crate::standards_review_run::ReviewOptions;
37use crate::transaction_git::GitContext;
38use crate::work_item::{Claimed, DONE, FAILED, WORK_DIRECTORY, WorkItem, WorkKind, claim_next_of, next_claim_expiry, requeue_expired_reporting};
39use std::path::{Path, PathBuf};
40use std::time::{Duration, Instant};
41
42/// The kinds of work the service runs itself.
43pub const SERVICE_KINDS: [WorkKind; 4] = [WorkKind::Push, WorkKind::Maintenance, WorkKind::Review, WorkKind::Trash];
44
45/// How many times an item is tried before it is given up on.
46pub const MAX_TRIES: u32 = 3;
47
48/// What a wake message brings: the repository to work on and the review credential that came with it, if any.
49pub type Wake = (PathBuf, Option<ReviewKey>);
50
51/// What a handler came to.
52#[derive(Clone, Debug, Eq, PartialEq)]
53pub enum Verdict {
54    /// The work was done, or there was nothing to do.
55    Done,
56    /// The work failed; the text is what the person should be told.
57    Failed(String),
58    /// Nothing was done and nothing is left to retry, but the person should be told why; the text is what they are told.
59    Skipped(String),
60    /// The push failed for a reason that can pass (a stall, the network, an LFS server error, a held lock, a remote that advanced again);
61    /// the text is what the person should be told, and the service tries the push again on its own, with growing pauses.
62    Retry(String),
63}
64
65/// What runs each kind of work.
66pub trait WorkHandlers {
67    /// Push the branch of these items, which share a repository and a branch; the last is the newest.
68    fn push(&self, items: &[WorkItem]) -> Verdict;
69    /// Run the repository's automatic upkeep.
70    fn maintenance(&self, item: &WorkItem) -> Verdict;
71    /// Review one landed commit into the repository's shadow log.
72    fn review(&self, item: &WorkItem) -> Verdict;
73    /// Empty the directory of left-out files a snapshot-made worktree set aside; a handler that does not know it skips.
74    fn trash(&self, _item: &WorkItem) -> Verdict {
75        return Verdict::Skipped(String::from("this handler does not empty trash directories"));
76    }
77}
78
79/// The work directory of a repository.
80pub fn work_directory(git_dir: &Path) -> PathBuf {
81    return git_dir.join(WORK_DIRECTORY);
82}
83
84/// What:
85///  Whether an item is one this service should act on for this repository.
86/// Why:
87///   Anyone who can write the work directory can write an item, so the fields that become a directory, a reference or a
88///  program argument are checked before they are used.
89///
90/// In TS you'd write (pseudocode):
91/// ```ts
92/// isAcceptable(item, gitDir)
93/// ```
94pub fn is_acceptable(item: &WorkItem, git_dir: &Path) -> bool {
95    return Path::new(item.repository.as_str()).is_absolute()
96        && Path::new(item.repository.as_str()).is_dir()
97        && Path::new(item.git_dir.as_str()) == git_dir
98        && item.branch_ref.starts_with("refs/")
99        && !item.branch_ref.contains(['\0', '\n', ' '])
100        && item.oid.len() >= 40
101        && item.oid.bytes().all(|byte| { return byte.is_ascii_hexdigit(); });
102}
103
104/// Moves the claims to `state` and stops renewing them; a claim that another instance took over is said, and left to it.
105fn settle_all(work_dir: &Path, held: &HeldClaims, claimed: &[Claimed], state: &str) {
106    for item in claimed {
107        if let Err(reason) = held.settle(work_dir, item.id.as_str(), state) {
108            crate::service_warn!("{reason}");
109        }
110    }
111}
112
113/// What:
114///  The claims of `claimed` that this instance still holds, before their work starts.
115/// Why:
116///   A claim whose renewal found it gone was taken over: another instance runs that item now, so running it here as well would
117///  do it twice. The claim is forgotten and the loss said, and nothing else is done for it.
118///
119/// In TS you'd write (pseudocode):
120/// ```ts
121/// claimed.filter((c) => held.isHeld(c.id))
122/// ```
123fn still_held(held: &HeldClaims, work_dir: &Path, claimed: Vec<Claimed>) -> Vec<Claimed> {
124    let (mine, lost): (Vec<Claimed>, Vec<Claimed>) = claimed.into_iter().partition(|item| { return held.is_held_on_disk(work_dir, item.id.as_str()); });
125    for item in lost {
126        crate::service_warn!("the claim on {} ran out before this instance could run it, so another instance took the work over and it is not run here",
127            item.id
128        );
129        held.release(item.id.as_str());
130    }
131    return mine;
132}
133
134/// What:
135///  Whether a common Git directory belongs to a repository with no main worktree, which `core.bare` in its own `config`, or in its
136///  `config.worktree` when `extensions.worktreeConfig` turns that file on, says.
137/// Why:
138///   A bare repository has no instance whose Git directory is the common one, so the instances of its linked worktrees are the
139///  only ones that can run the records left there; a repository with a main worktree has that instance, which owns them.
140///   Git moves `core.bare` into `config.worktree` when the extension is enabled (`git help config`, "extensions.worktreeConfig"),
141///  and reads that file after `config` only while the extension is on and the repository has a format version
142///  (`repo_read_layout.rs`), so the last `core.bare` of the two decides. An unreadable or unparsable file, or a spelling this
143///  reader leaves to Git, reads as not bare, and the records then stay for the main worktree's instance, which is the safe side
144///  of a wrong answer.
145///
146/// In TS you'd write (pseudocode):
147/// ```ts
148/// isBareCommonDirectory(commonDir)
149/// ```
150fn is_bare_common_directory(common: &Path) -> bool {
151    let Ok(text) = std::fs::read(common.join("config")) else {
152        return false;
153    };
154    let Some(mut entries) = parse_config_text(text.as_slice()) else {
155        return false;
156    };
157    let worktree_config: bool = last_entry(entries.as_slice(), "extensions.worktreeconfig").is_some_and(|entry| { return parse_bool(entry.value.as_deref()) == Some(true); })
158        && last_entry(entries.as_slice(), "core.repositoryformatversion").is_some();
159    if worktree_config {
160        // A missing file is an empty file; one that is there and cannot be read or parsed leaves the answer to Git.
161        match std::fs::read(common.join("config.worktree")) {
162            Ok(bytes) => match parse_config_text(bytes.as_slice()) {
163                Some(more) => entries.extend(more),
164                None => return false,
165            },
166            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
167            Err(_) => return false,
168        }
169    }
170    return last_entry(entries.as_slice(), "core.bare").is_some_and(|entry| { return parse_bool(entry.value.as_deref()) == Some(true); });
171}
172
173/// What:
174///  Leaves a notice for each push the service gave up on without ever finishing it: its holder was stopped, or lost, on every
175///  try.
176/// Why:
177///   Such an item moves to `failed` by recovery, not by a handler, so nothing else records why the commit was never backed up
178///  and the person would not be told. The notice is the one a skipped push leaves, because nothing failed and nothing was sent.
179///
180/// In TS you'd write (pseudocode):
181/// ```ts
182/// for (const item of givenUp.filter(isPush)) await recordPushSkip(gitDir, item, reason)
183/// ```
184fn tell_about_given_up(git_dir: &Path, given_up: &[WorkItem]) {
185    for item in given_up.iter().filter(|item| { return item.kind == WorkKind::Push; }) {
186        let reason: String = format!(
187            "cli-git: the commit service stopped {MAX_TRIES} times before it finished pushing {} (a restart, a kill or a crash each time), so it gave up on that push.",
188            item.branch_ref
189        );
190        if let Err(problem) = record_push_skip(git_dir, item.branch_ref.as_str(), item.oid.as_str(), reason.as_str()) {
191            crate::service_error!("the abandoned push of {} could not be recorded for the person: {problem}", item.oid);
192        }
193    }
194}
195
196/// What:
197///  What one drain of a directory hands back: how long until it must be looked at again for a claim another instance holds, and
198///  the changes it asks of the retry plan.
199/// Why:
200///   A drain runs on a worker and cannot change the plan, which the service loop owns; it returns what it did and the loop
201///  applies it (`PushRetries::apply`).
202///
203/// In TS you'd write (pseudocode):
204/// ```ts
205/// type Drained = { wait?: number; changes: RetryChange[] };
206/// ```
207#[derive(Debug, Default)]
208pub struct Drained {
209    /// How long until a claim another instance holds runs out its lease, or nothing when there is none.
210    pub wait: Option<Duration>,
211    /// The plans and the forgets the drain asks of the retry plan, in the order they were made.
212    pub changes: Vec<RetryChange>,
213}
214
215/// What:
216///  The retry state one drain works with: the first pause (nothing when a retryable push is to be reported as failed) and the
217///  changes it has asked of the plan so far.
218/// Why:
219///   One value is passed down to the places that plan or forget a retry, so none of them reads the plan or shares it.
220///
221/// In TS you'd write (pseudocode):
222/// ```ts
223/// type Replanning = { first?: number; changes: RetryChange[] };
224/// ```
225struct Replanning {
226    /// The pause before the first new try, or nothing for a drain that keeps no plan.
227    first: Option<Duration>,
228    /// What was asked of the plan.
229    changes: Vec<RetryChange>,
230}
231
232/// Methods over the state.
233impl Replanning {
234    /// Ask the plan to drop the branch: its push landed, or the person must act.
235    fn forget(&mut self, git_dir: &Path, branch_ref: &str) {
236        self.changes.push(RetryChange::Forget { git_dir: git_dir.to_path_buf(), branch_ref: branch_ref.to_string() });
237    }
238}
239
240/// What:
241///  Plan the next try of a push that failed for a reason that can pass, say so in the journal, and give the words the person's
242///  failure record carries.
243/// Why:
244///   The journal line names the cause, the number of the try that is planned and the pause before it, so the service's quiet
245///  between tries is explained; the person's record says the service keeps trying, so they do not push by hand for nothing.
246///  The plan is asked of the service loop (`Replanning`), which owns it.
247///
248/// In TS you'd write (pseudocode):
249/// ```ts
250/// const said = plannedRetry(gitDir, replanning, first, item, attempt, text);
251/// ```
252fn planned_retry(git_dir: &Path, replanning: &mut Replanning, first: Duration, item: &WorkItem, attempt: u32, text: &str) -> String {
253    let pause: Duration = pause_before(attempt, first);
254    replanning.changes.push(RetryChange::Plan { git_dir: git_dir.to_path_buf(), item: item.clone(), attempt, due: Instant::now() + pause });
255    let plan: String = format!(
256        "cli-git: the commit service tries this push again by itself in {} seconds (try {}), and keeps trying with longer pauses until the push lands or a conflict needs you.",
257        pause.as_secs(),
258        attempt + 1
259    );
260    let fields: [(&str, String); 3] = [
261        ("CLI_GIT_BRANCH", item.branch_ref.clone()),
262        ("CLI_GIT_OID", item.oid.clone()),
263        ("CLI_GIT_OUTCOME", String::from("retrying")),
264    ];
265    log_entry(
266        Priority::Warning,
267        format!("push will be tried again branch={:?} oid={} next_try={} in_seconds={} cause: {}", item.branch_ref, item.oid, attempt + 1, pause.as_secs(), text.lines().next().unwrap_or_default()).as_str(),
268        &fields,
269    );
270    return format!("{plan}\n{text}");
271}
272
273/// What:
274///  Push again every branch of this Git directory whose pause is over, with no new commit needed.
275/// Why:
276///   The work loop looks at the repository when `drain` tells it to (the wait it returns), and this is what the look does for
277///  a push the service stopped: it lands and the person's failure record goes, it is planned again with a longer pause, or it
278///  is given up on because a conflict needs the person. The due pushes come with the directory's request, taken from the plan
279///  by the service loop before the round.
280///
281/// In TS you'd write (pseudocode):
282/// ```ts
283/// for (const due of request.due.filter((d) => d.gitDir === gitDir)) await settleRetry(await handlers.push([due.item]));
284/// ```
285fn run_due_push_retries(git_dir: &Path, handlers: &dyn WorkHandlers, candidates: &[Due], replanning: &mut Replanning) {
286    for due in candidates.iter().filter(|candidate| { return candidate.git_dir == git_dir; }) {
287        let branch: &str = due.item.branch_ref.as_str();
288        let commit: &str = due.item.oid.as_str();
289        if !is_acceptable(&due.item, git_dir) {
290            crate::service_warn!("the push of {branch} is not tried again: its repository or branch record is no longer acceptable");
291            replanning.forget(git_dir, branch);
292            continue;
293        }
294        let pushing_since: std::time::SystemTime = std::time::SystemTime::now();
295        let began: Instant = Instant::now();
296        let verdict: Verdict = handlers.push(std::slice::from_ref(&due.item));
297        match verdict {
298            Verdict::Done => {
299                clear_push_failures_before(git_dir, branch, pushing_since);
300                replanning.forget(git_dir, branch);
301                crate::service_log!("the push of {branch:?} landed on try {} push_ms={}", due.attempt + 1, began.elapsed().as_millis());
302            }
303            Verdict::Retry(text) => {
304                let said: String = match replanning.first {
305                    Some(first) => planned_retry(git_dir, replanning, first, &due.item, due.attempt + 1, text.as_str()),
306                    None => {
307                        replanning.forget(git_dir, branch);
308                        text
309                    }
310                };
311                if let Err(reason) = record_push_failure(git_dir, branch, commit, said.as_str()) {
312                    crate::service_error!("the failed push of {commit} could not be recorded for the person: {reason}");
313                }
314            }
315            Verdict::Failed(text) => {
316                crate::service_warn!("the push of {branch:?} failed on try {} and needs you: {}", due.attempt + 1, text.lines().next().unwrap_or_default());
317                replanning.forget(git_dir, branch);
318                if let Err(reason) = record_push_failure(git_dir, branch, commit, text.as_str()) {
319                    crate::service_error!("the failed push of {commit} could not be recorded for the person: {reason}");
320                }
321            }
322            Verdict::Skipped(text) => {
323                replanning.forget(git_dir, branch);
324                if !text.is_empty() && let Err(reason) = record_push_skip(git_dir, branch, commit, text.as_str()) {
325                    crate::service_error!("the skipped push of {commit} could not be recorded for the person: {reason}");
326                }
327            }
328        }
329    }
330}
331
332/// How long a claim goes without a renewal before another instance runs its item again; the holder renews it three times in this
333/// span while it works (`service_claim_renewal.rs`), so a holder that is slow keeps its claim and one that died loses it
334/// after this long.
335///  Measured by killing a throwaway service during a claim: the service binary in a fixture repository claimed a push held by the
336///  remote's hook, was killed with SIGKILL 1.5 s after the claim (with its process group), and a new service was started and woken;
337///  the push was made again 119.57, 119.56 and 119.55 s after the kill in three runs (2026-11-03), so the cost of a kill is the lease
338///  to within half a second. No kill of a commit service appears in the user journal of this host (19 boots), so the kills a person
339///  meets were not counted. The renewal thread is independent of the landing (`service_claim_renewal.rs`), so a landing that stalled for
340///  119.8 s (the longest `total_ms` in the service journal) does not stop renewals; a stall of the whole process longer than the
341///  80 s a renewal may be late would. Kept at 120 s as the coordinator's pick under the owner's delegation, open to the owner's veto
342///  (`doc/planning/cli-git-shipped-limits-audit.md`, "Picked for these follow-ups", batch size and claim lease, option B).
343pub const CLAIM_LEASE: Duration = Duration::from_secs(120);
344
345/// What:
346///  Runs everything pending for one repository: takes back the claims whose holder stopped renewing them, claims every pending
347///  push, maintenance and review item, runs one push per branch and one upkeep per repository, settles the items, and leaves a
348///  notice for each failed or skipped push. Gives how long until a claim another instance still holds runs out its lease, or
349///  nothing when there is none.
350/// Why:
351///   Coalescing keeps a burst of commits to one push; a handler that fails leaves the items in `failed`, not in a loop. A
352///  claim found in this directory may belong to a live instance (a restarted service starts beside the one it replaces, and the
353///  linked worktrees of a bare repository share a directory), so only its lease says it was abandoned; the caller is told when
354///  to look again, because nothing else would wake it once the lease is over. This keeps no retry plan: a push that could pass
355///  is reported as failed, as for handlers that do not retry. The service's rounds use `drain_planned` instead.
356///
357/// In TS you'd write (pseudocode):
358/// ```ts
359/// const wait = await drain(gitDir, handlers, readClaimTime())
360/// ```
361pub fn drain(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime) -> Option<Duration> {
362    return drain_with_lease(git_dir, handlers, now, CLAIM_LEASE);
363}
364
365/// `drain` with the lease given, so a test can use a short one.
366pub fn drain_with_lease(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration) -> Option<Duration> {
367    return drain_with_renewal(git_dir, handlers, now, lease, Renewal::Timed);
368}
369
370/// `drain_with_lease` with the way its claims are kept fresh given,
371/// so a control can drive the renewal itself.
372pub fn drain_with_renewal(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration, renewal: Renewal<'_>) -> Option<Duration> {
373    return drain_planned_with(git_dir, handlers, now, lease, renewal, &Retrying::none()).wait;
374}
375
376/// `drain` for a service round: the pushes due again come with the request, and the changes the drain asks of the retry plan are
377/// given back with the wait, for the service loop to apply.
378pub fn drain_planned(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, retrying: &Retrying) -> Drained {
379    return drain_planned_with(git_dir, handlers, now, CLAIM_LEASE, Renewal::Timed, retrying);
380}
381
382/// `drain_planned` with the lease and the renewal given.
383pub fn drain_planned_with(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration, renewal: Renewal<'_>, retrying: &Retrying) -> Drained {
384    let work_dir: PathBuf = work_directory(git_dir);
385    let mut replanning: Replanning = Replanning { first: retrying.first, changes: Vec::new() };
386    tell_about_given_up(git_dir, requeue_expired_reporting(work_dir.as_path(), now, lease, MAX_TRIES).given_up.as_slice());
387    run_due_push_retries(git_dir, handlers, retrying.due.as_slice(), &mut replanning);
388    drain_pending(git_dir, handlers, lease, renewal, &mut replanning);
389    let wait: Option<Duration> = next_claim_expiry(work_dir.as_path(), &ClaimTime::read(), lease);
390    return Drained { wait, changes: replanning.changes };
391}
392
393/// What:
394///  Runs the records the releases before one service per worktree left in the common directory of a bare repository, from the
395///  instance of a linked worktree: nothing for a repository with a main worktree, for a main worktree and for a directory with
396///  no `commondir`.
397/// Why:
398///   Those releases kept every worktree's records in the common directory. A repository with a main worktree has an instance
399///  whose own directory that is, and it runs them (the main worktree's current records live there too), so a linked instance
400///  must not. A repository whose worktrees are all linked has no such instance, and its pushes would sit there with no notice.
401///   Several linked instances may drain the directory at once, so a claim is taken back only after its lease, which the holder
402///  renews while it works.
403///
404/// In TS you'd write (pseudocode):
405/// ```ts
406/// const wait = await drainCommonRecords(linkedGitDir, handlers, now)
407/// ```
408pub fn drain_common_records(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime) -> Option<Duration> {
409    return drain_common_records_with_lease(git_dir, handlers, now, CLAIM_LEASE);
410}
411
412/// `drain_common_records` with the lease given, so a test can use a short one.
413pub fn drain_common_records_with_lease(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration) -> Option<Duration> {
414    return drain_common_records_planned(git_dir, handlers, now, lease, None).wait;
415}
416
417/// `drain_common_records` for a service round: `first` is the pause before the first new try of a push that could pass (nothing
418/// keeps no plan), and the plans the drain asks for are keyed by the common directory, which the service loop wakes by itself
419/// when one is due; that wake drains the common directory as a directory of its own, which has no `commondir`.
420pub fn drain_common_records_planned(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration, first: Option<Duration>) -> Drained {
421    let Some(common) = common_directory_of(git_dir) else {
422        return Drained::default();
423    };
424    if common == git_dir || !is_bare_common_directory(common.as_path()) || !work_directory(common.as_path()).is_dir() {
425        return Drained::default();
426    }
427    let mut replanning: Replanning = Replanning { first, changes: Vec::new() };
428    tell_about_given_up(common.as_path(), requeue_expired_reporting(work_directory(common.as_path()).as_path(), now, lease, MAX_TRIES).given_up.as_slice());
429    drain_pending(common.as_path(), handlers, lease, Renewal::Timed, &mut replanning);
430    let wait: Option<Duration> = next_claim_expiry(work_directory(common.as_path()).as_path(), &ClaimTime::read(), lease);
431    return Drained { wait, changes: replanning.changes };
432}
433
434/// The common directory a linked worktree's Git directory names in its `commondir` file, or nothing for any other directory.
435pub(crate) fn common_directory_of(git_dir: &Path) -> Option<PathBuf> {
436    let text: String = std::fs::read_to_string(git_dir.join("commondir")).ok()?;
437    let named: &Path = Path::new(text.trim_end_matches(['\n', '\r']));
438    if named.as_os_str().is_empty() {
439        return None;
440    }
441    // Resolved by the text alone, so the result spells the path the way the worktree's own Git directory does, which is how the
442    //  records name their directory.
443    let mut resolved: PathBuf = PathBuf::new();
444    for part in git_dir.join(named).components() {
445        match part {
446            std::path::Component::ParentDir => {
447                if !resolved.pop() {
448                    return None;
449                }
450            }
451            std::path::Component::CurDir => {}
452            other => resolved.push(other.as_os_str()),
453        }
454    }
455    return Some(resolved);
456}
457
458/// Claims and runs every pending item of one work directory until none is left; the claims held are renewed every third of
459/// `lease`, so that none of them looks abandoned to another instance while it waits or runs.
460fn drain_pending(git_dir: &Path, handlers: &dyn WorkHandlers, lease: Duration, renewal: Renewal<'_>, replanning: &mut Replanning) {
461    let work_dir: PathBuf = work_directory(git_dir);
462    let (held, renewer) = held_claims(work_dir.as_path());
463    let _renewing: ClaimRenewal = match renewal {
464        Renewal::Timed => ClaimRenewal::start(renewer, renewal_interval(lease)),
465        Renewal::Driven(driver) => ClaimRenewal::driven(renewer, driver),
466    };
467    loop {
468        let mut claimed: Vec<Claimed> = Vec::new();
469        loop {
470            // Each claim carries the reading of the moment it is taken, not of the start of the round, which a long round makes
471            //  most of a lease old.
472            match claim_next_of(work_dir.as_path(), &ClaimTime::read(), &SERVICE_KINDS) {
473                Ok(Some(item)) => {
474                    held.hold(&item);
475                    claimed.push(item);
476                }
477                Ok(None) => break,
478                Err(reason) => {
479                    crate::service_error!("{reason}");
480                    break;
481                }
482            }
483        }
484        if claimed.is_empty() {
485            return;
486        }
487        let (acceptable, refused): (Vec<_>, Vec<_>) = claimed.into_iter().partition(|item| { return is_acceptable(&item.item, git_dir); });
488        settle_all(work_dir.as_path(), &held, &refused, FAILED);
489        let (pushes, rest): (Vec<_>, Vec<_>) = acceptable.into_iter().partition(|item| { return item.item.kind == WorkKind::Push; });
490        let (trashes, after_trash): (Vec<_>, Vec<_>) = rest.into_iter().partition(|item| { return item.item.kind == WorkKind::Trash; });
491        let (reviews, upkeep): (Vec<_>, Vec<_>) = after_trash.into_iter().partition(|item| { return item.item.kind == WorkKind::Review; });
492        let mut groups: Vec<((String, String), Vec<Claimed>)> = Vec::new();
493        for item in pushes {
494            let key = (item.item.repository.clone(), item.item.branch_ref.clone());
495            match groups.iter_mut().find(|(existing, _)| { return *existing == key; }) {
496                Some((_, members)) => members.push(item),
497                None => groups.push((key, vec![item])),
498            }
499        }
500        for (_, claimed_members) in groups {
501            let members: Vec<Claimed> = still_held(&held, work_dir.as_path(), claimed_members);
502            if members.is_empty() {
503                continue;
504            }
505            let items: Vec<WorkItem> = members.iter().map(|member| { return member.item.clone(); }).collect();
506            // The push carries what existed when it began; a notice written after this moment is about something newer.
507            let pushing_since: std::time::SystemTime = std::time::SystemTime::now();
508            let pushing: Instant = Instant::now();
509            let verdict: Verdict = handlers.push(items.as_slice());
510            if let Some(newest) = members.last() {
511                // One line for every push the service ran, so a branch that was not backed up (or was) is found by its name.
512                let (priority, outcome, said): (Priority, &str, &str) = match &verdict {
513                    Verdict::Done => (Priority::Info, "done", ""),
514                    Verdict::Failed(text) => (Priority::Warning, "failed", text.as_str()),
515                    Verdict::Retry(text) => (Priority::Warning, "retrying", text.as_str()),
516                    Verdict::Skipped(text) => (Priority::Info, "skipped", text.as_str()),
517                };
518                let fields: [(&str, String); 3] = [
519                    ("CLI_GIT_BRANCH", newest.item.branch_ref.clone()),
520                    ("CLI_GIT_OID", newest.item.oid.clone()),
521                    ("CLI_GIT_OUTCOME", outcome.to_string()),
522                ];
523                log_entry(
524                    priority,
525                    format!("pushed branch={:?} oid={} items={} outcome={outcome} push_ms={}{}{said}", newest.item.branch_ref, newest.item.oid, members.len(), pushing.elapsed().as_millis(), if said.is_empty() { "" } else { ": " }).as_str(),
526                    &fields,
527                );
528            }
529            match verdict {
530                Verdict::Done => {
531                    // A push of the branch went through, so an earlier failure to push it is no longer true. The records go
532                    //  before the items settle: a service that dies between the two pushes again, and nothing is lost.
533                    if let Some(newest) = members.last() {
534                        clear_push_failures_before(git_dir, newest.item.branch_ref.as_str(), pushing_since);
535                        replanning.forget(git_dir, newest.item.branch_ref.as_str());
536                    }
537                    settle_all(work_dir.as_path(), &held, &members, DONE);
538                }
539                Verdict::Retry(text) => {
540                    // Without handlers that retry, a retryable push is a failed one.
541                    let said: String = match (members.last(), replanning.first) {
542                        (Some(newest), Some(first)) => planned_retry(git_dir, replanning, first, &newest.item, 1, text.as_str()),
543                        _ => text,
544                    };
545                    if let Some(newest) = members.last()
546                        && let Err(reason) = record_push_failure(git_dir, newest.item.branch_ref.as_str(), newest.item.oid.as_str(), said.as_str())
547                    {
548                        crate::service_error!("the failed push of {} could not be recorded for the person: {reason}", newest.item.oid);
549                    }
550                    settle_all(work_dir.as_path(), &held, &members, FAILED);
551                }
552                Verdict::Failed(text) => {
553                    // A failure the service gives up on is no longer waiting for a new try.
554                    if let Some(newest) = members.last() {
555                        replanning.forget(git_dir, newest.item.branch_ref.as_str());
556                    }
557                    // The record goes before the items settle, so an item in `failed` always has its record.
558                    if let Some(newest) = members.last()
559                        && let Err(reason) = record_push_failure(git_dir, newest.item.branch_ref.as_str(), newest.item.oid.as_str(), text.as_str())
560                    {
561                        crate::service_error!("the failed push of {} could not be recorded for the person: {reason}", newest.item.oid);
562                    }
563                    settle_all(work_dir.as_path(), &held, &members, FAILED);
564                }
565                Verdict::Skipped(text) => {
566                    // Nothing was pushed, so an earlier failure of the branch is still true and stays. A skip with words leaves
567                    //  its own record before the items settle; one that could not be recorded settles as failed, so no item
568                    //  is done without the person having been told.
569                    let recorded: bool = match members.last() {
570                        Some(newest) if !text.is_empty() => match record_push_skip(git_dir, newest.item.branch_ref.as_str(), newest.item.oid.as_str(), text.as_str()) {
571                            Ok(()) => true,
572                            Err(reason) => {
573                                crate::service_error!("the skipped push of {} could not be recorded for the person: {reason}", newest.item.oid);
574                                false
575                            }
576                        },
577                        _ => true,
578                    };
579                    settle_all(work_dir.as_path(), &held, &members, if recorded { DONE } else { FAILED });
580                }
581            }
582        }
583        let mut upkept: Vec<String> = Vec::new();
584        for item in still_held(&held, work_dir.as_path(), upkeep) {
585            if !upkept.contains(&item.item.repository) {
586                upkept.push(item.item.repository.clone());
587                let upkeep_began: Instant = Instant::now();
588                match handlers.maintenance(&item.item) {
589                    Verdict::Failed(reason) | Verdict::Retry(reason) => crate::service_warn!("automatic maintenance did not run: {reason}"),
590                    Verdict::Done | Verdict::Skipped(_) => crate::service_log!("automatic maintenance ran maintenance_ms={}", upkeep_began.elapsed().as_millis()),
591                };
592            }
593            settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE);
594        }
595        // The left-out files of a snapshot-made worktree are emptied after the person's backup and the upkeep, before the shadow
596        //  reviews: freeing them is a bounded, local job.
597        for item in still_held(&held, work_dir.as_path(), trashes) {
598            match handlers.trash(&item.item) {
599                Verdict::Done => settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE),
600                Verdict::Failed(reason) | Verdict::Retry(reason) => {
601                    eprintln!("cli-git-service: the trash {} could not be emptied: {reason}", item.item.parent);
602                    settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), FAILED);
603                }
604                Verdict::Skipped(reason) => {
605                    eprintln!("cli-git-service: the trash {} was left as it is: {reason}", item.item.parent);
606                    settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE);
607                }
608            }
609        }
610        // Reviews go last: a push is the person's backup, a review is a shadow record, and each is bounded by its budget.
611        for item in still_held(&held, work_dir.as_path(), reviews) {
612            let review_began: Instant = Instant::now();
613            let verdict: Verdict = handlers.review(&item.item);
614            if matches!(verdict, Verdict::Done) {
615                crate::service_log!("the review of {} is done review_ms={}", item.item.oid, review_began.elapsed().as_millis());
616            }
617            match verdict {
618                Verdict::Done => settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE),
619                Verdict::Failed(reason) | Verdict::Retry(reason) => {
620                    crate::service_warn!("the review of {} failed: {reason}", item.item.oid);
621                    settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), FAILED);
622                }
623                Verdict::Skipped(reason) => {
624                    crate::service_log!("the review of {} was skipped: {reason}", item.item.oid);
625                    settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE);
626                }
627            }
628        }
629    }
630}
631
632/// What:
633///  Runs one round of wakes: holds each wake's review credential in memory for the round, drains each named repository once,
634///  repositories side by side, and drops the credentials when the round ends. Gives, for each repository that has a claim
635///  another instance holds, how long until that claim's lease runs out, in the order the repositories were named.
636/// Why:
637///   The credential is held only for the reviews of the round that brought it; a service that restarts before the round
638///  runs has none, and its reviews record the plain "no key" outcome. The service looks at such a repository again when the
639///  time is over, because the holder may have died and nothing else would wake the work. Repositories share nothing, so the
640///  round drains them side by side (`service_round.rs`); the credentials are passed to the workers as values in a copy of the
641///  handlers made for this round, so nothing is shared mutably and no lock is taken.
642///
643/// In TS you'd write (pseudocode):
644/// ```ts
645/// const round = handlers.forRound(keysOf(wakes)); await Promise.all(directories.map((dir) => drain(dir, round)));
646/// ```
647pub fn run_round(wakes: Vec<Wake>, handlers: &GitHandlers, retries: &mut PushRetries, now: &ClaimTime) -> Vec<(PathBuf, Duration)> {
648    return run_round_on(&HostResources::read(), wakes, handlers, retries, now);
649}
650
651/// `run_round` with the host's resources chosen by the caller, so a control states a host of any size.
652pub fn run_round_on(host: &HostResources, wakes: Vec<Wake>, handlers: &GitHandlers, retries: &mut PushRetries, now: &ClaimTime) -> Vec<(PathBuf, Duration)> {
653    let mut directories: Vec<PathBuf> = Vec::new();
654    let mut keys: std::collections::HashMap<PathBuf, ReviewKey> = std::collections::HashMap::new();
655    for (directory, review_key) in wakes {
656        if let Some(key) = review_key {
657            keys.insert(directory.clone(), key);
658        }
659        if !directories.contains(&directory) {
660            directories.push(directory);
661        }
662    }
663    let round: GitHandlers = handlers.for_round(keys);
664    return run_planned_round(host, directories.as_slice(), &round, retries, now);
665}
666
667/// What:
668///  Runs one round over `directories` with the retry plan: takes what is due for each directory out of the plan, hands it to
669///  that directory's worker as an owned value, applies the changes the workers ask once they are done, and gives for each
670///  directory when to look at it again, whichever is first of its leases and its plan.
671/// Why:
672///   The plan is owned by the caller and changed here on the caller's thread only, so the workers share nothing and no lock is
673///  taken. A directory whose drain planned a push under another directory (a common directory of a bare repository) is named
674///  too, so that its wake is not lost.
675///
676/// In TS you'd write (pseudocode):
677/// ```ts
678/// const due = dirs.map((d) => retries.takeDue(d)); const answers = await Promise.all(dirs.map((d, i) => drain(d, due[i]))); retries.apply(answers.flatMap(changes));
679/// ```
680pub fn run_planned_round<H: WorkHandlers + Sync>(host: &HostResources, directories: &[PathBuf], handlers: &H, retries: &mut PushRetries, now: &ClaimTime) -> Vec<(PathBuf, Duration)> {
681    let taken_at: Instant = Instant::now();
682    let first: Duration = retries.first();
683    let assignments: Vec<Assignment> = directories
684        .iter()
685        .map(|directory| { return Assignment { directory: directory.clone(), retrying: Retrying { first: Some(first), due: retries.take_due(directory.as_path(), taken_at) } }; })
686        .collect();
687    let answered: RoundAnswer = drain_directories(&RoundDrain { directories: assignments.as_slice(), handlers, now, host });
688    let mut named: Vec<PathBuf> = directories.to_vec();
689    for change in &answered.changes {
690        if let RetryChange::Plan { git_dir, .. } = change
691            && !named.contains(git_dir)
692        {
693            named.push(git_dir.clone());
694        }
695    }
696    retries.apply(answered.changes);
697    let applied_at: Instant = Instant::now();
698    let mut asked: Vec<(PathBuf, Duration)> = Vec::new();
699    for directory in named {
700        let lease: Option<Duration> = answered.waits.iter().find(|(named_directory, _)| { return *named_directory == directory; }).map(|(_, wait)| { return *wait; });
701        let planned: Option<Duration> = retries.next_wait(directory.as_path(), applied_at);
702        if let Some(wait) = [lease, planned].into_iter().flatten().min() {
703            asked.push((directory, wait));
704        }
705    }
706    return asked;
707}
708
709/// The handlers that run real Git.
710pub struct GitHandlers {
711    /// The context every Git child of the service runs in.
712    pub context: GitContext,
713    /// The review credential of each repository whose wake message carried one, for the round these handlers were made for. Held
714    /// in memory only and never changed after the handlers are made: a round makes its own copy (`for_round`) and drops it when
715    /// the round ends, so the workers of a round only read it and nothing waits for it.
716    review_keys: std::collections::HashMap<PathBuf, ReviewKey>,
717    /// Where tests point the review's gateway call, so no test reaches the real endpoint.
718    #[cfg(test)]
719    review_endpoint: Option<String>,
720    /// Variables a test adds to the environment of the push, such as a short stall window.
721    #[cfg(test)]
722    push_environment: Vec<(std::ffi::OsString, std::ffi::OsString)>,
723}
724
725/// Methods over the real handlers.
726impl GitHandlers {
727    /// Handlers that run real Git in `context`, with no review credential yet.
728    pub fn new(context: GitContext) -> GitHandlers {
729        return GitHandlers {
730            context,
731            review_keys: std::collections::HashMap::new(),
732            #[cfg(test)]
733            review_endpoint: None,
734            #[cfg(test)]
735            push_environment: Vec::new(),
736        };
737    }
738
739    /// The same handlers with variables added to the environment of every push; tests only.
740    #[cfg(test)]
741    pub fn with_push_environment(mut self, variables: &[(&str, &str)]) -> GitHandlers {
742        self.push_environment = variables.iter().map(|(name, value)| { return (std::ffi::OsString::from(name), std::ffi::OsString::from(value)); }).collect();
743        return self;
744    }
745
746    /// The same handlers with the review's gateway call sent to `endpoint`; tests only.
747    #[cfg(test)]
748    pub fn with_review_endpoint(mut self, endpoint: &str) -> GitHandlers {
749        self.review_endpoint = Some(endpoint.to_string());
750        return self;
751    }
752
753    /// A copy of these handlers that holds `keys` as the review credentials of one round; the copy is dropped with the round.
754    /// The test-only settings (the review endpoint, the push environment) are carried into the copy, so a round runs the way the
755    /// handlers it was made from are set to run.
756    pub fn for_round(&self, keys: std::collections::HashMap<PathBuf, ReviewKey>) -> GitHandlers {
757        return GitHandlers {
758            context: self.context.clone(),
759            review_keys: keys,
760            #[cfg(test)]
761            review_endpoint: self.review_endpoint.clone(),
762            #[cfg(test)]
763            push_environment: self.push_environment.clone(),
764        };
765    }
766
767    /// The credential held for this repository, if the round brought one.
768    fn review_key_of(&self, git_dir: &Path) -> Option<ReviewKey> {
769        return self.review_keys.get(git_dir).cloned();
770    }
771
772    fn push_git(&self, item: &WorkItem) -> PushGit {
773        let cwd: PathBuf = PathBuf::from(item.repository.as_str());
774        // Git writes a `~pid.lock` file beside every lock it takes when `core.lockfilePid` is on, which names the owner of a lock
775        //  that a stopped push leaves behind; the service's own environment does not turn it on.
776        #[cfg_attr(not(test), expect(unused_mut, reason = "only a test adds variables"))]
777        let mut inherited: Vec<(std::ffi::OsString, std::ffi::OsString)> = std::env::vars_os().collect();
778        #[cfg(test)]
779        inherited.extend(self.push_environment.iter().cloned());
780        return PushGit {
781            real_git: self.context.real_git.clone(),
782            lfs: read_config_snapshot(&self.context, cwd.as_path()).as_ref().is_none_or(uses_lfs),
783            cwd,
784            environment: environment_with_overlay(inherited.as_slice(), lockfile_pid_overlay(inherited.as_slice()).as_slice()),
785        };
786    }
787}
788
789/// Trait implementation for the real Git.
790impl WorkHandlers for GitHandlers {
791    fn push(&self, items: &[WorkItem]) -> Verdict {
792        let Some(newest) = items.last() else { return Verdict::Done };
793        let git: PushGit = self.push_git(newest);
794        let mut output: Vec<u8> = Vec::new();
795        return match push_once(&git, newest.branch_ref.as_str(), newest.oid.as_str(), &mut output) {
796            // A skip sent nothing: with words they are told; with none (no upstream and no origin) the repository has nowhere
797            //  to back up to and there is nothing to say, but an earlier failure of the branch is still true.
798            Ok(result) if result.outcome == AutoPushOutcome::Skipped => Verdict::Skipped(result.shown),
799            Ok(result) if result.exit_code == 0 => Verdict::Done,
800            // A conflict, a login the service does not have and a server hook's decline need the person; every other failure can pass
801            //  (a stall, the network, an LFS server error, a held lock).
802            Ok(result) => match final_failure(result.shown.as_str()) {
803                Some(cause) => Verdict::Failed(match plain_words(cause) {
804                    Some(words) => format!("{words}\n{}", result.shown),
805                    None => result.shown,
806                }),
807                None => {
808                    // A rejection by the server whose reason is not known is tried again, and named so that it can be learned.
809                    if let Some(line) = unrecognised_rejection(result.shown.as_str()) {
810                        crate::service_warn!("the server rejected the push for a reason this release does not know, so it is tried again: {line}");
811                    }
812                    Verdict::Retry(result.shown)
813                }
814            },
815            Err(error) => Verdict::Retry(format!("{error}")),
816        };
817    }
818
819    fn trash(&self, item: &WorkItem) -> Verdict {
820        return crate::worktree_trash::empty_trash(item);
821    }
822
823    fn maintenance(&self, item: &WorkItem) -> Verdict {
824        return match run_auto_maintenance_with(&self.context, Path::new(item.repository.as_str()), None) {
825            Ok(AutoMaintenanceOutcome::Exited(code)) if code != 0 => Verdict::Failed(format!("it exited with status {code}")),
826            Ok(_) => Verdict::Done,
827            Err(error) => Verdict::Failed(format!("{error:?}")),
828        };
829    }
830
831    fn review(&self, item: &WorkItem) -> Verdict {
832        let key: Option<ReviewKey> = self.review_key_of(Path::new(item.git_dir.as_str()));
833        #[cfg_attr(not(test), expect(unused_mut, reason = "only a test replaces the endpoint"))]
834        let mut options: ReviewOptions = options_for(item, &review_variables(key.as_ref()));
835        #[cfg(test)]
836        if let Some(endpoint) = self.review_endpoint.as_ref() {
837            options.endpoint = endpoint.clone();
838        }
839        return review_item(&self.context, item, options);
840    }
841}
842
843/// Work consumer controls stay out of the release executable.
844#[cfg(test)]
845#[path = "service_work_tests.rs"]
846mod tests;
847
848/// A push the service stopped is tried again on its own: the controls with a remote that stalls once.
849#[cfg(test)]
850#[path = "service_push_retry_tests.rs"]
851mod push_retry_tests;