steid
Sign in

@jamesgill /

steid

steid/src/infrastructure/cat_file.rs
Rust33.4 KBCode·Blame·History·Raw
1//! Long-lived `git cat-file` processes, pooled per repository.
2//!
3//! Stage 2 of the upgrade path in the Milestone 5 amendment to
4//! [0006](../../plans/decisions/0006-git-binary-behind-narrow-ports.md): every object
5//! read used to cost a fork/exec of ~11–12ms, and the answer is the one Gitaly and
6//! Gitea both reached for — keep the process alive and write another line to it.
7//!
8//! # Two processes, not one
9//!
10//! `git cat-file --batch-check` answers existence, type and size; `git cat-file
11//! --batch` answers contents. They are separate processes because a single `--batch`
12//! stream cannot be asked a question whose answer is not read: writing a spec commits
13//! the reader to consuming however many bytes the object holds, and "is this blob
14//! small enough to read?" is exactly the question that must be answered *before* the
15//! bytes are on the wire.
16//!
17//! **Not `--batch-command`**, which would collapse the pair into one process by letting
18//! each line name its own verb. It arrived in git **2.36**, and Ubuntu 22.04 — which
19//! `install.sh` deliberately supports — ships **2.34**. Two processes cost one more
20//! fork per repository per five minutes; a raised floor costs an operator their install.
21//!
22//! Input is NUL-delimited (`-z`, git 2.31, comfortably below the floor) rather than
23//! newline-delimited, because a git path may contain a newline and a newline in a spec
24//! would otherwise be read as the start of another request.
25//!
26//! # What this module guarantees
27//!
28//! - **Requests serialise per process.** One mutex per process, so a repository can
29//! answer a `--batch-check` and a `--batch` at once but never two of either.
30//! - **A desync is fatal to the process, not to the stream.** If a reply does not parse
31//! as the protocol says it must — a header for a different object, a body not
32//! followed by its newline — the process is killed and the request fails. Reading on
33//! past a reply we did not understand would answer the *next* request with this one's
34//! bytes, which is worse than an error by exactly as much as a wrong file is worse
35//! than a missing one.
36//! - **A crashed process is respawned once per request.** `pkill` on a batch process,
37//! or an OOM kill, shows up as EOF on the next read; the second failure in one
38//! request is a real fault and is reported.
39//! - **The timeout is the same one every other read pays.** A pool request that outruns
40//! it kills the process, so a stuck git cannot hold the mutex for the next visitor.
41
42use std::{
43 collections::HashMap,
44 io::ErrorKind,
45 path::{Path, PathBuf},
46 process::Stdio,
47 sync::Mutex,
48 time::{Duration, Instant},
49};
50
51use tokio::{
52 io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
53 process::{Child, ChildStderr, ChildStdin, ChildStdout},
54 sync::Mutex as AsyncMutex,
55};
56
57use crate::{
58 application::port::GitQueryError,
59 domain::ObjectId,
60 infrastructure::{
61 git::{count_spawn, git_command},
62 metrics::metrics,
63 },
64};
65
66/// How many repositories may hold live processes at once.
67///
68/// Two processes each, so the cap is 128 file descriptors' worth of git and whatever
69/// each has mapped of a pack index — the memory is the reason there is a cap at all,
70/// since a `cat-file` process holds its repository's idx files open. Sixty-four is
71/// chosen rather than derived: a personal instance has single figures of repositories,
72/// and the number only bites on an instance browsing far more than that at once, where
73/// the eviction it causes costs one fork to undo.
74const MAX_LIVE_REPOSITORIES: usize = 64;
75
76/// How long a repository's processes survive with nothing asking them anything.
77///
78/// Long enough that a person reading a repository never pays a spawn twice, short
79/// enough that a crawler walking a hundred repositories does not leave a hundred
80/// processes behind for the rest of the day.
81const IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60);
82
83/// The read budget for one exchange with a pooled process.
84///
85/// The same twenty seconds `git_query`'s one-shot commands get, and for the same
86/// reason: a request that never finishes holds a worker for as long as the client
87/// waits. It is enforced here as well as there because the pool is where the *next*
88/// request queues — a hung `cat-file` holding its mutex would otherwise stall a
89/// repository rather than one page.
90const GIT_TIMEOUT: Duration = Duration::from_secs(20);
91
92/// The words `cat-file` ends a line with when it did not resolve a spec.
93///
94/// `missing` covers the common cases; `ambiguous` is an abbreviated id matching more
95/// than one object, and `dangling` and `notdir` appear when following a `^{}` or a path
96/// through something that cannot hold one. None of them is a failure — they are the
97/// answer "no such thing here".
98const NOT_FOUND_MARKERS: [&str; 4] = ["missing", "ambiguous", "dangling", "notdir"];
99
100/// How many specs are written before the replies are read.
101///
102/// A pipe holds 64 KiB; writing more than that without reading deadlocks, with git
103/// blocked writing replies nobody is draining and us blocked writing specs git is not
104/// reading. Both bounds are well under it, and the byte bound is the one that matters
105/// because a spec can be a path of any length.
106const MAX_BATCH_SPECS: usize = 256;
107const MAX_BATCH_BYTES: usize = 8 * 1024;
108
109/// What `cat-file --batch-check` said about one object.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub(crate) struct ObjectInfo {
112 pub(crate) id: ObjectId,
113 pub(crate) kind: ObjectKind,
114 pub(crate) size: u64,
115}
116
117/// A git object's type, as its header spells it.
118///
119/// Distinct from [`EntryKind`](crate::domain::EntryKind), which is about what a tree
120/// entry *means* — the object store cannot tell a symlink from a file, because both are
121/// blobs.
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
123pub(crate) enum ObjectKind {
124 Blob,
125 Tree,
126 Commit,
127 Tag,
128}
129
130impl ObjectKind {
131 fn from_str(value: &str) -> Option<Self> {
132 match value {
133 "blob" => Some(Self::Blob),
134 "tree" => Some(Self::Tree),
135 "commit" => Some(Self::Commit),
136 "tag" => Some(Self::Tag),
137 _ => None,
138 }
139 }
140}
141
142/// Which of a repository's two processes a request is for.
143#[derive(Debug, Clone, Copy, PartialEq, Eq)]
144enum Mode {
145 Check,
146 Contents,
147}
148
149impl Mode {
150 fn flag(self) -> &'static str {
151 match self {
152 Self::Check => "--batch-check",
153 Self::Contents => "--batch",
154 }
155 }
156}
157
158/// One question for a pooled process.
159///
160/// An enum rather than a closure because the retry in [`CatFilePool::exchange`] runs the
161/// same question twice against two different processes, and a borrow that outlives one
162/// attempt is exactly what a closure cannot promise here.
163#[derive(Debug, Clone, Copy)]
164enum Request<'a> {
165 /// Specs for `--batch-check`, answered one line each.
166 Check(&'a [String]),
167 /// An object `--batch-check` has already described, answered with its bytes.
168 Contents(&'a ObjectInfo),
169}
170
171impl Request<'_> {
172 fn mode(self) -> Mode {
173 match self {
174 Self::Check(_) => Mode::Check,
175 Self::Contents(_) => Mode::Contents,
176 }
177 }
178}
179
180/// What a process answered.
181#[derive(Debug)]
182enum Reply {
183 Check(Vec<Option<ObjectInfo>>),
184 Contents(Vec<u8>),
185}
186
187/// The `git cat-file` processes for one repository.
188///
189/// A mutex per process rather than one around the pair: the two answer independent
190/// questions, and a tree listing that reads a tree from `--batch` and then asks
191/// `--batch-check` for its blobs' sizes has no reason to stop another request doing the
192/// same in the other order.
193#[derive(Debug, Default)]
194struct Slot {
195 check: AsyncMutex<Option<Batch>>,
196 contents: AsyncMutex<Option<Batch>>,
197}
198
199impl Slot {
200 fn process(&self, mode: Mode) -> &AsyncMutex<Option<Batch>> {
201 match mode {
202 Mode::Check => &self.check,
203 Mode::Contents => &self.contents,
204 }
205 }
206}
207
208/// What the pool holds for one repository path.
209#[derive(Debug)]
210struct Entry {
211 slot: std::sync::Arc<Slot>,
212 last_used: Instant,
213}
214
215/// Kept-alive `git cat-file` processes, keyed by repository path.
216///
217/// Lives in app context, not in the adapter: `DiskGitQuery` is constructed per request,
218/// and a pool built per request would spawn and kill a process per request — the fork
219/// cost back, plus a pool that never pools. The Milestone 5 amendment to 0006 names
220/// this as the one thing that had to be got right before the first batch adapter
221/// existed.
222#[derive(Debug, Default)]
223pub struct CatFilePool {
224 repos: Mutex<HashMap<PathBuf, Entry>>,
225}
226
227impl CatFilePool {
228 pub fn new() -> Self {
229 Self::default()
230 }
231
232 /// Asks `--batch-check` about one spec.
233 pub(crate) async fn info(
234 &self,
235 repo: &Path,
236 spec: &str,
237 ) -> Result<Option<ObjectInfo>, GitQueryError> {
238 Ok(self
239 .info_many(repo, std::slice::from_ref(&spec.to_owned()))
240 .await?
241 .pop()
242 .flatten())
243 }
244
245 /// Asks `--batch-check` about many specs, in as few round trips as a pipe allows.
246 ///
247 /// One answer per spec, in the order asked. This is what makes a tree listing's
248 /// blob sizes cheap: the binary tree format carries no sizes, so they are asked for
249 /// in bulk rather than one process — or one round trip — at a time.
250 pub(crate) async fn info_many(
251 &self,
252 repo: &Path,
253 specs: &[String],
254 ) -> Result<Vec<Option<ObjectInfo>>, GitQueryError> {
255 let mut answers = Vec::with_capacity(specs.len());
256
257 for chunk in chunks(specs) {
258 let Reply::Check(replies) = self.exchange(repo, Request::Check(chunk)).await? else {
259 unreachable!("a check is answered with a check")
260 };
261
262 answers.extend(replies);
263 }
264
265 Ok(answers)
266 }
267
268 /// Reads an object's bytes from `--batch`.
269 ///
270 /// Takes the [`ObjectInfo`] a `--batch-check` already produced rather than a spec,
271 /// for two reasons that are really one: the caller has to know the size before it
272 /// decides to read (an oversized blob is never fetched), and knowing the id, type
273 /// and size in advance turns the reply header into a check rather than a parse. A
274 /// header that disagrees is a desync, and the process does not survive it.
275 pub(crate) async fn contents(
276 &self,
277 repo: &Path,
278 info: &ObjectInfo,
279 ) -> Result<Vec<u8>, GitQueryError> {
280 let Reply::Contents(bytes) = self.exchange(repo, Request::Contents(info)).await? else {
281 unreachable!("a contents request is answered with contents")
282 };
283
284 Ok(bytes)
285 }
286
287 /// One exchange with a pooled process: spawn if needed, run, and decide what a
288 /// failure means.
289 ///
290 /// The three outcomes are deliberately not one. A desync fails the request and
291 /// kills the process, because the stream can no longer be trusted. A dead process
292 /// is respawned and the exchange retried **once**, because a `pkill` between two
293 /// requests is not the visitor's fault and is invisible to them. A timeout kills
294 /// the process and fails, because the mutex is held until it does.
295 async fn exchange(&self, repo: &Path, request: Request<'_>) -> Result<Reply, GitQueryError> {
296 // One exchange, whatever it costs: a `--batch-check` of forty specs is one round
297 // trip through one process, which is the whole point of the pool and the number
298 // worth watching beside `steid_git_processes_total`.
299 metrics().record_cat_file_request();
300
301 let mode = request.mode();
302 let slot = self.slot(repo);
303 let mut process = slot.process(mode).lock().await;
304
305 for attempt in 0..2 {
306 if process.is_none() {
307 // A spawn on the second attempt is a *re*spawn: the first attempt found
308 // a process that had gone. A spawn on the first is a cold slot, which is
309 // the pool working rather than a fault, and is not counted here.
310 if attempt == 1 {
311 metrics().record_cat_file_respawn();
312 }
313
314 *process = Some(Batch::spawn(repo, mode)?);
315 }
316
317 // Bound to a `let` rather than matched on directly: the future borrows the
318 // guard, and a match arm that replaces the process needs that borrow to have
319 // ended. A respawn makes a different `Batch`, which is what the second
320 // attempt re-borrows.
321 let outcome = {
322 let batch = process.as_mut().expect("just spawned");
323
324 tokio::time::timeout(GIT_TIMEOUT, batch.run(request)).await
325 };
326
327 match outcome {
328 Ok(Ok(value)) => return Ok(value),
329 Ok(Err(Desync::Protocol(message))) => {
330 // Killed rather than kept: the next request would otherwise read
331 // this one's leftovers as its own answer.
332 *process = None;
333 return Err(GitQueryError::new(format!(
334 "git cat-file {} spoke out of turn: {message}",
335 mode.flag()
336 )));
337 }
338 Ok(Err(Desync::Gone(message))) => {
339 *process = None;
340
341 if attempt == 1 {
342 return Err(GitQueryError::new(format!(
343 "git cat-file {} would not stay running: {message}",
344 mode.flag()
345 )));
346 }
347 }
348 Err(_elapsed) => {
349 *process = None;
350 return Err(GitQueryError::timed_out(GIT_TIMEOUT));
351 }
352 }
353 }
354
355 unreachable!("the loop either returns or retries once")
356 }
357
358 /// The slot for a repository, evicting what has gone stale on the way past.
359 ///
360 /// Eviction is lazy rather than a background task: the pool has no runtime of its
361 /// own, and a sweep on acquire runs exactly as often as the pool is used. A slot
362 /// somebody is still holding is never evicted — that is what the strong count is
363 /// read for — so the worst an eviction costs is one spawn.
364 fn slot(&self, repo: &Path) -> std::sync::Arc<Slot> {
365 let mut repos = self.repos.lock().expect("lock poisoned");
366 let now = Instant::now();
367
368 repos.retain(|_, entry| {
369 std::sync::Arc::strong_count(&entry.slot) > 1
370 || now.duration_since(entry.last_used) < IDLE_TIMEOUT
371 });
372
373 if let Some(entry) = repos.get_mut(repo) {
374 entry.last_used = now;
375 return entry.slot.clone();
376 }
377
378 if repos.len() >= MAX_LIVE_REPOSITORIES {
379 let oldest = repos
380 .iter()
381 .filter(|(_, entry)| std::sync::Arc::strong_count(&entry.slot) == 1)
382 .min_by_key(|(_, entry)| entry.last_used)
383 .map(|(path, _)| path.clone());
384
385 if let Some(oldest) = oldest {
386 repos.remove(&oldest);
387 }
388 }
389
390 let slot = std::sync::Arc::new(Slot::default());
391
392 repos.insert(
393 repo.to_path_buf(),
394 Entry {
395 slot: slot.clone(),
396 last_used: now,
397 },
398 );
399
400 slot
401 }
402
403 /// How many repositories currently hold a slot.
404 ///
405 /// Exposed as `steid_cat_file_pool_size` and read by the tests below. Sampled at
406 /// scrape time rather than mirrored into a counter, so it cannot drift from the map
407 /// it describes; a poisoned lock reports nothing rather than failing a scrape, since
408 /// the whole endpoint going 500 because one gauge could not be read is the wrong
409 /// trade.
410 #[must_use]
411 pub fn live_repositories(&self) -> usize {
412 self.repos.lock().map_or(0, |repos| repos.len())
413 }
414}
415
416/// Splits specs into groups small enough to write without reading.
417fn chunks(specs: &[String]) -> Vec<&[String]> {
418 let mut groups = Vec::new();
419 let mut start = 0;
420 let mut bytes = 0;
421
422 for (index, spec) in specs.iter().enumerate() {
423 let next = bytes + spec.len() + 1;
424
425 if index > start && (index - start >= MAX_BATCH_SPECS || next > MAX_BATCH_BYTES) {
426 groups.push(&specs[start..index]);
427 start = index;
428 bytes = spec.len() + 1;
429 } else {
430 bytes = next;
431 }
432 }
433
434 if start < specs.len() {
435 groups.push(&specs[start..]);
436 }
437
438 groups
439}
440
441/// Why an exchange did not produce an answer.
442#[derive(Debug)]
443enum Desync {
444 /// git said something the protocol does not allow. The stream is not trustworthy.
445 Protocol(String),
446 /// The process is not there any more — EOF, a broken pipe, a signal.
447 Gone(String),
448}
449
450/// One live `git cat-file` process and its pipes.
451#[derive(Debug)]
452struct Batch {
453 child: Child,
454 stdin: ChildStdin,
455 stdout: BufReader<ChildStdout>,
456 stderr: Option<ChildStderr>,
457}
458
459impl Batch {
460 /// Answers one question, leaving the stream positioned for the next.
461 async fn run(&mut self, request: Request<'_>) -> Result<Reply, Desync> {
462 match request {
463 Request::Check(specs) => self.check(specs).await.map(Reply::Check),
464 Request::Contents(info) => self.contents(info).await.map(Reply::Contents),
465 }
466 }
467
468 fn spawn(repo: &Path, mode: Mode) -> Result<Self, GitQueryError> {
469 let mut command = git_command();
470
471 command
472 .arg("-C")
473 .arg(repo)
474 .arg("cat-file")
475 .arg(mode.flag())
476 // NUL-delimited input, so a path containing a newline is one spec rather
477 // than two requests. git 2.31; the floor is 2.34.
478 .arg("-z")
479 .stdin(Stdio::piped())
480 .stdout(Stdio::piped())
481 .stderr(Stdio::piped())
482 // A pool entry dropped — evicted, killed after a desync, or the process
483 // shutting down — must take git with it, or the process outlives everything
484 // that could ever read from it.
485 .kill_on_drop(true);
486 count_spawn(&command);
487
488 let mut child = command
489 .spawn()
490 .map_err(|error| GitQueryError::new(format!("could not run git cat-file: {error}")))?;
491
492 let stdin = child.stdin.take().expect("stdin was piped");
493 let stdout = BufReader::new(child.stdout.take().expect("stdout was piped"));
494 let stderr = child.stderr.take();
495
496 Ok(Self {
497 child,
498 stdin,
499 stdout,
500 stderr,
501 })
502 }
503
504 /// Writes a group of specs and reads one reply line each.
505 async fn check(&mut self, specs: &[String]) -> Result<Vec<Option<ObjectInfo>>, Desync> {
506 let mut request = Vec::new();
507
508 for spec in specs {
509 request.extend_from_slice(spec.as_bytes());
510 request.push(0);
511 }
512
513 self.write(&request).await?;
514
515 let mut answers = Vec::with_capacity(specs.len());
516
517 for _ in specs {
518 let line = self.line().await?;
519
520 answers.push(parse_check_line(&line)?);
521 }
522
523 Ok(answers)
524 }
525
526 /// Writes one spec and reads the object it names.
527 async fn contents(&mut self, info: &ObjectInfo) -> Result<Vec<u8>, Desync> {
528 let mut request = info.id.as_str().as_bytes().to_vec();
529 request.push(0);
530
531 self.write(&request).await?;
532
533 let line = self.line().await?;
534
535 // The caller already knows what it asked for, so anything else is a stream that
536 // has slipped a reply — not an object that has changed.
537 let Some(header) = parse_check_line(&line)? else {
538 return Err(Desync::Protocol(format!(
539 "{} is missing from a repository that had just described it",
540 info.id.as_str()
541 )));
542 };
543
544 if &header != info {
545 return Err(Desync::Protocol(format!(
546 "asked for {:?} and was answered with {:?}",
547 info, header
548 )));
549 }
550
551 // One byte more than the object: git writes a newline after the body, and
552 // reading it is what leaves the stream positioned for the next request.
553 let mut body = vec![0u8; header.size as usize + 1];
554
555 self.stdout
556 .read_exact(&mut body)
557 .await
558 .map_err(|error| Desync::Gone(format!("reading {} bytes: {error}", header.size)))?;
559
560 if body.pop() != Some(b'\n') {
561 return Err(Desync::Protocol(format!(
562 "{} was not followed by a newline",
563 header.id.as_str()
564 )));
565 }
566
567 Ok(body)
568 }
569
570 async fn write(&mut self, request: &[u8]) -> Result<(), Desync> {
571 match self.stdin.write_all(request).await {
572 Ok(()) => self
573 .stdin
574 .flush()
575 .await
576 .map_err(|error| Desync::Gone(error.to_string())),
577 // A process that has died gives EPIPE here rather than at the read, which is
578 // the same fact arriving one step earlier.
579 Err(error) => Err(Desync::Gone(self.why_gone(error).await)),
580 }
581 }
582
583 async fn line(&mut self) -> Result<String, Desync> {
584 let mut line = String::new();
585
586 match self.stdout.read_line(&mut line).await {
587 // Zero bytes is EOF: git closed its stdout, which for a batch process means
588 // it exited.
589 Ok(0) => Err(Desync::Gone(
590 self.why_gone(ErrorKind::UnexpectedEof.into()).await,
591 )),
592 Ok(_) => Ok(line.trim_end_matches('\n').to_owned()),
593 Err(error) => Err(Desync::Gone(self.why_gone(error).await)),
594 }
595 }
596
597 /// What git said on its way out, if anything, for the message a visitor's 500 logs.
598 ///
599 /// Read only when the process is already known to be gone, and capped, because a
600 /// pipe nobody drains is a pipe that can fill.
601 async fn why_gone(&mut self, error: std::io::Error) -> String {
602 let mut said = String::new();
603
604 if let Some(stderr) = self.stderr.as_mut() {
605 let mut buffer = vec![0u8; 4096];
606
607 if let Ok(read) = stderr.read(&mut buffer).await {
608 said = String::from_utf8_lossy(&buffer[..read]).trim().to_owned();
609 }
610 }
611
612 let status = self.child.try_wait().ok().flatten();
613
614 match (said.is_empty(), status) {
615 (false, _) => said,
616 (true, Some(status)) => format!("{error} ({status})"),
617 (true, None) => error.to_string(),
618 }
619 }
620}
621
622/// Parses one `<id> <type> <size>` reply, or the not-found line git answers instead.
623///
624/// The marker is checked before the field count, because a not-found line echoes the
625/// spec back — and a spec naming a file with spaces in it has no fixed field count.
626fn parse_check_line(line: &str) -> Result<Option<ObjectInfo>, Desync> {
627 if line
628 .rsplit(' ')
629 .next()
630 .is_some_and(|last| NOT_FOUND_MARKERS.contains(&last))
631 {
632 return Ok(None);
633 }
634
635 let fields: Vec<&str> = line.split_whitespace().collect();
636
637 let [id, kind, size] = fields[..] else {
638 return Err(Desync::Protocol(format!(
639 "described an object in a shape we do not understand: {line:?}"
640 )));
641 };
642
643 Ok(Some(ObjectInfo {
644 // Validated rather than trusted. git's ids are trustworthy, but this is a parse
645 // of text whose layout we have assumed, and an id is about to appear in a URL —
646 // a misread field should stop here rather than surface as a broken link.
647 id: ObjectId::new(id)
648 .map_err(|error| Desync::Protocol(format!("named a bad object id: {error}")))?,
649 kind: ObjectKind::from_str(kind)
650 .ok_or_else(|| Desync::Protocol(format!("reported an unknown object type {kind:?}")))?,
651 size: size
652 .parse()
653 .map_err(|_| Desync::Protocol(format!("reported an unreadable size {size:?}")))?,
654 }))
655}
656
657#[cfg(test)]
658mod tests {
659 use std::process::Command;
660
661 use tempfile::TempDir;
662
663 use super::*;
664
665 /// A repository with one commit, one file and one directory.
666 fn fixture() -> TempDir {
667 let dir = tempfile::tempdir().expect("temp dir");
668 let path = dir.path();
669
670 run(path, &["init", "--quiet", "--initial-branch=main", "."]);
671 std::fs::create_dir(path.join("sub")).expect("mkdir");
672 std::fs::write(path.join("a.txt"), b"hello\n").expect("write");
673 std::fs::write(path.join("sub/b.txt"), b"nested\n").expect("write");
674 run(path, &["add", "-A"]);
675 run(
676 path,
677 &[
678 "-c",
679 "user.email=a@example.com",
680 "-c",
681 "user.name=A",
682 "commit",
683 "--quiet",
684 "-m",
685 "first",
686 ],
687 );
688
689 dir
690 }
691
692 fn run(dir: &Path, args: &[&str]) {
693 let status = Command::new("git")
694 .arg("-C")
695 .arg(dir)
696 .args(args)
697 .status()
698 .expect("git runs");
699
700 assert!(status.success(), "git {args:?} failed");
701 }
702
703 #[tokio::test]
704 async fn an_object_is_described_by_id_type_and_size() {
705 let dir = fixture();
706 let pool = CatFilePool::new();
707
708 let info = pool
709 .info(dir.path(), "HEAD:a.txt")
710 .await
711 .expect("asked")
712 .expect("found");
713
714 assert_eq!(info.kind, ObjectKind::Blob);
715 assert_eq!(info.size, 6);
716 }
717
718 #[tokio::test]
719 async fn a_spec_that_names_nothing_is_not_a_failure() {
720 let dir = fixture();
721 let pool = CatFilePool::new();
722
723 assert_eq!(
724 pool.info(dir.path(), "HEAD:nope").await.expect("asked"),
725 None
726 );
727 assert_eq!(
728 pool.info(dir.path(), "nosuchref").await.expect("asked"),
729 None
730 );
731 }
732
733 #[tokio::test]
734 async fn contents_come_back_whole() {
735 let dir = fixture();
736 let pool = CatFilePool::new();
737
738 let info = pool
739 .info(dir.path(), "HEAD:a.txt")
740 .await
741 .expect("asked")
742 .expect("found");
743
744 assert_eq!(
745 pool.contents(dir.path(), &info).await.expect("read"),
746 b"hello\n"
747 );
748 }
749
750 #[tokio::test]
751 async fn a_repository_is_asked_many_times_through_one_pair_of_processes() {
752 let dir = fixture();
753 let pool = CatFilePool::new();
754
755 for _ in 0..5 {
756 let info = pool
757 .info(dir.path(), "HEAD:a.txt")
758 .await
759 .expect("asked")
760 .expect("found");
761
762 pool.contents(dir.path(), &info).await.expect("read");
763 }
764
765 assert_eq!(pool.live_repositories(), 1, "one repository, one slot");
766
767 // The processes are the ones spawned on the first call: a respawn would have
768 // replaced the `Option`, and this reads the same two through the same slot.
769 let slot = pool.slot(dir.path());
770
771 assert!(slot.check.lock().await.is_some());
772 assert!(slot.contents.lock().await.is_some());
773 }
774
775 #[tokio::test]
776 async fn an_exchange_and_a_respawn_each_move_their_own_counter() {
777 // The counters are process-wide and tests run concurrently, so this asserts a
778 // rise rather than an absolute value.
779 let metrics = crate::infrastructure::metrics::metrics();
780 let requests = metrics.value_of("steid_cat_file_requests_total");
781 let respawns = metrics.value_of("steid_cat_file_respawns_total");
782
783 let dir = fixture();
784 let pool = CatFilePool::new();
785
786 pool.info(dir.path(), "HEAD:a.txt")
787 .await
788 .expect("asked")
789 .expect("found");
790
791 assert!(
792 metrics.value_of("steid_cat_file_requests_total") > requests,
793 "an exchange must be counted"
794 );
795 // Exactly what `pkill -f 'cat-file --batch-check'` does to a running instance.
796 {
797 let slot = pool.slot(dir.path());
798 let mut process = slot.check.lock().await;
799 let batch = process.as_mut().expect("spawned");
800
801 batch.child.start_kill().expect("kill");
802 batch.child.wait().await.expect("reaped");
803 }
804
805 pool.info(dir.path(), "HEAD:a.txt")
806 .await
807 .expect("respawned and asked")
808 .expect("found");
809
810 assert!(
811 metrics.value_of("steid_cat_file_respawns_total") > respawns,
812 "a process found dead and started again must be counted"
813 );
814 }
815
816 #[tokio::test]
817 async fn a_killed_process_is_replaced_on_the_next_request() {
818 let dir = fixture();
819 let pool = CatFilePool::new();
820
821 let info = pool
822 .info(dir.path(), "HEAD:a.txt")
823 .await
824 .expect("asked")
825 .expect("found");
826
827 // Exactly what `pkill -f 'cat-file --batch'` does to a running instance.
828 {
829 let slot = pool.slot(dir.path());
830 let mut process = slot.check.lock().await;
831 let batch = process.as_mut().expect("spawned");
832
833 batch.child.start_kill().expect("kill");
834 batch.child.wait().await.expect("reaped");
835 }
836
837 let again = pool
838 .info(dir.path(), "HEAD:a.txt")
839 .await
840 .expect("respawned and asked")
841 .expect("found");
842
843 assert_eq!(again, info);
844 }
845
846 #[tokio::test]
847 async fn a_reply_that_does_not_match_the_question_kills_the_process() {
848 let dir = fixture();
849 let pool = CatFilePool::new();
850
851 let mut info = pool
852 .info(dir.path(), "HEAD:a.txt")
853 .await
854 .expect("asked")
855 .expect("found");
856
857 // The size is what a desync looks like from here: git answers with the object's
858 // real header, which no longer matches what the caller believes it asked about.
859 info.size += 1;
860
861 let error = pool
862 .contents(dir.path(), &info)
863 .await
864 .expect_err("a mismatched header is not an answer");
865
866 assert!(
867 error.to_string().contains("spoke out of turn"),
868 "expected a desync, got {error}"
869 );
870
871 // And the process is gone rather than left holding an unread body.
872 let slot = pool.slot(dir.path());
873 assert!(slot.contents.lock().await.is_none());
874 }
875
876 #[tokio::test]
877 async fn the_stream_still_answers_correctly_after_a_desync() {
878 let dir = fixture();
879 let pool = CatFilePool::new();
880
881 let real = pool
882 .info(dir.path(), "HEAD:a.txt")
883 .await
884 .expect("asked")
885 .expect("found");
886
887 let mut wrong = real.clone();
888 wrong.size += 1;
889
890 pool.contents(dir.path(), &wrong).await.expect_err("desync");
891
892 // A respawned process, and the answer is the file's own bytes rather than the
893 // tail of the reply the killed one was part-way through.
894 assert_eq!(
895 pool.contents(dir.path(), &real).await.expect("read"),
896 b"hello\n"
897 );
898 }
899
900 #[tokio::test]
901 async fn a_repository_that_is_not_there_is_an_error() {
902 let dir = fixture();
903 let pool = CatFilePool::new();
904
905 let error = pool
906 .info(&dir.path().join("nowhere"), "HEAD")
907 .await
908 .expect_err("no repository, no answer");
909
910 assert!(
911 error.to_string().contains("would not stay running"),
912 "expected a spawn failure, got {error}"
913 );
914 }
915
916 #[tokio::test]
917 async fn the_pool_holds_at_most_its_cap() {
918 let pool = CatFilePool::new();
919 let dir = fixture();
920
921 // One real repository so the cap is exercised against something that answers,
922 // and the rest as distinct paths — the pool keys on the path, and a slot is
923 // created before any process is.
924 for index in 0..MAX_LIVE_REPOSITORIES + 10 {
925 pool.slot(&dir.path().join(format!("repo-{index}")));
926 }
927
928 assert_eq!(pool.live_repositories(), MAX_LIVE_REPOSITORIES);
929 }
930
931 #[tokio::test]
932 async fn the_repository_asked_about_most_recently_survives_eviction() {
933 let pool = CatFilePool::new();
934 let dir = fixture();
935 let kept = dir.path().join("repo-0");
936
937 pool.slot(&kept);
938
939 for index in 1..MAX_LIVE_REPOSITORIES {
940 pool.slot(&dir.path().join(format!("repo-{index}")));
941 }
942
943 // Touched again, so it is no longer the oldest when the cap next bites.
944 pool.slot(&kept);
945 pool.slot(&dir.path().join("one-too-many"));
946
947 let repos = pool.repos.lock().expect("lock poisoned");
948
949 assert_eq!(repos.len(), MAX_LIVE_REPOSITORIES);
950 assert!(repos.contains_key(&kept), "the most recent was evicted");
951 }
952
953 #[test]
954 fn specs_are_written_in_groups_a_pipe_can_hold() {
955 let many: Vec<String> = (0..MAX_BATCH_SPECS * 2 + 3)
956 .map(|index| format!("{index:040}"))
957 .collect();
958
959 let groups = chunks(&many);
960
961 assert!(groups.iter().all(|group| group.len() <= MAX_BATCH_SPECS));
962 assert!(groups.iter().all(
963 |group| group.iter().map(|spec| spec.len() + 1).sum::<usize>() <= MAX_BATCH_BYTES
964 ));
965 assert_eq!(
966 groups.iter().map(|group| group.len()).sum::<usize>(),
967 many.len()
968 );
969 }
970
971 #[test]
972 fn one_enormous_spec_is_still_a_group_of_its_own() {
973 let specs = vec!["a".repeat(MAX_BATCH_BYTES * 2), "b".to_owned()];
974 let groups = chunks(&specs);
975
976 assert_eq!(groups.len(), 2, "a spec too big to batch still gets asked");
977 }
978}