fakecloud_core/
container_image.rs

1//! Image pulls for the runtimes that launch containers: ECS (and Batch
2//! through it) and Lambda `PackageType=Image` functions pull on every launch
3//! with [`pull_image`]; EC2 instances and CodeBuild builds, whose images
4//! stand in for an AMI or a curated build image, pull only when the image is
5//! missing with [`ensure_image`].
6//!
7//! A bare `docker pull` always contacts the registry, even when the image is
8//! already in the local cache, so a momentary registry failure fails the
9//! launch. The common one is rate limiting: anonymous pulls from
10//! `public.ecr.aws` are capped per source IP, and a burst of task launches --
11//! or several processes sharing one NAT address -- gets
12//! `429 Too Many Requests` back.
13//!
14//! A transient failure (throttling, a registry 5xx, a network timeout) is
15//! retried with backoff, and falls back to the image already cached locally
16//! instead of failing the launch. A refusal is final: an image that no longer
17//! exists or a pull the registry denies fails at once with the registry's own
18//! error, even when a stale copy is cached. Otherwise an image deleted from
19//! ECR, or one a repository policy denies, would keep launching from the
20//! local copy -- which neither Fargate nor Lambda, having no per-host image
21//! cache, ever does.
22
23use std::path::Path;
24use std::time::Duration;
25
26use tokio::process::Command;
27
28/// Pull attempts made while the registry keeps failing transiently, before
29/// the last error is returned.
30const MAX_PULL_ATTEMPTS: u32 = 5;
31
32/// Delay before the first retry; doubles on each further retry.
33const BASE_RETRY_DELAY: Duration = Duration::from_secs(1);
34
35/// How an image became available for a launch.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub enum PulledImage {
38    /// The registry served the image.
39    Pulled,
40    /// The pull failed transiently but the image was already cached locally,
41    /// so the cached copy is used. Carries the pull's error for logging.
42    Cached { pull_error: String },
43    /// [`ensure_image`] found the image already cached and did not pull.
44    Present,
45}
46
47/// Pull `reference` with the container `cli`. A transient registry failure
48/// falls back to a locally cached copy, or is retried with backoff when
49/// nothing is cached; any other failure is returned at once. `docker_config`
50/// is exported as `DOCKER_CONFIG` for the pull so registry credentials
51/// resolve.
52///
53/// Returns the pull's stderr as the error.
54pub async fn pull_image(
55    cli: &str,
56    docker_config: Option<&Path>,
57    reference: &str,
58) -> Result<PulledImage, String> {
59    pull_image_with(cli, docker_config, reference, BASE_RETRY_DELAY).await
60}
61
62/// Make `reference` available locally, pulling it only when it is not
63/// already cached -- what `docker run` does with its implicit pull, but with
64/// [`pull_image`]'s retry when the registry fails transiently. `docker run`
65/// gives up on the first `429 Too Many Requests`, which on a host with an
66/// empty cache fails every launch that races the first pull of an image.
67pub async fn ensure_image(
68    cli: &str,
69    docker_config: Option<&Path>,
70    reference: &str,
71) -> Result<PulledImage, String> {
72    ensure_image_with(cli, docker_config, reference, BASE_RETRY_DELAY).await
73}
74
75async fn ensure_image_with(
76    cli: &str,
77    docker_config: Option<&Path>,
78    reference: &str,
79    base_delay: Duration,
80) -> Result<PulledImage, String> {
81    if image_cached(cli, docker_config, reference).await {
82        return Ok(PulledImage::Present);
83    }
84    pull_image_with(cli, docker_config, reference, base_delay).await
85}
86
87async fn pull_image_with(
88    cli: &str,
89    docker_config: Option<&Path>,
90    reference: &str,
91    base_delay: Duration,
92) -> Result<PulledImage, String> {
93    let mut delay = base_delay;
94    let mut attempt = 1;
95    loop {
96        let mut cmd = Command::new(cli);
97        if let Some(p) = docker_config {
98            cmd.env("DOCKER_CONFIG", p);
99        }
100        let out = cmd
101            .args(["pull", reference])
102            .output()
103            .await
104            .map_err(|e| format!("{cli} pull: {e}"))?;
105        if out.status.success() {
106            return Ok(PulledImage::Pulled);
107        }
108        let pull_error = String::from_utf8_lossy(&out.stderr).trim().to_string();
109        if !is_transient(&pull_error, reference) {
110            return Err(pull_error);
111        }
112        if image_cached(cli, docker_config, reference).await {
113            tracing::warn!(
114                image = %reference,
115                error = %pull_error,
116                "image pull failed transiently; using the locally cached image"
117            );
118            return Ok(PulledImage::Cached { pull_error });
119        }
120        if attempt >= MAX_PULL_ATTEMPTS {
121            return Err(pull_error);
122        }
123        tracing::info!(
124            image = %reference,
125            attempt,
126            retry_in_ms = delay.as_millis() as u64,
127            "image pull failed transiently; retrying"
128        );
129        tokio::time::sleep(delay).await;
130        delay *= 2;
131        attempt += 1;
132    }
133}
134
135/// Whether `reference` resolves to an image in the local cache. Runs with
136/// the same `DOCKER_CONFIG` as the pull: the config also selects the Docker
137/// context, so without it the check could consult a different daemon than
138/// the one that pulls and later runs the image.
139async fn image_cached(cli: &str, docker_config: Option<&Path>, reference: &str) -> bool {
140    let mut cmd = Command::new(cli);
141    if let Some(p) = docker_config {
142        cmd.env("DOCKER_CONFIG", p);
143    }
144    cmd.args(["image", "inspect", reference])
145        .stdout(std::process::Stdio::null())
146        .stderr(std::process::Stdio::null())
147        .status()
148        .await
149        .map(|s| s.success())
150        .unwrap_or(false)
151}
152
153/// Whether a pull error is one a later attempt could succeed past: the
154/// registry throttling the caller (the `429 Too Many Requests` status line,
155/// the `toomanyrequests` error code, ECR Public's `Rate exceeded`), any 5xx
156/// from the registry, or the network timing out or dropping the connection.
157///
158/// A refusal (not found, access denied, unauthorized) is never transient, and
159/// wins over transient wording in the same message. Both are matched only
160/// after the image's own name is removed from the message: the name is chosen
161/// by the user and quoted in the error, so a repository spelled like either
162/// kind of marker (`toomanyrequests/app`, `acme/access-denied-page`) must not
163/// flip the classification.
164fn is_transient(stderr: &str, reference: &str) -> bool {
165    const REFUSED: [&str; 6] = [
166        "manifest unknown",
167        "not found",
168        "denied",
169        "unauthorized",
170        "forbidden",
171        "does not exist",
172    ];
173    const TRANSIENT: [&str; 8] = [
174        "toomanyrequests",
175        "too many requests",
176        "rate exceeded",
177        "i/o timeout",
178        "tls handshake timeout",
179        "connection reset by peer",
180        "context deadline exceeded",
181        "request canceled while waiting for connection",
182    ];
183    let message = without_image_name(&stderr.to_ascii_lowercase(), reference);
184    if REFUSED.iter().any(|m| message.contains(m)) {
185        return false;
186    }
187    TRANSIENT.iter().any(|m| message.contains(m)) || has_server_error_status(&message)
188}
189
190/// `message` (lowercase) with each way an error can quote the image blanked
191/// out: the reference as given, every trailing path of its repository
192/// (`public.ecr.aws/acme/app`, `acme/app`, `app`), and its registry host --
193/// registries put the host and repository path in URLs, and Podman expands short names to
194/// `docker.io/library/<name>`. Only whole names are removed, bounded by
195/// characters a name cannot contain, so a short repository like `d` never
196/// cuts letters out of the surrounding words.
197fn without_image_name(message: &str, reference: &str) -> String {
198    let reference = reference.to_ascii_lowercase();
199    let untagged = reference.split('@').next().unwrap_or(&reference);
200    // A `:` after the last `/` starts the tag; one before it is a registry port.
201    let repository = match (untagged.rfind(':'), untagged.rfind('/')) {
202        (Some(colon), Some(slash)) if colon < slash => untagged,
203        (Some(colon), _) => &untagged[..colon],
204        (None, _) => untagged,
205    };
206    let mut names = vec![reference.as_str(), repository];
207    names.extend(
208        repository
209            .match_indices('/')
210            .map(|(i, _)| &repository[i + 1..]),
211    );
212    // The registry host, which URLs in the error quote on its own.
213    if let Some((host, _)) = repository.split_once('/') {
214        if host.contains(['.', ':']) || host == "localhost" {
215            names.push(host);
216        }
217    }
218    names.sort_by_key(|n| std::cmp::Reverse(n.len()));
219
220    let mut out = message.to_string();
221    for name in names.into_iter().filter(|n| !n.is_empty()) {
222        out = remove_whole(&out, name);
223    }
224    out
225}
226
227/// `haystack` with every occurrence of `name` that is not part of a longer
228/// name (flanked by a letter, digit, `.`, `_` or `-`) replaced by a space.
229fn remove_whole(haystack: &str, name: &str) -> String {
230    let is_name_char = |c: char| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-');
231    let mut out = String::with_capacity(haystack.len());
232    let mut rest = haystack;
233    while let Some(i) = rest.find(name) {
234        let end = i + name.len();
235        let before = rest[..i].chars().next_back();
236        let after = rest[end..].chars().next();
237        if before.is_some_and(is_name_char) || after.is_some_and(is_name_char) {
238            out.push_str(&rest[..end]);
239        } else {
240            out.push_str(&rest[..i]);
241            out.push(' ');
242        }
243        rest = &rest[end..];
244    }
245    out.push_str(rest);
246    out
247}
248
249/// Whether the message carries a 5xx HTTP status. Docker and Podman quote the
250/// status as `: 503 Service Unavailable`, `status: 500`, or
251/// `status code 502`; a three-digit number in that position from 500 to 599
252/// counts, whatever reason phrase follows.
253fn has_server_error_status(message: &str) -> bool {
254    let bytes = message.as_bytes();
255    ["status code ", "status: ", "status ", ": "]
256        .iter()
257        .flat_map(|prefix| message.match_indices(prefix).map(|(i, p)| i + p.len()))
258        .any(|start| {
259            let code = &bytes[start..bytes.len().min(start + 3)];
260            code.len() == 3
261                && code[0] == b'5'
262                && code.iter().all(u8::is_ascii_digit)
263                && !bytes
264                    .get(start + 3)
265                    .is_some_and(|c| c.is_ascii_alphanumeric())
266        })
267}
268
269#[cfg(all(test, unix))]
270mod tests {
271    use super::*;
272    use std::os::unix::fs::PermissionsExt;
273
274    /// A stand-in container CLI. `pull` fails with `pull_stderr` for the first
275    /// `pull_failures` calls and succeeds after; `image inspect` succeeds only
276    /// when `cached`. Every invocation is appended to `calls.log`.
277    struct FakeCli {
278        dir: tempfile::TempDir,
279    }
280
281    impl FakeCli {
282        fn new(pull_failures: u32, pull_stderr: &str, cached: bool) -> Self {
283            let dir = tempfile::tempdir().unwrap();
284            let script = format!(
285                r#"#!/bin/sh
286d="{dir}"
287echo "$*" >> "$d/calls.log"
288case "$1" in
289  pull)
290    n=$(cat "$d/pulls" 2>/dev/null || echo 0)
291    n=$((n + 1))
292    echo "$n" > "$d/pulls"
293    if [ "$n" -le {pull_failures} ]; then
294      echo '{pull_stderr}' >&2
295      exit 1
296    fi
297    exit 0 ;;
298  image)
299    [ "{cached}" = "true" ] && exit 0
300    echo 'Error: No such image' >&2
301    exit 1 ;;
302esac
303exit 2
304"#,
305                dir = dir.path().display(),
306            );
307            let path = dir.path().join("cli");
308            std::fs::write(&path, script).unwrap();
309            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
310            // Executing a file that another process holds open for writing
311            // fails with ETXTBSY. Tests run in parallel threads, and a child
312            // forked by another test while `fs::write` above had the script
313            // open inherits that descriptor until it execs. Run the script
314            // once, retrying until nothing holds it; after that no process can
315            // gain a writable descriptor to it again.
316            let mut attempts = 0;
317            loop {
318                match std::process::Command::new(&path).arg("probe").output() {
319                    Err(e)
320                        if e.kind() == std::io::ErrorKind::ExecutableFileBusy && attempts < 200 =>
321                    {
322                        attempts += 1;
323                        std::thread::sleep(Duration::from_millis(5));
324                    }
325                    Err(e) => panic!("fake CLI did not run: {e}"),
326                    Ok(_) => break,
327                }
328            }
329            let _ = std::fs::remove_file(dir.path().join("calls.log"));
330            Self { dir }
331        }
332
333        fn cli(&self) -> String {
334            self.dir.path().join("cli").display().to_string()
335        }
336
337        fn calls(&self) -> Vec<String> {
338            std::fs::read_to_string(self.dir.path().join("calls.log"))
339                .unwrap_or_default()
340                .lines()
341                .map(String::from)
342                .collect()
343        }
344
345        async fn pull(&self) -> Result<PulledImage, String> {
346            self.pull_ref("alpine:3.20").await
347        }
348
349        async fn pull_ref(&self, reference: &str) -> Result<PulledImage, String> {
350            pull_image_with(&self.cli(), None, reference, Duration::from_millis(1)).await
351        }
352
353        async fn ensure(&self) -> Result<PulledImage, String> {
354            ensure_image_with(&self.cli(), None, "alpine:3.20", Duration::from_millis(1)).await
355        }
356    }
357
358    fn is_transient_for_test(stderr: &str) -> bool {
359        is_transient(stderr, "alpine:3.20")
360    }
361
362    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";
363
364    #[tokio::test]
365    async fn ensure_image_uses_a_cached_image_without_contacting_the_registry() {
366        let cli = FakeCli::new(u32::MAX, THROTTLED, true);
367        assert_eq!(cli.ensure().await, Ok(PulledImage::Present));
368        assert_eq!(cli.calls(), ["image inspect alpine:3.20"]);
369    }
370
371    #[tokio::test]
372    async fn ensure_image_retries_a_throttled_first_pull() {
373        let cli = FakeCli::new(2, THROTTLED, false);
374        assert_eq!(cli.ensure().await, Ok(PulledImage::Pulled));
375        let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
376        assert_eq!(pulls, 3);
377    }
378
379    #[tokio::test]
380    async fn ensure_image_fails_on_a_refused_pull() {
381        let missing =
382            "Error response from daemon: manifest for alpine:3.20 not found: manifest unknown";
383        let cli = FakeCli::new(u32::MAX, missing, false);
384        assert_eq!(cli.ensure().await, Err(missing.to_string()));
385    }
386
387    #[tokio::test]
388    async fn a_successful_pull_needs_no_cache_check() {
389        let cli = FakeCli::new(0, "", false);
390        assert_eq!(cli.pull().await, Ok(PulledImage::Pulled));
391        assert_eq!(cli.calls(), ["pull alpine:3.20"]);
392    }
393
394    #[tokio::test]
395    async fn a_throttled_pull_uses_the_cached_image() {
396        let cli = FakeCli::new(u32::MAX, THROTTLED, true);
397        let got = cli.pull().await;
398        assert_eq!(
399            got,
400            Ok(PulledImage::Cached {
401                pull_error: THROTTLED.to_string()
402            })
403        );
404        assert_eq!(
405            cli.calls(),
406            ["pull alpine:3.20", "image inspect alpine:3.20"],
407            "a cached image is used at once, without retrying the pull"
408        );
409    }
410
411    #[tokio::test]
412    async fn a_rate_limited_pull_with_nothing_cached_is_retried() {
413        let cli = FakeCli::new(2, THROTTLED, false);
414        assert_eq!(cli.pull().await, Ok(PulledImage::Pulled));
415        let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
416        assert_eq!(pulls, 3);
417    }
418
419    #[tokio::test]
420    async fn retries_stop_after_the_attempt_cap() {
421        let cli = FakeCli::new(u32::MAX, THROTTLED, false);
422        assert_eq!(cli.pull().await, Err(THROTTLED.to_string()));
423        let pulls = cli.calls().iter().filter(|c| c.starts_with("pull")).count();
424        assert_eq!(pulls, MAX_PULL_ATTEMPTS as usize);
425    }
426
427    #[tokio::test]
428    async fn a_missing_image_fails_without_retrying() {
429        let missing =
430            "Error response from daemon: manifest for alpine:nope not found: manifest unknown";
431        let cli = FakeCli::new(u32::MAX, missing, false);
432        assert_eq!(cli.pull().await, Err(missing.to_string()));
433        assert_eq!(
434            cli.calls(),
435            ["pull alpine:3.20"],
436            "a refused pull is neither retried nor checked against the cache"
437        );
438    }
439
440    #[tokio::test]
441    async fn a_refused_pull_fails_even_with_a_stale_cached_copy() {
442        // The image was deleted from the registry, or a policy now denies the
443        // pull. A copy cached by an earlier launch must not be used.
444        for refused in [
445            "Error response from daemon: manifest for alpine:3.20 not found: manifest unknown",
446            "Error response from daemon: pull access denied for alpine, repository does not exist or may require authorization: denied",
447        ] {
448            let cli = FakeCli::new(u32::MAX, refused, true);
449            assert_eq!(cli.pull().await, Err(refused.to_string()));
450            assert_eq!(cli.calls(), ["pull alpine:3.20"]);
451        }
452    }
453
454    #[tokio::test]
455    async fn a_registry_server_error_uses_the_cached_image() {
456        let unavailable =
457            "Error response from daemon: received unexpected HTTP status: 503 Service Unavailable";
458        let cli = FakeCli::new(u32::MAX, unavailable, true);
459        assert_eq!(
460            cli.pull().await,
461            Ok(PulledImage::Cached {
462                pull_error: unavailable.to_string()
463            })
464        );
465    }
466
467    #[tokio::test]
468    async fn a_refused_pull_of_a_repository_named_like_a_marker_is_still_refused() {
469        // The message quotes the image name. A repository spelled like a
470        // throttling code must not make a refused pull look transient.
471        let refused = "Error response from daemon: manifest for toomanyrequests:latest not found: manifest unknown: manifest unknown";
472        let cli = FakeCli::new(u32::MAX, refused, true);
473        assert_eq!(
474            cli.pull_ref("toomanyrequests:latest").await,
475            Err(refused.to_string())
476        );
477        assert_eq!(cli.calls(), ["pull toomanyrequests:latest"]);
478    }
479
480    #[tokio::test]
481    async fn a_throttled_pull_of_a_repository_named_like_a_refusal_uses_the_cache() {
482        // Only removing the image name from the message keeps `denied` in the
483        // repository from reading as a refusal; without it this pull would
484        // fail instead of falling back to the cached copy.
485        let reference = "public.ecr.aws/acme/access-denied-page:1";
486        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";
487        let cli = FakeCli::new(u32::MAX, throttled, true);
488        assert_eq!(
489            cli.pull_ref(reference).await,
490            Ok(PulledImage::Cached {
491                pull_error: throttled.to_string()
492            })
493        );
494        assert_eq!(
495            cli.calls(),
496            [
497                format!("pull {reference}"),
498                format!("image inspect {reference}")
499            ]
500        );
501    }
502
503    #[test]
504    fn transient_detection_separates_retryable_from_refused() {
505        assert!(is_transient_for_test(THROTTLED));
506        assert!(is_transient_for_test(
507            "toomanyrequests: You have reached your pull rate limit."
508        ));
509        assert!(is_transient_for_test("Error: Rate exceeded"));
510        assert!(is_transient_for_test(
511            "received unexpected HTTP status: 502 Bad Gateway"
512        ));
513        assert!(is_transient_for_test(
514            "Get \"https://public.ecr.aws/v2/\": net/http: TLS handshake timeout"
515        ));
516        assert!(is_transient_for_test(
517            "read tcp 10.0.0.2:4431->1.2.3.4:443: read: connection reset by peer"
518        ));
519        assert!(!is_transient_for_test("manifest unknown"));
520        assert!(!is_transient_for_test("pull access denied for foo"));
521        assert!(!is_transient_for_test(
522            "unauthorized: authentication required"
523        ));
524        assert!(!is_transient_for_test(
525            "pull access denied for toomanyrequests, repository does not exist or may require authorization"
526        ));
527    }
528
529    #[test]
530    fn any_5xx_status_is_transient_whatever_its_reason_phrase() {
531        for msg in [
532            "received unexpected HTTP status: 500 Internal Server Error",
533            "unexpected status from GET request to https://r.example/v2/: 507 Insufficient Storage",
534            "unexpected status code 520",
535            "error pulling image: status: 599",
536            "unexpected status from HEAD request to https://r.example/v2/a/manifests/1: 503",
537        ] {
538            assert!(is_transient_for_test(msg), "{msg}");
539        }
540        for msg in [
541            // A registry port or a 4xx is not a server error.
542            "Get \"http://127.0.0.1:5000/v2/\": dial tcp 127.0.0.1:5000: connect: connection refused",
543            "unexpected status code 400 Bad Request",
544            "status: 5001",
545        ] {
546            assert!(!is_transient_for_test(msg), "{msg}");
547        }
548    }
549
550    #[test]
551    fn a_throttled_pull_of_a_repository_named_like_a_refusal_is_still_transient() {
552        let reference = "public.ecr.aws/acme/access-denied-page:1";
553        for msg in [
554            "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",
555            "Error response from daemon: toomanyrequests: Rate exceeded for public.ecr.aws/acme/access-denied-page:1",
556        ] {
557            assert!(is_transient(msg, reference), "{msg}");
558        }
559        // Its genuine refusals are still refusals.
560        assert!(!is_transient(
561            "Error response from daemon: manifest for public.ecr.aws/acme/access-denied-page:1 not found: manifest unknown",
562            reference
563        ));
564    }
565
566    #[test]
567    fn a_registry_host_named_like_a_refusal_does_not_hide_a_throttle() {
568        assert!(is_transient(
569            "Error response from daemon: unexpected status from HEAD request to https://denied.example/v2/app/manifests/1: 429 Too Many Requests",
570            "denied.example/app:1"
571        ));
572    }
573
574    #[test]
575    fn only_whole_names_are_removed() {
576        // A one-letter repository must not cut the `d` out of `denied`.
577        assert!(!is_transient(
578            "Error response from daemon: pull access denied for d, repository does not exist",
579            "d"
580        ));
581        assert_eq!(
582            without_image_name(
583                "pull access denied for docker.io/library/alpine",
584                "alpine:3.20"
585            ),
586            "pull access denied for docker.io/library/ "
587        );
588        assert_eq!(
589            without_image_name(
590                "get https://127.0.0.1:5000/v2/team/app/manifests/v1",
591                "127.0.0.1:5000/team/app:v1"
592            ),
593            "get https:// /v2/ /manifests/v1"
594        );
595    }
596}
fakecloud_core/container_image.rs

fakecloud_core/
container_image.rs
