refactor: make auth interceptor await initial credential refresh and remove manual delays in tests.

Author: westarle

src/integration-tests/src/observability/auth.rs

@@ -50,9 +50,15 @@ pub struct CloudTelemetryAuthInterceptor {
 impl CloudTelemetryAuthInterceptor {
     /// Creates a new `CloudTelemetryAuthInterceptor` and starts a background task to keep
     /// credentials refreshed.
-    pub fn new(credentials: Credentials) -> Self {
-        let (tx, rx) = watch::channel(None);
+    pub async fn new(credentials: Credentials) -> Self {
+        let (tx, mut rx) = watch::channel(None);
         tokio::spawn(refresh_task(credentials, tx));
+
+        // Wait for the first refresh to complete.
+        // We ignore the result because if the sender is dropped (unlikely),
+        // the interceptor will just fail requests, which is the correct behavior.
+        let _ = rx.changed().await;
+
         Self { rx }
     }
 }
@@ -139,6 +145,12 @@ mod tests {
     #[tokio::test]
     async fn test_interceptor_injects_headers() {
         let (tx, rx) = watch::channel(None);
+        // Manually construct because new() spawns a task we don't want here,
+        // or we could just use new() and let it spawn.
+        // But since we want to control the channel, we construct manually as before.
+        // Wait, the previous test code manually constructed it:
+        // let mut interceptor = CloudTelemetryAuthInterceptor { rx };
+        // So we don't need to change this test if it doesn't use new().
         let mut interceptor = CloudTelemetryAuthInterceptor { rx };
 
         // 1. Initial state (no headers)

src/integration-tests/src/observability/otlp.rs

@@ -101,7 +101,7 @@ impl CloudTelemetryTracerProviderBuilder {
                 .build()
                 .map_err(|e| TraceError::Other(e.into()))?,
         };
-        let interceptor = CloudTelemetryAuthInterceptor::new(credentials);
+        let interceptor = CloudTelemetryAuthInterceptor::new(credentials).await;
 
         let resource = opentelemetry_sdk::Resource::builder_empty()
             .with_attributes(vec![

src/integration-tests/tests/telemetry.rs

@@ -38,7 +38,6 @@ mod telemetry {
         // This requires GOOGLE_CLOUD_PROJECT to be set.
         let project_id = integration_tests::project_id()?;
         let service_name = "e2e-telemetry-test";
-        println!("Project ID: {}", project_id);
 
         // Configure OTLP provider (sends to telemetry.googleapis.com)
         // This uses ADC automatically from the environment.
@@ -53,28 +52,23 @@ mod telemetry {
             ))
             .set_default();
 
-        // Wait for credentials to be refreshed in the background task
-        println!("Waiting for credentials to initialize...");
-        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-
         // 3. Generate Trace
         let span_name = "e2e-showcase-test";
 
         // Start a root span
         let root_span = tracing::info_span!("e2e_root", "otel.name" = span_name);
         let trace_id = {
             let _enter = root_span.enter();
-            let otel_ctx = root_span.context();
-            let otel_span_ref = otel_ctx.span();
-            let span_context = otel_span_ref.span_context();
-            let trace_id = span_context.trace_id().to_string();
+            let trace_id = root_span
+                .context()
+                .span()
+                .span_context()
+                .trace_id()
+                .to_string();
 
             // Initialize showcase client pointing to local mock server
-            // We use anonymous credentials for the *client* because it's talking to httptest
-            let endpoint = format!("http://{}", echo_server.addr());
-
             let client = showcase::client::Echo::builder()
-                .with_endpoint(endpoint)
+                .with_endpoint(format!("http://{}", echo_server.addr()))
                 .with_credentials(auth::credentials::anonymous::Builder::new().build())
                 .with_tracing()
                 .build()
@@ -89,30 +83,24 @@ mod telemetry {
         // explicitly drop the span to end it
         drop(root_span);
 
-        println!("Generated Trace ID: {}", trace_id);
-        println!("Span Name: {}", span_name);
         println!(
-            "View in Console: https://console.cloud.google.com/traces/explorer;traceId={}?project={}",
+            "View generated trace in Console: https://console.cloud.google.com/traces/explorer;traceId={}?project={}",
             trace_id, project_id
         );
 
-        // 4. Flush
-        // Force flush
-        for _ in 0..5 {
-            let _ = provider.force_flush();
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-        }
-        println!("Spans flushed.");
+        // 4. Force flush to ensure spans are sent.
+        let _ = provider.force_flush();
 
         // 5. Verify (Poll Cloud Trace API)
         let client = TraceService::builder().build().await?;
 
-        println!("Polling for trace...");
+        // Because we are limited by quota, start with a backoff.
+        // Traces can take several minutes to propagate after they have been written.
+        // Implement a generous retry loop to account for this.
         let backoff_delays = [10, 60, 120, 120, 120];
         let mut trace = None;
 
         for delay in backoff_delays {
-            println!("Waiting {}s before polling...", delay);
             tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
 
             match client
@@ -145,9 +133,6 @@ mod telemetry {
 
         let trace = trace.ok_or_else(|| anyhow::anyhow!("Timed out waiting for trace"))?;
 
-        println!("Trace found!");
-        println!("Response: {:?}", trace);
-
         // 6. Assertions
         // Check for root span
         let root_found = trace.spans.iter().any(|s| s.name == span_name);