From cb71b4980a07ab0f9b349022227ee4d395e81de6 Mon Sep 17 00:00:00 2001 From: Thanatat Tamtan Date: Sat, 13 Jun 2026 13:57:06 +0700 Subject: [PATCH] feat: honor Retry-After, jitter backoff, bound response-header wait Three transport/retry hardening items from the review. - 429/503 Retry-After: flush() now reads the Retry-After header on a 429 (or 503), parses both delta-seconds and HTTP-date forms, clamps to RetryAfterMax (60s), and paces the next retry to it instead of the fixed 100ms-1s backoff. The value is carried to retryFlush via a worker-local; the retry sleep is now a closeSignal-interruptible timer, so a long Retry-After (or Close) cannot delay shutdown. Retry-After is deliberately ignored on the closing path. - Backoff jitter: both retry loops use equal jitter (sleep in [d/2, d], via math/rand/v2) so retries across workers spread out instead of arriving in lockstep waves that can re-tip a recovering server. - ResponseHeaderTimeout: the default transport now bounds the wait for response headers to getIngestTimeout()/3 (~5s default). A server that completes the request but stalls before responding frees the worker in ~5s instead of the full 15s deadline. Firing returns a transport error, already retryable. User-supplied clients via SetHTTPClient are untouched. 429/503 stay retryable; the failure taxonomy, fire-and-forget batching, ordering, and the at-least-once/settle invariants are unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- backpressure_internal_test.go | 66 ++++++++++++++++++++++++++++ backpressure_test.go | 73 ++++++++++++++++++++++++++++++ quickwit.go | 83 ++++++++++++++++++++++++++++++++++- 3 files changed, 220 insertions(+), 2 deletions(-) create mode 100644 backpressure_internal_test.go create mode 100644 backpressure_test.go diff --git a/backpressure_internal_test.go b/backpressure_internal_test.go new file mode 100644 index 0000000..41dc6fa --- /dev/null +++ b/backpressure_internal_test.go @@ -0,0 +1,66 @@ +package quickwit + +import ( + "net/http" + "testing" + "time" +) + +// The default transport bounds the wait for response headers to a third of the +// ingest timeout, so a server that stalls before responding frees the worker +// early instead of holding it for the full deadline. +func TestNewDefaultClient_ResponseHeaderTimeout(t *testing.T) { + c := NewClient("http://example/api/v1/test") + c.SetIngestTimeout(30 * time.Second) + + tr, ok := c.httpClient().Transport.(*http.Transport) + if !ok { + t.Fatalf("transport = %T, want *http.Transport", c.httpClient().Transport) + } + if want := 10 * time.Second; tr.ResponseHeaderTimeout != want { + t.Errorf("ResponseHeaderTimeout = %v, want %v (ingestTimeout/3)", tr.ResponseHeaderTimeout, want) + } +} + +func TestParseRetryAfter(t *testing.T) { + now := time.Date(2026, 6, 13, 12, 0, 0, 0, time.UTC) + httpDate := now.Add(5 * time.Second).UTC().Format(http.TimeFormat) + pastDate := now.Add(-5 * time.Second).UTC().Format(http.TimeFormat) + + cases := []struct { + name string + header string + want time.Duration + ok bool + }{ + {"seconds", "2", 2 * time.Second, true}, + {"seconds with space", " 3 ", 3 * time.Second, true}, + {"http date future", httpDate, 5 * time.Second, true}, + {"http date past", pastDate, 0, false}, + {"empty", "", 0, false}, + {"garbage", "soon", 0, false}, + {"zero", "0", 0, false}, + {"negative", "-1", 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := parseRetryAfter(tc.header, now) + if ok != tc.ok || got != tc.want { + t.Errorf("parseRetryAfter(%q) = (%v, %v), want (%v, %v)", tc.header, got, ok, tc.want, tc.ok) + } + }) + } +} + +func TestJitterBackoff(t *testing.T) { + if got := jitterBackoff(0); got != 0 { + t.Errorf("jitterBackoff(0) = %v, want 0", got) + } + const d = 100 * time.Millisecond + for i := 0; i < 1000; i++ { + got := jitterBackoff(d) + if got < d/2 || got > d { + t.Fatalf("jitterBackoff(%v) = %v, want within [%v, %v]", d, got, d/2, d) + } + } +} diff --git a/backpressure_test.go b/backpressure_test.go new file mode 100644 index 0000000..7cfba03 --- /dev/null +++ b/backpressure_test.go @@ -0,0 +1,73 @@ +package quickwit_test + +import ( + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/moonrhythm/quickwit" +) + +// Core: a 429 with Retry-After paces the next attempt to the server's request +// instead of the much shorter default backoff, and the record still delivers. +func TestIngest_HonorsRetryAfterOn429(t *testing.T) { + var mu sync.Mutex + var times []time.Time + var delivered []int + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + mu.Lock() + times = append(times, time.Now()) + first := len(times) == 1 + if !first { + delivered = append(delivered, parseIndices(body)...) + } + mu.Unlock() + + if first { + w.Header().Set("Retry-After", "1") // 1s, well above the 100ms default backoff + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + c := quickwit.NewClient(server.URL + "/api/v1/test") + c.SetConcurrent(1) + c.SetBatchSize(1) // the single item triggers a flush + defer c.Close() + + c.Ingest(map[string]any{"index": 0}) + + // Wait for the retry to land (normal operation, not shutdown) before Close, + // so the Retry-After sleep is not cut short by the close signal. + deadline := time.After(4 * time.Second) + for { + mu.Lock() + n := len(times) + mu.Unlock() + if n >= 2 { + break + } + select { + case <-deadline: + t.Fatal("retry did not arrive before timeout") + case <-time.After(20 * time.Millisecond): + } + } + + mu.Lock() + defer mu.Unlock() + gap := times[1].Sub(times[0]) + if gap < 700*time.Millisecond { + t.Errorf("retry gap = %v, want >= ~1s (Retry-After honored, not the 100ms backoff)", gap) + } + if len(delivered) != 1 || delivered[0] != 0 { + t.Errorf("delivered = %v, want [0]", delivered) + } +} diff --git a/quickwit.go b/quickwit.go index 0a42074..cfca9f1 100644 --- a/quickwit.go +++ b/quickwit.go @@ -8,8 +8,10 @@ import ( "fmt" "io" "log/slog" + "math/rand/v2" "net/http" "slices" + "strconv" "strings" "sync" "time" @@ -284,11 +286,57 @@ func (c *Client) newDefaultClient() *http.Client { t.MaxIdleConns = concurrent } + // Bound the time spent waiting for response headers to a fraction of the + // per-flush deadline. A server that completes TCP/TLS and reads the body but + // then stalls before responding (typical of an L4 load balancer fronting an + // overloaded indexer) frees the worker in ~1/3 of getIngestTimeout instead of + // tying it up for the full deadline. Firing returns a transport error, which + // is already retryable. + t.ResponseHeaderTimeout = c.getIngestTimeout() / 3 + // No client-level timeout: the per-request context deadline from // getIngestTimeout already bounds each flush, matching prior behavior. return &http.Client{Transport: t} } +// RetryAfterMax caps how long a Retry-After header can delay the next flush, so +// a buggy or hostile value cannot stall ingestion indefinitely. +const RetryAfterMax = 60 * time.Second + +// jitterBackoff returns a duration in [d/2, d] (equal jitter) so retries across +// workers spread out instead of arriving in lockstep waves. The floor of d/2 +// keeps the fast path from collapsing to a near-zero sleep. +func jitterBackoff(d time.Duration) time.Duration { + if d <= 0 { + return 0 + } + half := d / 2 + return half + time.Duration(rand.Int64N(int64(half)+1)) +} + +// parseRetryAfter parses an HTTP Retry-After header, which is either a number of +// seconds or an HTTP-date. now is passed in so the worker can use a single clock +// read. It returns ok=false when the header is absent or unparseable. +func parseRetryAfter(h string, now time.Time) (time.Duration, bool) { + h = strings.TrimSpace(h) + if h == "" { + return 0, false + } + if secs, err := strconv.Atoi(h); err == nil { + if secs <= 0 { + return 0, false + } + return time.Duration(secs) * time.Second, true + } + if t, err := http.ParseTime(h); err == nil { + if d := t.Sub(now); d > 0 { + return d, true + } + return 0, false + } + return 0, false +} + func (c *Client) getMaxDelay() time.Duration { if c.maxDelay <= 0 { return IngestMaxDelay @@ -549,6 +597,11 @@ func (c *Client) loop() { buffer := make([]ingestItem, 0, batchSize) var resetBatchSizeAfter time.Time + // retryAfter carries a server-requested backpressure delay (from a 429/503 + // Retry-After header) out of flush() and into retryFlush()'s sleep. It is + // reset at the top of every flush so it only reflects the most recent attempt. + var retryAfter time.Duration + endpoint := c.endpoint endpoint = strings.TrimSuffix(endpoint, "/") endpoint = endpoint + "/ingest" @@ -558,6 +611,7 @@ func (c *Client) loop() { return true } + retryAfter = 0 buf.Reset() // encodeFailures can only ever hold fire-and-forget items (ack == nil): @@ -665,6 +719,14 @@ func (c *Client) loop() { return true } + // Backpressure: on 429/503, honor Retry-After (clamped) so we pace to + // the server instead of hammering it on the fixed backoff. + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable { + if d, ok := parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()); ok { + retryAfter = min(d, RetryAfterMax) + } + } + // Retryable (5xx, 408, 425, 429, transport errors handled above). return false } @@ -738,7 +800,22 @@ func (c *Client) loop() { attempt++ slog.Info("quickwit: flush failed, retrying indefinitely", "attempt", attempt) - time.Sleep(backoff) + + // Equal-jittered backoff, raised to a server-requested Retry-After + // when present. The sleep is interruptible so Close (or a long + // Retry-After) cannot delay shutdown past the next close signal. + sleep := jitterBackoff(backoff) + if retryAfter > sleep { + sleep = retryAfter + } + timer := time.NewTimer(sleep) + select { + case <-timer.C: + case <-c.closeSignal: + timer.Stop() + goto closing + } + // Exponential backoff with a cap backoff = time.Duration(float64(backoff) * 1.5) if backoff > time.Second { @@ -761,7 +838,9 @@ func (c *Client) loop() { // Don't sleep on the last attempt if i < maxRetries-1 { slog.Info("quickwit: flush failed while closing, retrying", "attempt", i+1, "maxRetries", maxRetries) - time.Sleep(backoff) + // Jittered backoff; Retry-After is intentionally not honored here + // so a server-requested delay cannot stretch shutdown. + time.Sleep(jitterBackoff(backoff)) // Exponential backoff with a cap backoff = time.Duration(float64(backoff) * 1.5) if backoff > time.Second {