feat: add REDIS_OP_TIMEOUT for per-operation Redis timeout

README.md

@@ -1316,10 +1316,39 @@ The deployment type can be specified with the `REDIS_TYPE` / `REDIS_PERSECOND_TY
 
 ### Connection Timeout
 
-Controls the maximum duration for Redis connection establishment, read operations, and write operations.
-
-1. `REDIS_TIMEOUT`: sets the timeout for Redis connection and I/O operations. Default: `10s`
-1. `REDIS_PERSECOND_TIMEOUT`: sets the timeout for per-second Redis connection and I/O operations. Default: `10s`
+1. `REDIS_TIMEOUT`: sets the dial (connection establishment) timeout used when opening a new
+   Redis connection. It does **not** bound how long an individual command can take once
+   connected: the underlying radix v4 client has no built-in read/write timeout, so a command
+   sent on an established connection is only bounded by the context deadline passed to it (see
+   `REDIS_OP_TIMEOUT` below). Default: `10s`
+1. `REDIS_PERSECOND_TIMEOUT`: same as `REDIS_TIMEOUT`, for the per-second Redis pool. Default: `10s`
+
+### Per-Operation Timeout
+
+Controls how long a single Redis command or pipeline is allowed to park waiting on Redis before
+it is aborted. Unlike `REDIS_TIMEOUT` (dial-only), this bounds the actual command round-trip, so
+it protects against a Redis that is slow, paused, or unreachable *after* the connection has
+already been established, which is the case where `ratelimit` is most exposed: on the hot rate
+limit path, commands are issued with the inbound gRPC request's context, which typically carries
+no deadline of its own.
+
+1. `REDIS_OP_TIMEOUT`: wraps the context used for each Redis command/pipeline with a deadline of
+   this duration. `0` (default) disables this and preserves the current behavior of relying
+   solely on the caller's context. Recommended production value: a small duration such as `50ms`,
+   to bound per-call latency and prevent goroutine buildup during Redis slowness without
+   noticeably affecting p99 latency under healthy conditions.
+1. `REDIS_PERSECOND_OP_TIMEOUT`: same as `REDIS_OP_TIMEOUT`, for the per-second Redis pool.
+   Default: `0`
+
+Note the two timeouts operate on very different time scales: `REDIS_TIMEOUT` is a one-time,
+per-connection cost so it is typically set in seconds, while `REDIS_OP_TIMEOUT` applies to every
+command so it should be set much smaller (tens of milliseconds) to avoid adding latency to normal
+traffic.
+
+When a command exceeds `REDIS_OP_TIMEOUT`, the request fails fast with a Redis error rather than
+returning an over-limit or under-limit response: it is counted in the `redis_error` stat and
+returned to the caller as a gRPC error, the same as any other Redis failure. Expect this stat to
+rise (instead of requests hanging) during Redis slowness once this timeout is enabled.
 
 ### Pool On-Empty Behavior
 

src/redis/cache_impl.go

@@ -22,15 +22,15 @@ func NewRateLimiterCacheImplFromSettings(ctx context.Context, s settings.Setting
 			s.RedisPerSecondType, s.RedisPerSecondUrl, s.RedisPerSecondPoolSize, s.RedisPerSecondPipelineWindow, s.RedisPerSecondPipelineLimit, s.RedisTlsConfig, s.RedisHealthCheckActiveConnection, srv, s.RedisPerSecondTimeout,
 			s.RedisPerSecondPoolOnEmptyBehavior, s.RedisPerSecondSentinelAuth,
 			s.RedisStartupInitialInterval, s.RedisStartupMaxInterval, s.RedisStartupMaxElapsedTime,
-			s.RedisPerSecondClusterPipelineParallelism)
+			s.RedisPerSecondClusterPipelineParallelism, s.RedisPerSecondOpTimeout)
 		closer.Closers = append(closer.Closers, perSecondPool)
 	}
 
 	otherPool := newClientImpl(ctx, srv.Scope().Scope("redis_pool"), s.RedisTls, s.RedisAuth, s.RedisSocketType, s.RedisType, s.RedisUrl, s.RedisPoolSize,
 		s.RedisPipelineWindow, s.RedisPipelineLimit, s.RedisTlsConfig, s.RedisHealthCheckActiveConnection, srv, s.RedisTimeout,
 		s.RedisPoolOnEmptyBehavior, s.RedisSentinelAuth,
 		s.RedisStartupInitialInterval, s.RedisStartupMaxInterval, s.RedisStartupMaxElapsedTime,
-		s.RedisClusterPipelineParallelism)
+		s.RedisClusterPipelineParallelism, s.RedisOpTimeout)
 	closer.Closers = append(closer.Closers, otherPool)
 
 	return NewFixedRateLimitCacheImpl(

src/redis/driver_impl.go

@@ -74,6 +74,11 @@ type clientImpl struct {
 	stats                      poolStats
 	isCluster                  bool
 	clusterPipelineParallelism int
+	// opTimeout, when > 0, bounds how long a single Redis command (DoCmd) or
+	// pipeline (PipeDo) is allowed to park waiting on Redis by wrapping the
+	// context passed to the underlying radix client with a deadline. When 0,
+	// no deadline is added and callers' context governs command duration as before.
+	opTimeout time.Duration
 }
 
 func checkError(err error) {
@@ -147,18 +152,19 @@ func NewClientImpl(ctx context.Context, scope stats.Scope, useTls bool, auth, re
 	pipelineWindow time.Duration, pipelineLimit int, tlsConfig *tls.Config, healthCheckActiveConnection bool, srv server.Server,
 	timeout time.Duration, poolOnEmptyBehavior string, sentinelAuth string,
 	startupInitialInterval, startupMaxInterval, startupMaxElapsedTime time.Duration,
+	opTimeout time.Duration,
 ) Client {
 	return newClientImpl(ctx, scope, useTls, auth, redisSocketType, redisType, url, poolSize,
 		pipelineWindow, pipelineLimit, tlsConfig, healthCheckActiveConnection, srv,
 		timeout, poolOnEmptyBehavior, sentinelAuth,
-		startupInitialInterval, startupMaxInterval, startupMaxElapsedTime, 1)
+		startupInitialInterval, startupMaxInterval, startupMaxElapsedTime, 1, opTimeout)
 }
 
 func newClientImpl(ctx context.Context, scope stats.Scope, useTls bool, auth, redisSocketType, redisType, url string, poolSize int,
 	pipelineWindow time.Duration, pipelineLimit int, tlsConfig *tls.Config, healthCheckActiveConnection bool, srv server.Server,
 	timeout time.Duration, poolOnEmptyBehavior string, sentinelAuth string,
 	startupInitialInterval, startupMaxInterval, startupMaxElapsedTime time.Duration,
-	clusterPipelineParallelism int,
+	clusterPipelineParallelism int, opTimeout time.Duration,
 ) Client {
 	maskedUrl := utils.MaskCredentialsInUrl(url)
 	logger.Warnf("connecting to redis on %s with pool size %d", maskedUrl, poolSize)
@@ -332,11 +338,17 @@ func newClientImpl(ctx context.Context, scope stats.Scope, useTls bool, auth, re
 		stats:                      stats,
 		isCluster:                  isCluster,
 		clusterPipelineParallelism: effectivePipelineParallelism,
+		opTimeout:                  opTimeout,
 	}
 }
 
 func (c *clientImpl) DoCmd(rcv interface{}, cmd, key string, args ...interface{}) error {
 	ctx := context.Background()
+	if c.opTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, c.opTimeout)
+		defer cancel()
+	}
 	// Combine key and args into a single slice
 	allArgs := make([]interface{}, 0, 1+len(args))
 	allArgs = append(allArgs, key)
@@ -364,6 +376,12 @@ func (c *clientImpl) PipeAppend(pipeline Pipeline, rcv interface{}, cmd, key str
 }
 
 func (c *clientImpl) PipeDo(ctx context.Context, pipeline Pipeline) error {
+	if c.opTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, c.opTimeout)
+		defer cancel()
+	}
+
 	if c.isCluster {
 		// Cluster mode: group commands by key and execute each group as a pipeline.
 		// This ensures INCRBY + EXPIRE for the same key are pipelined together (same slot),

src/redis/driver_impl_test.go

@@ -44,11 +44,19 @@ type recordingRedisClient struct {
 	calls       []radix.Action
 	inFlight    int
 	maxInFlight int
+	// blockDelay, when > 0, makes Do block until ctx is done or blockDelay
+	// elapses, regardless of the action type. Used to simulate a Redis
+	// server that never responds (e.g. paused/unreachable).
+	blockDelay time.Duration
+	// lastCtx records the context passed to the most recent Do call, so
+	// tests can assert whether a deadline was attached to it.
+	lastCtx context.Context
 }
 
 func (c *recordingRedisClient) Do(ctx context.Context, action radix.Action) error {
 	c.mu.Lock()
 	c.calls = append(c.calls, action)
+	c.lastCtx = ctx
 	c.inFlight++
 	if c.inFlight > c.maxInFlight {
 		c.maxInFlight = c.inFlight
@@ -61,12 +69,29 @@ func (c *recordingRedisClient) Do(ctx context.Context, action radix.Action) erro
 		c.mu.Unlock()
 	}()
 
+	if c.blockDelay > 0 {
+		timer := time.NewTimer(c.blockDelay)
+		defer timer.Stop()
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-timer.C:
+			return nil
+		}
+	}
+
 	if action, ok := action.(*testAction); ok {
 		return action.Perform(ctx, nil)
 	}
 	return nil
 }
 
+func (c *recordingRedisClient) getLastCtx() context.Context {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.lastCtx
+}
+
 func (c *recordingRedisClient) Close() error {
 	return nil
 }
@@ -219,3 +244,136 @@ func TestExecuteGroupedPipelineBoundedParallelism(t *testing.T) {
 	assert.Equal(t, 3, fakeClient.callCount())
 	assert.Equal(t, 2, fakeClient.maxConcurrentCalls())
 }
+
+// --- opTimeout tests ---
+//
+// These verify that clientImpl.opTimeout bounds how long a single Redis
+// command/pipeline is allowed to park when the underlying Redis connection
+// never responds (simulated via recordingRedisClient.blockDelay), rather
+// than parking indefinitely on the caller's context (which may have no
+// deadline at all, as is the case on the hot DoLimit path today).
+
+func TestPipeDoReturnsWithinOpTimeoutWhenRedisHangs(t *testing.T) {
+	fakeClient := &recordingRedisClient{blockDelay: time.Hour}
+	client := &clientImpl{client: fakeClient, opTimeout: 30 * time.Millisecond}
+
+	start := time.Now()
+	err := client.PipeDo(context.Background(), Pipeline{
+		{Key: "a", Action: &testAction{key: "a"}},
+	})
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Less(t, elapsed, 500*time.Millisecond, "PipeDo should return promptly once opTimeout elapses instead of blocking indefinitely")
+}
+
+func TestDoCmdReturnsWithinOpTimeoutWhenRedisHangs(t *testing.T) {
+	fakeClient := &recordingRedisClient{blockDelay: time.Hour}
+	client := &clientImpl{client: fakeClient, opTimeout: 30 * time.Millisecond}
+
+	start := time.Now()
+	err := client.DoCmd(nil, "GET", "foo")
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Less(t, elapsed, 500*time.Millisecond, "DoCmd should return promptly once opTimeout elapses instead of blocking indefinitely")
+}
+
+func TestPipeDoAppliesOpTimeoutDeadlineToContext(t *testing.T) {
+	fakeClient := &recordingRedisClient{}
+	client := &clientImpl{client: fakeClient, opTimeout: 50 * time.Millisecond}
+
+	err := client.PipeDo(context.Background(), Pipeline{
+		{Key: "a", Action: &testAction{key: "a"}},
+	})
+	require.NoError(t, err)
+
+	ctx := fakeClient.getLastCtx()
+	require.NotNil(t, ctx)
+	deadline, ok := ctx.Deadline()
+	assert.True(t, ok, "expected ctx passed to the underlying client to carry a deadline when opTimeout > 0")
+	assert.True(t, time.Until(deadline) <= 50*time.Millisecond)
+}
+
+func TestPipeDoWithoutOpTimeoutLeavesContextUnbounded(t *testing.T) {
+	fakeClient := &recordingRedisClient{}
+	client := &clientImpl{client: fakeClient, opTimeout: 0}
+
+	err := client.PipeDo(context.Background(), Pipeline{
+		{Key: "a", Action: &testAction{key: "a"}},
+	})
+	require.NoError(t, err)
+
+	ctx := fakeClient.getLastCtx()
+	require.NotNil(t, ctx)
+	_, ok := ctx.Deadline()
+	assert.False(t, ok, "expected ctx to have no deadline when opTimeout == 0 (preserves current behavior)")
+}
+
+func TestDoCmdAppliesOpTimeoutDeadlineToContext(t *testing.T) {
+	fakeClient := &recordingRedisClient{}
+	client := &clientImpl{client: fakeClient, opTimeout: 50 * time.Millisecond}
+
+	err := client.DoCmd(nil, "GET", "foo")
+	require.NoError(t, err)
+
+	ctx := fakeClient.getLastCtx()
+	require.NotNil(t, ctx)
+	deadline, ok := ctx.Deadline()
+	assert.True(t, ok, "expected ctx passed to the underlying client to carry a deadline when opTimeout > 0")
+	assert.True(t, time.Until(deadline) <= 50*time.Millisecond)
+}
+
+func TestDoCmdWithoutOpTimeoutLeavesContextUnbounded(t *testing.T) {
+	fakeClient := &recordingRedisClient{}
+	client := &clientImpl{client: fakeClient, opTimeout: 0}
+
+	err := client.DoCmd(nil, "GET", "foo")
+	require.NoError(t, err)
+
+	ctx := fakeClient.getLastCtx()
+	require.NotNil(t, ctx)
+	_, ok := ctx.Deadline()
+	assert.False(t, ok, "expected ctx to have no deadline when opTimeout == 0 (preserves current behavior, matches context.Background() used today)")
+}
+
+// The opTimeout wrap happens before the isCluster branch in PipeDo, so it
+// should also bound the cluster grouped-pipeline path (executeGroupedPipeline
+// / doPipelineGroup), not just the single/sentinel pipeline. These two tests
+// cover that path explicitly with clusterPipelineParallelism > 1 so multiple
+// keys are grouped and dispatched concurrently via errgroup.
+
+func TestPipeDoClusterReturnsWithinOpTimeoutWhenRedisHangs(t *testing.T) {
+	fakeClient := &recordingRedisClient{blockDelay: time.Hour}
+	client := &clientImpl{client: fakeClient, opTimeout: 30 * time.Millisecond, isCluster: true, clusterPipelineParallelism: 2}
+
+	start := time.Now()
+	err := client.PipeDo(context.Background(), Pipeline{
+		{Key: "a", Action: &testAction{key: "a"}},
+		{Key: "b", Action: &testAction{key: "b"}},
+	})
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Less(t, elapsed, 500*time.Millisecond, "cluster PipeDo should return promptly once opTimeout elapses instead of blocking indefinitely")
+}
+
+func TestPipeDoClusterAppliesOpTimeoutDeadlineToContext(t *testing.T) {
+	fakeClient := &recordingRedisClient{}
+	client := &clientImpl{client: fakeClient, opTimeout: 50 * time.Millisecond, isCluster: true, clusterPipelineParallelism: 2}
+
+	err := client.PipeDo(context.Background(), Pipeline{
+		{Key: "a", Action: &testAction{key: "a"}},
+		{Key: "b", Action: &testAction{key: "b"}},
+	})
+	require.NoError(t, err)
+
+	ctx := fakeClient.getLastCtx()
+	require.NotNil(t, ctx)
+	deadline, ok := ctx.Deadline()
+	assert.True(t, ok, "expected ctx passed to the underlying client to carry a deadline in cluster mode when opTimeout > 0")
+	assert.True(t, time.Until(deadline) <= 50*time.Millisecond)
+}

src/settings/settings.go

@@ -183,10 +183,21 @@ type Settings struct {
 	RedisPerSecondClusterPipelineParallelism int `envconfig:"REDIS_PERSECOND_CLUSTER_PIPELINE_PARALLELISM" default:"1"`
 	// Enable healthcheck to check Redis Connection. If there is no active connection, healthcheck failed.
 	RedisHealthCheckActiveConnection bool `envconfig:"REDIS_HEALTH_CHECK_ACTIVE_CONNECTION" default:"false"`
-	// RedisTimeout sets the timeout for Redis connection and I/O operations.
+	// RedisTimeout sets the dial/connection timeout used when establishing a new Redis
+	// connection (net.Dialer.Timeout). It does NOT bound per-command read/write latency;
+	// once connected, radix v4 has no built-in read/write timeout, so individual commands
+	// are only bounded by the context deadline passed to them. See RedisOpTimeout.
 	RedisTimeout time.Duration `envconfig:"REDIS_TIMEOUT" default:"10s"`
-	// RedisPerSecondTimeout sets the timeout for per-second Redis connection and I/O operations.
+	// RedisPerSecondTimeout sets the dial/connection timeout for the per-second Redis pool.
+	// See RedisTimeout for details; it does not bound per-command latency.
 	RedisPerSecondTimeout time.Duration `envconfig:"REDIS_PERSECOND_TIMEOUT" default:"10s"`
+	// RedisOpTimeout wraps the context passed to each Redis command (DoCmd/PipeDo) with a
+	// deadline of this duration, bounding how long a single call can park when Redis is slow
+	// or unreachable. 0 (default) disables this and preserves prior behavior, where commands
+	// are bounded only by the caller's inbound context (which may have no deadline at all).
+	RedisOpTimeout time.Duration `envconfig:"REDIS_OP_TIMEOUT" default:"0"`
+	// RedisPerSecondOpTimeout is the equivalent of RedisOpTimeout for the per-second Redis pool.
+	RedisPerSecondOpTimeout time.Duration `envconfig:"REDIS_PERSECOND_OP_TIMEOUT" default:"0"`
 
 	// RedisPoolOnEmptyBehavior controls what happens when Redis connection pool is empty.
 	// NOTE: In radix v4, the pool ALWAYS blocks when empty (WAIT behavior).

src/settings/settings_test.go

@@ -3,6 +3,7 @@ package settings
 import (
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -143,3 +144,26 @@ func TestRedisPoolOnEmptyBehavior_IndependentConfiguration(t *testing.T) {
 	// Per-second pool configured differently
 	assert.Equal(t, "CREATE", settings.RedisPerSecondPoolOnEmptyBehavior)
 }
+
+// Tests for RedisOpTimeout / RedisPerSecondOpTimeout
+func TestRedisOpTimeout_Default(t *testing.T) {
+	os.Unsetenv("REDIS_OP_TIMEOUT")
+	os.Unsetenv("REDIS_PERSECOND_OP_TIMEOUT")
+
+	settings := NewSettings()
+
+	assert.Equal(t, time.Duration(0), settings.RedisOpTimeout)
+	assert.Equal(t, time.Duration(0), settings.RedisPerSecondOpTimeout)
+}
+
+func TestRedisOpTimeout_Configured(t *testing.T) {
+	os.Setenv("REDIS_OP_TIMEOUT", "50ms")
+	os.Setenv("REDIS_PERSECOND_OP_TIMEOUT", "25ms")
+	defer os.Unsetenv("REDIS_OP_TIMEOUT")
+	defer os.Unsetenv("REDIS_PERSECOND_OP_TIMEOUT")
+
+	settings := NewSettings()
+
+	assert.Equal(t, 50*time.Millisecond, settings.RedisOpTimeout)
+	assert.Equal(t, 25*time.Millisecond, settings.RedisPerSecondOpTimeout)
+}

test/redis/bench_test.go

@@ -44,7 +44,7 @@ func BenchmarkParallelDoLimit(b *testing.B) {
 		return func(b *testing.B) {
 			statsStore := gostats.NewStore(gostats.NewNullSink(), false)
 			sm := stats.NewMockStatManager(statsStore)
-			client := redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", "127.0.0.1:6379", poolSize, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0)
+			client := redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", "127.0.0.1:6379", poolSize, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0, 0)
 			defer client.Close()
 
 			cache := redis.NewFixedRateLimitCacheImpl(client, nil, utils.NewTimeSourceImpl(), rand.New(utils.NewLockedSource(time.Now().Unix())), 10, nil, 0.8, "", sm, true)