This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 46caa7c82 fix(connectors): honor Retry-After on every retried status
(#4192)
46caa7c82 is described below
commit 46caa7c82bd3a329bc319b135dd50779d38fa1c5
Author: Ryan Huang <[email protected]>
AuthorDate: Sun Oct 11 00:15:18 2026 +0800
fix(connectors): honor Retry-After on every retried status (#4192)
---
.claude/skills/connector-sdk/SKILL.md | 2 +-
core/connectors/sdk/README.md | 2 +
core/connectors/sdk/src/retry.rs | 195 +++++++++++++++++-----
core/connectors/sinks/influxdb_sink/README.md | 5 +-
core/connectors/sinks/quickwit_sink/README.md | 4 +-
core/connectors/sources/influxdb_source/README.md | 4 +-
6 files changed, 163 insertions(+), 49 deletions(-)
diff --git a/.claude/skills/connector-sdk/SKILL.md
b/.claude/skills/connector-sdk/SKILL.md
index 8ad3a2791..440e9a1ec 100644
--- a/.claude/skills/connector-sdk/SKILL.md
+++ b/.claude/skills/connector-sdk/SKILL.md
@@ -154,7 +154,7 @@ Plugin authors call this on every consumed message. The
implementation in `lib.r
- `CircuitBreaker`: threshold + cooldown, `try_lock()` on the success path to
avoid hot-path contention.
- `retry_async(policy, context, should_retry, op)`: the retry loop for
anything failing as `Err`. Owns attempt counting, backoff and the per-retry
log; returns `RetryFailure { error, attempts, exhausted }` so callers write
their own terminal log.
- `retry_backoff(base, retry, max)`: backoff only, for a loop that computes
its own delay. `retry` is 1-based. `exponential_backoff` is the 0-based
primitive underneath and applies no jitter, so call it directly only when the
delay must be exact (`source.rs::nack_retry_delay`, whose tests assert exact
values).
-- `HttpRetryMiddleware`: integrates with `reqwest-middleware`. Retries 429 +
5xx + network errors. Honors `Retry-After`.
+- `HttpRetryMiddleware`: integrates with `reqwest-middleware`. Retries 429 +
5xx + network errors. Honors `Retry-After` on 429 and 5xx as a floor over the
computed backoff, capped at a one-hour `MAX_RETRY_AFTER`, not `max_delay`.
- `max_retries` = **total attempts** including the first try, not extra
retries. Document if you change this convention. `meilisearch_sink` is the
standing exception: its `max_retries` / `max_open_retries` count retries
*after* the first, as its README states.
- New helpers must take `Duration` (not `u64 millis`) on the public API.
Internal computation uses `humantime` parsing of `String`.
diff --git a/core/connectors/sdk/README.md b/core/connectors/sdk/README.md
index f058291d4..f046397dc 100644
--- a/core/connectors/sdk/README.md
+++ b/core/connectors/sdk/README.md
@@ -54,6 +54,8 @@ value.static = "hello"
`retry_async` runs an operation that fails with `Err` and retries it while
`should_retry` accepts the error. It owns attempt counting, backoff and the
per-retry log, and returns `RetryFailure { error, attempts, exhausted }` so the
caller logs the terminal failure. `retry_backoff` computes a single delay for a
loop that cannot use `retry_async`, such as `HttpRetryMiddleware`, which
retries on an `Ok` response rather than an `Err`. Its `retry` argument is
1-based.
+`HttpRetryMiddleware` reads a server's `Retry-After` header on every status it
retries, which is 429 and any 5xx. The header can lengthen the computed backoff
but never shorten it, so a zero or a short value keeps the backoff. `max_delay`
bounds only the computed backoff. The header has its own one-hour ceiling, and
a value past it is drawn from the jitter window just below it. Only the
integer-seconds form is read, so an HTTP-date or unparsable value falls back to
the computed backoff.
+
Two symbols were removed and one field changed meaning, each in a way that
breaks out-of-tree plugins, so those plugins must be rebuilt against the
current source:
| Removed | Replacement |
diff --git a/core/connectors/sdk/src/retry.rs b/core/connectors/sdk/src/retry.rs
index 362eebbcd..7ff788613 100644
--- a/core/connectors/sdk/src/retry.rs
+++ b/core/connectors/sdk/src/retry.rs
@@ -37,6 +37,7 @@ use anyhow::anyhow;
use http::Extensions;
use humantime::Duration as HumanDuration;
use rand::RngExt as _;
+use reqwest::header::{HeaderMap, RETRY_AFTER};
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, Middleware,
Next};
use std::fmt;
use std::future::Future;
@@ -168,8 +169,8 @@ pub fn exponential_backoff(base: Duration, attempt: u32,
max_delay: Duration) ->
}
/// Parse a `Retry-After` header value (integer seconds).
-/// Returns `None` for HTTP-date values — callers should fall back to their
-/// own backoff strategy.
+/// Returns `None` for any value that is not integer seconds, including the
+/// HTTP-date form — callers should fall back to their own backoff strategy.
pub fn parse_retry_after(value: &str) -> Option<Duration> {
if let Ok(secs) = value.trim().parse::<u64>() {
return Some(Duration::from_secs(secs));
@@ -352,11 +353,43 @@ pub fn is_transient_status(status: reqwest::StatusCode)
-> bool {
status == reqwest::StatusCode::TOO_MANY_REQUESTS ||
status.is_server_error()
}
+/// Ceiling on a server-supplied `Retry-After`, separate from and much larger
+/// than `max_delay`, which bounds only the backoff we compute ourselves. It
+/// guards against an absurd value rather than tuning a real rate-limit window.
+const MAX_RETRY_AFTER: Duration = Duration::from_secs(60 * 60);
+
+/// Read the delay a response's `Retry-After` header asks for, before any cap.
+///
+/// Returns `None` when the header is absent, unreadable, or in the HTTP-date
+/// form that [`parse_retry_after`] does not accept.
+fn requested_retry_after(headers: &HeaderMap) -> Option<Duration> {
+ headers
+ .get(RETRY_AFTER)
+ .and_then(|value| value.to_str().ok())
+ .and_then(parse_retry_after)
+}
+
+/// The wait before a retry whose response carried `Retry-After`.
+///
+/// The header can only lengthen the computed `backoff`, never shorten it, so a
+/// `Retry-After: 0` or a short window keeps the backoff the operator
configured.
+/// A value past `ceiling` is drawn from the jitter window below it instead of
+/// landing on it, so instances throttled together do not retry in lockstep.
+fn retry_after_delay(requested: Duration, backoff: Duration, ceiling:
Duration) -> Duration {
+ let honored = if requested > ceiling {
+ retry_backoff(ceiling, 1, ceiling)
+ } else {
+ requested
+ };
+ honored.max(backoff)
+}
+
/// Per-request retry middleware for HTTP connectors.
///
/// Wraps a `reqwest-middleware` stack and retries transient failures
/// (HTTP 429, 5xx, network errors) with exponential back-off and ±20 % jitter.
-/// A `Retry-After` response header on a 429 overrides the calculated delay.
+/// A `Retry-After` response header on a retried status can lengthen the
+/// calculated delay, up to a one-hour ceiling of its own.
///
/// The `log_prefix` parameter identifies the connector in log messages
/// (e.g. `"InfluxDB"`, `"Elasticsearch"`), allowing this middleware to be
@@ -420,29 +453,30 @@ impl Middleware for HttpRetryMiddleware {
return Ok(response);
}
- // Parse Retry-After on 429 before falling back to our own
- // calculated backoff.
- let retry_after = if status ==
reqwest::StatusCode::TOO_MANY_REQUESTS {
- response
- .headers()
- .get("Retry-After")
- .and_then(|value| value.to_str().ok())
- .and_then(parse_retry_after)
- } else {
- None
- };
-
attempts += 1;
if is_transient_status(status) && attempts <
self.policy.max_attempts {
+ let backoff = self.policy.backoff(attempts);
+ let requested =
requested_retry_after(response.headers());
+ let delay = requested.map_or(backoff, |requested| {
+ retry_after_delay(requested, backoff,
MAX_RETRY_AFTER)
+ });
// Consume the error body for logging, then retry.
let body_text =
response.text().await.unwrap_or_default();
- let delay = retry_after.unwrap_or_else(||
self.policy.backoff(attempts));
- warn!(
- "{} transient error {status} \
- (attempt {attempts}/{}): {body_text}. \
- Retrying in {delay:?}...",
- self.log_prefix, self.policy.max_attempts
- );
+ match requested.filter(|requested| *requested >
MAX_RETRY_AFTER) {
+ Some(requested) => warn!(
+ "{} transient error {status} \
+ (attempt {attempts}/{}): {body_text}. \
+ Retry-After asked for {requested:?}, past the
\
+ {MAX_RETRY_AFTER:?} ceiling. Retrying in
{delay:?}...",
+ self.log_prefix, self.policy.max_attempts
+ ),
+ None => warn!(
+ "{} transient error {status} \
+ (attempt {attempts}/{}): {body_text}. \
+ Retrying in {delay:?}...",
+ self.log_prefix, self.policy.max_attempts
+ ),
+ }
tokio::time::sleep(delay).await;
current_req = match next_req {
Some(r) => r,
@@ -816,34 +850,109 @@ mod tests {
);
}
- #[tokio::test]
- async fn
given_a_retry_after_should_take_precedence_over_the_computed_backoff() {
- let server = mock_then_ok(429, &[("Retry-After", "1")]).await;
- // Backoff bounded far below the header, so honoring it is visible.
- let client = retry_client(3, Duration::from_millis(10),
Duration::from_millis(50));
-
- let started = Instant::now();
- let response = client.get(server.uri()).send().await.unwrap();
- let elapsed = started.elapsed();
+ fn retry_after_header(value: &str) -> HeaderMap {
+ let mut headers = HeaderMap::new();
+ headers.insert(RETRY_AFTER, value.parse().unwrap());
+ headers
+ }
- assert_eq!(response.status(), 200);
+ #[test]
+ fn given_a_retry_after_should_read_the_full_requested_delay() {
+ // The middleware applies the cap, and logs the full value when it
does.
assert_eq!(
- server.received_requests().await.unwrap().len(),
- 2,
- "expected exactly one retry"
+ requested_retry_after(&retry_after_header("86400")),
+ Some(Duration::from_secs(86400))
);
+ }
+
+ #[test]
+ fn
given_an_unusable_retry_after_should_fall_back_to_the_computed_backoff() {
+ for (header, why) in [
+ (
+ "Wed, 21 Oct 2026 07:28:00 GMT",
+ "the HTTP-date form is not parsed yet",
+ ),
+ (
+ "not-a-number",
+ "an unparsable header must not stop the retry",
+ ),
+ ] {
+ assert_eq!(
+ requested_retry_after(&retry_after_header(header)),
+ None,
+ "{why}"
+ );
+ }
+ }
+
+ #[test]
+ fn
given_a_retry_after_past_the_ceiling_should_draw_from_the_jitter_window_below_it()
{
+ let ceiling = Duration::from_secs(10);
+ let delays: Vec<Duration> = (0..100)
+ .map(|_| {
+ retry_after_delay(
+ Duration::from_secs(86400),
+ Duration::from_millis(10),
+ ceiling,
+ )
+ })
+ .collect();
+
+ for delay in &delays {
+ assert!(
+ *delay >= ceiling.mul_f64(JITTER_LOW) && *delay <= ceiling,
+ "a day-long header must land in the jitter window under
{ceiling:?}, got {delay:?}"
+ );
+ }
assert!(
- elapsed >= Duration::from_millis(900),
- "used the computed backoff instead of Retry-After: waited
{elapsed:?}"
- );
- // The header asks for 1s. Anything far past it means the middleware
- // added its own backoff on top instead of honoring the header.
- assert!(
- elapsed < Duration::from_millis(1500),
- "waited {elapsed:?}, past the 1s the header asked for"
+ delays.iter().any(|delay| *delay < ceiling),
+ "every draw landed on {ceiling:?}, so throttled instances would
retry in lockstep"
);
}
+ #[test]
+ fn given_a_retry_after_shorter_than_the_backoff_should_keep_the_backoff() {
+ let backoff = Duration::from_secs(5);
+
+ for requested in [Duration::ZERO, Duration::from_secs(1)] {
+ assert_eq!(
+ retry_after_delay(requested, backoff, MAX_RETRY_AFTER),
+ backoff,
+ "a {requested:?} header must not shorten the computed backoff"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn
given_a_retry_after_past_max_delay_should_be_honored_on_every_retried_status() {
+ // RFC 9110 allows Retry-After on any 5xx, and InfluxDB OSS documents
it
+ // on 503. `max_delay` bounds only the computed backoff, so the header
+ // must outrank it.
+ for status in [429, 503] {
+ let server = mock_then_ok(status, &[("Retry-After", "1")]).await;
+ let client = retry_client(3, Duration::from_millis(10),
Duration::from_millis(100));
+
+ let started = Instant::now();
+ let response = client.get(server.uri()).send().await.unwrap();
+ let elapsed = started.elapsed();
+
+ assert_eq!(response.status(), 200);
+ assert_eq!(
+ server.received_requests().await.unwrap().len(),
+ 2,
+ "expected exactly one retry on {status}"
+ );
+ assert!(
+ elapsed >= Duration::from_millis(900),
+ "ignored Retry-After on {status} and used the computed
backoff: waited {elapsed:?}"
+ );
+ assert!(
+ elapsed < Duration::from_millis(1500),
+ "waited {elapsed:?} on {status}, past the 1s the header asked
for"
+ );
+ }
+ }
+
#[test]
fn given_a_retry_number_should_back_off_from_the_base_delay() {
let policy = policy(8);
diff --git a/core/connectors/sinks/influxdb_sink/README.md
b/core/connectors/sinks/influxdb_sink/README.md
index 1ab1b5b91..361e61da1 100644
--- a/core/connectors/sinks/influxdb_sink/README.md
+++ b/core/connectors/sinks/influxdb_sink/README.md
@@ -156,8 +156,9 @@ The sink uses a layered design:
A serialization error rejects its whole chunk. Later chunks are still
attempted after failures,
and the first error is returned after the loop. `max_retries = 3` allows two
retries; `0` and
`1` both allow one attempt. Backoff starts at `retry_delay`, doubles with ±20%
jitter and is
-capped by `retry_max_delay`. Integer-seconds `Retry-After` on a 429 overrides
that cap;
-HTTP-date values are ignored. Startup uses its own attempt budget/cap and
retries any failed
+capped by `retry_max_delay`. Integer-seconds `Retry-After` on a 429 or any 5xx
can
+lengthen the calculated delay past `retry_max_delay`, up to one hour, but
never shortens
+it. HTTP-date and unparsable values fall back to the calculated delay. Startup
uses its own attempt budget/cap and retries any failed
health check. Invalid duration strings warn and fall back to `1s`.
The runtime records a plugin error and continues polling with consumer
auto-commit. Failed
diff --git a/core/connectors/sinks/quickwit_sink/README.md
b/core/connectors/sinks/quickwit_sink/README.md
index b64575721..a5158d58f 100644
--- a/core/connectors/sinks/quickwit_sink/README.md
+++ b/core/connectors/sinks/quickwit_sink/README.md
@@ -11,7 +11,7 @@ The Quickwit connector sends data to the Quickwit API using
HTTP. It checks read
| `verbose_logging` | `false` | Log received and ingested message counts at
`info` instead of `debug`. |
| `max_retries` | `3` | Total HTTP attempts including the first; `0` and `1`
both allow one attempt. |
| `retry_delay` | `"1s"` | Base exponential delay for HTTP retries and
readiness probes. |
-| `retry_max_delay` | `"5s"` | Cap for calculated HTTP retry delays; a valid
`Retry-After` on HTTP 429 overrides it. |
+| `retry_max_delay` | `"5s"` | Cap for calculated HTTP retry delays. A valid
`Retry-After` on HTTP 429 or any 5xx can exceed it, up to one hour. |
| `max_open_retries` | `10` | One initial probe per readiness check, with up
to `max_open_retries - 1` retries shared between them; `0` and `1` disable
retries. |
| `open_retry_max_delay` | `"30s"` | Maximum delay between readiness probes. |
| `timeout` | `"30s"` | Timeout per HTTP attempt; retries and waits can extend
the complete operation. |
@@ -103,7 +103,7 @@ The examples use `mode: dynamic` to retain wrapper fields.
With `mode: strict`,
## Delivery semantics
-Transient HTTP failures, including 429, can retry a request that Quickwit
already accepted. Quickwit ingest has no deduplication key, so these retries
can produce duplicate documents. Set `max_retries = 1` to disable the sink's
HTTP retry loop. Calculated delays use exponential backoff with jitter; a valid
`Retry-After` on HTTP 429 replaces the calculated delay.
+Transient HTTP failures, including 429, can retry a request that Quickwit
already accepted. Quickwit ingest has no deduplication key, so these retries
can produce duplicate documents. Set `max_retries = 1` to disable the sink's
HTTP retry loop. Calculated delays use exponential backoff with jitter. A valid
`Retry-After` on HTTP 429 or any 5xx can lengthen the calculated delay past
`retry_max_delay`, up to one hour.
Service readiness retries any failed health probe. After verifying or creating
the index, the sink probes its ingest endpoint with an empty body before
accepting messages.
This is intentional: legacy index metadata can exist before the ingest queue
is ready, while the read-only `/tail` endpoint checks only legacy ingestion.
diff --git a/core/connectors/sources/influxdb_source/README.md
b/core/connectors/sources/influxdb_source/README.md
index 337f61e84..2de54a27b 100644
--- a/core/connectors/sources/influxdb_source/README.md
+++ b/core/connectors/sources/influxdb_source/README.md
@@ -112,7 +112,9 @@ circuit_breaker_threshold = 5 # consecutive failures
before circuit trips
circuit_breaker_cool_down = "30s" # cooldown before queries resume
```
-Attempt counts are clamped to at least one. Query retries use exponential
backoff with jitter; integer-seconds `Retry-After` on HTTP 429 can override the
delay cap. Other HTTP errors and malformed successful response bodies fail the
poll. Startup retries failed authenticated `GET /health` requests. Invalid
duration strings warn and fall back to `1s`, while zero durations are accepted.
`batch_size = 0` is clamped to `1`. Without `initial_offset`, the cursor starts
at `1970-01-01T00:00:00Z`.
+Attempt counts are clamped to at least one. Query retries use exponential
backoff with jitter. Integer-seconds `Retry-After` on HTTP 429 or any 5xx can
lengthen the calculated delay past `retry_max_delay`, up to one hour.
+
+HTTP 429 and 5xx fail the poll once the last attempt fails. Other HTTP errors
and malformed successful response bodies fail it right away. Startup retries
failed authenticated `GET /health` requests. Invalid duration strings warn and
fall back to `1s`, while zero durations are accepted. `batch_size = 0` is
clamped to `1`. Without `initial_offset`, the cursor starts at
`1970-01-01T00:00:00Z`.
## Query Template Placeholders
