1use super::auto_push::push_once;
19use super::journald_native::Priority;
20use super::auto_push_single_flight::PushGit;
21use super::config_snapshot::read_config_snapshot;
22use super::lfs_pre_push::uses_lfs;
23use super::maintenance_run::{AutoMaintenanceOutcome, run_auto_maintenance_with};
24use super::auto_push::AutoPushOutcome;
25use super::claim_time::ClaimTime;
26use super::push_failure_notice::{clear_push_failures_before, record_push_failure, record_push_skip};
27use super::repo_read_config_text::{last_entry, parse_bool, parse_config_text};
28use super::service_claim_renewal::{ClaimRenewal, HeldClaims, Renewal, renewal_interval};
29use super::service_review::{BUDGET, options_for, review_item, review_variables};
30use super::service_review_key::ReviewKey;
31use super::service_timing::log_entry;
32use super::standards_review_run::ReviewOptions;
33use super::transaction_git::GitContext;
34use super::work_item::{Claimed, DONE, FAILED, WORK_DIRECTORY, WorkItem, WorkKind, claim_next_of, next_claim_expiry, requeue_expired_reporting};
35use std::path::{Path, PathBuf};
36use std::time::{Duration, Instant};
37
38pub const SERVICE_KINDS: [WorkKind; 3] = [WorkKind::Push, WorkKind::Maintenance, WorkKind::Review];
40
41pub const MAX_TRIES: u32 = 3;
43
44pub type Wake = (PathBuf, Option<ReviewKey>);
46
47#[derive(Clone, Debug, Eq, PartialEq)]
49pub enum Verdict {
50 Done,
52 Failed(String),
54 Skipped(String),
56}
57
58pub trait WorkHandlers {
60 fn push(&self, items: &[WorkItem]) -> Verdict;
62 fn maintenance(&self, item: &WorkItem) -> Verdict;
64 fn review(&self, item: &WorkItem) -> Verdict;
66}
67
68pub fn work_directory(git_dir: &Path) -> PathBuf {
70 return git_dir.join(WORK_DIRECTORY);
71}
72
73pub fn is_acceptable(item: &WorkItem, git_dir: &Path) -> bool {
84 return Path::new(item.repository.as_str()).is_absolute()
85 && Path::new(item.repository.as_str()).is_dir()
86 && Path::new(item.git_dir.as_str()) == git_dir
87 && item.branch_ref.starts_with("refs/")
88 && !item.branch_ref.contains(['\0', '\n', ' '])
89 && item.oid.len() >= 40
90 && item.oid.bytes().all(|byte| { return byte.is_ascii_hexdigit(); });
91}
92
93fn settle_all(work_dir: &Path, held: &HeldClaims, claimed: &[Claimed], state: &str) {
95 for item in claimed {
96 if let Err(reason) = held.settle(work_dir, item.id.as_str(), state) {
97 crate::service_warn!("{reason}");
98 }
99 }
100}
101
102fn still_held(held: &HeldClaims, work_dir: &Path, claimed: Vec<Claimed>) -> Vec<Claimed> {
113 let (mine, lost): (Vec<Claimed>, Vec<Claimed>) = claimed.into_iter().partition(|item| { return held.is_held_on_disk(work_dir, item.id.as_str()); });
114 for item in lost {
115 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",
116 item.id
117 );
118 held.release(item.id.as_str());
119 }
120 return mine;
121}
122
123fn is_bare_common_directory(common: &Path) -> bool {
140 let Ok(text) = std::fs::read(common.join("config")) else {
141 return false;
142 };
143 let Some(mut entries) = parse_config_text(text.as_slice()) else {
144 return false;
145 };
146 let worktree_config: bool = last_entry(entries.as_slice(), "extensions.worktreeconfig").is_some_and(|entry| { return parse_bool(entry.value.as_deref()) == Some(true); })
147 && last_entry(entries.as_slice(), "core.repositoryformatversion").is_some();
148 if worktree_config {
149 match std::fs::read(common.join("config.worktree")) {
151 Ok(bytes) => match parse_config_text(bytes.as_slice()) {
152 Some(more) => entries.extend(more),
153 None => return false,
154 },
155 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
156 Err(_) => return false,
157 }
158 }
159 return last_entry(entries.as_slice(), "core.bare").is_some_and(|entry| { return parse_bool(entry.value.as_deref()) == Some(true); });
160}
161
162fn tell_about_given_up(git_dir: &Path, given_up: &[WorkItem]) {
174 for item in given_up.iter().filter(|item| { return item.kind == WorkKind::Push; }) {
175 let reason: String = format!(
176 "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.",
177 item.branch_ref
178 );
179 if let Err(problem) = record_push_skip(git_dir, item.branch_ref.as_str(), item.oid.as_str(), reason.as_str()) {
180 crate::service_error!("the abandoned push of {} could not be recorded for the person: {problem}", item.oid);
181 }
182 }
183}
184
185pub const CLAIM_LEASE: Duration = Duration::from_secs(120);
189
190pub fn drain(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime) -> Option<Duration> {
206 return drain_with_lease(git_dir, handlers, now, CLAIM_LEASE);
207}
208
209pub fn drain_with_lease(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration) -> Option<Duration> {
211 return drain_with_renewal(git_dir, handlers, now, lease, Renewal::Timed);
212}
213
214pub fn drain_with_renewal(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration, renewal: Renewal<'_>) -> Option<Duration> {
217 let work_dir: PathBuf = work_directory(git_dir);
218 tell_about_given_up(git_dir, requeue_expired_reporting(work_dir.as_path(), now, lease, MAX_TRIES).given_up.as_slice());
219 drain_pending(git_dir, handlers, lease, renewal);
220 return next_claim_expiry(work_dir.as_path(), &ClaimTime::read(), lease);
221}
222
223pub fn drain_common_records(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime) -> Option<Duration> {
239 return drain_common_records_with_lease(git_dir, handlers, now, CLAIM_LEASE);
240}
241
242pub fn drain_common_records_with_lease(git_dir: &Path, handlers: &dyn WorkHandlers, now: &ClaimTime, lease: Duration) -> Option<Duration> {
244 let common = common_directory_of(git_dir)?;
245 if common == git_dir || !is_bare_common_directory(common.as_path()) || !work_directory(common.as_path()).is_dir() {
246 return None;
247 }
248 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());
249 drain_pending(common.as_path(), handlers, lease, Renewal::Timed);
250 return next_claim_expiry(work_directory(common.as_path()).as_path(), &ClaimTime::read(), lease);
251}
252
253fn common_directory_of(git_dir: &Path) -> Option<PathBuf> {
255 let text: String = std::fs::read_to_string(git_dir.join("commondir")).ok()?;
256 let named: &Path = Path::new(text.trim_end_matches(['\n', '\r']));
257 if named.as_os_str().is_empty() {
258 return None;
259 }
260 let mut resolved: PathBuf = PathBuf::new();
263 for part in git_dir.join(named).components() {
264 match part {
265 std::path::Component::ParentDir => {
266 if !resolved.pop() {
267 return None;
268 }
269 }
270 std::path::Component::CurDir => {}
271 other => resolved.push(other.as_os_str()),
272 }
273 }
274 return Some(resolved);
275}
276
277fn drain_pending(git_dir: &Path, handlers: &dyn WorkHandlers, lease: Duration, renewal: Renewal<'_>) {
280 let work_dir: PathBuf = work_directory(git_dir);
281 let held: HeldClaims = HeldClaims::default();
282 let _renewing: Option<ClaimRenewal> = match renewal {
283 Renewal::Timed => Some(ClaimRenewal::start(work_dir.clone(), held.clone(), renewal_interval(lease))),
284 Renewal::Driven(driver) => {
285 driver.attach(work_dir.clone(), held.clone());
286 None
287 }
288 };
289 loop {
290 let mut claimed: Vec<Claimed> = Vec::new();
291 loop {
292 match claim_next_of(work_dir.as_path(), &ClaimTime::read(), &SERVICE_KINDS) {
295 Ok(Some(item)) => {
296 held.hold(&item);
297 claimed.push(item);
298 }
299 Ok(None) => break,
300 Err(reason) => {
301 crate::service_error!("{reason}");
302 break;
303 }
304 }
305 }
306 if claimed.is_empty() {
307 return;
308 }
309 let (acceptable, refused): (Vec<_>, Vec<_>) = claimed.into_iter().partition(|item| { return is_acceptable(&item.item, git_dir); });
310 settle_all(work_dir.as_path(), &held, &refused, FAILED);
311 let (pushes, rest): (Vec<_>, Vec<_>) = acceptable.into_iter().partition(|item| { return item.item.kind == WorkKind::Push; });
312 let (reviews, upkeep): (Vec<_>, Vec<_>) = rest.into_iter().partition(|item| { return item.item.kind == WorkKind::Review; });
313 let mut groups: Vec<((String, String), Vec<Claimed>)> = Vec::new();
314 for item in pushes {
315 let key = (item.item.repository.clone(), item.item.branch_ref.clone());
316 match groups.iter_mut().find(|(existing, _)| { return *existing == key; }) {
317 Some((_, members)) => members.push(item),
318 None => groups.push((key, vec![item])),
319 }
320 }
321 for (_, claimed_members) in groups {
322 let members: Vec<Claimed> = still_held(&held, work_dir.as_path(), claimed_members);
323 if members.is_empty() {
324 continue;
325 }
326 let items: Vec<WorkItem> = members.iter().map(|member| { return member.item.clone(); }).collect();
327 let pushing_since: std::time::SystemTime = std::time::SystemTime::now();
329 let pushing: Instant = Instant::now();
330 let verdict: Verdict = handlers.push(items.as_slice());
331 if let Some(newest) = members.last() {
332 let (priority, outcome, said): (Priority, &str, &str) = match &verdict {
334 Verdict::Done => (Priority::Info, "done", ""),
335 Verdict::Failed(text) => (Priority::Warning, "failed", text.as_str()),
336 Verdict::Skipped(text) => (Priority::Info, "skipped", text.as_str()),
337 };
338 let fields: [(&str, String); 3] = [
339 ("CLI_GIT_BRANCH", newest.item.branch_ref.clone()),
340 ("CLI_GIT_OID", newest.item.oid.clone()),
341 ("CLI_GIT_OUTCOME", outcome.to_string()),
342 ];
343 log_entry(
344 priority,
345 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(),
346 &fields,
347 );
348 }
349 match verdict {
350 Verdict::Done => {
351 if let Some(newest) = members.last() {
354 clear_push_failures_before(git_dir, newest.item.branch_ref.as_str(), pushing_since);
355 }
356 settle_all(work_dir.as_path(), &held, &members, DONE);
357 }
358 Verdict::Failed(text) => {
359 if let Some(newest) = members.last()
361 && let Err(reason) = record_push_failure(git_dir, newest.item.branch_ref.as_str(), newest.item.oid.as_str(), text.as_str())
362 {
363 crate::service_error!("the failed push of {} could not be recorded for the person: {reason}", newest.item.oid);
364 }
365 settle_all(work_dir.as_path(), &held, &members, FAILED);
366 }
367 Verdict::Skipped(text) => {
368 let recorded: bool = match members.last() {
372 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()) {
373 Ok(()) => true,
374 Err(reason) => {
375 crate::service_error!("the skipped push of {} could not be recorded for the person: {reason}", newest.item.oid);
376 false
377 }
378 },
379 _ => true,
380 };
381 settle_all(work_dir.as_path(), &held, &members, if recorded { DONE } else { FAILED });
382 }
383 }
384 }
385 let mut upkept: Vec<String> = Vec::new();
386 for item in still_held(&held, work_dir.as_path(), upkeep) {
387 if !upkept.contains(&item.item.repository) {
388 upkept.push(item.item.repository.clone());
389 let upkeep_began: Instant = Instant::now();
390 match handlers.maintenance(&item.item) {
391 Verdict::Failed(reason) => crate::service_warn!("automatic maintenance did not run: {reason}"),
392 Verdict::Done | Verdict::Skipped(_) => crate::service_log!("automatic maintenance ran maintenance_ms={}", upkeep_began.elapsed().as_millis()),
393 };
394 }
395 settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE);
396 }
397 for item in still_held(&held, work_dir.as_path(), reviews) {
399 let review_began: Instant = Instant::now();
400 let verdict: Verdict = handlers.review(&item.item);
401 if matches!(verdict, Verdict::Done) {
402 crate::service_log!("the review of {} is done review_ms={}", item.item.oid, review_began.elapsed().as_millis());
403 }
404 match verdict {
405 Verdict::Done => settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE),
406 Verdict::Failed(reason) => {
407 crate::service_warn!("the review of {} failed: {reason}", item.item.oid);
408 settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), FAILED);
409 }
410 Verdict::Skipped(reason) => {
411 crate::service_log!("the review of {} was skipped: {reason}", item.item.oid);
412 settle_all(work_dir.as_path(), &held, std::slice::from_ref(&item), DONE);
413 }
414 }
415 }
416 }
417}
418
419pub fn run_round(wakes: Vec<Wake>, handlers: &GitHandlers, now: &ClaimTime) -> Vec<(PathBuf, Duration)> {
433 let mut directories: Vec<PathBuf> = Vec::new();
434 for (directory, review_key) in wakes {
435 if let Some(key) = review_key {
436 handlers.remember_review_key(directory.as_path(), key);
437 }
438 if !directories.contains(&directory) {
439 directories.push(directory);
440 }
441 }
442 let mut retries: Vec<(PathBuf, Duration)> = Vec::new();
443 for directory in directories {
444 let waits: [Option<Duration>; 2] = [drain(directory.as_path(), handlers, now), drain_common_records(directory.as_path(), handlers, now)];
445 if let Some(wait) = waits.into_iter().flatten().min() {
446 retries.push((directory.clone(), wait));
447 }
448 handlers.forget_review_key(directory.as_path());
449 }
450 return retries;
451}
452
453pub struct GitHandlers {
455 pub context: GitContext,
457 review_keys: std::sync::Mutex<std::collections::HashMap<PathBuf, ReviewKey>>,
460 #[cfg(test)]
462 review_endpoint: Option<String>,
463}
464
465impl GitHandlers {
467 pub fn new(context: GitContext) -> GitHandlers {
469 return GitHandlers {
470 context,
471 review_keys: std::sync::Mutex::new(std::collections::HashMap::new()),
472 #[cfg(test)]
473 review_endpoint: None,
474 };
475 }
476
477 #[cfg(test)]
479 pub fn with_review_endpoint(mut self, endpoint: &str) -> GitHandlers {
480 self.review_endpoint = Some(endpoint.to_string());
481 return self;
482 }
483
484 pub fn remember_review_key(&self, git_dir: &Path, key: ReviewKey) {
486 self.review_keys.lock().unwrap_or_else(std::sync::PoisonError::into_inner).insert(git_dir.to_path_buf(), key);
487 }
488
489 pub fn forget_review_key(&self, git_dir: &Path) {
491 self.review_keys.lock().unwrap_or_else(std::sync::PoisonError::into_inner).remove(git_dir);
492 }
493
494 fn review_key_of(&self, git_dir: &Path) -> Option<ReviewKey> {
495 return self.review_keys.lock().unwrap_or_else(std::sync::PoisonError::into_inner).get(git_dir).cloned();
496 }
497
498 fn push_git(&self, item: &WorkItem) -> PushGit {
499 let cwd: PathBuf = PathBuf::from(item.repository.as_str());
500 return PushGit {
501 real_git: self.context.real_git.clone(),
502 lfs: read_config_snapshot(&self.context, cwd.as_path()).as_ref().is_none_or(uses_lfs),
503 cwd,
504 environment: std::env::vars_os().collect(),
505 };
506 }
507}
508
509impl WorkHandlers for GitHandlers {
511 fn push(&self, items: &[WorkItem]) -> Verdict {
512 let Some(newest) = items.last() else { return Verdict::Done };
513 let git: PushGit = self.push_git(newest);
514 let mut output: Vec<u8> = Vec::new();
515 return match push_once(&git, newest.branch_ref.as_str(), newest.oid.as_str(), &mut output) {
516 Ok(result) if result.outcome == AutoPushOutcome::Skipped => Verdict::Skipped(result.shown),
519 Ok(result) if result.exit_code == 0 => Verdict::Done,
520 Ok(result) => Verdict::Failed(result.shown),
521 Err(error) => Verdict::Failed(format!("{error}")),
522 };
523 }
524
525 fn maintenance(&self, item: &WorkItem) -> Verdict {
526 return match run_auto_maintenance_with(&self.context, Path::new(item.repository.as_str()), None) {
527 Ok(AutoMaintenanceOutcome::Exited(code)) if code != 0 => Verdict::Failed(format!("it exited with status {code}")),
528 Ok(_) => Verdict::Done,
529 Err(error) => Verdict::Failed(format!("{error:?}")),
530 };
531 }
532
533 fn review(&self, item: &WorkItem) -> Verdict {
534 let key: Option<ReviewKey> = self.review_key_of(Path::new(item.git_dir.as_str()));
535 #[cfg_attr(not(test), expect(unused_mut, reason = "only a test replaces the endpoint"))]
536 let mut options: ReviewOptions = options_for(item, &review_variables(key.as_ref()));
537 #[cfg(test)]
538 if let Some(endpoint) = self.review_endpoint.as_ref() {
539 options.endpoint = endpoint.clone();
540 }
541 return review_item(&self.context, item, options, BUDGET);
542 }
543}
544
545#[cfg(test)]
547#[path = "service_work_tests.rs"]
548mod tests;