feat(hedging): support hedging in order to reduce latency when quering endpoints (#1081)

* feat(hedging): support hedging in order to reduce latency when quering endpoints

Signed-off-by: Ahmet DEMIR <me@ahmet2mir.eu>

* fix(hedging): move fields to struct, remove ratelimit, fail on sethedging func if not enabled, restore retry, allow unsafe funcs

Signed-off-by: Ahmet DEMIR <me@ahmet2mir.eu>

* fix(hedging): adapt suggestions

Signed-off-by: Ahmet DEMIR <me@ahmet2mir.eu>

---------

Signed-off-by: Ahmet DEMIR <me@ahmet2mir.eu>

Author: ahmet2mir

BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "curl.go",
         "debug.go",
         "digest.go",
+        "hedging.go",
         "load_balancer.go",
         "middleware.go",
         "multipart.go",
@@ -42,6 +43,7 @@ go_test(
         "context_test.go",
         "curl_test.go",
         "digest_test.go",
+        "hedging_test.go",
         "load_balancer_test.go",
         "middleware_test.go",
         "multipart_test.go",

client.go

@@ -227,6 +227,7 @@ type Client struct {
 	contentDecompressers     map[string]ContentDecompresser
 	certWatcherStopChan      chan bool
 	circuitBreaker           *CircuitBreaker
+	hedging                  *hedgingConfig
 }
 
 // CertWatcherOptions allows configuring a watcher that reloads dynamically TLS certs.
@@ -1469,6 +1470,205 @@ func (c *Client) AddRetryHooks(hooks ...RetryHookFunc) *Client {
 	return c
 }
 
+// isHedgingEnabled method returns true if hedging is enabled and get client clock.
+func (c *Client) IsHedgingEnabled() bool {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	return c.isHedgingEnabled()
+}
+
+// isHedgingEnabled method returns true if hedging is enabled.
+func (c *Client) isHedgingEnabled() bool {
+	return c.hedging != nil && c.hedging.enabled
+}
+
+// SetHedgingDelay method sets the delay between hedged requests.
+func (c *Client) SetHedgingDelay(delay time.Duration) *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.isHedgingEnabled() {
+		c.hedging.delay = delay
+		return c
+	}
+	c.log.Errorf("SetHedgingDelay: %v", ErrHedgingDisabled)
+	return c
+}
+
+// HedgingDelay method returns the configured hedging delay.
+func (c *Client) HedgingDelay() time.Duration {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	if c.isHedgingEnabled() {
+		return c.hedging.delay
+	}
+	return hedgingDefaultDelay
+}
+
+// SetHedgingUpTo method sets maximum concurrent hedged requests.
+func (c *Client) SetHedgingUpTo(upTo int) *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.isHedgingEnabled() {
+		c.hedging.upTo = upTo
+		return c
+	}
+	c.log.Errorf("SetHedgingUpTo: %v", ErrHedgingDisabled)
+	return c
+}
+
+// HedgingUpTo method returns the maximum concurrent requests.
+func (c *Client) HedgingUpTo() int {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	if c.isHedgingEnabled() {
+		return c.hedging.upTo
+	}
+	return hedgingDefaultUpTo
+}
+
+// SetHedgingMaxPerSecond method sets rate limit for hedged requests.
+func (c *Client) SetHedgingMaxPerSecond(maxPerSecond float64) *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.isHedgingEnabled() {
+		c.hedging.maxPerSecond = maxPerSecond
+		return c
+	}
+	c.log.Errorf("SetHedgingMaxPerSecond: %v", ErrHedgingDisabled)
+	return c
+}
+
+// HedgingMaxPerSecond method returns the hedging rate limit.
+func (c *Client) HedgingMaxPerSecond() float64 {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	if c.isHedgingEnabled() {
+		return c.hedging.maxPerSecond
+	}
+	return hedgingDefaultMaxPerSecond
+}
+
+// SetHedgingAllowNonReadOnly method allows hedging for non-read-only HTTP methods.
+// By default, only read-only methods (GET, HEAD, OPTIONS, TRACE) are hedged.
+// NOTE:
+//   - Use this with caution as hedging write operations can lead to duplicates.
+func (c *Client) SetHedgingAllowNonReadOnly(allow bool) *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.isHedgingEnabled() {
+		c.hedging.allowNonReadOnly = allow
+		// Re-wrap to apply new settings
+		c.unwrapHedgingTransport()
+		c.wrapTransportWithHedging()
+		return c
+	}
+	c.log.Errorf("SetHedgingAllowNonReadOnly: %v", ErrHedgingDisabled)
+	return c
+}
+
+// IsHedgingAllowNonReadOnly method returns true if hedging is enabled for non-read-only methods.
+func (c *Client) IsHedgingAllowNonReadOnly() bool {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	if c.isHedgingEnabled() {
+		return c.hedging.allowNonReadOnly
+	}
+	return hedgingDefaultAllowNonReadOnly
+}
+
+// EnableHedging method enables hedging with the given configuration.
+//
+// Hedging sends multiple concurrent requests with staggered delays and returns
+// the first response to complete to reduce tail latency. Only read-only HTTP methods
+// (GET, HEAD, OPTIONS, TRACE) are hedged by default unless SetHedgingAllowNonReadOnly is used.
+//
+//		client.EnableHedging(
+//		    50*time.Millisecond,   // delay between requests
+//		    3,                     // max 3 concurrent requests
+//		    10.0,                  // max 10 hedged requests per second
+//		)
+//	 Last one come from rate package, to use fractional rates, e.g.
+//	 - 0.1 = 1 request every 10 seconds
+//	 - 0.5 = 1 request every 2 seconds
+//	 - 1.0 = 1 request per second
+//	 - 2.5 = 2.5 requests per second (5 requests every 2 seconds)
+//	 - 10.0 = 10 requests per second
+func (c *Client) EnableHedging(delay time.Duration, upTo int, maxPerSecond float64) *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.hedging == nil {
+		c.hedging = &hedgingConfig{}
+	}
+
+	c.hedging.delay = delay
+	c.hedging.upTo = upTo
+	c.hedging.maxPerSecond = maxPerSecond
+	c.hedging.enabled = true
+
+	// Disable retry by default when hedging is enabled.
+	// Users can re-enable retry if they want it as a fallback mechanism.
+	if c.retryCount > 0 {
+		c.log.Warnf("Disabling retry (count: %d) as hedging is now enabled. You can re-enable retry with SetRetryCount() if you want it as a fallback.", c.retryCount)
+		c.retryCount = 0
+	}
+
+	c.wrapTransportWithHedging()
+
+	return c
+}
+
+// DisableHedging method disables hedging.
+func (c *Client) DisableHedging() *Client {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.isHedgingEnabled() {
+		c.hedging.enabled = false
+	}
+
+	c.unwrapHedgingTransport()
+
+	return c
+}
+
+func (c *Client) wrapTransportWithHedging() {
+	if c.hedging == nil || !c.hedging.enabled {
+		return
+	}
+
+	currentTransport := c.httpClient.Transport
+	if currentTransport == nil {
+		currentTransport = createTransport(nil, nil)
+	}
+
+	// Already set
+	if _, ok := currentTransport.(*hedgingTransport); ok {
+		return
+	}
+
+	// Calculate rate delay: if maxPerSecond is 10, delay is 100ms (1s / 10)
+	var rateDelay time.Duration
+	if c.hedging.maxPerSecond > 0 {
+		rateDelay = time.Duration(float64(time.Second) / c.hedging.maxPerSecond)
+	}
+
+	c.httpClient.Transport = &hedgingTransport{
+		transport:        currentTransport,
+		delay:            c.hedging.delay,
+		upTo:             c.hedging.upTo,
+		rateDelay:        rateDelay,
+		allowNonReadOnly: c.hedging.allowNonReadOnly,
+	}
+}
+
+func (c *Client) unwrapHedgingTransport() {
+	if ht, ok := c.httpClient.Transport.(*hedgingTransport); ok {
+		c.httpClient.Transport = ht.transport
+	}
+}
+
 // TLSClientConfig method returns the [tls.Config] from underlying client transport
 // otherwise returns nil
 func (c *Client) TLSClientConfig() *tls.Config {

client_test.go

@@ -1533,3 +1533,126 @@ func TestClientOnCloseMultipleHooks(t *testing.T) {
 	assertNil(t, err)
 	assertEqual(t, []string{"first", "second", "third"}, executionOrder)
 }
+
+func TestClientHedgingBasic(t *testing.T) {
+	var attemptCount int32
+	ts := createHedgingTestServer(t, &attemptCount, 0, 0)
+	defer ts.Close()
+
+	c := dcnl()
+	c.EnableHedging(20*time.Millisecond, 3, 0)
+
+	resp, err := c.R().Get(ts.URL + "/hedging-slow-first")
+	assertError(t, err)
+	assertEqual(t, http.StatusOK, resp.StatusCode())
+	assertNotEqual(t, "", resp.String())
+}
+
+func TestClientHedgingDisable(t *testing.T) {
+	var attemptCount int32
+	ts := createHedgingTestServer(t, &attemptCount, 0, 0)
+	defer ts.Close()
+
+	c := dcnl()
+	c.EnableHedging(20*time.Millisecond, 3, 0)
+	assertEqual(t, true, c.IsHedgingEnabled())
+
+	c.DisableHedging()
+	assertEqual(t, false, c.IsHedgingEnabled())
+
+	resp, err := c.R().Get(ts.URL + "/hedging-slow-first")
+	assertError(t, err)
+	assertEqual(t, http.StatusOK, resp.StatusCode())
+}
+
+func TestClientHedgingNil(t *testing.T) {
+	c := dcnl()
+	c.hedging = nil
+	c.wrapTransportWithHedging()
+
+	assertEqual(t, false, c.IsHedgingEnabled())
+
+	_, ok := c.httpClient.Transport.(*hedgingTransport)
+	if ok {
+		t.Error("Transport shouldn't be hedgingTransport when hedging is nil")
+	}
+}
+
+func TestClientHedgingMutualExclusionWithRetry(t *testing.T) {
+	c := dcnl()
+
+	// Set retry first
+	c.SetRetryCount(2)
+	assertEqual(t, 2, c.RetryCount())
+
+	// Enable hedging should disable retry by default
+	c.EnableHedging(50*time.Millisecond, 3, 0)
+	assertEqual(t, 0, c.RetryCount())
+
+	// But user can re-enable retry as fallback
+	c.SetRetryCount(1)
+	assertEqual(t, 1, c.RetryCount())
+	assertEqual(t, true, c.IsHedgingEnabled())
+
+	// Disable hedging
+	c.DisableHedging()
+	assertEqual(t, false, c.IsHedgingEnabled())
+	assertEqual(t, 1, c.RetryCount()) // Retry count should remain
+}
+
+func TestClientHedgingConfiguration(t *testing.T) {
+	c := dcnl()
+
+	// Setters require hedging to be enabled first
+	assertEqual(t, false, c.IsHedgingEnabled())
+
+	c.EnableHedging(50*time.Millisecond, 3, 10.0)
+
+	assertEqual(t, true, c.IsHedgingEnabled())
+	assertEqual(t, 50*time.Millisecond, c.HedgingDelay())
+	assertEqual(t, 3, c.HedgingUpTo())
+	assertEqual(t, 10.0, c.HedgingMaxPerSecond())
+
+	// Now we can update individual settings
+	c.SetHedgingDelay(100 * time.Millisecond)
+	assertEqual(t, 100*time.Millisecond, c.HedgingDelay())
+
+	c.SetHedgingUpTo(5)
+	assertEqual(t, 5, c.HedgingUpTo())
+
+	c.SetHedgingMaxPerSecond(20.0)
+	assertEqual(t, 20.0, c.HedgingMaxPerSecond())
+}
+
+func TestClientHedgingWithRateLimit(t *testing.T) {
+	var attemptCount int32
+	ts := createHedgingTestServer(t, &attemptCount, 0, 0)
+	defer ts.Close()
+
+	c := dcnl()
+	c.EnableHedging(10*time.Millisecond, 10, 5.0)
+
+	resp, err := c.R().Get(ts.URL + "/hedging-slow-all")
+	assertError(t, err)
+	assertEqual(t, http.StatusOK, resp.StatusCode())
+}
+
+func TestClientHedgingSafeMethodsOnly(t *testing.T) {
+	ts := createGetServer(t)
+	defer ts.Close()
+
+	c := dcnl()
+	c.EnableHedging(20*time.Millisecond, 3, 0)
+
+	resp, err := c.R().Get(ts.URL + "/")
+	assertError(t, err)
+	assertEqual(t, http.StatusOK, resp.StatusCode())
+
+	resp2, err2 := c.R().Head(ts.URL + "/")
+	assertError(t, err2)
+	assertEqual(t, http.StatusOK, resp2.StatusCode())
+
+	resp3, err3 := c.R().Options(ts.URL + "/")
+	assertError(t, err3)
+	assertEqual(t, http.StatusOK, resp3.StatusCode())
+}