test(linux): retry a fixture's start when its image pull fails

linux/crates/ssh/tests/openssh_session.rs

@@ -9,9 +9,9 @@ use secrecy::SecretString;
 use tablepro_core::credentials::{CredentialInteraction, PromptPurpose, PromptReply};
 use tablepro_core::{LivenessPolicy, NetworkEndpoint, SshFailure, TransportError, TransportRoute};
 use tablepro_ssh::{SshAuth, SshConfig, SshDestination, SshRuntimeCell, SshServices, SshSession, forget_host_key};
+use tablepro_test_fixtures::start_container;
 use tablepro_test_support::FakePrompter;
 use testcontainers::core::{IntoContainerPort, WaitFor};
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 use tokio::io::AsyncReadExt;
 use tokio_util::sync::CancellationToken;
@@ -45,12 +45,13 @@ fn write_config(config: &Path, known_hosts: &Path, strict_host_key_checking: &st
 }
 
 async fn start() -> TestResult<Fixture> {
-    let container = GenericImage::new("alpine", "3.22")
-        .with_exposed_port(2222.tcp())
-        .with_wait_for(WaitFor::message_on_stderr("Server listening on"))
-        .with_cmd(["sh", "-c", SSHD_SCRIPT])
-        .start()
-        .await?;
+    let container = start_container(|| {
+        GenericImage::new("alpine", "3.22")
+            .with_exposed_port(2222.tcp())
+            .with_wait_for(WaitFor::message_on_stderr("Server listening on"))
+            .with_cmd(["sh", "-c", SSHD_SCRIPT])
+    })
+    .await?;
     let host = container.get_host().await?.to_string();
     let port = container.get_host_port_ipv4(2222).await?;
 

linux/crates/test-fixtures/src/cassandra_fixture.rs

@@ -1,12 +1,12 @@
 use std::time::Duration;
 
 use testcontainers::core::IntoContainerPort;
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, ContainerRequest, GenericImage, ImageExt};
 use tokio::time::Instant;
 
 use crate::container_file::with_files;
 use crate::container_report::{container_state, log_tail};
+use crate::container_start::start_container;
 use crate::{
     CassandraFlavour, CassandraTls, ContainerFile, ExecOutput, FixtureCredentials, FixtureError, PkiFile, TestNetwork,
     TestPki,
@@ -166,29 +166,34 @@ impl CassandraFixture {
         // ScyllaDB's image exposes "9042" with no protocol, so publishing every
         // exposed port gives Docker 28 two 9042/tcp bindings on one host port.
         // Host port 0 lets Docker pick one.
-        let mut request: ContainerRequest<GenericImage> =
-            image.with_cmd(command).with_mapped_port(0, Self::CONTAINER_PORT.tcp());
-        if flavour == CassandraFlavour::Cassandra {
-            // A joining node waits out the ring delay to learn who owns
-            // which tokens, so only the first node skips it.
-            let options = match &seed {
-                None => Self::fast_start_options(),
-                Some(_) => "-Dcassandra.superuser_setup_delay_ms=0",
-            };
-            request = request
-                .with_env_var("MAX_HEAP_SIZE", CASSANDRA_HEAP)
-                .with_env_var("HEAP_NEWSIZE", CASSANDRA_NEW_HEAP)
-                .with_env_var("JVM_EXTRA_OPTS", options);
-            if let Some(seed) = &seed {
-                request = request.with_env_var("CASSANDRA_SEEDS", seed.as_str());
-            }
-        }
         let network_alias = network.as_ref().map(|(_, alias)| alias.clone());
-        if let Some((network, alias)) = &network {
-            request = request.with_network(network.name()).with_container_name(alias.as_str());
-        }
         let files = Self::files(flavour, server_version, tls, &pki, &password)?;
-        let container = with_files(request, files).start().await?;
+        let container = start_container(|| {
+            let mut request: ContainerRequest<GenericImage> = image
+                .clone()
+                .with_cmd(command.clone())
+                .with_mapped_port(0, Self::CONTAINER_PORT.tcp());
+            if flavour == CassandraFlavour::Cassandra {
+                // A joining node waits out the ring delay to learn who owns
+                // which tokens, so only the first node skips it.
+                let options = match &seed {
+                    None => Self::fast_start_options(),
+                    Some(_) => "-Dcassandra.superuser_setup_delay_ms=0",
+                };
+                request = request
+                    .with_env_var("MAX_HEAP_SIZE", CASSANDRA_HEAP)
+                    .with_env_var("HEAP_NEWSIZE", CASSANDRA_NEW_HEAP)
+                    .with_env_var("JVM_EXTRA_OPTS", options);
+                if let Some(seed) = &seed {
+                    request = request.with_env_var("CASSANDRA_SEEDS", seed.as_str());
+                }
+            }
+            if let Some((network, alias)) = &network {
+                request = request.with_network(network.name()).with_container_name(alias.as_str());
+            }
+            with_files(request, files.clone())
+        })
+        .await?;
         let host = container.get_host().await?.to_string();
         let port = container.get_host_port_ipv4(Self::CONTAINER_PORT).await?;
         let fixture = Self {

linux/crates/test-fixtures/src/clickhouse_fixture.rs

@@ -1,14 +1,14 @@
 use std::time::Duration;
 
 use testcontainers::core::IntoContainerPort;
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, ContainerRequest, GenericImage, ImageExt, TestcontainersError};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::time::Instant;
 
 use crate::container_file::with_files;
 use crate::container_report::{container_state, log_tail};
+use crate::container_start::start_container;
 use crate::{
     ClickHouseVersion, ContainerFile, ExecOutput, FixtureCredentials, FixtureError, PkiFile, ServerVersion,
     TestNetwork, TestPki,
@@ -60,19 +60,22 @@ impl ClickHouseFixture {
         let pki = TestPki::generate()?;
         let password = FixtureCredentials::generate_password();
         let version = ClickHouseVersion::selected()?;
-        let mut request: ContainerRequest<GenericImage> = version
-            .image()
-            .generic()
-            .with_exposed_port(HTTP_PORT.tcp())
-            .with_exposed_port(HTTPS_PORT.tcp())
-            .into();
         let network_alias = network.map(|network| network.container_name(Self::NETWORK_ROLE));
-        if let (Some(network), Some(alias)) = (network, &network_alias) {
-            request = request.with_network(network.name()).with_container_name(alias);
-        }
         let mut files = Self::files(&pki, &password)?;
         files.extend(extra_files);
-        let container = with_files(request, files).start().await?;
+        let container = start_container(|| {
+            let mut request: ContainerRequest<GenericImage> = version
+                .image()
+                .generic()
+                .with_exposed_port(HTTP_PORT.tcp())
+                .with_exposed_port(HTTPS_PORT.tcp())
+                .into();
+            if let (Some(network), Some(alias)) = (network, &network_alias) {
+                request = request.with_network(network.name()).with_container_name(alias);
+            }
+            with_files(request, files.clone())
+        })
+        .await?;
         let (host, http_port, https_port) = match published(&container).await {
             Ok(published) => published,
             Err(error) => {

linux/crates/test-fixtures/src/container_start.rs

@@ -0,0 +1,60 @@
+use std::time::Duration;
+
+use testcontainers::core::error::ClientError;
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, ContainerRequest, Image, TestcontainersError};
+
+use crate::FixtureError;
+
+const START_ATTEMPTS: u32 = 3;
+const PULL_RETRY_DELAY: Duration = Duration::from_secs(5);
+
+/// Starts the container `request` builds, building and starting it again
+/// when pulling its image failed. testcontainers pulls once, so a pull the
+/// registry cut short ("bytes remaining on stream") failed every test that
+/// needed the image. A request is consumed by its start, hence the builder.
+pub async fn start_container<I: Image>(
+    request: impl Fn() -> ContainerRequest<I>,
+) -> Result<ContainerAsync<I>, FixtureError> {
+    let mut attempt = 1;
+    loop {
+        match request().start().await {
+            Err(TestcontainersError::Client(ClientError::PullImage { .. })) if attempt < START_ATTEMPTS => {
+                attempt += 1;
+                tokio::time::sleep(PULL_RETRY_DELAY).await;
+            }
+            started => return Ok(started?),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Instant;
+
+    use testcontainers::GenericImage;
+
+    use super::*;
+
+    #[tokio::test]
+    #[ignore = "requires docker"]
+    async fn a_pull_that_keeps_failing_is_tried_three_times_and_reported() {
+        let started = Instant::now();
+
+        let error = start_container(|| GenericImage::new("alpine", "tablepro-no-such-tag").into())
+            .await
+            .err()
+            .unwrap();
+
+        let FixtureError::Container(TestcontainersError::Client(ClientError::PullImage { descriptor, .. })) = &error
+        else {
+            panic!("unexpected error: {error}");
+        };
+        assert_eq!(descriptor, "alpine:tablepro-no-such-tag");
+        assert!(
+            started.elapsed() >= PULL_RETRY_DELAY * (START_ATTEMPTS - 1),
+            "took {:?}",
+            started.elapsed()
+        );
+    }
+}

linux/crates/test-fixtures/src/lib.rs

@@ -6,6 +6,7 @@ mod clickhouse_fixture;
 mod clickhouse_version;
 mod container_file;
 mod container_report;
+mod container_start;
 mod exec_output;
 mod fixture_credentials;
 mod fixture_error;
@@ -51,6 +52,7 @@ pub use cassandra_version::CassandraVersion;
 pub use clickhouse_fixture::ClickHouseFixture;
 pub use clickhouse_version::ClickHouseVersion;
 pub use container_file::ContainerFile;
+pub use container_start::start_container;
 pub use exec_output::ExecOutput;
 pub use fixture_credentials::FixtureCredentials;
 pub use fixture_error::FixtureError;

linux/crates/test-fixtures/src/mongo_fixture.rs

@@ -1,10 +1,10 @@
 use std::time::Duration;
 
 use testcontainers::core::{IntoContainerPort, WaitFor};
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, ContainerRequest, GenericImage, ImageExt};
 
 use crate::container_file::with_files;
+use crate::container_start::start_container;
 use crate::{
     ContainerFile, ExecOutput, FixtureCredentials, FixtureError, MongoVersion, PkiFile, ServerVersion, TestNetwork,
     TestPki,
@@ -100,22 +100,26 @@ impl MongoFixture {
         let pki = TestPki::generate()?;
         let password = FixtureCredentials::generate_password();
         let version = MongoVersion::selected()?;
-        let mut request: ContainerRequest<GenericImage> = version
-            .image()
-            .generic()
-            .with_wait_for(WaitFor::message_on_stdout(READY))
-            .with_exposed_port(PORT.tcp())
-            .with_exposed_port(CERTIFICATE_PORT.tcp())
-            .with_entrypoint("sh")
-            .with_cmd([format!("{CONFIG_DIR}/start.sh")])
-            .with_env_var("TABLEPRO_PASSWORD", &password)
-            .with_env_var("TABLEPRO_TLS", if listening == Listening::Tls { "1" } else { "0" })
-            .with_startup_timeout(READY_TIMEOUT);
         let network_alias = network.map(|network| network.container_name(Self::NETWORK_ROLE));
-        if let (Some(network), Some(alias)) = (network, &network_alias) {
-            request = request.with_network(network.name()).with_container_name(alias);
-        }
-        let container = with_files(request, Self::files(&pki)?).start().await?;
+        let files = Self::files(&pki)?;
+        let container = start_container(|| {
+            let mut request: ContainerRequest<GenericImage> = version
+                .image()
+                .generic()
+                .with_wait_for(WaitFor::message_on_stdout(READY))
+                .with_exposed_port(PORT.tcp())
+                .with_exposed_port(CERTIFICATE_PORT.tcp())
+                .with_entrypoint("sh")
+                .with_cmd([format!("{CONFIG_DIR}/start.sh")])
+                .with_env_var("TABLEPRO_PASSWORD", &password)
+                .with_env_var("TABLEPRO_TLS", if listening == Listening::Tls { "1" } else { "0" })
+                .with_startup_timeout(READY_TIMEOUT);
+            if let (Some(network), Some(alias)) = (network, &network_alias) {
+                request = request.with_network(network.name()).with_container_name(alias);
+            }
+            with_files(request, files.clone())
+        })
+        .await?;
         let host = container.get_host().await?.to_string();
         let port = container.get_host_port_ipv4(PORT).await?;
         let certificate_port = match listening {

linux/crates/test-fixtures/src/mssql_fixture.rs

@@ -1,11 +1,11 @@
 use std::time::Duration;
 
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, ImageExt};
 use testcontainers_modules::mssql_server::MssqlServer;
 use uuid::Uuid;
 
 use crate::container_file::with_files;
+use crate::container_start::start_container;
 use crate::{
     ContainerFile, ExecOutput, FixtureCredentials, FixtureError, MssqlVersion, PkiFile, ServerVersion, TestPki,
 };
@@ -32,16 +32,20 @@ impl MssqlFixture {
         let password = format!("Tp1!{}", Uuid::new_v4().simple());
         let version = MssqlVersion::selected()?;
         let image = version.image();
-        let request = MssqlServer::default()
-            .with_accept_eula()
-            .with_sa_password(&password)
-            .with_name(image.name)
-            .with_tag(image.tag)
-            .with_startup_timeout(STARTUP_TIMEOUT)
-            // SQL Server otherwise takes 80% of the host's memory, and a
-            // container left behind by a killed test keeps all of it.
-            .with_env_var("MSSQL_MEMORY_LIMIT_MB", MEMORY_LIMIT_MB);
-        let container = with_files(request, Self::files(&pki)?).start().await?;
+        let files = Self::files(&pki)?;
+        let container = start_container(|| {
+            let request = MssqlServer::default()
+                .with_accept_eula()
+                .with_sa_password(&password)
+                .with_name(image.name)
+                .with_tag(image.tag)
+                .with_startup_timeout(STARTUP_TIMEOUT)
+                // SQL Server otherwise takes 80% of the host's memory, and a
+                // container left behind by a killed test keeps all of it.
+                .with_env_var("MSSQL_MEMORY_LIMIT_MB", MEMORY_LIMIT_MB);
+            with_files(request, files.clone())
+        })
+        .await?;
         let host = container.get_host().await?.to_string();
         let port = container.get_host_port_ipv4(CONTAINER_PORT).await?;
         Ok(Self {

linux/crates/test-fixtures/src/mysql_fixture.rs

@@ -1,8 +1,8 @@
 use testcontainers::core::IntoContainerPort;
-use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 
 use crate::container_file::with_files;
+use crate::container_start::start_container;
 use crate::{
     ContainerFile, ExecOutput, FixtureCredentials, FixtureError, MysqlFlavour, MysqlOptions, PkiFile, TestNetwork,
     TestPki,
@@ -51,19 +51,22 @@ impl MysqlFixture {
             .into_iter()
             .fold(release.generic(), GenericImage::with_wait_for)
             .with_exposed_port(Self::CONTAINER_PORT.tcp());
-        let mut request = image
-            .with_env_var(format!("{prefix}_ROOT_PASSWORD"), &root_password)
-            .with_env_var(format!("{prefix}_DATABASE"), DATABASE)
-            .with_env_var(format!("{prefix}_USER"), PASSWORD_USER)
-            .with_env_var(format!("{prefix}_PASSWORD"), &password)
-            .with_cmd(Self::server_arguments(&options));
         let network_alias = network.map(|network| network.container_name(Self::NETWORK_ROLE));
-        if let (Some(network), Some(alias)) = (network, &network_alias) {
-            request = request.with_network(network.name()).with_container_name(alias);
-        }
-        let container = with_files(request, Self::files(&pki, &options, &certificate_password)?)
-            .start()
-            .await?;
+        let files = Self::files(&pki, &options, &certificate_password)?;
+        let container = start_container(|| {
+            let mut request = image
+                .clone()
+                .with_env_var(format!("{prefix}_ROOT_PASSWORD"), &root_password)
+                .with_env_var(format!("{prefix}_DATABASE"), DATABASE)
+                .with_env_var(format!("{prefix}_USER"), PASSWORD_USER)
+                .with_env_var(format!("{prefix}_PASSWORD"), &password)
+                .with_cmd(Self::server_arguments(&options));
+            if let (Some(network), Some(alias)) = (network, &network_alias) {
+                request = request.with_network(network.name()).with_container_name(alias);
+            }
+            with_files(request, files.clone())
+        })
+        .await?;
         let host = container.get_host().await?.to_string();
         let port = container.get_host_port_ipv4(Self::CONTAINER_PORT).await?;
         Ok(Self {