refactor(compose-runner): reduce duplicated failure paths

testing-framework/runners/compose/src/deployer/clients.rs

@@ -27,16 +27,19 @@ impl ClientBuilder {
         host: &str,
         environment: &mut StackEnvironment,
     ) -> Result<NodeClients, ComposeRunnerError> {
-        match build_node_clients_with_ports(descriptors, host_ports, host) {
+        let clients = match build_node_clients_with_ports(descriptors, host_ports, host) {
-            Ok(clients) => Ok(clients),
+            Ok(clients) => clients,
             Err(err) => {
-                environment
+                return Err(fail_deploy_step(
-                    .fail("failed to construct node api clients")
+                    environment,
-                    .await;
+                    "failed to construct node api clients",
-                tracing::warn!(error = ?err, host, "failed to build node clients");
+                    "failed to build node clients",
-                Err(err.into())
+                    err,
+                )
+                .await);
             }
-        }
+        };
+        Ok(clients)
     }
 
     pub async fn start_block_feed(
@@ -44,16 +47,33 @@ impl ClientBuilder {
         node_clients: &NodeClients,
         environment: &mut StackEnvironment,
     ) -> Result<(BlockFeed, BlockFeedTask), ComposeRunnerError> {
-        match spawn_block_feed_with_retry(node_clients).await {
+        let pair = match spawn_block_feed_with_retry(node_clients).await {
-            Ok(pair) => {
+            Ok(pair) => pair,
-                info!("block feed connected to validator");
-                Ok(pair)
-            }
             Err(err) => {
-                environment.fail("failed to initialize block feed").await;
+                return Err(fail_deploy_step(
-                tracing::warn!(error = ?err, "block feed initialization failed");
+                    environment,
-                Err(err)
+                    "failed to initialize block feed",
+                    "block feed initialization failed",
+                    err,
+                )
+                .await);
             }
-        }
+        };
+        info!("block feed connected to validator");
+        Ok(pair)
     }
 }
+
+async fn fail_deploy_step<E>(
+    environment: &mut StackEnvironment,
+    reason: &str,
+    log_message: &str,
+    error: E,
+) -> ComposeRunnerError
+where
+    E: std::fmt::Debug + Into<ComposeRunnerError>,
+{
+    environment.fail(reason).await;
+    tracing::warn!(error = ?error, "{log_message}");
+    error.into()
+}

testing-framework/runners/compose/src/deployer/readiness.rs

@@ -18,37 +18,56 @@ impl ReadinessChecker {
         host_ports: &HostPortMapping,
         environment: &mut StackEnvironment,
     ) -> Result<(), ComposeRunnerError> {
-        info!(
+        let validator_ports = host_ports.validator_api_ports();
-            ports = ?host_ports.validator_api_ports(),
+        info!(ports = ?validator_ports, "waiting for validator HTTP endpoints");
-            "waiting for validator HTTP endpoints"
+        if let Err(err) = ensure_validators_ready_with_ports(&validator_ports).await {
-        );
+            return fail_readiness_step(
-        if let Err(err) =
+                environment,
-            ensure_validators_ready_with_ports(&host_ports.validator_api_ports()).await
+                "validator readiness failed",
-        {
+                "validator readiness failed",
-            environment.fail("validator readiness failed").await;
+                err,
-            tracing::warn!(error = ?err, "validator readiness failed");
+            )
-            return Err(err.into());
+            .await;
         }
 
-        info!(
+        let executor_ports = host_ports.executor_api_ports();
-            ports = ?host_ports.executor_api_ports(),
+        info!(ports = ?executor_ports, "waiting for executor HTTP endpoints");
-            "waiting for executor HTTP endpoints"
+        if let Err(err) = ensure_executors_ready_with_ports(&executor_ports).await {
-        );
+            return fail_readiness_step(
-        if let Err(err) = ensure_executors_ready_with_ports(&host_ports.executor_api_ports()).await
+                environment,
-        {
+                "executor readiness failed",
-            environment.fail("executor readiness failed").await;
+                "executor readiness failed",
-            tracing::warn!(error = ?err, "executor readiness failed");
+                err,
-            return Err(err.into());
+            )
+            .await;
         }
 
         info!("waiting for remote service readiness");
         if let Err(err) = ensure_remote_readiness_with_ports(descriptors, host_ports).await {
-            environment.fail("remote readiness probe failed").await;
+            return fail_readiness_step(
-            tracing::warn!(error = ?err, "remote readiness probe failed");
+                environment,
-            return Err(err.into());
+                "remote readiness probe failed",
+                "remote readiness probe failed",
+                err,
+            )
+            .await;
         }
 
         info!("compose readiness checks passed");
         Ok(())
     }
 }
+
+async fn fail_readiness_step<E>(
+    environment: &mut StackEnvironment,
+    reason: &str,
+    log_message: &str,
+    error: E,
+) -> Result<(), ComposeRunnerError>
+where
+    E: std::fmt::Debug + Into<ComposeRunnerError>,
+{
+    environment.fail(reason).await;
+    tracing::warn!(error = ?error, "{log_message}");
+    Err(error.into())
+}

testing-framework/runners/compose/src/lifecycle/cleanup.rs

@@ -95,9 +95,7 @@ impl CleanupGuard for RunnerCleanup {
 
         self.teardown_compose();
 
-        if let Some(mut handle) = self.cfgsync.take() {
+        self.shutdown_cfgsync();
-            handle.shutdown();
-        }
     }
 }
 
@@ -114,4 +112,10 @@ impl RunnerCleanup {
 
         info!("compose preserve flag set; skipping docker compose down");
     }
+
+    fn shutdown_cfgsync(&mut self) {
+        if let Some(mut handle) = self.cfgsync.take() {
+            handle.shutdown();
+        }
+    }
 }