fix(gcp auth): OBE-12220 lazy token fetch to prevent startup crash loop

OBE-12220_review_findings.md

@@ -0,0 +1,148 @@
+# OBE-12220 Review Findings
+
+## Context
+
+This change makes GCP OAuth token fetching lazy — previously tokens were fetched eagerly at
+`build()` time, causing crash loops when credentials were expired at restart. Tokens are now
+fetched by a background `token_regenerator` task that fires immediately on startup and retries
+on failure.
+
+Reviewed with Opus. AWS auth patterns studied for comparison.
+
+---
+
+## Issues
+
+### HIGH — Health checks race the background token fetch (sinks)
+
+Every GCP sink constructs the healthcheck future and then spawns the token regenerator:
+
+```rust
+// same pattern in gcp/pubsub.rs, stackdriver/logs, stackdriver/metrics,
+// gcs cloud_storage, gcp_chronicle
+let healthcheck = healthcheck(client, uri, sink.auth.clone()).boxed();
+sink.auth.spawn_regenerate_token(&APP_INFO);   // token not yet fetched
+```
+
+The healthcheck calls `auth.apply()` → `make_token()` → `None` (no token yet) → no
+`Authorization` header → GCP returns 401 → healthcheck fails. With healthchecks enabled
+(the default), the crash loop becomes a healthcheck failure at startup instead of being
+fully resolved.
+
+The `gcp_pubsub` **source** is handled correctly — the watch channel + stream restart
+covers the race. **Sinks are the weak spot.**
+
+Fix options:
+- **Option A**: Await the first watch signal (with timeout) inside the healthcheck before
+  sending the request, treating a missing token as "pending" rather than fatal.
+- **Option B**: Gate sink readiness on the first `watch::Receiver::changed()` signal.
+- **Option C** (lowest touch): Treat 401 as a retryable/non-fatal response in the
+  healthcheck instead of a hard failure.
+
+### MEDIUM — First real sink request sends no auth header
+
+For HTTP sinks, `apply()` silently omits the `Authorization` header when `make_token()`
+returns `None`. Recovery depends on each sink's retry logic classifying 401 as retryable.
+That is not uniform across the affected sinks — needs verification:
+
+- `GcsRetryLogic` (`src/sinks/gcs_common/`)
+- Stackdriver logs/metrics retry logic
+- Chronicle retry logic
+
+If any classify 401 as non-retryable, the first batch of events is dropped on startup.
+
+### MEDIUM — `map_or` guard is dead code masking an invariant
+
+```rust
+// gcp.rs — after Ok(()) from regenerate_token:
+let expires_in = inner.token.read().unwrap()
+    .as_ref()
+    .map_or(METADATA_TOKEN_ERROR_RETRY_SECS, |t| t.expires_in() as u64);
+```
+
+`regenerate_token` unconditionally writes `Some(token)` before returning `Ok`, so the
+`None` fallback arm is unreachable under the current design. If a future change broke this
+invariant, the fallback (retry every 2s) would silently hammer the token endpoint rather
+than surfacing the bug.
+
+Prefer:
+```rust
+let expires_in = inner.token.read().unwrap()
+    .as_ref()
+    .expect("token present after successful regenerate")
+    .expires_in() as u64;
+```
+
+### LOW — No exponential backoff on repeated fetch failures
+
+With invalid static credentials (e.g. bad service account JSON), the loop retries every
+`METADATA_TOKEN_ERROR_RETRY_SECS` (2s) forever against `oauth2.googleapis.com`. The old
+code failed fast at build time. Consider exponential backoff with a cap for persistent
+auth failures.
+
+### LOW — `from_file` and `new_implicit` are needlessly `async`
+
+Both constructors do zero async work after the token fetch was removed. Drop `async` for
+accuracy. `GcpAuthConfig::build` will also no longer need to be `async` unless another
+reason requires it.
+
+### Formatting regression (unrelated)
+
+`src/sources/gcp_pubsub.rs` — the `#[snafu(display(...))]` attribute on the `Endpoint`
+variant lost its indentation:
+
+```rust
+// broken
+#[snafu(display("Could not create endpoint: {}", source))]
+```
+
+Fix with `make fmt` before merge.
+
+---
+
+## Test Coverage Gaps
+
+1. **No test that the background task delivers a token and fires the watch signal.**
+   The core async behavior (token transitions `None → Some`, watch receiver observes
+   `changed()`) is completely untested.
+
+2. **No test for the sink healthcheck race.** The HIGH finding above has no regression
+   test.
+
+3. **Invalid credentials path is untested.** `fails_missing_creds` was the only test
+   covering this. Its deletion leaves the error path for bad/expired credentials with
+   zero test coverage.
+
+---
+
+## AWS Comparison
+
+AWS uses `IdentityCache::lazy()` across all auth variants (`src/aws/auth.rs:205-225`).
+Credentials are never fetched at build time — only on the first request. There is no
+background refresh task; the AWS SDK refreshes inline via `provide_credentials()` on
+every signing call, with expiry managed inside the identity cache.
+
+The key difference: AWS delegates "no token yet" to the SDK, which blocks the first
+request for up to `load_timeout` (default 5s) waiting for credentials. The first request
+does not go out unauthenticated. GCP's new approach sends immediately with no header,
+which is what creates the sink healthcheck and first-request 401 problems.
+
+The GCP source path (gcp_pubsub) converges on the same safety property via a different
+mechanism — the stream fails, `changed()` fires when the token arrives, and the stream
+restarts authenticated. The sink path lacks an equivalent gate.
+
+---
+
+## Summary
+
+| Finding | Severity | Affects |
+|---------|----------|---------|
+| Healthcheck races token fetch | HIGH | All GCP sinks |
+| First request sends no auth header | MEDIUM | All GCP HTTP sinks |
+| `map_or` dead fallback masks invariant | MEDIUM | `token_regenerator` |
+| No backoff on auth failures | LOW | `token_regenerator` |
+| Needless `async` on constructors | LOW | `from_file`, `new_implicit` |
+| Formatting regression | LOW | `gcp_pubsub.rs` |
+| Missing async-token test | Gap | `gcp.rs` tests |
+| Missing healthcheck race test | Gap | sink integration tests |
+| Invalid creds path untested | Gap | `gcp.rs` tests |

lib/vector-core/src/gcp.rs

@@ -103,7 +103,7 @@ pub struct GcpAuthConfig {
 }
 
 impl GcpAuthConfig {
-    pub async fn build(&self, scope: Scope, app_info: &AppInfo) -> crate::Result<GcpAuthenticator> {
+    pub async fn build(&self, scope: Scope) -> crate::Result<GcpAuthenticator> {
         Ok(if self.skip_authentication {
             GcpAuthenticator::None
         } else {
@@ -112,7 +112,7 @@ impl GcpAuthConfig {
             match (&creds_path, &self.api_key) {
                 (Some(path), _) => GcpAuthenticator::from_file(path, scope).await?,
                 (None, Some(api_key)) => GcpAuthenticator::from_api_key(api_key.inner())?,
-                (None, None) => GcpAuthenticator::new_implicit(app_info).await?,
+                (None, None) => GcpAuthenticator::new_implicit().await?,
             }
         })
     }
@@ -128,21 +128,24 @@ pub enum GcpAuthenticator {
 #[derive(Debug)]
 pub struct InnerCreds {
     creds: Option<(Credentials, Scope)>,
-    token: RwLock<Token>,
+    token: RwLock<Option<Token>>,
 }
 
 impl GcpAuthenticator {
     async fn from_file(path: &str, scope: Scope) -> crate::Result<Self> {
         let creds = Credentials::from_file(path).context(InvalidCredentialsSnafu)?;
-        let token = RwLock::new(fetch_token(&creds, &scope).await?);
         let creds = Some((creds, scope));
-        Ok(Self::Credentials(Arc::new(InnerCreds { creds, token })))
+        Ok(Self::Credentials(Arc::new(InnerCreds {
+            creds,
+            token: RwLock::new(None),
+        })))
     }
 
-    async fn new_implicit(app_info: &AppInfo) -> crate::Result<Self> {
-        let token = RwLock::new(get_token_implicit(app_info).await?);
-        let creds = None;
-        Ok(Self::Credentials(Arc::new(InnerCreds { creds, token })))
+    async fn new_implicit() -> crate::Result<Self> {
+        Ok(Self::Credentials(Arc::new(InnerCreds {
+            creds: None,
+            token: RwLock::new(None),
+        })))
     }
 
     fn from_api_key(api_key: &str) -> crate::Result<Self> {
@@ -154,7 +157,7 @@ impl GcpAuthenticator {
 
     pub fn make_token(&self) -> Option<String> {
         match self {
-            Self::Credentials(inner) => Some(inner.make_token()),
+            Self::Credentials(inner) => inner.make_token(),
             Self::ApiKey(_) | Self::None => None,
         }
     }
@@ -198,33 +201,35 @@ impl GcpAuthenticator {
     async fn token_regenerator(self, sender: watch::Sender<()>, app_info: &'static AppInfo) {
         match self {
             Self::Credentials(inner) => {
-                let expires_in = inner.token.read().unwrap().expires_in() as u64;
-                let mut deadline =
-                    Duration::from_secs(expires_in - METADATA_TOKEN_EXPIRY_MARGIN_SECS);
                 loop {
-                    tokio::time::sleep(deadline).await;
-                    debug!("Renewing GCP authentication token.");
+                    debug!("Fetching GCP authentication token.");
                     match inner.regenerate_token(app_info).await {
                         Ok(()) => {
                             sender.send_replace(());
-                            let expires_in = inner.token.read().unwrap().expires_in() as u64;
+                            let expires_in = inner
+                                .token
+                                .read()
+                                .unwrap()
+                                .as_ref()
+                                .map_or(METADATA_TOKEN_ERROR_RETRY_SECS, |t| t.expires_in() as u64);
                             // Rather than an expected fresh token, the Metadata Server may return
                             // the same (cached) token during the last 300 seconds of its lifetime.
                             // This scenario is handled by retrying the token refresh after the
                             // METADATA_TOKEN_ERROR_RETRY_SECS period when a fresh token is expected
-                            let new_deadline = if expires_in <= METADATA_TOKEN_EXPIRY_MARGIN_SECS {
+                            let deadline = if expires_in <= METADATA_TOKEN_EXPIRY_MARGIN_SECS {
                                 METADATA_TOKEN_ERROR_RETRY_SECS
                             } else {
                                 expires_in - METADATA_TOKEN_EXPIRY_MARGIN_SECS
                             };
-                            deadline = Duration::from_secs(new_deadline);
+                            tokio::time::sleep(Duration::from_secs(deadline)).await;
                         }
                         Err(error) => {
                             error!(
-                                message = "Failed to update GCP authentication token.",
+                                message = "Failed to fetch GCP authentication token.",
                                 %error
                             );
-                            deadline = Duration::from_secs(METADATA_TOKEN_ERROR_RETRY_SECS);
+                            tokio::time::sleep(Duration::from_secs(METADATA_TOKEN_ERROR_RETRY_SECS))
+                                .await;
                         }
                     }
                 }
@@ -245,13 +250,15 @@ impl InnerCreds {
             Some((creds, scope)) => fetch_token(creds, scope).await?,
             None => get_token_implicit(app_info).await?,
         };
-        *self.token.write().unwrap() = token;
+        *self.token.write().unwrap() = Some(token);
         Ok(())
     }
 
-    fn make_token(&self) -> String {
+    fn make_token(&self) -> Option<String> {
         let token = self.token.read().unwrap();
-        format!("{} {}", token.token_type(), token.access_token())
+        token
+            .as_ref()
+            .map(|t| format!("{} {}", t.token_type(), t.access_token()))
     }
 }
 
@@ -307,10 +314,13 @@ mod tests {
     use crate::assert_downcast_matches;
 
     #[tokio::test]
-    async fn fails_missing_creds() {
-        let error = build_auth("").await.expect_err("build failed to error");
-        assert_downcast_matches!(error, GcpError, GcpError::GetImplicitToken { .. });
-        // This should be a more relevant error
+    async fn defers_implicit_auth_when_no_creds() {
+        // With lazy token fetching, building with no credentials succeeds immediately.
+        // The first token fetch attempt happens in the background via token_regenerator.
+        let auth = build_auth("").await.expect("build should succeed with deferred auth");
+        assert!(matches!(auth, GcpAuthenticator::Credentials(..)));
+        // No token yet — make_token returns None until the background fetch succeeds.
+        assert!(auth.make_token().is_none());
     }
 
     #[tokio::test]
@@ -369,10 +379,6 @@ mod tests {
 
     async fn build_auth(toml: &str) -> crate::Result<GcpAuthenticator> {
         let config: GcpAuthConfig = toml::from_str(toml).expect("Invalid TOML");
-        let app_info = vector_common::AppInfo {
-            name: "vector",
-            version: String::from("0.44.5"),
-        };
-        config.build(Scope::Compute, &app_info).await
+        config.build(Scope::Compute).await
     }
 }

src/sinks/gcp/cloud_storage.rs

@@ -238,7 +238,7 @@ impl SinkConfig for GcsSinkConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let auth = self
             .auth
-            .build(Scope::DevStorageReadWrite, &APP_INFO)
+            .build(Scope::DevStorageReadWrite)
             .await?;
         let base_url = format!("{}/{}/", self.endpoint, self.bucket);
         let tls = TlsSettings::from_options(self.tls.as_ref())?;

src/sinks/gcp/pubsub.rs

@@ -164,7 +164,7 @@ struct PubsubSink {
 impl PubsubSink {
     async fn from_config(config: &PubsubConfig) -> crate::Result<Self> {
         // We only need to load the credentials if we are not targeting an emulator.
-        let auth = config.auth.build(Scope::PubSub, &APP_INFO).await?;
+        let auth = config.auth.build(Scope::PubSub).await?;
 
         let uri_base = format!(
             "{}/v1/projects/{}/topics/{}",

src/sinks/gcp/stackdriver/logs/config.rs

@@ -202,8 +202,7 @@ impl_generate_config_from_default!(StackdriverConfig);
 #[typetag::serde(name = "gcp_stackdriver_logs")]
 impl SinkConfig for StackdriverConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let app_info = crate::app_info();
-        let auth = self.auth.build(Scope::LoggingWrite, &app_info).await?;
+        let auth = self.auth.build(Scope::LoggingWrite).await?;
 
         let request_builder = StackdriverLogsRequestBuilder {
             encoder: StackdriverLogsEncoder::new(

src/sinks/gcp/stackdriver/metrics/config.rs

@@ -95,7 +95,7 @@ impl_generate_config_from_default!(StackdriverConfig);
 #[typetag::serde(name = "gcp_stackdriver_metrics")]
 impl SinkConfig for StackdriverConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let auth = self.auth.build(Scope::MonitoringWrite, &APP_INFO).await?;
+        let auth = self.auth.build(Scope::MonitoringWrite).await?;
 
         let healthcheck = healthcheck().boxed();
         let started = chrono::Utc::now();

src/sinks/gcp_chronicle/chronicle_unstructured.rs

@@ -241,7 +241,7 @@ impl SinkConfig for ChronicleUnstructuredConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let creds = self
             .auth
-            .build(Scope::MalachiteIngestion, &APP_INFO)
+            .build(Scope::MalachiteIngestion)
             .await?;
 
         let tls = TlsSettings::from_options(self.tls.as_ref())?;

src/sources/gcp_pubsub.rs

@@ -95,7 +95,7 @@ mod proto {
 pub(crate) enum PubsubError {
     #[snafu(display("Invalid endpoint URI: {}", source))]
     Uri { source: InvalidUri },
-    #[snafu(display("Could not create endpoint: {}", source))]
+#[snafu(display("Could not create endpoint: {}", source))]
     Endpoint { source: tonic::transport::Error },
     #[snafu(display("Could not set up endpoint TLS settings: {}", source))]
     EndpointTls { source: tonic::transport::Error },
@@ -266,7 +266,7 @@ impl SourceConfig for PubsubConfig {
             }
         };
 
-        let auth = self.auth.build(Scope::PubSub, &APP_INFO).await?;
+        let auth = self.auth.build(Scope::PubSub).await?;
 
         let mut uri: Uri = self.endpoint.parse().context(UriSnafu)?;
         auth.apply_uri(&mut uri);