1use std::path::Path;
24use std::time::Duration;
25
26use tokio::process::Command;
27
28const MAX_PULL_ATTEMPTS: u32 = 5;
31
32const BASE_RETRY_DELAY: Duration = Duration::from_secs(1);
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum RegistryTransport {
38 Https,
40 PlainHttp,
46}
47
48impl RegistryTransport {
49 pub fn for_local_rewrite(rewritten: bool) -> Self {
53 if rewritten {
54 Self::PlainHttp
55 } else {
56 Self::Https
57 }
58 }
59}
60
61#[derive(Debug, Clone, PartialEq, Eq)]
63pub enum PulledImage {
64 Pulled,
66 Cached { pull_error: String },
69 Present,
71}
72
73pub async fn pull_image(
82 cli: &str,
83 docker_config: Option<&Path>,
84 reference: &str,
85 transport: RegistryTransport,
86) -> Result<PulledImage, String> {
87 pull_image_with(cli, docker_config, reference, transport, BASE_RETRY_DELAY).await
88}
89
90pub async fn ensure_image(
96 cli: &str,
97 docker_config: Option<&Path>,
98 reference: &str,
99) -> Result<PulledImage, String> {
100 ensure_image_with(cli, docker_config, reference, BASE_RETRY_DELAY).await
101}
102
103async fn ensure_image_with(
104 cli: &str,
105 docker_config: Option<&Path>,
106 reference: &str,
107 base_delay: Duration,
108) -> Result<PulledImage, String> {
109 if image_cached(cli, docker_config, reference).await {
110 return Ok(PulledImage::Present);
111 }
112 pull_image_with(
113 cli,
114 docker_config,
115 reference,
116 RegistryTransport::Https,
117 base_delay,
118 )
119 .await
120}
121
122async fn pull_image_with(
123 cli: &str,
124 docker_config: Option<&Path>,
125 reference: &str,
126 transport: RegistryTransport,
127 base_delay: Duration,
128) -> Result<PulledImage, String> {
129 let podman = transport == RegistryTransport::PlainHttp
132 && crate::container_net::is_podman_async(cli).await;
133 let mut delay = base_delay;
134 let mut attempt = 1;
135 loop {
136 let mut cmd = Command::new(cli);
137 if let Some(p) = docker_config {
138 cmd.env("DOCKER_CONFIG", p);
139 }
140 let out = cmd
141 .args(pull_args(podman, reference, transport))
142 .output()
143 .await
144 .map_err(|e| format!("{cli} pull: {e}"))?;
145 if out.status.success() {
146 return Ok(PulledImage::Pulled);
147 }
148 let pull_error = String::from_utf8_lossy(&out.stderr).trim().to_string();
149 if !is_transient(&pull_error, reference) {
150 return Err(pull_error);
151 }
152 if image_cached(cli, docker_config, reference).await {
153 tracing::warn!(
154 image = %reference,
155 error = %pull_error,
156 "image pull failed transiently; using the locally cached image"
157 );
158 return Ok(PulledImage::Cached { pull_error });
159 }
160 if attempt >= MAX_PULL_ATTEMPTS {
161 return Err(pull_error);
162 }
163 tracing::info!(
164 image = %reference,
165 attempt,
166 retry_in_ms = delay.as_millis() as u64,
167 "image pull failed transiently; retrying"
168 );
169 tokio::time::sleep(delay).await;
170 delay *= 2;
171 attempt += 1;
172 }
173}
174
175fn pull_args(podman: bool, reference: &str, transport: RegistryTransport) -> Vec<String> {
180 let mut args = vec!["pull".to_string()];
181 if transport == RegistryTransport::PlainHttp && podman {
182 args.push("--tls-verify=false".to_string());
183 }
184 args.push(reference.to_string());
185 args
186}
187
188async fn image_cached(cli: &str, docker_config: Option<&Path>, reference: &str) -> bool {
193 let mut cmd = Command::new(cli);
194 if let Some(p) = docker_config {
195 cmd.env("DOCKER_CONFIG", p);
196 }
197 cmd.args(["image", "inspect", reference])
198 .stdout(std::process::Stdio::null())
199 .stderr(std::process::Stdio::null())
200 .status()
201 .await
202 .map(|s| s.success())
203 .unwrap_or(false)
204}
205
206fn is_transient(stderr: &str, reference: &str) -> bool {
218 const REFUSED: [&str; 6] = [
219 "manifest unknown",
220 "not found",
221 "denied",
222 "unauthorized",
223 "forbidden",
224 "does not exist",
225 ];
226 const TRANSIENT: [&str; 8] = [
227 "toomanyrequests",
228 "too many requests",
229 "rate exceeded",
230 "i/o timeout",
231 "tls handshake timeout",
232 "connection reset by peer",
233 "context deadline exceeded",
234 "request canceled while waiting for connection",
235 ];
236 let message = without_image_name(&stderr.to_ascii_lowercase(), reference);
237 if REFUSED.iter().any(|m| message.contains(m)) {
238 return false;
239 }
240 TRANSIENT.iter().any(|m| message.contains(m)) || has_server_error_status(&message)
241}
242
243fn without_image_name(message: &str, reference: &str) -> String {
251 let reference = reference.to_ascii_lowercase();
252 let untagged = reference.split('@').next().unwrap_or(&reference);
253 let repository = match (untagged.rfind(':'), untagged.rfind('/')) {
255 (Some(colon), Some(slash)) if colon < slash => untagged,
256 (Some(colon), _) => &untagged[..colon],
257 (None, _) => untagged,
258 };
259 let mut names = vec![reference.as_str(), repository];
260 names.extend(
261 repository
262 .match_indices('/')
263 .map(|(i, _)| &repository[i + 1..]),
264 );
265 if let Some((host, _)) = repository.split_once('/') {
267 if host.contains(['.', ':']) || host == "localhost" {
268 names.push(host);
269 }
270 }
271 names.sort_by_key(|n| std::cmp::Reverse(n.len()));
272
273 let mut out = message.to_string();
274 for name in names.into_iter().filter(|n| !n.is_empty()) {
275 out = remove_whole(&out, name);
276 }
277 out
278}
279
280fn remove_whole(haystack: &str, name: &str) -> String {
283 let is_name_char = |c: char| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-');
284 let mut out = String::with_capacity(haystack.len());
285 let mut rest = haystack;
286 while let Some(i) = rest.find(name) {
287 let end = i + name.len();
288 let before = rest[..i].chars().next_back();
289 let after = rest[end..].chars().next();
290 if before.is_some_and(is_name_char) || after.is_some_and(is_name_char) {
291 out.push_str(&rest[..end]);
292 } else {
293 out.push_str(&rest[..i]);
294 out.push(' ');
295 }
296 rest = &rest[end..];
297 }
298 out.push_str(rest);
299 out
300}
301
302fn has_server_error_status(message: &str) -> bool {
307 let bytes = message.as_bytes();
308 ["status code ", "status: ", "status ", ": "]
309 .iter()
310 .flat_map(|prefix| message.match_indices(prefix).map(|(i, p)| i + p.len()))
311 .any(|start| {
312 let code = &bytes[start..bytes.len().min(start + 3)];
313 code.len() == 3
314 && code[0] == b'5'
315 && code.iter().all(u8::is_ascii_digit)
316 && !bytes
317 .get(start + 3)
318 .is_some_and(|c| c.is_ascii_alphanumeric())
319 })
320}
321
322#[cfg(all(test, unix))]
323mod tests {
324 use super::*;
325 use std::os::unix::fs::PermissionsExt;
326
327 struct FakeCli {
331 dir: tempfile::TempDir,
332 name: String,
333 }
334
335 impl FakeCli {
336 fn new(pull_failures: u32, pull_stderr: &str, cached: bool) -> Self {
337 Self::named("cli", pull_failures, pull_stderr, cached)
338 }
339
340 fn named(name: &str, pull_failures: u32, pull_stderr: &str, cached: bool) -> Self {
343 let version = if name.contains("podman") {
344 "podman version 5.2.0"
345 } else {
346 "Docker version 27.3.1, build ce12230"
347 };
348 Self::with_version(name, version, pull_failures, pull_stderr, cached)
349 }
350
351 fn with_version(
354 name: &str,
355 version: &str,
356 pull_failures: u32,
357 pull_stderr: &str,
358 cached: bool,
359 ) -> Self {
360 let dir = tempfile::tempdir().unwrap();
361 let script = format!(
362 r#"#!/bin/sh
363d="{dir}"
364if [ "$1" = --version ]; then
365 echo '{version}'
366 exit 0
367fi
368echo "$*" >> "$d/calls.log"
369case "$1" in
370 pull)
371 n=$(cat "$d/pulls" 2>/dev/null || echo 0)
372 n=$((n + 1))
373 echo "$n" > "$d/pulls"
374 if [ "$n" -le {pull_failures} ]; then
375 echo '{pull_stderr}' >&2
376 exit 1
377 fi
378 exit 0 ;;
379 image)
380 [ "{cached}" = "true" ] && exit 0
381 echo 'Error: No such image' >&2
382 exit 1 ;;
383esac
384exit 2
385"#,
386 dir = dir.path().display(),
387 );
388 let path = dir.path().join(name);
389 std::fs::write(&path, script).unwrap();
390 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
391 let mut attempts = 0;
398 loop {
399 match std::process::Command::new(&path).arg("probe").output() {
400 Err(e)
401 if e.kind() == std::io::ErrorKind::ExecutableFileBusy && attempts < 200 =>
402 {
403 attempts += 1;
404 std::thread::sleep(Duration::from_millis(5));
405 }
406 Err(e) => panic!("fake CLI did not run: {e}"),
407 Ok(_) => break,
408 }
409 }
410 let _ = std::fs::remove_file(dir.path().join("calls.log"));
411 Self {
412 dir,
413 name: name.to_string(),
414 }
415 }
416
417 fn cli(&self) -> String {
418 self.dir.path().join(&self.name).display().to_string()
419 }
420
421 fn calls(&self) -> Vec<String> {
422 std::fs::read_to_string(self.dir.path().join("calls.log"))
423 .unwrap_or_default()
424 .lines()
425 .map(String::from)
426 .collect()
427 }
428
429 async fn pull(&self) -> Result<PulledImage, String> {
430 self.pull_ref("alpine:3.20").await
431 }
432
433 async fn pull_ref(&self, reference: &str) -> Result<PulledImage, String> {
434 self.pull_via(reference, RegistryTransport::Https).await
435 }
436
437 async fn pull_via(
438 &self,
439 reference: &str,
440 transport: RegistryTransport,
441 ) -> Result<PulledImage, String> {
442 pull_image_with(
443 &self.cli(),
444 None,
445 reference,
446 transport,
447 Duration::from_millis(1),
448 )
449 .await
450 }
451
452 async fn ensure(&self) -> Result<PulledImage, String> {
453 ensure_image_with(&self.cli(), None, "alpine:3.20", Duration::from_millis(1)).await
454 }
455 }
456
457 fn is_transient_for_test(stderr: &str) -> bool {
458 is_transient(stderr, "alpine:3.20")
459 }
460
461 const THROTTLED: &str = "Error response from daemon: unexpected status from HEAD request to https://public.ecr.aws/v2/docker/library/alpine/manifests/3.20: 429 Too Many Requests";
462
463 #[tokio::test]
464 async fn ensure_image_uses_a_cached_image_without_contacting_the_registry() {
465 let cli = FakeCli::new(u32::MAX, THROTTLED, true);
466 assert_eq!(cli.ensure().await, Ok(PulledImage::Present));
467 assert_eq!(cli.calls(), ["image inspect alpine:3.20"]);
468 }
469
470 #[tokio::test]
471 async fn ensure_image_retries_a_throttled_first_pull() {
472 let cli = FakeCli::new(2, THROTTLED, false);
473 assert_eq!(cli.ensure().await, Ok(PulledImage::Pulled));
474 let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
475 assert_eq!(pulls, 3);
476 }
477
478 #[tokio::test]
479 async fn ensure_image_fails_on_a_refused_pull() {
480 let missing =
481 "Error response from daemon: manifest for alpine:3.20 not found: manifest unknown";
482 let cli = FakeCli::new(u32::MAX, missing, false);
483 assert_eq!(cli.ensure().await, Err(missing.to_string()));
484 }
485
486 #[tokio::test]
487 async fn podman_pulls_fakecloud_registry_over_plain_http() {
488 let cli = FakeCli::named("podman", 0, "", false);
491 let got = cli
492 .pull_via("127.0.0.1:4566/test:healthy", RegistryTransport::PlainHttp)
493 .await;
494 assert_eq!(got, Ok(PulledImage::Pulled));
495 assert_eq!(
496 cli.calls(),
497 ["pull --tls-verify=false 127.0.0.1:4566/test:healthy"]
498 );
499 }
500
501 #[tokio::test]
502 async fn podman_keeps_tls_for_upstream_registries() {
503 let cli = FakeCli::named("podman", 0, "", false);
504 let got = cli
505 .pull_via(
506 "public.ecr.aws/docker/library/alpine:3.20",
507 RegistryTransport::Https,
508 )
509 .await;
510 assert_eq!(got, Ok(PulledImage::Pulled));
511 assert_eq!(
512 cli.calls(),
513 ["pull public.ecr.aws/docker/library/alpine:3.20"]
514 );
515 }
516
517 #[tokio::test]
518 async fn podman_docker_shim_pulls_fakecloud_registry_over_plain_http() {
519 let cli = FakeCli::with_version("docker", "podman version 5.2.0", 0, "", false);
522 let got = cli
523 .pull_via("127.0.0.1:4566/test:healthy", RegistryTransport::PlainHttp)
524 .await;
525 assert_eq!(got, Ok(PulledImage::Pulled));
526 assert_eq!(
527 cli.calls(),
528 ["pull --tls-verify=false 127.0.0.1:4566/test:healthy"]
529 );
530 }
531
532 #[tokio::test]
533 async fn docker_pulls_fakecloud_registry_without_a_tls_flag() {
534 let cli = FakeCli::named("docker", 0, "", false);
537 let got = cli
538 .pull_via("127.0.0.1:4566/test:healthy", RegistryTransport::PlainHttp)
539 .await;
540 assert_eq!(got, Ok(PulledImage::Pulled));
541 assert_eq!(cli.calls(), ["pull 127.0.0.1:4566/test:healthy"]);
542 }
543
544 #[test]
545 fn transport_follows_the_local_rewrite() {
546 assert_eq!(
547 RegistryTransport::for_local_rewrite(true),
548 RegistryTransport::PlainHttp
549 );
550 assert_eq!(
551 RegistryTransport::for_local_rewrite(false),
552 RegistryTransport::Https
553 );
554 }
555
556 #[tokio::test]
557 async fn a_successful_pull_needs_no_cache_check() {
558 let cli = FakeCli::new(0, "", false);
559 assert_eq!(cli.pull().await, Ok(PulledImage::Pulled));
560 assert_eq!(cli.calls(), ["pull alpine:3.20"]);
561 }
562
563 #[tokio::test]
564 async fn a_throttled_pull_uses_the_cached_image() {
565 let cli = FakeCli::new(u32::MAX, THROTTLED, true);
566 let got = cli.pull().await;
567 assert_eq!(
568 got,
569 Ok(PulledImage::Cached {
570 pull_error: THROTTLED.to_string()
571 })
572 );
573 assert_eq!(
574 cli.calls(),
575 ["pull alpine:3.20", "image inspect alpine:3.20"],
576 "a cached image is used at once, without retrying the pull"
577 );
578 }
579
580 #[tokio::test]
581 async fn a_rate_limited_pull_with_nothing_cached_is_retried() {
582 let cli = FakeCli::new(2, THROTTLED, false);
583 assert_eq!(cli.pull().await, Ok(PulledImage::Pulled));
584 let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
585 assert_eq!(pulls, 3);
586 }
587
588 #[tokio::test]
589 async fn retries_stop_after_the_attempt_cap() {
590 let cli = FakeCli::new(u32::MAX, THROTTLED, false);
591 assert_eq!(cli.pull().await, Err(THROTTLED.to_string()));
592 let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
593 assert_eq!(pulls, MAX_PULL_ATTEMPTS as usize);
594 }
595
596 #[tokio::test]
597 async fn a_missing_image_fails_without_retrying() {
598 let missing =
599 "Error response from daemon: manifest for alpine:nope not found: manifest unknown";
600 let cli = FakeCli::new(u32::MAX, missing, false);
601 assert_eq!(cli.pull().await, Err(missing.to_string()));
602 assert_eq!(
603 cli.calls(),
604 ["pull alpine:3.20"],
605 "a refused pull is neither retried nor checked against the cache"
606 );
607 }
608
609 #[tokio::test]
610 async fn a_refused_pull_fails_even_with_a_stale_cached_copy() {
611 for refused in [
614 "Error response from daemon: manifest for alpine:3.20 not found: manifest unknown",
615 "Error response from daemon: pull access denied for alpine, repository does not exist or may require authorization: denied",
616 ] {
617 let cli = FakeCli::new(u32::MAX, refused, true);
618 assert_eq!(cli.pull().await, Err(refused.to_string()));
619 assert_eq!(cli.calls(), ["pull alpine:3.20"]);
620 }
621 }
622
623 #[tokio::test]
624 async fn a_registry_server_error_uses_the_cached_image() {
625 let unavailable =
626 "Error response from daemon: received unexpected HTTP status: 503 Service Unavailable";
627 let cli = FakeCli::new(u32::MAX, unavailable, true);
628 assert_eq!(
629 cli.pull().await,
630 Ok(PulledImage::Cached {
631 pull_error: unavailable.to_string()
632 })
633 );
634 }
635
636 #[tokio::test]
637 async fn a_refused_pull_of_a_repository_named_like_a_marker_is_still_refused() {
638 let refused = "Error response from daemon: manifest for toomanyrequests:latest not found: manifest unknown: manifest unknown";
641 let cli = FakeCli::new(u32::MAX, refused, true);
642 assert_eq!(
643 cli.pull_ref("toomanyrequests:latest").await,
644 Err(refused.to_string())
645 );
646 assert_eq!(cli.calls(), ["pull toomanyrequests:latest"]);
647 }
648
649 #[tokio::test]
650 async fn a_throttled_pull_of_a_repository_named_like_a_refusal_uses_the_cache() {
651 let reference = "public.ecr.aws/acme/access-denied-page:1";
655 let throttled = "Error response from daemon: unexpected status from HEAD request to https://public.ecr.aws/v2/acme/access-denied-page/manifests/1: 429 Too Many Requests";
656 let cli = FakeCli::new(u32::MAX, throttled, true);
657 assert_eq!(
658 cli.pull_ref(reference).await,
659 Ok(PulledImage::Cached {
660 pull_error: throttled.to_string()
661 })
662 );
663 assert_eq!(
664 cli.calls(),
665 [
666 format!("pull {reference}"),
667 format!("image inspect {reference}")
668 ]
669 );
670 }
671
672 #[test]
673 fn transient_detection_separates_retryable_from_refused() {
674 assert!(is_transient_for_test(THROTTLED));
675 assert!(is_transient_for_test(
676 "toomanyrequests: You have reached your pull rate limit."
677 ));
678 assert!(is_transient_for_test("Error: Rate exceeded"));
679 assert!(is_transient_for_test(
680 "received unexpected HTTP status: 502 Bad Gateway"
681 ));
682 assert!(is_transient_for_test(
683 "Get \"https://public.ecr.aws/v2/\": net/http: TLS handshake timeout"
684 ));
685 assert!(is_transient_for_test(
686 "read tcp 10.0.0.2:4431->1.2.3.4:443: read: connection reset by peer"
687 ));
688 assert!(!is_transient_for_test("manifest unknown"));
689 assert!(!is_transient_for_test("pull access denied for foo"));
690 assert!(!is_transient_for_test(
691 "unauthorized: authentication required"
692 ));
693 assert!(!is_transient_for_test(
694 "pull access denied for toomanyrequests, repository does not exist or may require authorization"
695 ));
696 }
697
698 #[test]
699 fn any_5xx_status_is_transient_whatever_its_reason_phrase() {
700 for msg in [
701 "received unexpected HTTP status: 500 Internal Server Error",
702 "unexpected status from GET request to https://r.example/v2/: 507 Insufficient Storage",
703 "unexpected status code 520",
704 "error pulling image: status: 599",
705 "unexpected status from HEAD request to https://r.example/v2/a/manifests/1: 503",
706 ] {
707 assert!(is_transient_for_test(msg), "{msg}");
708 }
709 for msg in [
710 "Get \"http://127.0.0.1:5000/v2/\": dial tcp 127.0.0.1:5000: connect: connection refused",
712 "unexpected status code 400 Bad Request",
713 "status: 5001",
714 ] {
715 assert!(!is_transient_for_test(msg), "{msg}");
716 }
717 }
718
719 #[test]
720 fn a_throttled_pull_of_a_repository_named_like_a_refusal_is_still_transient() {
721 let reference = "public.ecr.aws/acme/access-denied-page:1";
722 for msg in [
723 "Error response from daemon: unexpected status from HEAD request to https://public.ecr.aws/v2/acme/access-denied-page/manifests/1: 429 Too Many Requests",
724 "Error response from daemon: toomanyrequests: Rate exceeded for public.ecr.aws/acme/access-denied-page:1",
725 ] {
726 assert!(is_transient(msg, reference), "{msg}");
727 }
728 assert!(!is_transient(
730 "Error response from daemon: manifest for public.ecr.aws/acme/access-denied-page:1 not found: manifest unknown",
731 reference
732 ));
733 }
734
735 #[test]
736 fn a_registry_host_named_like_a_refusal_does_not_hide_a_throttle() {
737 assert!(is_transient(
738 "Error response from daemon: unexpected status from HEAD request to https://denied.example/v2/app/manifests/1: 429 Too Many Requests",
739 "denied.example/app:1"
740 ));
741 }
742
743 #[test]
744 fn only_whole_names_are_removed() {
745 assert!(!is_transient(
747 "Error response from daemon: pull access denied for d, repository does not exist",
748 "d"
749 ));
750 assert_eq!(
751 without_image_name(
752 "pull access denied for docker.io/library/alpine",
753 "alpine:3.20"
754 ),
755 "pull access denied for docker.io/library/ "
756 );
757 assert_eq!(
758 without_image_name(
759 "get https://127.0.0.1:5000/v2/team/app/manifests/v1",
760 "127.0.0.1:5000/team/app:v1"
761 ),
762 "get https:// /v2/ /manifests/v1"
763 );
764 }
765}