fix(gcp_pubsub source): recover from stalled streams and reduce error noise - #25901
fix(gcp_pubsub source): recover from stalled streams and reduce error noise#25901SamyDjemai wants to merge 20 commits into
Conversation
github-actions
Bot
added
docs review on hold
|
All contributors have signed the CLA ✍️ ✅ |
This comment has been minimized.
This comment has been minimized.
|
I have read the CLA Document and I hereby sign the CLA |
github-actions
Bot
added
the
docs review on hold
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7e6e1a0236
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that 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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 47479d311e
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that 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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1f0b4d2881
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that 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.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d45eddce26
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5526793559
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bc31918cc9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // 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 => { |
There was a problem hiding this comment.
When end-to-end source acknowledgements are disabled, which is the default, handle_response still auto-acks by awaiting the bounded ack_ids.send(ids) path. If the request stream is the stalled side and that 8-slot channel fills, the task is parked inside handle_response, so this newly added idle-timeout branch is never polled and the source can still hang until restart. Please make the auto-ack send non-blocking/reconnect as well, or otherwise keep the idle timer active while waiting on that channel.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Good catch — this was the one remaining ack path that still blocked. Fixed in 3f992d5: the auto-ack branch in handle_response now uses a non-blocking try_send and reconnects on a full channel, matching the finalizer and keepalive paths, so the idle timeout can always recover a stalled request stream. Dropped ack IDs are redelivered under at-least-once.
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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3f992d59c8
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
…meout 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.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
vector/src/sources/gcp_pubsub.rs
Line 657 in a4e78d7
Route startup stream failures through retry threshold
When Pub/Sub returns a transient Unavailable before response headers are received, this startup branch still emits a component error on every retry and never updates consecutive_failures; the new max_retry_errors threshold only applies later in stream.next(). This matters for idle or low-volume subscriptions because this code already notes that streaming_pull may sit in startup until data is available, so the same “no messages” failures can continue to flood failed_streaming_pull logs/metrics instead of being suppressed until the configured threshold.
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| match ack_ids_sender.try_send(Vec::new()) { | ||
| Ok(()) => {} | ||
| Err(mpsc::error::TrySendError::Full(_)) => { | ||
| debug!("Keepalive skipped: request stream has acks queued."); | ||
| } |
There was a problem hiding this comment.
If a user configures keepalive_secs: 0 and the request stream stops draining, this new non-blocking full-queue path returns immediately; the surrounding tokio::time::sleep(self.keepalive) is then ready again on the next loop, so the task can hot-loop and repeatedly log this debug message instead of yielding until space is available. The previous awaited send backpressured on a full channel, so this regression only shows up with zero keepalive and a stalled/full ack queue.
Useful? React with 👍 / 👎.
| // 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; |
There was a problem hiding this comment.
I think the backpressure/blocking is desirable here -- perhaps the ack sender should be a different loop/worker, so the channel provides a bounded buffer, but the actual ack work would not block the main stream loop.
There was a problem hiding this comment.
Ah okay, was unsure since Codex bounced back and forth between blocking and non-blocking. Implemented in c2e578a
Sending acks directly from the select loop meant a stalled request stream either blocked the loop (starving the idle timeout) or forced dropping acks. Forward them through a worker task instead: the loop enqueues without blocking and the worker absorbs backpressure with a bounded send, so a backed-up request stream applies flow control instead of stalling or dropping. The worker exits when the stream is torn down.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c2e578a6c1
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 66e33f972a
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
…ssure Replace the ad-hoc try_send/reconnect-on-full handoff with the aws_sqs worker pattern: a dedicated task owns the finalizer ack stream and forwards receipts to the bounded request channel with a blocking send. A slow request stream backpressures the finalizer instead of blocking the main loop or growing an unbounded queue. pending_acks is shared so the loop still gates shutdown and keepalive on in-flight acks.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c6cc12f9f1
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // Auto-ack: a send error means the request stream is gone; the | ||
| // caller reconnects on the next `stream.next()`. | ||
| None => { | ||
| ack_ids.send(ids).await.ok(); |
There was a problem hiding this comment.
With the default acknowledgements-disabled configuration, if the outbound request stream stalls while responses continue arriving, this bounded 8-slot channel eventually fills and the awaited send parks the main tokio::select! loop, so the new idle timeout is never polled and the source remains hung until restart. Fresh evidence after the earlier 3f992d5 fix is that the final bounded-worker rewrite restored ack_ids.send(ids).await here while routing only finalized acknowledgements through the worker.
Useful? React with 👍 / 👎.
| pending_acks.fetch_sub(1, Ordering::Relaxed); | ||
| if status == BatchStatus::Delivered | ||
| && ack_ids_sender.send(receipts).await.is_err() |
There was a problem hiding this comment.
With acknowledgements enabled, if the request channel is full when the worker receives the last finalization, this decrements the count to zero before the acknowledgement has been enqueued and while the following send is blocked. The main task can then take the guarded shutdown branch—or classify this as an inactive stream at the keepalive branch—and drop the request stream, causing that send to fail and the already-delivered message to be redelivered instead of flushing remaining acknowledgements as intended; keep the acknowledgement pending until its send completes or fails.
Useful? React with 👍 / 👎.
| let (finalizer, mut ack_stream) = | ||
| Finalizer::maybe_new(self.acknowledgements, Some(self.shutdown.clone())); | ||
| let mut pending_acks = 0; | ||
| let pending_acks = Arc::new(AtomicUsize::new(0)); | ||
| { | ||
| let ack_ids_sender = ack_ids_sender.clone(); | ||
| let pending_acks = Arc::clone(&pending_acks); | ||
| crate::spawn_in_current_span(async move { |
There was a problem hiding this comment.
With the default acknowledgements-disabled configuration, Finalizer::maybe_new(false, ...) returns an EmptyStream whose poll_next is permanently pending, but this detached worker is still spawned unconditionally. It therefore never observes shutdown or channel closure and never exits; every successful stream setup followed by an idle reconnect, transient error, or configuration reload leaves another Tokio task and channel allocation behind, producing unbounded task growth during long-running error/reconnect cycles.
Useful? React with 👍 / 👎.
- Only spawn the ack worker when acknowledgements are enabled; otherwise the finalizer's EmptyStream never resolves and each reconnect leaks a task. - Decrement pending_acks after the worker's forward send resolves so the shutdown/keepalive guards don't drop the stream mid-flush. - Make the main-loop auto-ack and keepalive sends non-blocking; a full channel means the request stream stalled, so reconnect instead of parking the select loop and starving the idle timeout.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3196b32193
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| Finalizer::maybe_new(true, Some(self.shutdown.clone())); | ||
| let ack_ids_sender = ack_ids_sender.clone(); | ||
| let pending_acks = Arc::clone(&pending_acks); | ||
| crate::spawn_in_current_span(async move { |
There was a problem hiding this comment.
When acknowledgements are enabled and a stream exits while a downstream batch is still pending—for example, through the new idle-timeout branch—this detached task outlives run_once: dropping the finalizer closes its producer, but UnorderedFinalizer waits for outstanding BatchStatusReceivers before ack_stream ends. Because the task handle is discarded, reconnects can strand additional workers holding receipts and a cloned ShutdownSignal; if a batch never finalizes, even one such worker can prevent source shutdown completion indefinitely. Fresh evidence beyond the fixed acknowledgements-disabled case is that the enabled path still spawns an unmanaged worker that drains pending finalizers after its gRPC stream has already been torn down.
Useful? React with 👍 / 👎.
| // 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(&self, error: tonic::Status) -> State { |
There was a problem hiding this comment.
When Pub/Sub returns a transient status before the response stream is established—for example, an Unavailable status in the initial StreamingPull response—Tonic returns it from client.streaming_pull(...), so this new handler is never called and the setup branch emits GcpPubsubStreamingPullError on every retry. This bypasses max_retry_errors and the shared success reset, meaning that with concurrent streams a healthy stream can keep receiving events while a sibling floods component errors at retry_delay_secs cadence; route setup-time statuses through the same retry classification.
Useful? React with 👍 / 👎.
…eshold - Abort the ack worker when run_once returns (via a local AbortOnDrop guard), so a reconnect can't strand workers that keep draining a finalizer and holding a ShutdownSignal after their request stream is gone (which could otherwise block source shutdown indefinitely). - Route connect and streaming-pull setup errors through the same consecutive-failure threshold as stream errors, so a transient setup status on an idle subscription no longer floods failed_streaming_pull / failed_connecting on every retry.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7b4d1826cc
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| finalizer | ||
| .as_ref() | ||
| .expect("Finalizer must have been set up for acknowledgements") | ||
| .add(ids, notifier); |
There was a problem hiding this comment.
With source acknowledgements enabled, if the outbound request stream stops draining while Pub/Sub continues delivering responses, the ack worker blocks on its bounded send but this call never backpressures the producer: UnorderedFinalizer::add writes each (notifier, ids) pair to the unbounded channel in lib/vector-common/src/finalizer.rs. Each incoming response also resets idle_deadline, so a busy subscription can grow that queue indefinitely and eventually exhaust memory. Fresh evidence beyond the earlier handoff finding is that the final bounded-worker rewrite moved the unbounded ingress into FinalizerSet rather than removing it; bound this handoff or reconnect when it cannot accept more entries.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
The unbounded channel is in lib/vector-common, meaning it's used by all ack-enabled sources. This seems out of scope
| debug!( | ||
| message = "Transient error, retrying.", | ||
| consecutive_failures, | ||
| max_retry_errors = self.max_retry_errors, | ||
| ); |
There was a problem hiding this comment.
For the first max_retry_errors - 1 connect, setup, or receive failures, the caller does not emit its error and this debug record includes only counters, so the actual transport error or gRPC status is discarded. With concurrent streams, successful responses can repeatedly reset the shared counter and keep a persistently failing sibling below the threshold indefinitely, leaving no actionable record of the failure even at debug level; pass the error and operation context into this path.
Useful? React with 👍 / 👎.
| let sent = ack_ids_sender.send(receipts).await.is_ok(); | ||
| pending_acks.fetch_sub(1, Ordering::Relaxed); |
There was a problem hiding this comment.
When shutdown arrives while pending_acks is nonzero, the shutdown branch's guard prevents its future from being polled; if the worker then forwards the last acknowledgement and decrements the atomic to zero, that atomic update does not wake the tokio::select! task to reevaluate the guard. On an otherwise idle subscription, shutdown is therefore delayed until the keepalive or idle timer fires—60 seconds by default and potentially much longer when configured—which can exceed an orchestrator's grace period and force-kill Vector. Notify the main loop when the count reaches zero or receive worker completions through a polled channel.
Useful? React with 👍 / 👎.
| None => match ack_ids.try_send(ids) { | ||
| Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {} | ||
| Err(mpsc::error::TrySendError::Full(_)) => return Some(State::RetryDelay), |
There was a problem hiding this comment.
With acknowledgements disabled, TrySendError::Closed means the outbound request stream has already dropped its receiver, so these ack IDs cannot be sent. Treating that result as success lets the source continue consuming any buffered or still-arriving responses, silently dropping every corresponding acknowledgement; those responses also keep resetting the idle deadline, so reconnect need not happen promptly. Return RetryNow for the closed case as the keepalive path already does.
Useful? React with 👍 / 👎.
- Reconnect (RetryNow) when the auto-ack channel is closed instead of silently dropping acknowledgements. - Log the underlying error and operation for suppressed transient failures so there is an actionable debug record below the threshold. - Notify the main loop when pending acks drain to zero so shutdown is not delayed until the next keepalive or idle timer fires.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4ce2770e30
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| .add(ids, notifier); | ||
| *pending_acks += 1; | ||
| pending_acks.fetch_add(1, Ordering::Relaxed); |
There was a problem hiding this comment.
With acknowledgements enabled, a fast downstream—or a response that decodes to no events—can make the notifier ready before this call. Because FinalizerSet::add exposes it to the concurrently running ack worker before pending_acks is incremented, the worker can execute fetch_sub while the counter is zero, wrapping it to usize::MAX; the stream then appears to have acknowledgements pending indefinitely, preventing prompt shutdown and inactive-stream cleanup. Increment the counter before making the finalizer visible to the worker.
Useful? React with 👍 / 👎.
The worker can finalize and decrement pending_acks the instant a batch is added (fast downstream or a zero-event response), so incrementing after add() races and can underflow the counter to usize::MAX, leaving the stream permanently marked as having acks pending. Increment first.
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 119f7d96d9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| let sent = status != BatchStatus::Delivered | ||
| // Keep the ack pending until its send resolves, so the | ||
| // main task's guards don't drop the stream mid-flush. | ||
| || ack_ids_sender.send(receipts).await.is_ok(); |
There was a problem hiding this comment.
When acknowledgements are enabled and the outbound gRPC request stream stops draining, this worker blocks on the bounded channel, but the main response loop continues adding entries to FinalizerSet's unbounded input and incrementing pending_acks. With sustained incoming Pub/Sub traffic, memory can therefore grow without backpressure until the next keepalive or idle timeout—an arbitrarily long period for configurations with a large keepalive_secs. Propagate the full-channel condition to the main loop or otherwise bound the finalizer queue rather than isolating the blocking send in this worker.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Same as the other thread, the unbounded channel is in lib/vector-common, shared by all ack-enabled sources.
| None => match ack_ids.try_send(ids) { | ||
| Ok(()) => {} | ||
| Err(mpsc::error::TrySendError::Full(_)) => return Some(State::RetryDelay), | ||
| Err(mpsc::error::TrySendError::Closed(_)) => return Some(State::RetryNow), |
There was a problem hiding this comment.
Under a burst of responses or temporary request-side backpressure, a full eight-entry ack channel does not prove that the stream is stalled; it can simply mean the consumer has not polled the request body yet. Returning here tears down the stream after the events were already delivered downstream but before these IDs were acknowledged, causing redelivery and potentially repeated reconnect/duplicate loops under sustained load. The previous awaited send safely applied backpressure, so automatic acknowledgements should be queued without treating ordinary channel saturation as a connection failure.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
This reverses the earlier P1 in this review that made me switch to non-blocking. Going back to a blocking send reintroduces that hang: it parks the select loop so the idle timeout can't recover a stalled stream. Maybe a human maintainer can help here? @milas?
2 participants
Summary
The
gcp_pubsubsource could stop processing messages until Vector was restarted, and floods the logs with misleadingfailed_fetching_eventserrors on idle/low-volume/bursty subscriptions. This addresses both:keepaliveoption, mirroring thevectorsink).idle_timeout_secs(default900) to restart a stream that stops delivering responses without erroring.max_retry_errors(default5) consecutive failures with no successful fetch in between; transient failures are logged atdebug. A successful fetch resets the counter.How did you test this PR?
cargo fmt --check,cargo clippy -- -D warnings, unit tests, andvdev check eventsall pass (run in a Rust 1.95 container).Testing caveat, in the interest of transparency: the runtime behavior here is hard to cover with automated tests. The
code=8a75"no messages" response is a real-GCP behavior that the Pub/Sub emulator does not reproduce, and exercising the stalled-connection recovery (HTTP/2 keepalive + idle timeout) would require fault-injection infrastructure that doesn't currently exist in the repo. The existing gcp_pubsub emulator tests are also disabled (#24133). Unit tests here therefore only cover the config validation; the retry-escalation and connection-resilience changes are validated by reasoning and would benefit from manual verification against a real GCP subscription.Change Type
Is this a breaking change?
Does this PR include user facing changes?
no-changeloglabel to this PR.References
Per Vector's AI policy: this PR was substantially generated with an AI agent (Claude Opus 4.8). I have reviewed, understood, and tested the changes, and take ownership of them.