fix: back off Consul CAS retries during KV outages

CHANGELOG.md

@@ -15,6 +15,7 @@
 * [BUGFIX] Store Gateway: Fix the Parquet bucket store ignoring query sharding, which made every shard return all series. #7892
 * [BUGFIX] Querier: Fix the querier crashing instead of reporting the error when a blocks store set subservice fails. `blocksStoreReplicationSet` never initialised its `FailureWatcher`, so the failure callback dereferenced a nil pointer on the goroutine the service manager runs listeners on, and the resulting panic's stack trace buried the error logged for the actual cause. #7896
 * [BUGFIX] Compactor: Delete block deletion markers left in the global `markers/` location of a tenant after the block they refer to was removed. They were skipped at every bucket index update and kept the tenant location in the bucket forever. #7890
+* [BUGFIX] Consul: Back off CAS retries after failed reads or writes to reduce request amplification during outages, and honor context cancellation while waiting to retry. #7893
 
 ## 1.22.0 in progress
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

pkg/ring/kv/consul/client.go

@@ -30,10 +30,21 @@ const (
 var (
 	writeOptions = &consul.WriteOptions{}
 
+	// backoffConfig is used by the long-running watch loops.
 	backoffConfig = backoff.Config{
 		MinBackoff: 1 * time.Second,
 		MaxBackoff: 1 * time.Minute,
 	}
+
+	// casBackoffConfig is used between CAS attempts after a failed read or
+	// write. It is kept short because CAS callers (e.g. ring heartbeats and
+	// shutdown paths) may not have a deadline: with the default 10 attempts
+	// the total wait is less than 13s (200ms+400ms+800ms+1.6s+5*2s upper
+	// bounds), well below the default 1m ring heartbeat timeout.
+	casBackoffConfig = backoff.Config{
+		MinBackoff: 100 * time.Millisecond,
+		MaxBackoff: 2 * time.Second,
+	}
 )
 
 // Config to create a ConsulClient
@@ -181,16 +192,34 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r
 	}
 
 	index := uint64(0)
+	// Keep error backoff across attempts: a successful read doesn't imply
+	// that writes have recovered. Conflicts and callback retries don't wait.
+	errorBackoff := backoff.New(ctx, casBackoffConfig)
+	retryAfterError := false
 	for i := 0; i < retries; i++ {
+		if retryAfterError {
+			errorBackoff.Wait()
+			retryAfterError = false
+		}
 		if i > 0 && sleepBeforeRetry > 0 {
-			time.Sleep(sleepBeforeRetry)
+			timer := time.NewTimer(sleepBeforeRetry)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return ctx.Err()
+			case <-timer.C:
+			}
+		}
+		if err := ctx.Err(); err != nil {
+			return err
 		}
 
 		// Get with default options - don't want stale data to compare with
 		options := &consul.QueryOptions{}
 		kvp, _, err := c.kv.Get(key, options.WithContext(ctx))
 		if err != nil {
 			level.Error(c.logger).Log("msg", "error getting key", "key", key, "err", err)
+			retryAfterError = true
 			continue
 		}
 		var intermediate any
@@ -224,13 +253,17 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r
 			level.Error(c.logger).Log("msg", "error serialising value", "key", key, "err", err)
 			continue
 		}
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		ok, _, err := c.kv.CAS(&consul.KVPair{
 			Key:         key,
 			Value:       bytes,
 			ModifyIndex: index,
 		}, writeOptions.WithContext(ctx))
 		if err != nil {
 			level.Error(c.logger).Log("msg", "error CASing", "key", key, "err", err)
+			retryAfterError = true
 			continue
 		}
 		if !ok {
@@ -239,6 +272,9 @@ func (c *Client) cas(ctx context.Context, key string, f func(in any) (out any, r
 		}
 		return nil
 	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	return fmt.Errorf("failed to CAS %s", key)
 }
 

pkg/ring/kv/consul/client_test.go

@@ -4,9 +4,14 @@ import (
 	"context"
 	"crypto/x509"
 	"crypto/x509/pkix"
+	"encoding/json"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"path/filepath"
 	"strconv"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,6 +22,7 @@ import (
 
 	"github.com/cortexproject/cortex/integration/ca"
 	"github.com/cortexproject/cortex/pkg/ring/kv/codec"
+	"github.com/cortexproject/cortex/pkg/util/backoff"
 	"github.com/cortexproject/cortex/pkg/util/tls"
 )
 
@@ -263,3 +269,157 @@ type testLogger struct {
 func (l testLogger) Log(keyvals ...any) error {
 	return nil
 }
+
+func TestCASBackoff(t *testing.T) {
+	requests := make(chan time.Time, 2)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests <- time.Now()
+		http.Error(w, "unavailable", http.StatusTooManyRequests)
+	}))
+	defer server.Close()
+
+	c, err := NewClient(Config{Host: server.URL, MaxCasRetries: 2}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry())
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	err = c.CAS(ctx, "test", func(any) (any, bool, error) {
+		t.Fatal("callback called after failed read")
+		return nil, false, nil
+	})
+	require.EqualError(t, err, "failed to CAS test")
+	require.Len(t, requests, 2)
+	first, second := <-requests, <-requests
+	require.GreaterOrEqual(t, second.Sub(first), casBackoffConfig.MinBackoff)
+}
+
+// setCASBackoff overrides the CAS backoff for the duration of a test.
+func setCASBackoff(t *testing.T, cfg backoff.Config) {
+	t.Helper()
+	orig := casBackoffConfig
+	casBackoffConfig = cfg
+	t.Cleanup(func() { casBackoffConfig = orig })
+}
+
+// casTestServer returns a fake Consul server whose reads always succeed and
+// whose writes are answered by onWrite. It records the time of every write.
+func casTestServer(t *testing.T, onWrite func(w http.ResponseWriter, n int)) (*httptest.Server, func() []time.Time) {
+	t.Helper()
+	var (
+		mtx    sync.Mutex
+		writes []time.Time
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.Method {
+		case http.MethodGet:
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode([]consul.KVPair{{Key: "test", Value: []byte("value"), ModifyIndex: 1}})
+		case http.MethodPut:
+			mtx.Lock()
+			writes = append(writes, time.Now())
+			n := len(writes)
+			mtx.Unlock()
+			onWrite(w, n)
+		default:
+			http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server, func() []time.Time {
+		mtx.Lock()
+		defer mtx.Unlock()
+		return append([]time.Time(nil), writes...)
+	}
+}
+
+func TestCASBackoffAfterFailedWrites(t *testing.T) {
+	server, writes := casTestServer(t, func(w http.ResponseWriter, _ int) {
+		http.Error(w, "unavailable", http.StatusTooManyRequests)
+	})
+
+	c, err := NewClient(Config{Host: server.URL, MaxCasRetries: 4}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry())
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	calls := 0
+	err = c.CAS(ctx, "test", func(in any) (any, bool, error) {
+		calls++
+		require.Equal(t, "value", in)
+		return "updated", false, nil
+	})
+	require.EqualError(t, err, "failed to CAS test")
+	// Every read succeeds, so every attempt reaches the write.
+	require.Equal(t, 4, calls)
+
+	ts := writes()
+	require.Len(t, ts, 4)
+	// The backoff is not reset by the successful reads in between, so the
+	// minimum delay between writes keeps doubling.
+	minDelay := casBackoffConfig.MinBackoff
+	for i := 1; i < len(ts); i++ {
+		require.GreaterOrEqual(t, ts[i].Sub(ts[i-1]), minDelay, "delay before write %d", i+1)
+		minDelay *= 2
+	}
+}
+
+func TestCASBackoffContextCanceledDuringWait(t *testing.T) {
+	// Make the wait long enough that returning early can only be explained
+	// by the context cancellation.
+	setCASBackoff(t, backoff.Config{MinBackoff: time.Minute, MaxBackoff: time.Minute})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		http.Error(w, "unavailable", http.StatusTooManyRequests)
+	}))
+	defer server.Close()
+
+	c, err := NewClient(Config{Host: server.URL}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry())
+	require.NoError(t, err)
+
+	// Cancel shortly after the first read has failed, while CAS is waiting.
+	go func() {
+		for requests.Load() == 0 {
+			time.Sleep(10 * time.Millisecond)
+		}
+		time.Sleep(200 * time.Millisecond)
+		cancel()
+	}()
+
+	start := time.Now()
+	err = c.CAS(ctx, "test", func(any) (any, bool, error) {
+		t.Fatal("callback called after failed read")
+		return nil, false, nil
+	})
+	require.ErrorIs(t, err, context.Canceled)
+	require.Less(t, time.Since(start), 10*time.Second)
+	require.Equal(t, int32(1), requests.Load())
+}
+
+func TestCASConflictRetriedWithoutBackoff(t *testing.T) {
+	// Any error backoff would make the test exceed its deadline.
+	setCASBackoff(t, backoff.Config{MinBackoff: time.Minute, MaxBackoff: time.Minute})
+
+	const conflicts = 3
+	server, writes := casTestServer(t, func(w http.ResponseWriter, n int) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(n > conflicts)
+	})
+
+	c, err := NewClient(Config{Host: server.URL}, codec.String{}, testLogger{}, prometheus.NewPedanticRegistry())
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	calls := 0
+	err = c.CAS(ctx, "test", func(any) (any, bool, error) {
+		calls++
+		return "updated", false, nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, conflicts+1, calls)
+	require.Len(t, writes(), conflicts+1)
+}