feat(redis source): add reconnecting pubsub session for channel data_type#25892
feat(redis source): add reconnecting pubsub session for channel data_type#25892artusmode wants to merge 7 commits into
data_type#25892Conversation
…_type`
- Rework channel source to maintain a pubsub session that auto-reconnects
and re-subscribes after Redis disconnects.
- Add shutdown-aware backoff and best-effort unsubscribe on exit.
- Log when the connection is re-established ("re-established and resubscribed").
Resolves vectordotdev#22615
Follow-up to the reconnecting pub/sub work: - Replace the ad-hoc `backoff_exponential` helper with the shared `common::backoff::ExponentialBackoff` (from_millis(2).factor(250), capped at 30s), matching the strategy used by other reconnecting sources such as `aws_s3`/`sqs`. Backoff resets on a successful connect. - Emit internal events for reconnection so operators can observe/alert: `RedisConnectionError` (increments `component_errors_total` with error_type=connection_failed) on connect/subscribe failure, and `RedisConnectionEstablished` (increments `connection_established_total`) on initial connect and recovery. - Unify the `list` data_type source (`list.rs`) onto the same shared `ExponentialBackoff`, removing its duplicate `backoff_exponential` helper and making its retry sleep shutdown-aware. - Add a changelog fragment. Co-authored-by: Gibran <gibran.zaman@finaccel.co>
github-actions
Bot
added
the
domain: sources
|
All contributors have signed the CLA ✍️ ✅ |
This comment has been minimized.
This comment has been minimized.
![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: aa85aeb2f1
ℹ️ 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".
| loop { | ||
| // connect + SUBSCRIBE | ||
| let mut pubsub_conn = | ||
| match connect_and_subscribe(&client, &endpoint, &channel).await { |
There was a problem hiding this comment.
When the channel source is in the reconnect loop and a Redis connect/subscribe attempt stalls, for example against a black-holed TCP/TLS endpoint, this await is not raced with shutdown. A shutdown signal that arrives after the backoff completes or during the connect will not be observed until the Redis client/OS connect attempt returns, so the source misses graceful shutdown and Vector waits for the force-shutdown deadline. Please wrap connect_and_subscribe in a tokio::select! with the shutdown signal and/or apply a per-attempt timeout.
Useful? React with 👍 / 👎.
| // `aws_s3`/`sqs`. Reset once a value is successfully received. | ||
| let mut backoff = ExponentialBackoff::from_millis(2) | ||
| .factor(250) | ||
| .max_delay(Duration::from_secs(30)); |
There was a problem hiding this comment.
For existing data_type = "list" configurations, this reused backoff now grows to 30 seconds after repeated Redis I/O errors. During a Redis outage the list source can reach that cap, and once Redis is back it may still wait up to 30 seconds before the next BLPOP/BRPOP; the removed helper capped retries at 1 second, so this unrelated list change is a significant recovery regression. Keep the list source's prior cap or avoid changing its retry policy while adding channel reconnects.
Useful? React with 👍 / 👎.
| match end_reason { | ||
| SessionEnd::Shutdown | SessionEnd::DownstreamClosed => { | ||
| // shutting down cleanly, or downstream closed: stop for good. | ||
| let _ = pubsub_conn.unsubscribe(&channel).await; |
There was a problem hiding this comment.
When shutdown is selected while the Pub/Sub session is still connected, this branch now waits for Redis to acknowledge UNSUBSCRIBE before the source can exit. If Redis is slow or the TCP connection is half-open during shutdown, graceful shutdown can block until the force-shutdown deadline; the previous take_until(shutdown) path stopped immediately by dropping the connection. Since the component is already stopping, avoid a network round trip here or race it with the shutdown/force path.
Useful? React with 👍 / 👎.
- Race `connect_and_subscribe` with the shutdown signal (biased select) so a connect against a black-holed endpoint can't block graceful shutdown until the force-shutdown deadline. - Drop the explicit `UNSUBSCRIBE` round trips on the shutdown/disconnect paths; dropping the pubsub connection closes it and Redis releases the subscription automatically, so shutdown no longer waits on a network round trip. - Keep the `list` source's retry backoff capped at 1s (its prior behavior) rather than the channel source's 30s cap, avoiding a recovery-latency regression for existing `data_type = "list"` users.
9f6f014 to
ac72174
Compare
|
I have read the CLA Document and I hereby sign the CLA |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ac72174894
ℹ️ 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".
| }; | ||
|
|
||
| // Connected: reset backoff so the next outage starts from the shortest delay. | ||
| backoff.reset(); |
There was a problem hiding this comment.
When Redis or an intervening load balancer accepts SUBSCRIBE and then immediately closes the connection, this reset makes every SessionEnd::Disconnected retry use the first 500 ms delay again, so the reconnect loop never backs off exponentially during a flapping outage. Keep the backoff advanced until the session has stayed healthy long enough or has actually delivered data; otherwise the source can produce an endless fast reconnect/log loop despite the intended capped exponential backoff.
Useful? React with 👍 / 👎.
| Err(err) => { | ||
| // failed to connect or SUBSCRIBE | ||
| emit!(RedisConnectionError::from(err.into_source())); | ||
| reconnecting = true; |
There was a problem hiding this comment.
For channel sources with a permanent Redis error such as bad credentials or ACLs that reject SUBSCRIBE, this branch now logs and retries forever inside the source future instead of returning the Connection/Subscribe build error as before. That makes an invalid configuration appear to start successfully and only surface as repeated runtime errors; keep transient reconnects, but preserve fail-fast behavior for non-recoverable connect/subscribe errors.
Useful? React with 👍 / 👎.
Second round of automated review feedback: - Do the initial connect + SUBSCRIBE during source build and fail fast on non-recoverable errors (bad auth/ACL/config) by returning the error, instead of entering the reconnect loop. An invalid configuration now surfaces at startup rather than looking like it started and only erroring at runtime. Transient (I/O) errors are still tolerated and handled by the reconnect loop. - Only reset the reconnect backoff after a session has actually delivered a message, not merely on a successful connect. A connection that is accepted and then immediately dropped (a flapping outage) now keeps backing off exponentially instead of hot-looping at the 500ms floor. - Classify permanent vs transient connect errors during reconnect too: a non-recoverable error stops the source rather than retrying forever.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 780538f5c0
ℹ️ 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".
| /// `SUBSCRIBE`) is treated as permanent, so an invalid configuration fails fast instead of | ||
| /// silently retrying forever. | ||
| fn is_transient(error: &redis::RedisError) -> bool { | ||
| error.kind() == redis::ErrorKind::IoError |
There was a problem hiding this comment.
When Redis is coming back from a restart and accepts TCP connections but replies to SUBSCRIBE with a server-side retryable error such as LOADING/BUSYLOADING or TRYAGAIN, redis-rs reports that as a non-I/O Redis error. Because this helper only treats IoError as transient, the reconnect path will return Err(()) (or fail the source build) instead of backing off and retrying, so data_type = "channel" can still stop permanently during the Redis-restart scenario this change is meant to recover from. Please classify retryable Redis server states as transient while keeping auth/ACL errors fatal.
Useful? React with 👍 / 👎.
Use redis-rs's own `RetryMethod` classification instead of only treating `ErrorKind::IoError` as transient. A Redis node coming back from a restart may accept the TCP connection but reply to `SUBSCRIBE` with a server-side retryable error such as `LOADING`/`BUSYLOADING`/`TRYAGAIN`, which redis-rs reports as a non-I/O error. The previous check treated those as permanent and stopped the source, defeating the restart recovery this change is meant to provide. Now only `RetryMethod::NoRetry` (auth/ACL/config rejections) is fatal; all other errors, including those retryable server states, back off and reconnect.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 48254413e6
ℹ️ 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".
| fn is_transient(error: &redis::RedisError) -> bool { | ||
| !matches!(error.retry_method(), redis::RetryMethod::NoRetry) |
There was a problem hiding this comment.
Fresh evidence for the earlier fail-fast concern: this new RetryMethod-based classification still treats bad Redis credentials as transient because redis-rs 0.32.4 maps ErrorKind::AuthenticationFailed to RetryMethod::Reconnect, not NoRetry. With an invalid password, the initial get_async_pubsub() failure is accepted as retryable, build() returns Ok, and the source retries forever instead of surfacing the configuration error; handle AuthenticationFailed explicitly as permanent before falling back to retry_method().
Useful? React with 👍 / 👎.
| // A message was delivered downstream, so the connection is | ||
| // healthy: reset the reconnect backoff. Resetting only here | ||
| // (not on connect) means a connection that drops before | ||
| // delivering anything keeps backing off exponentially. | ||
| backoff.reset(); |
There was a problem hiding this comment.
With this reset only happening after a message is delivered downstream, a low-volume channel can reconnect successfully and remain healthy for hours without ever resetting a backoff that was previously driven to the 30s cap. If Redis drops again during that quiet period, the source still waits up to 30s before its first resubscribe attempt, leaving the Pub/Sub source unsubscribed and losing messages despite the prior session being stable; reset after a minimum healthy session duration (or another connection-health signal) rather than only on delivered data.
Useful? React with 👍 / 👎.
… sessions Third round of automated review feedback: - redis-rs maps `ErrorKind::AuthenticationFailed` to `RetryMethod::Reconnect`, so the `RetryMethod`-based classifier treated bad credentials as transient and retried forever. Treat `AuthenticationFailed` as permanent explicitly so an invalid password/ACL fails the source build as intended. - Reset the reconnect backoff after a session has either delivered a message or stayed connected for a minimum healthy duration, rather than only on delivered data. A low-volume channel that stays healthy but quiet now clears a backoff that a previous flapping period drove up to the cap, so a later drop resubscribes promptly instead of waiting up to 30s.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: de4b3cba8d
ℹ️ 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".
| if error.kind() == redis::ErrorKind::AuthenticationFailed { | ||
| return false; | ||
| } | ||
| !matches!(error.retry_method(), redis::RetryMethod::NoRetry) |
There was a problem hiding this comment.
For rediss:// configs with an invalid certificate or TLS hostname, redis-rs reports the handshake failure as ErrorKind::IoError with retry_method() returning RetryImmediately, so this predicate treats it as transient. That makes a permanent TLS misconfiguration build successfully and retry forever instead of surfacing the config error like the previous build-time connect did; classify TLS/handshake failures as permanent before falling back to retry_method() or switch to an allowlist of known transient Redis/socket errors.
Useful? React with 👍 / 👎.
| warn!( | ||
| endpoint, | ||
| channel, "Redis pubsub stream ended; will reconnect." |
There was a problem hiding this comment.
When Redis closes an established Pub/Sub connection and the next reconnect succeeds on the first attempt, this branch only logs a warning and returns Disconnected; no component_errors_total is emitted because RedisConnectionError is only used for connect/subscribe failures. In that common Redis-restart/load-balancer-close scenario, Vector is unsubscribed and can lose Pub/Sub messages, but metric-based alerts see no component error. Emit a disconnect internal event before returning so successful reconnects still record the failure.
Useful? React with 👍 / 👎.
Replace the error-classification approach with the source's original build-time semantics, which is both simpler and more robust: - The initial connect + SUBSCRIBE fails fast on ANY error (auth, TLS, ACL, unreachable, LOADING), surfacing misconfiguration at startup as the source did before reconnect support. This avoids fragile transient-vs-permanent classification: redis-rs reports TLS handshake failures and transient network timeouts both as IoError/RetryImmediately, so no kind/retry_method predicate can separate them. Removing `is_transient` also drops the auth special-case. - Once the source is running, every reconnect failure is retried; a permanent problem was already ruled out by the successful build-time connect. Failures are still recorded via RedisConnectionError for alerting. - Emit a new RedisConnectionDropped internal event (component_errors_total) when an established connection drops, so alerts fire even when the reconnect succeeds on the first attempt.
2 participants
Supersedes #24100 (rebased onto master; original work by @gibranbadrul, preserved as the first commit). Closes #22615.
The
redissource withdata_type = "channel"currently stops receiving after any Redis connection drop and only recovers on a Vector restart. This adds a reconnect loop that re-connects and re-subscribes automatically, with exponential backoff.Addresses the review feedback on #24100:
common::backoff::ExponentialBackoff(from_millis(2).factor(250), capped at 30s), matchingaws_s3/sqs, instead of a bespoke helper. Backoff resets on a successful connect.RedisConnectionError→component_errors_total(error_type=connection_failed) on connect/subscribe failure, andRedisConnectionEstablished→connection_established_totalon initial connect and recovery.Also unifies the
listdata_type source onto the same sharedExponentialBackoff(removing its duplicatebackoff_exponential) and makes its retry sleep shutdown-aware.