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 13 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://cdn.github.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 👍 / 👎.
1 participant
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.