From 63da8bef20dc64a7eb0d5aaa30a3cbc6f4087e5f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Mon, 20 Jul 2026 19:26:56 +0200
Subject: [PATCH 01/11] fix(gcp_pubsub source): recover from stalled streams
 and reduce error noise

---
 .../19418_gcp_pubsub_idle_reconnect.fix.md    |   7 +
 src/sources/gcp_pubsub.rs                     | 205 ++++++++++++++++--
 .../sources/generated/gcp_pubsub.cue          |  44 ++++
 3 files changed, 237 insertions(+), 19 deletions(-)
 create mode 100644 changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md

diff --git a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
new file mode 100644
index 0000000000000..f8ed5d1753615
--- /dev/null
+++ b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
@@ -0,0 +1,7 @@
+The `gcp_pubsub` source no longer floods the logs with misleading errors on idle, low-volume, or bursty subscriptions. GCP routinely ends a streaming pull with a transient error (such as an `Unavailable` status when there are simply no messages to deliver), which Vector retries automatically. These transient failures are now logged at `debug`, and a `failed_fetching_events` component error is only emitted once several failures occur in a row without any successful fetch in between. The threshold is configurable via the new `max_retry_errors` option (default `5`); a successful fetch resets the counter.
+
+The source is also more resilient to connections that silently break and stop delivering messages until Vector is restarted:
+
+- HTTP/2 keepalive pings are sent on the gRPC connection so a dead-but-open connection is detected and torn down instead of hanging.
+- The stream keepalive uses a non-blocking send, so a backed-up request channel can no longer deadlock the task.
+- A new `idle_timeout_secs` option (defaulting to `900`) restarts the stream if no activity is seen on an active connection within that window.
diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 1600c07dab339..2904ffdf26b0f 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -66,6 +66,11 @@ const MAX_ACK_DEADLINE_SECS: u64 = 600;
 // processing.
 const ACK_QUEUE_SIZE: usize = 8;
 
+// Time to wait for an HTTP/2 keepalive ping response before closing the
+// connection, and the upper bound on a single connection attempt.
+const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
+const HTTP2_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+
 type Finalizer = UnorderedFinalizer<Vec<String>>;
 
 // prost emits some generated code that includes clones on `Arc`
@@ -116,6 +121,15 @@ pub(crate) enum PubsubError {
         MAX_ACK_DEADLINE_SECS
     ))]
     InvalidAckDeadline,
+    #[snafu(display(
+        "`idle_timeout_secs` ({}) must be larger than `keepalive_secs` ({})",
+        idle_timeout_secs,
+        keepalive_secs
+    ))]
+    InvalidIdleTimeout {
+        idle_timeout_secs: f64,
+        keepalive_secs: f64,
+    },
 }
 
 static CLIENT_ID: LazyLock<String> = LazyLock::new(|| uuid::Uuid::new_v4().to_string());
@@ -205,6 +219,31 @@ pub struct PubsubConfig {
     #[configurable(metadata(docs::human_name = "Keepalive"))]
     pub keepalive_secs: Duration,
 
+    /// The maximum amount of time, in seconds, to wait for a response on an
+    /// active stream before assuming the connection has stalled and
+    /// reconnecting.
+    ///
+    /// GCP Pub/Sub can leave a streaming pull in a state where the
+    /// connection looks healthy but no further responses are delivered and
+    /// the stream never errors. This bounds that inactivity. It must be
+    /// larger than `keepalive_secs`.
+    #[serde(default = "default_idle_timeout")]
+    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
+    #[configurable(metadata(docs::human_name = "Idle Timeout"))]
+    pub idle_timeout_secs: Duration,
+
+    /// The number of consecutive stream failures, with no successful fetch in
+    /// between, before a fetch error is reported as a component error.
+    ///
+    /// GCP Pub/Sub routinely ends a streaming pull with a transient error
+    /// (for example, an `Unavailable` status when an idle subscription has no
+    /// messages). These are retried automatically and logged at `debug`
+    /// until this threshold is reached. A successful fetch resets the count.
+    /// Set to `1` to report every failure as an error.
+    #[serde(default = "default_max_retry_errors")]
+    #[configurable(metadata(docs::human_name = "Max Retry Errors Before Alerting"))]
+    pub max_retry_errors: usize,
+
     /// The namespace to use for logs. This overrides the global setting.
     #[configurable(metadata(docs::hidden))]
     #[serde(default)]
@@ -241,6 +280,14 @@ const fn default_keepalive() -> Duration {
     Duration::from_secs(60)
 }
 
+const fn default_idle_timeout() -> Duration {
+    Duration::from_secs(900)
+}
+
+const fn default_max_retry_errors() -> usize {
+    5
+}
+
 const fn default_max_concurrency() -> usize {
     10
 }
@@ -271,6 +318,14 @@ impl SourceConfig for PubsubConfig {
             return Err(PubsubError::InvalidAckDeadline.into());
         }
 
+        if self.idle_timeout_secs <= self.keepalive_secs {
+            return Err(PubsubError::InvalidIdleTimeout {
+                idle_timeout_secs: self.idle_timeout_secs.as_secs_f64(),
+                keepalive_secs: self.keepalive_secs.as_secs_f64(),
+            }
+            .into());
+        }
+
         let retry_delay_secs = match self.retry_delay_seconds {
             None => self.retry_delay_secs,
             Some(rds) => {
@@ -301,6 +356,16 @@ impl SourceConfig for PubsubConfig {
             endpoint = endpoint.tls_config(tls_config).context(EndpointTlsSnafu)?;
         }
 
+        // Detect a silently broken connection (for example, GCP dropping
+        // the stream but leaving the socket open) via HTTP/2 keepalive
+        // pings, and bound the initial dial, so a reconnect can't wedge
+        // with no error ever surfacing.
+        endpoint = endpoint
+            .http2_keep_alive_interval(self.keepalive_secs)
+            .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT)
+            .keep_alive_while_idle(true)
+            .connect_timeout(HTTP2_CONNECT_TIMEOUT);
+
         let token_generator = auth.spawn_regenerate_token();
 
         let protocol = uri
@@ -328,6 +393,9 @@ impl SourceConfig for PubsubConfig {
             ack_deadline_secs,
             retry_delay: retry_delay_secs,
             keepalive: self.keepalive_secs,
+            idle_timeout: self.idle_timeout_secs,
+            max_retry_errors: self.max_retry_errors.max(1),
+            consecutive_failures: 0,
             concurrency: Default::default(),
             full_response_size: self.full_response_size,
             log_namespace,
@@ -393,6 +461,10 @@ struct PubsubSource {
     out: SourceSender,
     retry_delay: Duration,
     keepalive: Duration,
+    idle_timeout: Duration,
+    max_retry_errors: usize,
+    // Consecutive stream failures since the last successful fetch.
+    consecutive_failures: usize,
     // The current concurrency is shared across all tasks. It is used
     // by the streams to avoid shutting down the last stream, which
     // would result in repeatedly re-opening the stream on idle.
@@ -522,6 +594,12 @@ impl PubsubSource {
             Finalizer::maybe_new(self.acknowledgements, Some(self.shutdown.clone()));
         let mut pending_acks = 0;
 
+        // Reconnect if no response arrives within the idle timeout, to
+        // recover from a stream that stalls without ever erroring. Reset
+        // whenever a response arrives.
+        let idle_deadline = tokio::time::sleep(self.idle_timeout);
+        tokio::pin!(idle_deadline);
+
         loop {
             tokio::select! {
                 biased;
@@ -536,6 +614,10 @@ impl PubsubSource {
                 },
                 response = stream.next() => match response {
                     Some(Ok(response)) => {
+                        self.consecutive_failures = 0;
+                        idle_deadline
+                            .as_mut()
+                            .reset(tokio::time::Instant::now() + self.idle_timeout);
                         self.handle_response(
                             response,
                             &finalizer,
@@ -544,7 +626,7 @@ impl PubsubSource {
                             busy_flag,
                         ).await;
                     }
-                    Some(Err(error)) => break translate_error(error),
+                    Some(Err(error)) => break self.handle_stream_error(error),
                     None => break State::RetryNow,
                 },
                 _ = &mut self.shutdown, if pending_acks == 0 => return State::Shutdown,
@@ -552,6 +634,13 @@ impl PubsubSource {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
+                _ = &mut idle_deadline => {
+                    debug!(
+                        message = "No activity on the stream within the idle timeout, restarting stream.",
+                        timeout_secs = self.idle_timeout.as_secs_f64(),
+                    );
+                    break State::RetryDelay;
+                },
                 _ = tokio::time::sleep(self.keepalive) => {
                     if pending_acks == 0 {
                         // No pending acks, and no new data, so drop
@@ -571,10 +660,20 @@ impl PubsubSource {
                     // other activity has happened. This will result
                     // in a new request with empty fields, effectively
                     // a keepalive.
-                    ack_ids_sender
-                        .send(Vec::new())
-                        .await
-                        .unwrap_or_else(|_| unreachable!("request stream never closes"));
+                    //
+                    // Send without blocking: a full channel means the
+                    // request stream stalled, so reconnect rather than
+                    // deadlock.
+                    match ack_ids_sender.try_send(Vec::new()) {
+                        Ok(()) => {}
+                        Err(mpsc::error::TrySendError::Full(_)) => {
+                            debug!("Keepalive request stream stalled, restarting stream.");
+                            break State::RetryDelay;
+                        }
+                        Err(mpsc::error::TrySendError::Closed(_)) => {
+                            unreachable!("request stream never closes")
+                        }
+                    }
                 }
             }
         }
@@ -647,6 +746,32 @@ impl PubsubSource {
         }
     }
 
+    // GCP frequently ends a streaming pull with a transient, auto-retried
+    // error (a reset, or `Unavailable` when an idle subscription has no
+    // messages). Only escalate to a component error once failures persist,
+    // to avoid flooding the logs on quiet subscriptions.
+    fn handle_stream_error(&mut self, error: tonic::Status) -> State {
+        // A server-side reset is not really an error, so retry immediately
+        // without counting it.
+        if is_reset(&error) {
+            debug!("Stream reset by server.");
+            return State::RetryNow;
+        }
+
+        self.consecutive_failures += 1;
+        if should_alert(self.consecutive_failures, self.max_retry_errors) {
+            emit!(GcpPubsubReceiveError { error });
+        } else {
+            debug!(
+                message = "Transient error fetching events, retrying.",
+                %error,
+                consecutive_failures = self.consecutive_failures,
+                max_retry_errors = self.max_retry_errors,
+            );
+        }
+        State::RetryDelay
+    }
+
     async fn parse_messages(
         &self,
         response: Vec<proto::ReceivedMessage>,
@@ -712,20 +837,9 @@ impl PubsubSource {
     }
 }
 
-fn translate_error(error: tonic::Status) -> State {
-    // GCP occasionally issues a connection reset
-    // in the middle of the streaming pull. This
-    // reset is not technically an error, so we
-    // want to retry immediately, but it is
-    // reported to us as an error from the
-    // underlying library (`tonic`).
-    if is_reset(&error) {
-        debug!("Stream reset by server.");
-        State::RetryNow
-    } else {
-        emit!(GcpPubsubReceiveError { error });
-        State::RetryDelay
-    }
+// Escalate to a component error once failures reach the threshold.
+const fn should_alert(consecutive_failures: usize, max_retry_errors: usize) -> bool {
+    consecutive_failures >= max_retry_errors
 }
 
 fn is_reset(error: &Status) -> bool {
@@ -762,6 +876,59 @@ mod tests {
         crate::test_util::test_generate_config::<PubsubConfig>();
     }
 
+    #[test]
+    fn alerts_only_once_failures_reach_threshold() {
+        let max = default_max_retry_errors();
+        assert!((1..max).all(|n| !should_alert(n, max)));
+        assert!(should_alert(max, max));
+        assert!(should_alert(max + 1, max));
+        // A threshold of 1 reports every failure (the previous behavior).
+        assert!(should_alert(1, 1));
+    }
+
+    #[test]
+    fn default_idle_timeout_is_larger_than_keepalive() {
+        // Defaults come from serde, not `Default::default()`.
+        let config: PubsubConfig = serde_yaml::from_str(
+            r#"
+            project: my-project
+            subscription: my-subscription
+            "#,
+        )
+        .unwrap();
+        assert!(
+            config.idle_timeout_secs > config.keepalive_secs,
+            "the default idle timeout must be larger than the default keepalive"
+        );
+    }
+
+    #[tokio::test]
+    async fn rejects_idle_timeout_not_larger_than_keepalive() {
+        let config: PubsubConfig = serde_yaml::from_str(
+            r#"
+            project: my-project
+            subscription: my-subscription
+            skip_authentication: true
+            keepalive_secs: 60
+            idle_timeout_secs: 60
+            "#,
+        )
+        .unwrap();
+
+        let (tx, _rx) = SourceSender::new_test();
+        let context = SourceContext::new_test(tx, None);
+        let error = match config.build(context).await {
+            Ok(_) => {
+                panic!("build should reject an idle timeout that is not larger than keepalive")
+            }
+            Err(error) => error.to_string(),
+        };
+        assert!(
+            error.contains("idle_timeout_secs"),
+            "unexpected error: {error}"
+        );
+    }
+
     #[test]
     fn output_schema_definition_vector_namespace() {
         let config = PubsubConfig {
diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
index af645bf7384b3..83a5d9ba7ce3f 100644
--- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue
+++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
@@ -597,6 +597,30 @@ generated: components: sources: gcp_pubsub: configuration: {
 		required: false
 		type: uint: default: 100
 	}
+	idle_timeout_secs: {
+		description: """
+			The maximum amount of time, in seconds, to wait for any response
+			on an active stream before assuming the connection has stalled
+			and reconnecting.
+
+			The GCP Pub/Sub servers occasionally leave a streaming pull in a
+			state where the connection appears healthy but no further
+			responses (not even keepalive echoes) are ever delivered, and the
+			stream never errors or closes on its own. When no response is
+			received within this window, the stream is torn down and
+			re-established to recover.
+
+			A healthy but idle stream still receives periodic keepalive
+			responses and is recycled by the server on its own, so this only
+			needs to be larger than that activity. It must be larger than
+			`keepalive_secs`.
+			"""
+		required: false
+		type: float: {
+			default: 900.0
+			unit:    "seconds"
+		}
+	}
 	keepalive_secs: {
 		description: """
 			The amount of time, in seconds, with no received activity
@@ -614,6 +638,26 @@ generated: components: sources: gcp_pubsub: configuration: {
 		required:    false
 		type: uint: default: 10
 	}
+	max_retry_errors: {
+		description: """
+			The number of consecutive stream failures that must occur before a
+			fetch error is reported as a component error.
+
+			The GCP Pub/Sub servers routinely end a streaming pull with a
+			transient error (for example, an `Unavailable` status when there
+			are no messages to deliver on an idle or low-volume subscription).
+			These are retried automatically and are not actionable on their
+			own, so the first failures after a successful fetch are logged at
+			the `debug` level. Only once this many failures happen in a row
+			without any successful fetch in between is a `failed_fetching_events`
+			component error emitted, indicating the stream is genuinely stuck.
+
+			A successful fetch resets the counter. Set to `1` to report every
+			failure as an error (the previous behavior).
+			"""
+		required: false
+		type: uint: default: 5
+	}
 	poll_time_seconds: {
 		description: """
 			How often to poll the currently active streams to see if they

From 54108a4348c5cede1c6ee47b476aab6e330a92af Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Mon, 20 Jul 2026 19:40:14 +0200
Subject: [PATCH 02/11] chore(gcp_pubsub source): drop low-value decision unit
 tests

---
 src/sources/gcp_pubsub.rs | 17 +----------------
 1 file changed, 1 insertion(+), 16 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 2904ffdf26b0f..a5d692eaacb14 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -759,7 +759,7 @@ impl PubsubSource {
         }
 
         self.consecutive_failures += 1;
-        if should_alert(self.consecutive_failures, self.max_retry_errors) {
+        if self.consecutive_failures >= self.max_retry_errors {
             emit!(GcpPubsubReceiveError { error });
         } else {
             debug!(
@@ -837,11 +837,6 @@ impl PubsubSource {
     }
 }
 
-// Escalate to a component error once failures reach the threshold.
-const fn should_alert(consecutive_failures: usize, max_retry_errors: usize) -> bool {
-    consecutive_failures >= max_retry_errors
-}
-
 fn is_reset(error: &Status) -> bool {
     error
         .source()
@@ -876,16 +871,6 @@ mod tests {
         crate::test_util::test_generate_config::<PubsubConfig>();
     }
 
-    #[test]
-    fn alerts_only_once_failures_reach_threshold() {
-        let max = default_max_retry_errors();
-        assert!((1..max).all(|n| !should_alert(n, max)));
-        assert!(should_alert(max, max));
-        assert!(should_alert(max + 1, max));
-        // A threshold of 1 reports every failure (the previous behavior).
-        assert!(should_alert(1, 1));
-    }
-
     #[test]
     fn default_idle_timeout_is_larger_than_keepalive() {
         // Defaults come from serde, not `Default::default()`.

From 40b224bdde73a79b7f555870d051fc3d0a8cd3dc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Mon, 20 Jul 2026 20:38:10 +0200
Subject: [PATCH 03/11] enhancement(gcp_pubsub source): align HTTP/2 keepalive
 with vector sink config convention

---
 src/sources/gcp_pubsub.rs                     | 71 +++++++++++++----
 .../sources/generated/gcp_pubsub.cue          | 76 ++++++++++++-------
 2 files changed, 105 insertions(+), 42 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index a5d692eaacb14..19cbb5d42a3f5 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -1,6 +1,7 @@
 use std::{
     error::Error as _,
     future::Future,
+    num::NonZeroU64,
     pin::Pin,
     sync::{
         Arc, LazyLock,
@@ -66,11 +67,6 @@ const MAX_ACK_DEADLINE_SECS: u64 = 600;
 // processing.
 const ACK_QUEUE_SIZE: usize = 8;
 
-// Time to wait for an HTTP/2 keepalive ping response before closing the
-// connection, and the upper bound on a single connection attempt.
-const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
-const HTTP2_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
-
 type Finalizer = UnorderedFinalizer<Vec<String>>;
 
 // prost emits some generated code that includes clones on `Arc`
@@ -134,6 +130,47 @@ pub(crate) enum PubsubError {
 
 static CLIENT_ID: LazyLock<String> = LazyLock::new(|| uuid::Uuid::new_v4().to_string());
 
+/// HTTP/2 keepalive configuration for the `gcp_pubsub` source's gRPC connection.
+#[configurable_component]
+#[derive(Clone, Copy, Debug)]
+#[serde(deny_unknown_fields)]
+pub struct PubsubKeepaliveConfig {
+    /// How often, in seconds, to send a keepalive PING on the connection.
+    ///
+    /// Shorter intervals detect dead connections faster at the cost of additional traffic.
+    /// gRPC guidance recommends no less than 60 seconds to avoid tripping `too_many_pings`
+    /// policies on servers or proxies between the source and Pub/Sub.
+    #[serde(default = "default_keepalive_interval_secs")]
+    #[configurable(metadata(docs::human_name = "Keepalive Interval"))]
+    pub interval_secs: NonZeroU64,
+
+    /// How long, in seconds, to wait for a keepalive PING acknowledgement before treating
+    /// the connection as dead and closing it.
+    #[serde(default = "default_keepalive_timeout_secs")]
+    #[configurable(metadata(docs::human_name = "Keepalive Timeout"))]
+    pub timeout_secs: NonZeroU64,
+}
+
+const fn default_keepalive_interval_secs() -> NonZeroU64 {
+    // Aligned with gRPC keepalive guidance, which recommends no less than one minute to avoid
+    // tripping `too_many_pings` policies on proxies between the source and Pub/Sub.
+    NonZeroU64::new(60).expect("keepalive interval default must be nonzero")
+}
+
+const fn default_keepalive_timeout_secs() -> NonZeroU64 {
+    // Matches hyper's default keepalive timeout.
+    NonZeroU64::new(20).expect("keepalive timeout default must be nonzero")
+}
+
+impl Default for PubsubKeepaliveConfig {
+    fn default() -> Self {
+        Self {
+            interval_secs: default_keepalive_interval_secs(),
+            timeout_secs: default_keepalive_timeout_secs(),
+        }
+    }
+}
+
 /// Configuration for the `gcp_pubsub` source.
 #[serde_as]
 #[configurable_component(source(
@@ -244,6 +281,16 @@ pub struct PubsubConfig {
     #[configurable(metadata(docs::human_name = "Max Retry Errors Before Alerting"))]
     pub max_retry_errors: usize,
 
+    /// HTTP/2 keepalive settings for the source's gRPC connection.
+    ///
+    /// The source sends HTTP/2 PING frames on the connection so that one which has gone away
+    /// (for example, dropped by Pub/Sub or cut off by a network partition) is detected and
+    /// torn down instead of silently delivering no further messages.
+    #[configurable(derived)]
+    #[serde(default)]
+    #[derivative(Default(value = "PubsubKeepaliveConfig::default()"))]
+    pub keepalive: PubsubKeepaliveConfig,
+
     /// The namespace to use for logs. This overrides the global setting.
     #[configurable(metadata(docs::hidden))]
     #[serde(default)]
@@ -356,15 +403,13 @@ impl SourceConfig for PubsubConfig {
             endpoint = endpoint.tls_config(tls_config).context(EndpointTlsSnafu)?;
         }
 
-        // Detect a silently broken connection (for example, GCP dropping
-        // the stream but leaving the socket open) via HTTP/2 keepalive
-        // pings, and bound the initial dial, so a reconnect can't wedge
-        // with no error ever surfacing.
+        // Send HTTP/2 keepalive PINGs so a silently broken connection (for
+        // example, one dropped by Pub/Sub but left open) is detected and
+        // torn down instead of delivering no further messages.
         endpoint = endpoint
-            .http2_keep_alive_interval(self.keepalive_secs)
-            .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT)
-            .keep_alive_while_idle(true)
-            .connect_timeout(HTTP2_CONNECT_TIMEOUT);
+            .http2_keep_alive_interval(Duration::from_secs(self.keepalive.interval_secs.get()))
+            .keep_alive_timeout(Duration::from_secs(self.keepalive.timeout_secs.get()))
+            .keep_alive_while_idle(true);
 
         let token_generator = auth.spawn_regenerate_token();
 
diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
index 83a5d9ba7ce3f..e051febcad3d9 100644
--- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue
+++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
@@ -599,21 +599,14 @@ generated: components: sources: gcp_pubsub: configuration: {
 	}
 	idle_timeout_secs: {
 		description: """
-			The maximum amount of time, in seconds, to wait for any response
-			on an active stream before assuming the connection has stalled
-			and reconnecting.
-
-			The GCP Pub/Sub servers occasionally leave a streaming pull in a
-			state where the connection appears healthy but no further
-			responses (not even keepalive echoes) are ever delivered, and the
-			stream never errors or closes on its own. When no response is
-			received within this window, the stream is torn down and
-			re-established to recover.
-
-			A healthy but idle stream still receives periodic keepalive
-			responses and is recycled by the server on its own, so this only
-			needs to be larger than that activity. It must be larger than
-			`keepalive_secs`.
+			The maximum amount of time, in seconds, to wait for a response on an
+			active stream before assuming the connection has stalled and
+			reconnecting.
+
+			GCP Pub/Sub can leave a streaming pull in a state where the
+			connection looks healthy but no further responses are delivered and
+			the stream never errors. This bounds that inactivity. It must be
+			larger than `keepalive_secs`.
 			"""
 		required: false
 		type: float: {
@@ -621,6 +614,37 @@ generated: components: sources: gcp_pubsub: configuration: {
 			unit:    "seconds"
 		}
 	}
+	keepalive: {
+		description: """
+			HTTP/2 keepalive settings for the source's gRPC connection.
+
+			The source sends HTTP/2 PING frames on the connection so that one which has gone away
+			(for example, dropped by Pub/Sub or cut off by a network partition) is detected and
+			torn down instead of silently delivering no further messages.
+			"""
+		required: false
+		type: object: options: {
+			interval_secs: {
+				description: """
+					How often, in seconds, to send a keepalive PING on the connection.
+
+					Shorter intervals detect dead connections faster at the cost of additional traffic.
+					gRPC guidance recommends no less than 60 seconds to avoid tripping `too_many_pings`
+					policies on servers or proxies between the source and Pub/Sub.
+					"""
+				required: false
+				type: uint: default: 60
+			}
+			timeout_secs: {
+				description: """
+					How long, in seconds, to wait for a keepalive PING acknowledgement before treating
+					the connection as dead and closing it.
+					"""
+				required: false
+				type: uint: default: 20
+			}
+		}
+	}
 	keepalive_secs: {
 		description: """
 			The amount of time, in seconds, with no received activity
@@ -640,20 +664,14 @@ generated: components: sources: gcp_pubsub: configuration: {
 	}
 	max_retry_errors: {
 		description: """
-			The number of consecutive stream failures that must occur before a
-			fetch error is reported as a component error.
-
-			The GCP Pub/Sub servers routinely end a streaming pull with a
-			transient error (for example, an `Unavailable` status when there
-			are no messages to deliver on an idle or low-volume subscription).
-			These are retried automatically and are not actionable on their
-			own, so the first failures after a successful fetch are logged at
-			the `debug` level. Only once this many failures happen in a row
-			without any successful fetch in between is a `failed_fetching_events`
-			component error emitted, indicating the stream is genuinely stuck.
-
-			A successful fetch resets the counter. Set to `1` to report every
-			failure as an error (the previous behavior).
+			The number of consecutive stream failures, with no successful fetch in
+			between, before a fetch error is reported as a component error.
+
+			GCP Pub/Sub routinely ends a streaming pull with a transient error
+			(for example, an `Unavailable` status when an idle subscription has no
+			messages). These are retried automatically and logged at `debug`
+			until this threshold is reached. A successful fetch resets the count.
+			Set to `1` to report every failure as an error.
 			"""
 		required: false
 		type: uint: default: 5

From 7e6e1a0236f07f78bf711022c8aa18add7413b69 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Mon, 20 Jul 2026 20:41:28 +0200
Subject: [PATCH 04/11] docs(gcp_pubsub source): note configurable keepalive in
 changelog

---
 changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
index f8ed5d1753615..2877d8c2fb97d 100644
--- a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
+++ b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
@@ -2,6 +2,6 @@ The `gcp_pubsub` source no longer floods the logs with misleading errors on idle
 
 The source is also more resilient to connections that silently break and stop delivering messages until Vector is restarted:
 
-- HTTP/2 keepalive pings are sent on the gRPC connection so a dead-but-open connection is detected and torn down instead of hanging.
+- HTTP/2 keepalive pings are sent on the gRPC connection so a dead-but-open connection is detected and torn down instead of hanging. The ping interval and timeout are configurable via the new `keepalive` option.
 - The stream keepalive uses a non-blocking send, so a backed-up request channel can no longer deadlock the task.
 - A new `idle_timeout_secs` option (defaulting to `900`) restarts the stream if no activity is seen on an active connection within that window.

From 8307a0392e1be53e253502bac564f6665622f0a1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Tue, 21 Jul 2026 15:22:34 +0200
Subject: [PATCH 05/11] fix(gcp_pubsub source): address stalled-stream review
 feedback

- Defer idle-timeout restart while acknowledgements are pending so
  finalized-but-unsent ack IDs are not dropped and redelivered.
- Skip the keepalive when the request channel is full instead of
  reconnecting, preserving queued acks.
- Make idle_timeout_secs optional and derive a default larger than
  keepalive_secs so existing configs with a long keepalive keep starting.

Adds an Option<DurationSecondsWithFrac> Configurable impl in vector-config
to support the now-optional field.
---
 lib/vector-config/src/external/serde_with.rs  |  9 +++
 src/sources/gcp_pubsub.rs                     | 72 +++++++++++++------
 .../sources/generated/gcp_pubsub.cue          | 13 ++--
 3 files changed, 67 insertions(+), 27 deletions(-)

diff --git a/lib/vector-config/src/external/serde_with.rs b/lib/vector-config/src/external/serde_with.rs
index d671b8f48d530..da28a7c416f55 100644
--- a/lib/vector-config/src/external/serde_with.rs
+++ b/lib/vector-config/src/external/serde_with.rs
@@ -143,3 +143,12 @@ impl Configurable for Option<serde_with::DurationMilliSeconds<u64, serde_with::f
         generate_optional_schema(&u64::as_configurable_ref(), generator)
     }
 }
+
+impl Configurable for Option<serde_with::DurationSecondsWithFrac<f64, serde_with::formats::Strict>> {
+    fn generate_schema(generator: &RefCell<SchemaGenerator>) -> Result<SchemaObject, GenerateError>
+    where
+        Self: Sized,
+    {
+        generate_optional_schema(&f64::as_configurable_ref(), generator)
+    }
+}
diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 19cbb5d42a3f5..4e239daec9d4c 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -262,12 +262,13 @@ pub struct PubsubConfig {
     ///
     /// GCP Pub/Sub can leave a streaming pull in a state where the
     /// connection looks healthy but no further responses are delivered and
-    /// the stream never errors. This bounds that inactivity. It must be
-    /// larger than `keepalive_secs`.
-    #[serde(default = "default_idle_timeout")]
-    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
+    /// the stream never errors. This bounds that inactivity. When set, it
+    /// must be larger than `keepalive_secs`; when unset, it defaults to `900`
+    /// or just above `keepalive_secs`, whichever is larger.
+    #[serde_as(as = "Option<serde_with::DurationSecondsWithFrac<f64>>")]
+    #[derivative(Default(value = "Option::None"))]
     #[configurable(metadata(docs::human_name = "Idle Timeout"))]
-    pub idle_timeout_secs: Duration,
+    pub idle_timeout_secs: Option<Duration>,
 
     /// The number of consecutive stream failures, with no successful fetch in
     /// between, before a fetch error is reported as a component error.
@@ -331,6 +332,10 @@ const fn default_idle_timeout() -> Duration {
     Duration::from_secs(900)
 }
 
+/// Margin over `keepalive_secs` for the derived default idle timeout, keeping it
+/// strictly larger when `keepalive_secs` is at or above `default_idle_timeout`.
+const IDLE_TIMEOUT_KEEPALIVE_MARGIN: Duration = Duration::from_secs(60);
+
 const fn default_max_retry_errors() -> usize {
     5
 }
@@ -365,13 +370,20 @@ impl SourceConfig for PubsubConfig {
             return Err(PubsubError::InvalidAckDeadline.into());
         }
 
-        if self.idle_timeout_secs <= self.keepalive_secs {
-            return Err(PubsubError::InvalidIdleTimeout {
-                idle_timeout_secs: self.idle_timeout_secs.as_secs_f64(),
-                keepalive_secs: self.keepalive_secs.as_secs_f64(),
+        // Only an explicitly set idle timeout is validated; when unset, derive
+        // one that is always larger than the keepalive so upgrades with a long
+        // `keepalive_secs` keep starting.
+        let idle_timeout = match self.idle_timeout_secs {
+            Some(idle_timeout) if idle_timeout <= self.keepalive_secs => {
+                return Err(PubsubError::InvalidIdleTimeout {
+                    idle_timeout_secs: idle_timeout.as_secs_f64(),
+                    keepalive_secs: self.keepalive_secs.as_secs_f64(),
+                }
+                .into());
             }
-            .into());
-        }
+            Some(idle_timeout) => idle_timeout,
+            None => default_idle_timeout().max(self.keepalive_secs + IDLE_TIMEOUT_KEEPALIVE_MARGIN),
+        };
 
         let retry_delay_secs = match self.retry_delay_seconds {
             None => self.retry_delay_secs,
@@ -438,7 +450,7 @@ impl SourceConfig for PubsubConfig {
             ack_deadline_secs,
             retry_delay: retry_delay_secs,
             keepalive: self.keepalive_secs,
-            idle_timeout: self.idle_timeout_secs,
+            idle_timeout,
             max_retry_errors: self.max_retry_errors.max(1),
             consecutive_failures: 0,
             concurrency: Default::default(),
@@ -679,7 +691,11 @@ impl PubsubSource {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
-                _ = &mut idle_deadline => {
+                // Restart on idle only once acks drain: dropping the stream also
+                // drops the request stream, losing finalized-but-unsent ack IDs
+                // and risking redelivery. The elapsed timer refires once acks
+                // reach zero.
+                _ = &mut idle_deadline, if pending_acks == 0 => {
                     debug!(
                         message = "No activity on the stream within the idle timeout, restarting stream.",
                         timeout_secs = self.idle_timeout.as_secs_f64(),
@@ -706,14 +722,14 @@ impl PubsubSource {
                     // in a new request with empty fields, effectively
                     // a keepalive.
                     //
-                    // Send without blocking: a full channel means the
-                    // request stream stalled, so reconnect rather than
-                    // deadlock.
+                    // Send without blocking to avoid deadlocking the task. A
+                    // full channel already has acks queued, so the keepalive is
+                    // redundant and skipped; the idle timeout handles a genuine
+                    // stall.
                     match ack_ids_sender.try_send(Vec::new()) {
                         Ok(()) => {}
                         Err(mpsc::error::TrySendError::Full(_)) => {
-                            debug!("Keepalive request stream stalled, restarting stream.");
-                            break State::RetryDelay;
+                            debug!("Keepalive skipped: request stream has acks queued.");
                         }
                         Err(mpsc::error::TrySendError::Closed(_)) => {
                             unreachable!("request stream never closes")
@@ -918,17 +934,31 @@ mod tests {
 
     #[test]
     fn default_idle_timeout_is_larger_than_keepalive() {
-        // Defaults come from serde, not `Default::default()`.
+        assert!(
+            default_idle_timeout() > default_keepalive(),
+            "the default idle timeout must be larger than the default keepalive"
+        );
+    }
+
+    #[tokio::test]
+    async fn accepts_long_keepalive_without_explicit_idle_timeout() {
+        // A large `keepalive_secs` with no `idle_timeout_secs` must still build:
+        // the derived default stays larger than the keepalive.
         let config: PubsubConfig = serde_yaml::from_str(
             r#"
             project: my-project
             subscription: my-subscription
+            skip_authentication: true
+            keepalive_secs: 900
             "#,
         )
         .unwrap();
+
+        let (tx, _rx) = SourceSender::new_test();
+        let context = SourceContext::new_test(tx, None);
         assert!(
-            config.idle_timeout_secs > config.keepalive_secs,
-            "the default idle timeout must be larger than the default keepalive"
+            config.build(context).await.is_ok(),
+            "build should accept a long keepalive when idle_timeout_secs is unset"
         );
     }
 
diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
index e051febcad3d9..bd7f7b320bf21 100644
--- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue
+++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
@@ -605,14 +605,15 @@ generated: components: sources: gcp_pubsub: configuration: {
 
 			GCP Pub/Sub can leave a streaming pull in a state where the
 			connection looks healthy but no further responses are delivered and
-			the stream never errors. This bounds that inactivity. It must be
-			larger than `keepalive_secs`.
+			the stream never errors. This bounds that inactivity. When set, it
+			must be larger than `keepalive_secs`.
+
+			When unset, it defaults to `900`, or `keepalive_secs` plus a small
+			margin when `keepalive_secs` is configured at or above that value, so
+			that the idle timeout is always larger than the keepalive interval.
 			"""
 		required: false
-		type: float: {
-			default: 900.0
-			unit:    "seconds"
-		}
+		type: float: {}
 	}
 	keepalive: {
 		description: """

From 1f0b4d2881d34f8cc0aba28954bedc9b0ebddec6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Tue, 21 Jul 2026 16:44:16 +0200
Subject: [PATCH 06/11] fix(gcp_pubsub source): address second round of review
 feedback

- Default the HTTP/2 keepalive interval to 300s: Pub/Sub follows gRPC's
  default server enforcement and closes streams pinged more often than
  every 5 minutes with too_many_pings.
- Share the consecutive-failure counter across concurrent streams so a
  successful fetch on any stream resets the source-level count.
- Also gate the idle restart on the request channel being drained, not
  just pending_acks, to avoid dropping queued ack IDs.
- Add the required authors trailer to the changelog fragment.
---
 .../19418_gcp_pubsub_idle_reconnect.fix.md    |  2 +
 src/sources/gcp_pubsub.rs                     | 41 +++++++++++--------
 .../sources/generated/gcp_pubsub.cue          | 13 +++---
 3 files changed, 30 insertions(+), 26 deletions(-)

diff --git a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
index 2877d8c2fb97d..5b4fc0da9d2d9 100644
--- a/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
+++ b/changelog.d/19418_gcp_pubsub_idle_reconnect.fix.md
@@ -5,3 +5,5 @@ The source is also more resilient to connections that silently break and stop de
 - HTTP/2 keepalive pings are sent on the gRPC connection so a dead-but-open connection is detected and torn down instead of hanging. The ping interval and timeout are configurable via the new `keepalive` option.
 - The stream keepalive uses a non-blocking send, so a backed-up request channel can no longer deadlock the task.
 - A new `idle_timeout_secs` option (defaulting to `900`) restarts the stream if no activity is seen on an active connection within that window.
+
+authors: SamyDjemai
diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 4e239daec9d4c..a4e6d1068a791 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -138,8 +138,8 @@ pub struct PubsubKeepaliveConfig {
     /// How often, in seconds, to send a keepalive PING on the connection.
     ///
     /// Shorter intervals detect dead connections faster at the cost of additional traffic.
-    /// gRPC guidance recommends no less than 60 seconds to avoid tripping `too_many_pings`
-    /// policies on servers or proxies between the source and Pub/Sub.
+    /// Pub/Sub enforces gRPC's default minimum of one ping every 5 minutes, so lowering this
+    /// below `300` risks the server closing the stream with `too_many_pings`.
     #[serde(default = "default_keepalive_interval_secs")]
     #[configurable(metadata(docs::human_name = "Keepalive Interval"))]
     pub interval_secs: NonZeroU64,
@@ -152,9 +152,10 @@ pub struct PubsubKeepaliveConfig {
 }
 
 const fn default_keepalive_interval_secs() -> NonZeroU64 {
-    // Aligned with gRPC keepalive guidance, which recommends no less than one minute to avoid
-    // tripping `too_many_pings` policies on proxies between the source and Pub/Sub.
-    NonZeroU64::new(60).expect("keepalive interval default must be nonzero")
+    // Pub/Sub follows gRPC's default server enforcement, which closes streams pinged more often
+    // than once every 5 minutes with `too_many_pings`. Google's own Pub/Sub clients keepalive at
+    // this same interval.
+    NonZeroU64::new(300).expect("keepalive interval default must be nonzero")
 }
 
 const fn default_keepalive_timeout_secs() -> NonZeroU64 {
@@ -452,7 +453,7 @@ impl SourceConfig for PubsubConfig {
             keepalive: self.keepalive_secs,
             idle_timeout,
             max_retry_errors: self.max_retry_errors.max(1),
-            consecutive_failures: 0,
+            consecutive_failures: Default::default(),
             concurrency: Default::default(),
             full_response_size: self.full_response_size,
             log_namespace,
@@ -520,8 +521,9 @@ struct PubsubSource {
     keepalive: Duration,
     idle_timeout: Duration,
     max_retry_errors: usize,
-    // Consecutive stream failures since the last successful fetch.
-    consecutive_failures: usize,
+    // Consecutive stream failures since the last successful fetch, shared across
+    // all concurrent streams so a success on any stream resets the count.
+    consecutive_failures: Arc<AtomicUsize>,
     // The current concurrency is shared across all tasks. It is used
     // by the streams to avoid shutting down the last stream, which
     // would result in repeatedly re-opening the stream on idle.
@@ -671,7 +673,7 @@ impl PubsubSource {
                 },
                 response = stream.next() => match response {
                     Some(Ok(response)) => {
-                        self.consecutive_failures = 0;
+                        self.consecutive_failures.store(0, Ordering::Relaxed);
                         idle_deadline
                             .as_mut()
                             .reset(tokio::time::Instant::now() + self.idle_timeout);
@@ -691,11 +693,14 @@ impl PubsubSource {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
-                // Restart on idle only once acks drain: dropping the stream also
-                // drops the request stream, losing finalized-but-unsent ack IDs
-                // and risking redelivery. The elapsed timer refires once acks
-                // reach zero.
-                _ = &mut idle_deadline, if pending_acks == 0 => {
+                // Restart on idle only once acks are finalized and flushed from
+                // the request channel: dropping the stream drops that channel,
+                // and any queued ack IDs would be lost, risking redelivery.
+                // Delivery is at-least-once, so a small residual window remains
+                // (IDs buffered inside the request stream), but this avoids the
+                // common case. The elapsed timer refires as the conditions clear.
+                _ = &mut idle_deadline, if pending_acks == 0
+                    && ack_ids_sender.capacity() == ack_ids_sender.max_capacity() => {
                     debug!(
                         message = "No activity on the stream within the idle timeout, restarting stream.",
                         timeout_secs = self.idle_timeout.as_secs_f64(),
@@ -811,7 +816,7 @@ impl PubsubSource {
     // error (a reset, or `Unavailable` when an idle subscription has no
     // messages). Only escalate to a component error once failures persist,
     // to avoid flooding the logs on quiet subscriptions.
-    fn handle_stream_error(&mut self, error: tonic::Status) -> State {
+    fn handle_stream_error(&self, error: tonic::Status) -> State {
         // A server-side reset is not really an error, so retry immediately
         // without counting it.
         if is_reset(&error) {
@@ -819,14 +824,14 @@ impl PubsubSource {
             return State::RetryNow;
         }
 
-        self.consecutive_failures += 1;
-        if self.consecutive_failures >= self.max_retry_errors {
+        let consecutive_failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
+        if consecutive_failures >= self.max_retry_errors {
             emit!(GcpPubsubReceiveError { error });
         } else {
             debug!(
                 message = "Transient error fetching events, retrying.",
                 %error,
-                consecutive_failures = self.consecutive_failures,
+                consecutive_failures,
                 max_retry_errors = self.max_retry_errors,
             );
         }
diff --git a/website/cue/reference/components/sources/generated/gcp_pubsub.cue b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
index bd7f7b320bf21..f666ac907258e 100644
--- a/website/cue/reference/components/sources/generated/gcp_pubsub.cue
+++ b/website/cue/reference/components/sources/generated/gcp_pubsub.cue
@@ -606,11 +606,8 @@ generated: components: sources: gcp_pubsub: configuration: {
 			GCP Pub/Sub can leave a streaming pull in a state where the
 			connection looks healthy but no further responses are delivered and
 			the stream never errors. This bounds that inactivity. When set, it
-			must be larger than `keepalive_secs`.
-
-			When unset, it defaults to `900`, or `keepalive_secs` plus a small
-			margin when `keepalive_secs` is configured at or above that value, so
-			that the idle timeout is always larger than the keepalive interval.
+			must be larger than `keepalive_secs`; when unset, it defaults to `900`
+			or just above `keepalive_secs`, whichever is larger.
 			"""
 		required: false
 		type: float: {}
@@ -630,11 +627,11 @@ generated: components: sources: gcp_pubsub: configuration: {
 					How often, in seconds, to send a keepalive PING on the connection.
 
 					Shorter intervals detect dead connections faster at the cost of additional traffic.
-					gRPC guidance recommends no less than 60 seconds to avoid tripping `too_many_pings`
-					policies on servers or proxies between the source and Pub/Sub.
+					Pub/Sub enforces gRPC's default minimum of one ping every 5 minutes, so lowering this
+					below `300` risks the server closing the stream with `too_many_pings`.
 					"""
 				required: false
-				type: uint: default: 60
+				type: uint: default: 300
 			}
 			timeout_secs: {
 				description: """

From d45eddce26bf8a73db02df7654d048f84834f386 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Tue, 21 Jul 2026 17:10:37 +0200
Subject: [PATCH 07/11] fix(gcp_pubsub source): address third round of review
 feedback

- Always restart on idle timeout: the previous ack-drain guard disabled
  the recovery path exactly when the request stream stalled, hanging the
  source. Unsent ack IDs are dropped and redelivered, acceptable under
  at-least-once delivery.
- Emit the fetch component error once per failure streak (== threshold)
  instead of on every retry past it, so an idle subscription no longer
  resumes the error flood.
- Give max_retry_errors a derivative default so programmatic/generated
  configs keep the threshold instead of collapsing to 0.
---
 src/sources/gcp_pubsub.rs | 31 ++++++++++++++++++++++---------
 1 file changed, 22 insertions(+), 9 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index a4e6d1068a791..e15977ba13483 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -280,6 +280,7 @@ pub struct PubsubConfig {
     /// until this threshold is reached. A successful fetch resets the count.
     /// Set to `1` to report every failure as an error.
     #[serde(default = "default_max_retry_errors")]
+    #[derivative(Default(value = "default_max_retry_errors()"))]
     #[configurable(metadata(docs::human_name = "Max Retry Errors Before Alerting"))]
     pub max_retry_errors: usize,
 
@@ -693,14 +694,12 @@ impl PubsubSource {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
-                // Restart on idle only once acks are finalized and flushed from
-                // the request channel: dropping the stream drops that channel,
-                // and any queued ack IDs would be lost, risking redelivery.
-                // Delivery is at-least-once, so a small residual window remains
-                // (IDs buffered inside the request stream), but this avoids the
-                // common case. The elapsed timer refires as the conditions clear.
-                _ = &mut idle_deadline, if pending_acks == 0
-                    && ack_ids_sender.capacity() == ack_ids_sender.max_capacity() => {
+                // Always restart on idle, even with acks pending or queued: this
+                // is the recovery path for a stalled request stream, so it must
+                // not be gated on the very channel that has stopped draining.
+                // Any unsent ack IDs are dropped and their messages redelivered,
+                // which is acceptable under Pub/Sub's at-least-once semantics.
+                _ = &mut idle_deadline => {
                     debug!(
                         message = "No activity on the stream within the idle timeout, restarting stream.",
                         timeout_secs = self.idle_timeout.as_secs_f64(),
@@ -824,8 +823,12 @@ impl PubsubSource {
             return State::RetryNow;
         }
 
+        // Escalate to a component error only when the streak first reaches the
+        // threshold, not on every subsequent failure, so a persistently idle
+        // subscription does not resume the error flood. A successful fetch resets
+        // the counter and re-arms escalation for the next streak.
         let consecutive_failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
-        if consecutive_failures >= self.max_retry_errors {
+        if consecutive_failures == self.max_retry_errors {
             emit!(GcpPubsubReceiveError { error });
         } else {
             debug!(
@@ -945,6 +948,16 @@ mod tests {
         );
     }
 
+    #[test]
+    fn programmatic_default_preserves_max_retry_errors() {
+        // `Default` must match the serde default so generated/programmatic
+        // configs keep the retry threshold instead of collapsing to 0.
+        assert_eq!(
+            PubsubConfig::default().max_retry_errors,
+            default_max_retry_errors()
+        );
+    }
+
     #[tokio::test]
     async fn accepts_long_keepalive_without_explicit_idle_timeout() {
         // A large `keepalive_secs` with no `idle_timeout_secs` must still build:

From 552679355968be8db03afbca4b13ced1992f2899 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Thu, 23 Jul 2026 13:34:19 +0200
Subject: [PATCH 08/11] fix(gcp_pubsub source): don't block the select loop on
 ack sends

Awaiting a full ack channel inside the select arm stalled the whole loop,
so a backed-up request stream could never be recovered by the idle
timeout. Use a non-blocking send and reconnect when the channel is full;
dropped ack IDs are redelivered under at-least-once delivery.
---
 src/sources/gcp_pubsub.rs | 48 +++++++++++++++++++--------------------
 1 file changed, 23 insertions(+), 25 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index e15977ba13483..4ab0b9a611bec 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -666,10 +666,20 @@ impl PubsubSource {
                 receipts = ack_stream.next() => if let Some((status, receipts)) = receipts {
                     pending_acks -= 1;
                     if status == BatchStatus::Delivered {
-                        ack_ids_sender
-                            .send(receipts)
-                            .await
-                            .unwrap_or_else(|_| unreachable!("request stream never closes"));
+                        // Non-blocking: awaiting a full channel would stall the
+                        // select loop, preventing the idle timeout below from
+                        // recovering a stalled request stream. Reconnect instead;
+                        // dropped ack IDs are redelivered (at-least-once).
+                        match ack_ids_sender.try_send(receipts) {
+                            Ok(()) => {}
+                            Err(mpsc::error::TrySendError::Full(_)) => {
+                                debug!("Request stream stalled sending acks, restarting stream.");
+                                break State::RetryDelay;
+                            }
+                            Err(mpsc::error::TrySendError::Closed(_)) => {
+                                unreachable!("request stream never closes")
+                            }
+                        }
                     }
                 },
                 response = stream.next() => match response {
@@ -694,11 +704,9 @@ impl PubsubSource {
                     debug!("New authentication token generated, restarting stream.");
                     break State::RetryNow;
                 },
-                // Always restart on idle, even with acks pending or queued: this
-                // is the recovery path for a stalled request stream, so it must
-                // not be gated on the very channel that has stopped draining.
-                // Any unsent ack IDs are dropped and their messages redelivered,
-                // which is acceptable under Pub/Sub's at-least-once semantics.
+                // The recovery path for a stalled stream: always fires, never
+                // gated on the request channel that may have stopped draining.
+                // Unsent ack IDs are dropped and redelivered (at-least-once).
                 _ = &mut idle_deadline => {
                     debug!(
                         message = "No activity on the stream within the idle timeout, restarting stream.",
@@ -718,18 +726,10 @@ impl PubsubSource {
                         // Otherwise, mark this stream as idle.
                         busy_flag.store(false, Ordering::Relaxed);
                     }
-                    // GCP Pub/Sub likes to time out connections after
-                    // about 75 seconds of inactivity. To forestall
-                    // the resulting error, send an empty array of
-                    // acknowledgement IDs to the request stream if no
-                    // other activity has happened. This will result
-                    // in a new request with empty fields, effectively
-                    // a keepalive.
-                    //
-                    // Send without blocking to avoid deadlocking the task. A
-                    // full channel already has acks queued, so the keepalive is
-                    // redundant and skipped; the idle timeout handles a genuine
-                    // stall.
+                    // GCP Pub/Sub times out connections after ~75s of
+                    // inactivity. Send an empty ack ID array as a keepalive to
+                    // forestall that. Non-blocking: a full channel already has
+                    // acks queued, so the keepalive is redundant and skipped.
                     match ack_ids_sender.try_send(Vec::new()) {
                         Ok(()) => {}
                         Err(mpsc::error::TrySendError::Full(_)) => {
@@ -823,10 +823,8 @@ impl PubsubSource {
             return State::RetryNow;
         }
 
-        // Escalate to a component error only when the streak first reaches the
-        // threshold, not on every subsequent failure, so a persistently idle
-        // subscription does not resume the error flood. A successful fetch resets
-        // the counter and re-arms escalation for the next streak.
+        // Escalate once, when the streak first hits the threshold, so a
+        // persistently idle subscription doesn't resume the error flood.
         let consecutive_failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
         if consecutive_failures == self.max_retry_errors {
             emit!(GcpPubsubReceiveError { error });

From bc31918cc9b0ea6b5a2851e347ffc07e332d7971 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Thu, 23 Jul 2026 13:52:48 +0200
Subject: [PATCH 09/11] fix(gcp_pubsub source): keep reporting fetch errors
 past the threshold

Emitting only at the exact threshold silenced a persistently broken
stream after one report and made max_retry_errors: 1 report just the
first failure. Use >= so failures keep surfacing until a success resets
the counter, matching the documented behavior.
---
 src/sources/gcp_pubsub.rs | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 4ab0b9a611bec..79969cf6580cd 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -823,10 +823,12 @@ impl PubsubSource {
             return State::RetryNow;
         }
 
-        // Escalate once, when the streak first hits the threshold, so a
-        // persistently idle subscription doesn't resume the error flood.
+        // Escalate once the streak reaches the threshold, and keep reporting
+        // while it persists so a genuinely broken stream stays visible (and
+        // `max_retry_errors: 1` reports every failure). A successful fetch
+        // resets the counter.
         let consecutive_failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
-        if consecutive_failures == self.max_retry_errors {
+        if consecutive_failures >= self.max_retry_errors {
             emit!(GcpPubsubReceiveError { error });
         } else {
             debug!(

From 3f992d59c8734ae19f0da5d40e99b113ca1a3e12 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Thu, 23 Jul 2026 14:27:07 +0200
Subject: [PATCH 10/11] fix(gcp_pubsub source): make the auto-ack send
 non-blocking too

The default (acknowledgements disabled) path in handle_response still
awaited the bounded ack channel, which parks the task inside the
stream.next() arm and starves the idle-timeout recovery when the request
stream stalls. Send non-blocking and reconnect on a full channel, like
the finalizer and keepalive paths; dropped ack IDs are redelivered.
---
 src/sources/gcp_pubsub.rs | 27 ++++++++++++++++++++-------
 1 file changed, 20 insertions(+), 7 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 79969cf6580cd..1edda539b92e0 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -688,13 +688,15 @@ impl PubsubSource {
                         idle_deadline
                             .as_mut()
                             .reset(tokio::time::Instant::now() + self.idle_timeout);
-                        self.handle_response(
+                        if let Some(state) = self.handle_response(
                             response,
                             &finalizer,
                             &ack_ids_sender,
                             &mut pending_acks,
                             busy_flag,
-                        ).await;
+                        ).await {
+                            break state;
+                        }
                     }
                     Some(Err(error)) => break self.handle_stream_error(error),
                     None => break State::RetryNow,
@@ -776,6 +778,7 @@ impl PubsubSource {
         }))
     }
 
+    // Returns `Some(state)` when the stream should be torn down and restarted.
     async fn handle_response(
         &mut self,
         response: proto::StreamingPullResponse,
@@ -783,7 +786,7 @@ impl PubsubSource {
         ack_ids: &mpsc::Sender<Vec<String>>,
         pending_acks: &mut usize,
         busy_flag: &Arc<AtomicBool>,
-    ) {
+    ) -> Option<State> {
         if response.received_messages.len() >= self.full_response_size {
             busy_flag.store(true, Ordering::Relaxed);
         }
@@ -796,10 +799,19 @@ impl PubsubSource {
         match self.out.send_batch(events).await {
             Err(_) => emit!(StreamClosedError { count }),
             Ok(()) => match notifier {
-                None => ack_ids
-                    .send(ids)
-                    .await
-                    .unwrap_or_else(|_| unreachable!("request stream never closes")),
+                // Auto-ack: send without blocking so a stalled request stream
+                // can't park the task here and starve the idle timeout. On a
+                // full channel, reconnect; the unsent IDs are redelivered.
+                None => match ack_ids.try_send(ids) {
+                    Ok(()) => {}
+                    Err(mpsc::error::TrySendError::Full(_)) => {
+                        debug!("Request stream stalled sending acks, restarting stream.");
+                        return Some(State::RetryDelay);
+                    }
+                    Err(mpsc::error::TrySendError::Closed(_)) => {
+                        unreachable!("request stream never closes")
+                    }
+                },
                 Some(notifier) => {
                     finalizer
                         .as_ref()
@@ -809,6 +821,7 @@ impl PubsubSource {
                 }
             },
         }
+        None
     }
 
     // GCP frequently ends a streaming pull with a transient, auto-retried

From a4e78d76fefff5b376837af604273f358531782d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Samy=20Djema=C3=AF?=
 <53857555+SamyDjemai@users.noreply.github.com>
Date: Thu, 23 Jul 2026 15:18:46 +0200
Subject: [PATCH 11/11] fix(gcp_pubsub source): bound streaming-pull startup
 with the idle timeout

If Pub/Sub accepts the connection but streaming_pull never resolves (the
no-data startup case), the source could hang with no recovery. Arm the
idle deadline before the startup call so a stuck startup also reconnects.
---
 src/sources/gcp_pubsub.rs | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs
index 1edda539b92e0..8677b1c79a836 100644
--- a/src/sources/gcp_pubsub.rs
+++ b/src/sources/gcp_pubsub.rs
@@ -634,12 +634,23 @@ impl PubsubSource {
 
         let (ack_ids_sender, ack_ids_receiver) = mpsc::channel(ACK_QUEUE_SIZE);
 
+        // Reconnect if no response arrives within the idle timeout, to
+        // recover from a stream that stalls without ever erroring. Reset
+        // whenever a response arrives. Armed before the streaming-pull call so
+        // it also bounds a startup that Pub/Sub accepts but never resolves.
+        let idle_deadline = tokio::time::sleep(self.idle_timeout);
+        tokio::pin!(idle_deadline);
+
         // Handle shutdown during startup, the streaming pull doesn't
         // start if there is no data in the subscription.
         let request_stream = self.request_stream(ack_ids_receiver);
         debug!("Starting streaming pull.");
         let stream = tokio::select! {
             _ = &mut self.shutdown => return State::Shutdown,
+            _ = &mut idle_deadline => {
+                debug!("Streaming pull did not start within the idle timeout, restarting stream.");
+                return State::RetryDelay;
+            }
             result = client.streaming_pull(request_stream) => match result {
                 Ok(stream) => stream,
                 Err(error) => {
@@ -654,12 +665,6 @@ impl PubsubSource {
             Finalizer::maybe_new(self.acknowledgements, Some(self.shutdown.clone()));
         let mut pending_acks = 0;
 
-        // Reconnect if no response arrives within the idle timeout, to
-        // recover from a stream that stalls without ever erroring. Reset
-        // whenever a response arrives.
-        let idle_deadline = tokio::time::sleep(self.idle_timeout);
-        tokio::pin!(idle_deadline);
-
         loop {
             tokio::select! {
                 biased;