diff --git a/README.md b/README.md index f2e8d6df..041c3c9f 100644 --- a/README.md +++ b/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 diff --git a/src/redis/cache_impl.go b/src/redis/cache_impl.go index f7b7072c..2b402aaa 100644 --- a/src/redis/cache_impl.go +++ b/src/redis/cache_impl.go @@ -22,7 +22,7 @@ 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) } @@ -30,7 +30,7 @@ func NewRateLimiterCacheImplFromSettings(ctx context.Context, s settings.Setting 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( diff --git a/src/redis/driver_impl.go b/src/redis/driver_impl.go index 6b93d02a..f89f3934 100644 --- a/src/redis/driver_impl.go +++ b/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), diff --git a/src/redis/driver_impl_test.go b/src/redis/driver_impl_test.go index 57a6864c..1d75809f 100644 --- a/src/redis/driver_impl_test.go +++ b/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) +} diff --git a/src/settings/settings.go b/src/settings/settings.go index 2129cf80..f0150b88 100644 --- a/src/settings/settings.go +++ b/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). diff --git a/src/settings/settings_test.go b/src/settings/settings_test.go index c6ab2fd0..6b5961b2 100644 --- a/src/settings/settings_test.go +++ b/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) +} diff --git a/test/redis/bench_test.go b/test/redis/bench_test.go index b055dc4e..886633bb 100644 --- a/test/redis/bench_test.go +++ b/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) diff --git a/test/redis/driver_impl_test.go b/test/redis/driver_impl_test.go index ecfa5f69..4c00e8fb 100644 --- a/test/redis/driver_impl_test.go +++ b/test/redis/driver_impl_test.go @@ -41,7 +41,7 @@ func testNewClientImpl(t *testing.T, pipelineWindow time.Duration, pipelineLimit // Use a short maxElapsedTime so failing connection tests don't hang in retry loops. mkRedisClient := func(auth, addr string) redis.Client { - return redis.NewClientImpl(context.Background(), statsStore, false, auth, "tcp", "single", addr, 1, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 100*time.Millisecond) + return redis.NewClientImpl(context.Background(), statsStore, false, auth, "tcp", "single", addr, 1, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 100*time.Millisecond, 0) } t.Run("connection refused", func(t *testing.T) { @@ -119,7 +119,7 @@ func TestDoCmd(t *testing.T) { statsStore := stats.NewStore(stats.NewNullSink(), false) mkRedisClient := func(addr string) redis.Client { - return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, 0, 0, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0) + return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, 0, 0, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0, 0) } t.Run("SETGET ok", func(t *testing.T) { @@ -164,7 +164,7 @@ func testPipeDo(t *testing.T, pipelineWindow time.Duration, pipelineLimit int) f statsStore := stats.NewStore(stats.NewNullSink(), false) mkRedisClient := func(addr string) redis.Client { - return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0) + return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, pipelineWindow, pipelineLimit, nil, false, nil, 10*time.Second, "", "", time.Second, 30*time.Second, 0, 0) } t.Run("SETGET ok", func(t *testing.T) { @@ -232,7 +232,7 @@ func TestPoolOnEmptyBehavior(t *testing.T) { // Helper to create client with specific on-empty behavior mkRedisClientWithBehavior := func(addr, behavior string) redis.Client { - return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, 0, 0, nil, false, nil, 10*time.Second, behavior, "", time.Second, 30*time.Second, 0) + return redis.NewClientImpl(context.Background(), statsStore, false, "", "tcp", "single", addr, 1, 0, 0, nil, false, nil, 10*time.Second, behavior, "", time.Second, 30*time.Second, 0, 0) } t.Run("default behavior (empty string)", func(t *testing.T) { @@ -357,7 +357,7 @@ func TestNewClientImplSentinel(t *testing.T) { // Pass nil for tlsConfig - we can't test TLS without a real TLS server, // but we can verify the code path is executed (logs will show TLS is enabled) // Use a short maxElapsedTime so failing connection tests don't hang in retry loops. - return redis.NewClientImpl(context.Background(), statsStore, useTls, auth, "tcp", "sentinel", url, 1, 0, 0, nil, false, nil, timeout, "", sentinelAuth, time.Second, 30*time.Second, 100*time.Millisecond) + return redis.NewClientImpl(context.Background(), statsStore, useTls, auth, "tcp", "sentinel", url, 1, 0, 0, nil, false, nil, timeout, "", sentinelAuth, time.Second, 30*time.Second, 100*time.Millisecond, 0) } t.Run("invalid url format - missing sentinel addresses", func(t *testing.T) {