mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-20 05:56:56 +00:00
perf(resp): pool redis connections instead of one mutex-guarded conn
respClient held a single net.Conn behind one mutex, serializing every caller's whole round trip. Interpreted-harness audit measured this as the throughput ceiling: 1,458 req/s Redis-backed vs 7,811 req/s on memoryStore for the same workload, latency scaling near-linearly with concurrency (869us@conc=1 -> 10.81ms@conc=16). Replace the single connection with a pool of independently usable connections (respConn slots in a buffered channel), each acquired for one caller's whole round trip and returned via a deferred send back onto the pool. A broken connection is discarded (closeConn) but its slot always returns to the pool, so pool size never shrinks after an error. Default size is GOMAXPROCS clamped to [4, 8] (defaultRespPoolSize); RedisConfig.PoolSize lets an operator override it explicitly, per the house rule that an explicit override always wins over self-tuning. Preserves exactly: the 2s respCallTimeout call deadline, reconnect- once-on-error semantics, nested-array rejection, the RESP size bounds, and every existing pipelining contract (incrMulti, incrAndGetMulti, getMulti, cache get/set) — nothing outside resp.go references a connection field directly. Interpreted-harness prototype (variants/resp_pool8.go, pool=8) measured at a realistic 200us Redis RTT, conc=16: p50 24.06ms -> 3.24ms (-86%), throughput 663 -> 4,670 req/s (+599%, +-1% over 3 reps); a pool=1 control matched HEAD exactly, confirming the win is parallelism, not refactor noise. cache_test.go and llmgateway_test.go: adjust white-box tests that issue multiple sequential calls against one respClient scripted with runFakeRESPServer's single-connection model — force pool size 1 there (newRESPClientPool) so the test still pins one connection's exact wire sequence; production callers are unaffected.
This commit is contained in:
+15
-2
@@ -301,9 +301,19 @@ func TestBuildResponseCache_EnabledWithRedis_ReturnsWorkingCache(t *testing.T) {
|
|||||||
// newTestResponseCache builds a responseCache over ln's fake server, with
|
// newTestResponseCache builds a responseCache over ln's fake server, with
|
||||||
// nowFn pinned to a fixed instant so logCacheError's rate-limit window is
|
// nowFn pinned to a fixed instant so logCacheError's rate-limit window is
|
||||||
// deterministic. logged collects every logf call's formatted message.
|
// deterministic. logged collects every logf call's formatted message.
|
||||||
|
//
|
||||||
|
// Pool size 1 (newRESPClientPool, not newRESPClient's self-tuned
|
||||||
|
// default): a caller issuing more than one lookup/store against the
|
||||||
|
// returned cache expects every command on the SAME connection, matching
|
||||||
|
// runFakeRESPServer's strict one-connection-at-a-time script — with the
|
||||||
|
// default pool (size >1, finding 1), a second call could acquire a
|
||||||
|
// different, not-yet-dialled slot instead of reusing the first
|
||||||
|
// connection, sending a second SELECT the fixed script never expects.
|
||||||
|
// Production callers (llmgateway.go's buildRedisClient) are unaffected —
|
||||||
|
// they always go through newRESPClient's own default.
|
||||||
func newTestResponseCache(t *testing.T, ln net.Listener) (c *responseCache, logged *[]string) {
|
func newTestResponseCache(t *testing.T, ln net.Listener) (c *responseCache, logged *[]string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
client := newRESPClient(ln.Addr().String(), "", 0)
|
client := newRESPClientPool(ln.Addr().String(), "", 0, 1)
|
||||||
msgs := []string{}
|
msgs := []string{}
|
||||||
spy := func(format string, args ...any) { msgs = append(msgs, fmt.Sprintf(format, args...)) }
|
spy := func(format string, args ...any) { msgs = append(msgs, fmt.Sprintf(format, args...)) }
|
||||||
c = newResponseCache(client, time.Minute, defaultCacheMaxBodyBytes, spy, spy)
|
c = newResponseCache(client, time.Minute, defaultCacheMaxBodyBytes, spy, spy)
|
||||||
@@ -678,7 +688,10 @@ func TestResponseCache_Store_EffectiveTTLDiffersPerGroup_SETCarriesGroupTTL(t *t
|
|||||||
{wantArgs: []string{"SET", "k-override", string(val1), "EX", "30"}, reply: []byte("+OK\r\n")},
|
{wantArgs: []string{"SET", "k-override", string(val1), "EX", "30"}, reply: []byte("+OK\r\n")},
|
||||||
{wantArgs: []string{"SET", "k-inherit", string(val2), "EX", "60"}, reply: []byte("+OK\r\n")},
|
{wantArgs: []string{"SET", "k-inherit", string(val2), "EX", "60"}, reply: []byte("+OK\r\n")},
|
||||||
})
|
})
|
||||||
client := newRESPClient(ln.Addr().String(), "", 0)
|
// Pool size 1 — see newTestResponseCache's own doc comment: two
|
||||||
|
// sequential store() calls below must land on runFakeRESPServer's one
|
||||||
|
// scripted connection, not cycle across separate pool slots.
|
||||||
|
client := newRESPClientPool(ln.Addr().String(), "", 0, 1)
|
||||||
noop := func(string, ...any) {}
|
noop := func(string, ...any) {}
|
||||||
c := newResponseCache(client, time.Minute, defaultCacheMaxBodyBytes, noop, noop) // global ttl = 60s
|
c := newResponseCache(client, time.Minute, defaultCacheMaxBodyBytes, noop, noop) // global ttl = 60s
|
||||||
c.nowFn = time.Now
|
c.nowFn = time.Now
|
||||||
|
|||||||
+63
-1
@@ -196,6 +196,19 @@ type RedisConfig struct {
|
|||||||
Address string `json:"address"`
|
Address string `json:"address"`
|
||||||
Password string `json:"password,omitempty"`
|
Password string `json:"password,omitempty"`
|
||||||
DB int `json:"db,omitempty"`
|
DB int `json:"db,omitempty"`
|
||||||
|
// PoolSize overrides respClient's self-tuned connection pool size
|
||||||
|
// (defaultRespPoolSize, resp.go: GOMAXPROCS clamped to [4, 8]) — perf
|
||||||
|
// finding 1, 2026-08-2x audit. Left at 0 (the default), the pool
|
||||||
|
// self-tunes; an operator who sets this explicitly always wins over
|
||||||
|
// that auto-tuning (house engineering rule), e.g. to open more
|
||||||
|
// connections than a low-GOMAXPROCS container would otherwise pick
|
||||||
|
// for a Redis known to have plenty of headroom, or fewer against a
|
||||||
|
// connection-constrained managed Redis tier. A negative value, or one
|
||||||
|
// above respPoolConfigMax (resp.go), is a construction error
|
||||||
|
// (buildRedisClient, below) — negative can never mean "no pool", and
|
||||||
|
// an unbounded value would eagerly allocate that many *respConn
|
||||||
|
// structs at construction before a single request arrives.
|
||||||
|
PoolSize int `json:"poolSize,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// RetryConfig configures same-provider retry on transient upstream
|
// RetryConfig configures same-provider retry on transient upstream
|
||||||
@@ -309,7 +322,14 @@ type Gateway struct {
|
|||||||
// before using it, rather than responseCache having its own
|
// before using it, rather than responseCache having its own
|
||||||
// always-disabled zero value.
|
// always-disabled zero value.
|
||||||
cache *responseCache
|
cache *responseCache
|
||||||
name string
|
// redisClient is the same instance newGateway hands to both the
|
||||||
|
// limiter's redisStore and the response cache (its own doc comment,
|
||||||
|
// below, explains why it's built once and shared) — kept here too,
|
||||||
|
// on Gateway itself, purely so Close (below) has something to release
|
||||||
|
// pooled connections through. nil when config.Redis is absent, same
|
||||||
|
// as buildRedisClient's own nil-for-unconfigured contract.
|
||||||
|
redisClient *respClient
|
||||||
|
name string
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates the middleware. NOTE: no tail call — Yaegi zeroes
|
// New creates the middleware. NOTE: no tail call — Yaegi zeroes
|
||||||
@@ -322,6 +342,35 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
|
|||||||
return g, nil
|
return g, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Close releases resources this Gateway instance holds open: currently,
|
||||||
|
// the pooled Redis connections behind redisClient (perf finding 1),
|
||||||
|
// shared between the limiter's redisStore and the response cache — a
|
||||||
|
// nil-safe no-op when Redis is not configured.
|
||||||
|
//
|
||||||
|
// Review fix (Should-Fix 4, 2026-08-2x): nothing in this codebase calls
|
||||||
|
// Close automatically. Traefik's plugin contract for a locally-loaded
|
||||||
|
// middleware is exactly `func New(...) (http.Handler, error)`, above —
|
||||||
|
// the returned value satisfies only http.Handler, with no Shutdown/Close
|
||||||
|
// interface Traefik itself checks for or invokes on a discarded
|
||||||
|
// instance. README.md's "Config hot-reload" section already documents
|
||||||
|
// this repository has no evidence of Traefik's teardown mechanism for an
|
||||||
|
// old instance at config-reload swap time; a plugin-side Close, however
|
||||||
|
// correct, has nowhere in Traefik's own contract to be wired to. Close
|
||||||
|
// exists for direct Go callers — tests, or a future framework/harness
|
||||||
|
// that DOES hold such a hook — to release pooled connections
|
||||||
|
// deterministically instead of depending solely on GC finalizing the
|
||||||
|
// underlying net.Conns. A discarded, never-Closed Gateway still leaks at
|
||||||
|
// most respPoolMax (8) idle connections rather than the pre-pool single
|
||||||
|
// connection — a real but bounded regression the pool accepts (perf
|
||||||
|
// finding 1), because nothing in Traefik's own contract gives this
|
||||||
|
// plugin a way to do better.
|
||||||
|
func (g *Gateway) Close() error {
|
||||||
|
if g.redisClient == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return g.redisClient.Close()
|
||||||
|
}
|
||||||
|
|
||||||
func newGateway(ctx context.Context, next http.Handler, config *Config, name string) (*Gateway, error) {
|
func newGateway(ctx context.Context, next http.Handler, config *Config, name string) (*Gateway, error) {
|
||||||
if config == nil || len(config.Providers) == 0 {
|
if config == nil || len(config.Providers) == 0 {
|
||||||
return nil, errors.New("llmgateway: at least one provider must be configured")
|
return nil, errors.New("llmgateway: at least one provider must be configured")
|
||||||
@@ -351,6 +400,7 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
g.redisClient = redisClient
|
||||||
lim := newConfiguredLimiter(config, redisClient)
|
lim := newConfiguredLimiter(config, redisClient)
|
||||||
lim.logf = g.errorf
|
lim.logf = g.errorf
|
||||||
g.limiter = lim
|
g.limiter = lim
|
||||||
@@ -410,12 +460,24 @@ func buildRedisClient(rc *RedisConfig) (*respClient, error) {
|
|||||||
if rc.DB < 0 {
|
if rc.DB < 0 {
|
||||||
return nil, fmt.Errorf("llmgateway: redis: db must not be negative, got %d", rc.DB)
|
return nil, fmt.Errorf("llmgateway: redis: db must not be negative, got %d", rc.DB)
|
||||||
}
|
}
|
||||||
|
if rc.PoolSize < 0 {
|
||||||
|
return nil, fmt.Errorf("llmgateway: redis: poolSize must not be negative, got %d", rc.PoolSize)
|
||||||
|
}
|
||||||
|
if rc.PoolSize > respPoolConfigMax {
|
||||||
|
return nil, fmt.Errorf("llmgateway: redis: poolSize must not exceed %d, got %d", respPoolConfigMax, rc.PoolSize)
|
||||||
|
}
|
||||||
|
|
||||||
password, err := resolveSecret(rc.Password)
|
password, err := resolveSecret(rc.Password)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("llmgateway: redis: %w", err)
|
return nil, fmt.Errorf("llmgateway: redis: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PoolSize left at 0 keeps newRESPClient's own self-tuned default
|
||||||
|
// (defaultRespPoolSize, resp.go); an explicit positive value always
|
||||||
|
// overrides it (RedisConfig.PoolSize's own doc comment).
|
||||||
|
if rc.PoolSize > 0 {
|
||||||
|
return newRESPClientPool(rc.Address, password, rc.DB, rc.PoolSize), nil
|
||||||
|
}
|
||||||
return newRESPClient(rc.Address, password, rc.DB), nil
|
return newRESPClient(rc.Address, password, rc.DB), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -604,6 +604,57 @@ func TestNewGateway_RedisConfigured_WiresRedisStoreAndFailOpen(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestGateway_Close_NoRedis_NilSafeNoOp is the review fix (Should-Fix 4,
|
||||||
|
// 2026-08-2x): a Gateway built with no Redis configured has a nil
|
||||||
|
// redisClient, and Close must be a nil-safe no-op, not a panic.
|
||||||
|
func TestGateway_Close_NoRedis_NilSafeNoOp(t *testing.T) {
|
||||||
|
cfg := CreateConfig()
|
||||||
|
cfg.Providers = map[string]*ProviderConfig{"openai": {Type: "openai", APIKey: "k"}}
|
||||||
|
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
|
||||||
|
h, err := New(context.Background(), next, cfg, "llmgw")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("New: %v", err)
|
||||||
|
}
|
||||||
|
gw := h.(*Gateway)
|
||||||
|
if gw.redisClient != nil {
|
||||||
|
t.Fatal("gw.redisClient != nil, want nil (no Redis configured)")
|
||||||
|
}
|
||||||
|
if err := gw.Close(); err != nil {
|
||||||
|
t.Errorf("Close: %v, want nil", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGateway_Close_WithRedis_DrainsPooledConnections is the review fix
|
||||||
|
// (Should-Fix 4, 2026-08-2x) end-to-end proof: a Gateway built with Redis
|
||||||
|
// configured wires redisClient, and Close drains its connection pool
|
||||||
|
// (every slot, dialled or not — construction never eagerly dials, so
|
||||||
|
// this exercises Close's drain path rather than a real socket close, but
|
||||||
|
// proves the wiring: gw.Close() actually reaches gw.redisClient.Close(),
|
||||||
|
// not a stub that does nothing).
|
||||||
|
func TestGateway_Close_WithRedis_DrainsPooledConnections(t *testing.T) {
|
||||||
|
cfg := CreateConfig()
|
||||||
|
cfg.Providers = map[string]*ProviderConfig{"openai": {Type: "openai", APIKey: "k"}}
|
||||||
|
cfg.Redis = &RedisConfig{Address: "127.0.0.1:0", PoolSize: 3}
|
||||||
|
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
|
||||||
|
h, err := New(context.Background(), next, cfg, "llmgw")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("New: %v", err)
|
||||||
|
}
|
||||||
|
gw := h.(*Gateway)
|
||||||
|
if gw.redisClient == nil {
|
||||||
|
t.Fatal("gw.redisClient = nil, want a non-nil client (Redis configured)")
|
||||||
|
}
|
||||||
|
if got := len(gw.redisClient.free); got != 3 {
|
||||||
|
t.Fatalf("len(redisClient.free) before Close = %d, want 3", got)
|
||||||
|
}
|
||||||
|
if err := gw.Close(); err != nil {
|
||||||
|
t.Fatalf("Close: %v", err)
|
||||||
|
}
|
||||||
|
if got := len(gw.redisClient.free); got != 0 {
|
||||||
|
t.Errorf("len(redisClient.free) after Close = %d, want 0 (drained)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestNewGateway_RedisFailOpenNilDefaultsTrue asserts a configured Redis
|
// TestNewGateway_RedisFailOpenNilDefaultsTrue asserts a configured Redis
|
||||||
// block with FailOpen left nil defaults to true, matching the no-Redis
|
// block with FailOpen left nil defaults to true, matching the no-Redis
|
||||||
// default.
|
// default.
|
||||||
@@ -649,6 +700,71 @@ func TestNewGateway_RedisConfigured_NegativeDB_ReturnsConstructorError(t *testin
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestBuildRedisClient_NegativePoolSize_ReturnsConstructorError mirrors
|
||||||
|
// the negative-DB case above: RedisConfig.PoolSize must fail construction
|
||||||
|
// immediately, same as an empty address or a negative db, rather than
|
||||||
|
// reaching newRESPClientPool with a value it would otherwise silently
|
||||||
|
// clamp up to 1.
|
||||||
|
func TestBuildRedisClient_NegativePoolSize_ReturnsConstructorError(t *testing.T) {
|
||||||
|
_, err := buildRedisClient(&RedisConfig{Address: "127.0.0.1:0", PoolSize: -1})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("want constructor error for a negative Redis poolSize")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestBuildRedisClient_PoolSizeAboveCeiling_ReturnsConstructorError is the
|
||||||
|
// review fix (Should-Fix 1, 2026-08-2x): PoolSize had no upper bound —
|
||||||
|
// PoolSize: 1<<20 was accepted and eagerly allocated over a million
|
||||||
|
// *respConn structs at construction. A value above respPoolConfigMax must
|
||||||
|
// fail construction immediately, mirroring the negative-poolSize check
|
||||||
|
// above.
|
||||||
|
func TestBuildRedisClient_PoolSizeAboveCeiling_ReturnsConstructorError(t *testing.T) {
|
||||||
|
_, err := buildRedisClient(&RedisConfig{Address: "127.0.0.1:0", PoolSize: respPoolConfigMax + 1})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("want constructor error for a Redis poolSize above respPoolConfigMax")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestBuildRedisClient_PoolSizeAtCeiling_Allowed asserts the ceiling
|
||||||
|
// itself is inclusive — respPoolConfigMax is a valid, accepted value, not
|
||||||
|
// an off-by-one rejection.
|
||||||
|
func TestBuildRedisClient_PoolSizeAtCeiling_Allowed(t *testing.T) {
|
||||||
|
client, err := buildRedisClient(&RedisConfig{Address: "127.0.0.1:0", PoolSize: respPoolConfigMax})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("buildRedisClient: %v, want no error at exactly respPoolConfigMax", err)
|
||||||
|
}
|
||||||
|
if got := cap(client.free); got != respPoolConfigMax {
|
||||||
|
t.Errorf("cap(client.free) = %d, want %d", got, respPoolConfigMax)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestBuildRedisClient_PoolSize_ZeroUsesSelfTunedDefault_ExplicitOverrides
|
||||||
|
// is perf finding 1's config-wiring proof: RedisConfig.PoolSize left at 0
|
||||||
|
// gets defaultRespPoolSize's own self-tuned pool, but any explicit
|
||||||
|
// positive value always overrides it — the house engineering rule
|
||||||
|
// (self-tuning over operator knobs, explicit override always wins).
|
||||||
|
func TestBuildRedisClient_PoolSize_ZeroUsesSelfTunedDefault_ExplicitOverrides(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
poolSize int
|
||||||
|
wantPoolSize int
|
||||||
|
}{
|
||||||
|
{name: "zero uses the self-tuned default", poolSize: 0, wantPoolSize: defaultRespPoolSize()},
|
||||||
|
{name: "explicit override wins", poolSize: 3, wantPoolSize: 3},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
t.Run(c.name, func(t *testing.T) {
|
||||||
|
client, err := buildRedisClient(&RedisConfig{Address: "127.0.0.1:0", PoolSize: c.poolSize})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("buildRedisClient: %v", err)
|
||||||
|
}
|
||||||
|
if got := cap(client.free); got != c.wantPoolSize {
|
||||||
|
t.Errorf("cap(client.free) = %d, want %d", got, c.wantPoolSize)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestNewGateway_RedisPassword_ResolvedViaSecret is an end-to-end proof
|
// TestNewGateway_RedisPassword_ResolvedViaSecret is an end-to-end proof
|
||||||
// that newGateway resolves Config.Redis.Password through resolveSecret
|
// that newGateway resolves Config.Redis.Password through resolveSecret
|
||||||
// (env:/file:/literal) before handing it to respClient: an "env:" password
|
// (env:/file:/literal) before handing it to respClient: an "env:" password
|
||||||
|
|||||||
@@ -6,9 +6,9 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
|
"runtime"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -20,9 +20,7 @@ import (
|
|||||||
// conn.SetDeadline, not reset before each individual read or write — a
|
// conn.SetDeadline, not reset before each individual read or write — a
|
||||||
// server that accepts the connection and then never answers must cost
|
// server that accepts the connection and then never answers must cost
|
||||||
// this one call ~respCallTimeout, not respCallTimeout multiplied by every
|
// this one call ~respCallTimeout, not respCallTimeout multiplied by every
|
||||||
// handshake and command step it happens to perform (which, with the
|
// handshake and command step it happens to perform.
|
||||||
// mutex serializing every caller behind one connection, previously turned
|
|
||||||
// one hung server into a many-times-respCallTimeout stall per request).
|
|
||||||
const (
|
const (
|
||||||
respDialTimeout = 2 * time.Second
|
respDialTimeout = 2 * time.Second
|
||||||
respCallTimeout = 2 * time.Second
|
respCallTimeout = 2 * time.Second
|
||||||
@@ -48,6 +46,63 @@ const respMaxArrayLen = 1024
|
|||||||
// bytes before erroring.
|
// bytes before erroring.
|
||||||
const respMaxLineLen = 64 << 10
|
const respMaxLineLen = 64 << 10
|
||||||
|
|
||||||
|
// respPoolMin/respPoolMax bound the self-tuned connection pool size
|
||||||
|
// (house engineering rule: self-tuning over operator knobs, explicit
|
||||||
|
// override always wins — see RedisConfig.PoolSize, llmgateway.go's
|
||||||
|
// buildRedisClient). A single mutex-guarded connection was measured to be
|
||||||
|
// THE throughput ceiling under Yaegi interpretation (perf audit,
|
||||||
|
// 2026-08-2x): 1,458 req/s Redis-backed vs 7,811 req/s on memoryStore for
|
||||||
|
// the identical workload, latency scaling ~linearly with concurrency
|
||||||
|
// (869us@conc=1 -> 10.81ms@conc=16) — near-perfect serialization, since
|
||||||
|
// pipeline held c.mu for the whole round trip. A pool of independent
|
||||||
|
// connections, each handed to exactly one caller at a time, lets N
|
||||||
|
// callers run N round trips concurrently instead of queueing behind one.
|
||||||
|
// respPoolMax=8 is the exact size the audit's variants/resp_pool8.go
|
||||||
|
// prototype validated: at a realistic 200us Redis RTT, conc=16, p50
|
||||||
|
// 24.06ms -> 3.24ms (-86%), throughput 663 -> 4,670 req/s (+599%, ±1%
|
||||||
|
// over 3 reps); a pool=1 control matched HEAD exactly, proving the win is
|
||||||
|
// parallelism, not refactor noise. respPoolMin=4 keeps a low-core-count
|
||||||
|
// replica (GOMAXPROCS clamps low under a small CPU request/limit) from
|
||||||
|
// opening fewer connections than there are usable goroutine-scheduling
|
||||||
|
// slots to fill them.
|
||||||
|
const (
|
||||||
|
respPoolMin = 4
|
||||||
|
respPoolMax = 8
|
||||||
|
)
|
||||||
|
|
||||||
|
// respPoolConfigMax bounds RedisConfig.PoolSize (llmgateway.go's
|
||||||
|
// buildRedisClient rejects anything above it outright, mirroring the
|
||||||
|
// existing negative-DB/negative-poolSize checks): an unbounded override
|
||||||
|
// would let PoolSize: 1<<20 eagerly allocate over a million *respConn
|
||||||
|
// structs at construction (verified in review, 2026-08-2x) before a
|
||||||
|
// single request ever arrives. 256 is far beyond any real single-instance
|
||||||
|
// Redis/Dragonfly connection budget while staying well above
|
||||||
|
// respPoolMax, leaving room for a legitimately high explicit override.
|
||||||
|
const respPoolConfigMax = 256
|
||||||
|
|
||||||
|
// defaultRespPoolSize returns the self-tuned connection pool size: the
|
||||||
|
// host's GOMAXPROCS, clamped to [respPoolMin, respPoolMax]. Used whenever
|
||||||
|
// RedisConfig.PoolSize is left at its zero value; an operator who sets it
|
||||||
|
// explicitly always overrides this (newRESPClientPool, below).
|
||||||
|
func defaultRespPoolSize() int {
|
||||||
|
return clampPoolSize(runtime.GOMAXPROCS(0))
|
||||||
|
}
|
||||||
|
|
||||||
|
// clampPoolSize clamps n to [respPoolMin, respPoolMax]. Split out of
|
||||||
|
// defaultRespPoolSize so the clamp itself is testable against every
|
||||||
|
// branch directly, independent of the test process's own real
|
||||||
|
// GOMAXPROCS.
|
||||||
|
func clampPoolSize(n int) int {
|
||||||
|
switch {
|
||||||
|
case n < respPoolMin:
|
||||||
|
return respPoolMin
|
||||||
|
case n > respPoolMax:
|
||||||
|
return respPoolMax
|
||||||
|
default:
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// respErr is a RESP2 error reply ("-message\r\n"). It implements error so a
|
// respErr is a RESP2 error reply ("-message\r\n"). It implements error so a
|
||||||
// caller inspecting a decoded reply can type-assert for it the same way as
|
// caller inspecting a decoded reply can type-assert for it the same way as
|
||||||
// any other error.
|
// any other error.
|
||||||
@@ -56,7 +111,7 @@ const respMaxLineLen = 64 << 10
|
|||||||
// string"): yaegi v0.16.1 conflates a defined string type with plain
|
// string"): yaegi v0.16.1 conflates a defined string type with plain
|
||||||
// string in a type assertion — v.(respErr) on an any holding a genuine
|
// string in a type assertion — v.(respErr) on an any holding a genuine
|
||||||
// plain string (e.g. decodeReply's "+OK" success reply, itself a string)
|
// plain string (e.g. decodeReply's "+OK" success reply, itself a string)
|
||||||
// incorrectly reports ok=true, so every successful handshakeLocked call
|
// incorrectly reports ok=true, so every successful handshakeOn call
|
||||||
// (AUTH/SELECT) was misread as a failed one under real Traefik. A struct
|
// (AUTH/SELECT) was misread as a failed one under real Traefik. A struct
|
||||||
// has no such ambiguity: verified empirically under real Traefik (Task
|
// has no such ambiguity: verified empirically under real Traefik (Task
|
||||||
// 15's integration suite, cross-replica rate limiting via Redis).
|
// 15's integration suite, cross-replica rate limiting via Redis).
|
||||||
@@ -65,33 +120,98 @@ type respErr struct{ msg string }
|
|||||||
// Error implements the error interface.
|
// Error implements the error interface.
|
||||||
func (e respErr) Error() string { return e.msg }
|
func (e respErr) Error() string { return e.msg }
|
||||||
|
|
||||||
|
// respConn is one connection in respClient's pool: a lazily dialled TCP
|
||||||
|
// connection plus its buffered reader, or the zero value for a slot that
|
||||||
|
// has never dialled yet (or whose previous connection was discarded after
|
||||||
|
// an error — see closeConn). Owned by exactly one caller at a time,
|
||||||
|
// between an acquire and its matching release back onto respClient.free.
|
||||||
|
type respConn struct {
|
||||||
|
conn net.Conn
|
||||||
|
r *bufio.Reader
|
||||||
|
}
|
||||||
|
|
||||||
// respClient is a minimal stdlib-only RESP2 client for a single Redis-
|
// respClient is a minimal stdlib-only RESP2 client for a single Redis-
|
||||||
// compatible server: one mutex-guarded TCP connection, lazily dialled on
|
// compatible server: a pool of independently usable TCP connections, each
|
||||||
// first use and reconnected at most once when a command round trip fails
|
// lazily dialled on first use and reconnected at most once when a command
|
||||||
// for a reason other than a timeout (see pipeline). It exists because the
|
// round trip fails for a reason other than a timeout (see pipeline). It
|
||||||
// plugin runs interpreted under Yaegi with only the Go standard library
|
// exists because the plugin runs interpreted under Yaegi with only the Go
|
||||||
// available, so a full Redis client library is not an option.
|
// standard library available, so a full Redis client library is not an
|
||||||
|
// option.
|
||||||
//
|
//
|
||||||
// mu serializes every operation — do and pipeline hold it for their whole
|
// free holds every currently-idle connection slot; a caller acquires one
|
||||||
// round trip, including any reconnect — since RESP is not multiplexed:
|
// (blocking, bounded by its call deadline, if the pool is momentarily
|
||||||
// interleaving two callers' writes on one connection would corrupt both
|
// exhausted) for its whole round trip and returns it via a deferred send
|
||||||
// replies.
|
// back onto free — RESP is not multiplexed, so two callers interleaving
|
||||||
|
// writes on the SAME connection would corrupt both replies, but distinct
|
||||||
|
// connections have no such restriction and can run fully concurrently.
|
||||||
type respClient struct {
|
type respClient struct {
|
||||||
conn net.Conn
|
free chan *respConn
|
||||||
r *bufio.Reader
|
|
||||||
addr string
|
addr string
|
||||||
password string
|
password string
|
||||||
mu sync.Mutex
|
|
||||||
db int
|
db int
|
||||||
}
|
}
|
||||||
|
|
||||||
// newRESPClient returns a respClient for addr. It does not connect until
|
// newRESPClient returns a respClient for addr, sized by defaultRespPoolSize
|
||||||
// the first do or pipeline call. password, if non-empty, is sent via AUTH
|
// — the self-tuned pool size used everywhere an explicit override
|
||||||
// on every new connection; db is always sent via SELECT on every new
|
// (RedisConfig.PoolSize) is absent. It does not connect until the first
|
||||||
// connection (including db 0), since a Redis-compatible proxy is not
|
// do or pipeline call. password, if non-empty, is sent via AUTH on every
|
||||||
// guaranteed to default a fresh connection to database 0.
|
// new connection; db is always sent via SELECT on every new connection
|
||||||
|
// (including db 0), since a Redis-compatible proxy is not guaranteed to
|
||||||
|
// default a fresh connection to database 0.
|
||||||
func newRESPClient(addr, password string, db int) *respClient {
|
func newRESPClient(addr, password string, db int) *respClient {
|
||||||
return &respClient{addr: addr, password: password, db: db}
|
return newRESPClientPool(addr, password, db, defaultRespPoolSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
// newRESPClientPool is newRESPClient with an explicit poolSize, used by
|
||||||
|
// llmgateway.go's buildRedisClient when RedisConfig.PoolSize is set
|
||||||
|
// (house rule: an explicit override always wins over the self-tuned
|
||||||
|
// default). poolSize below 1 is clamped to 1 — a zero or negative value
|
||||||
|
// would leave free empty, making every acquire block forever.
|
||||||
|
func newRESPClientPool(addr, password string, db, poolSize int) *respClient {
|
||||||
|
if poolSize < 1 {
|
||||||
|
poolSize = 1
|
||||||
|
}
|
||||||
|
c := &respClient{
|
||||||
|
addr: addr,
|
||||||
|
password: password,
|
||||||
|
db: db,
|
||||||
|
free: make(chan *respConn, poolSize),
|
||||||
|
}
|
||||||
|
for i := 0; i < poolSize; i++ {
|
||||||
|
c.free <- &respConn{}
|
||||||
|
}
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close closes every currently idle pooled connection (a non-blocking
|
||||||
|
// receive per slot, so it never waits on one still checked out by an
|
||||||
|
// in-flight caller — the same "don't call Close concurrently with
|
||||||
|
// in-flight use" contract any io.Closer implicitly carries) and always
|
||||||
|
// returns nil — closeConn already discards each connection's own Close
|
||||||
|
// error, matching every other close in this client.
|
||||||
|
//
|
||||||
|
// Review fix (Should-Fix 4, 2026-08-2x): the pool (perf finding 1) turned
|
||||||
|
// a discarded-instance leak of one idle connection into up to
|
||||||
|
// respPoolMax (8). Nothing in this codebase calls Close automatically —
|
||||||
|
// Traefik's plugin contract for a locally-loaded middleware is exactly
|
||||||
|
// `func New(...) (http.Handler, error)` (llmgateway.go); the returned
|
||||||
|
// value satisfies only http.Handler, with no Shutdown/Close interface
|
||||||
|
// Traefik itself checks for or invokes on a discarded instance (see
|
||||||
|
// Gateway.Close's own doc comment for the full picture, including why
|
||||||
|
// this repository has no evidence such a hook could even be wired to).
|
||||||
|
// Close exists so a respClient built and discarded outside that contract
|
||||||
|
// — tests, or a future framework/harness that does hold a lifecycle hook
|
||||||
|
// — has a correct, deterministic way to release pooled connections
|
||||||
|
// rather than depend solely on GC finalizing the underlying net.Conns.
|
||||||
|
func (c *respClient) Close() error {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case pc := <-c.free:
|
||||||
|
closeConn(pc)
|
||||||
|
default:
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// do sends one command and returns its decoded reply. It is equivalent to
|
// do sends one command and returns its decoded reply. It is equivalent to
|
||||||
@@ -104,42 +224,77 @@ func (c *respClient) do(args ...string) (any, error) {
|
|||||||
return replies[0], nil
|
return replies[0], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// pipeline sends every command in cmds on one round trip and returns their
|
// pipeline sends every command in cmds on one round trip, over one pooled
|
||||||
|
// connection acquired for the duration of the call, and returns their
|
||||||
// decoded replies in the same order.
|
// decoded replies in the same order.
|
||||||
//
|
//
|
||||||
// One absolute deadline is computed here, at call entry, and shared by
|
// One absolute deadline is computed here, at call entry, and shared by
|
||||||
// both the first attempt and (if it happens) the one retry — not a fresh
|
// acquiring a connection, the first attempt, and (if it happens) the one
|
||||||
// respCallTimeout for each. On a non-timeout I/O error (connection reset,
|
// retry — not a fresh respCallTimeout for each. On a non-timeout I/O error
|
||||||
// EOF from a peer that closed mid-command) it closes the connection and
|
// (connection reset, EOF from a peer that closed mid-command) it closes
|
||||||
// retries the whole pipeline once against the same deadline. On a timeout
|
// the connection and retries the whole pipeline once against the same
|
||||||
// specifically, it does not retry: a server that accepted the connection
|
// deadline, on the SAME pooled slot (a fresh dial reusing that slot, not a
|
||||||
// and then went silent would just be given the same non-answer a second
|
// second slot acquired from the pool). On a timeout specifically, it does
|
||||||
// time, doubling the caller's wait for nothing. Either way, a failure
|
// not retry: a server that accepted the connection and then went silent
|
||||||
// leaves the connection closed so the next call lazily reconnects fresh
|
// would just be given the same non-answer a second time, doubling the
|
||||||
// rather than retrying against a connection already known bad.
|
// caller's wait for nothing. Either way, a failure leaves that connection
|
||||||
|
// closed (closeConn) so the next call to acquire this slot lazily
|
||||||
|
// reconnects fresh rather than reusing a connection already known bad —
|
||||||
|
// the slot itself always returns to the pool via the deferred send below,
|
||||||
|
// whether or not its connection survived the call.
|
||||||
func (c *respClient) pipeline(cmds [][]string) ([]any, error) {
|
func (c *respClient) pipeline(cmds [][]string) ([]any, error) {
|
||||||
deadline := time.Now().Add(respCallTimeout)
|
deadline := time.Now().Add(respCallTimeout)
|
||||||
|
|
||||||
c.mu.Lock()
|
pc, err := c.acquire(deadline)
|
||||||
defer c.mu.Unlock()
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer func() { c.free <- pc }()
|
||||||
|
|
||||||
replies, err := c.attemptPipelineLocked(cmds, deadline)
|
replies, err := c.attemptPipelineOn(pc, cmds, deadline)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return replies, nil
|
return replies, nil
|
||||||
}
|
}
|
||||||
c.closeLocked()
|
closeConn(pc)
|
||||||
if isTimeout(err) {
|
if isTimeout(err) {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
replies, err = c.attemptPipelineLocked(cmds, deadline)
|
replies, err = c.attemptPipelineOn(pc, cmds, deadline)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.closeLocked()
|
closeConn(pc)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return replies, nil
|
return replies, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// acquire takes one idle connection slot from c.free, waiting no longer
|
||||||
|
// than deadline. The non-blocking check first is an optimization, not a
|
||||||
|
// correctness requirement of the blocking select below — with a buffered
|
||||||
|
// channel a receive that CAN succeed immediately always does, so this
|
||||||
|
// merely skips allocating a timer on the (overwhelmingly common) case
|
||||||
|
// where a slot is already free.
|
||||||
|
func (c *respClient) acquire(deadline time.Time) (*respConn, error) {
|
||||||
|
select {
|
||||||
|
case pc := <-c.free:
|
||||||
|
return pc, nil
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
wait := time.Until(deadline)
|
||||||
|
if wait <= 0 {
|
||||||
|
return nil, respDeadlineExceededErr{}
|
||||||
|
}
|
||||||
|
t := time.NewTimer(wait)
|
||||||
|
defer t.Stop()
|
||||||
|
select {
|
||||||
|
case pc := <-c.free:
|
||||||
|
return pc, nil
|
||||||
|
case <-t.C:
|
||||||
|
return nil, respDeadlineExceededErr{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// isTimeout reports whether err is (or wraps) a net.Error that timed out.
|
// isTimeout reports whether err is (or wraps) a net.Error that timed out.
|
||||||
func isTimeout(err error) bool {
|
func isTimeout(err error) bool {
|
||||||
var netErr net.Error
|
var netErr net.Error
|
||||||
@@ -161,22 +316,24 @@ func dialTimeoutFor(deadline time.Time) time.Duration {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// respDeadlineExceededErr signals that a call's deadline had already
|
// respDeadlineExceededErr signals that a call's deadline had already
|
||||||
// elapsed before an attempt could even dial — e.g. this call queued
|
// elapsed before an attempt could even acquire a pool slot or dial — e.g.
|
||||||
// behind respClient's mutex for the whole budget while an earlier call
|
// this call queued behind every one of respClient's pool slots being busy
|
||||||
// was stuck. It implements net.Error so pipeline's isTimeout check treats
|
// for the whole budget while earlier calls were stuck. It implements
|
||||||
// it the same as any other timed-out I/O: no retry, since retrying an
|
// net.Error so pipeline's isTimeout check treats it the same as any other
|
||||||
// already-elapsed deadline can only fail the same way again.
|
// timed-out I/O: no retry, since retrying an already-elapsed deadline can
|
||||||
|
// only fail the same way again.
|
||||||
type respDeadlineExceededErr struct{}
|
type respDeadlineExceededErr struct{}
|
||||||
|
|
||||||
func (respDeadlineExceededErr) Error() string { return "resp: call deadline already elapsed" }
|
func (respDeadlineExceededErr) Error() string { return "resp: call deadline already elapsed" }
|
||||||
func (respDeadlineExceededErr) Timeout() bool { return true }
|
func (respDeadlineExceededErr) Timeout() bool { return true }
|
||||||
func (respDeadlineExceededErr) Temporary() bool { return true }
|
func (respDeadlineExceededErr) Temporary() bool { return true }
|
||||||
|
|
||||||
// attemptPipelineLocked runs one full attempt of cmds over c's connection,
|
// attemptPipelineOn runs one full attempt of cmds over pc, connecting
|
||||||
// connecting first if needed, with every read and write bound by
|
// first if needed, with every read and write bound by deadline. Callers
|
||||||
// deadline. Callers must hold c.mu.
|
// own pc exclusively for the duration of this call (acquired from
|
||||||
func (c *respClient) attemptPipelineLocked(cmds [][]string, deadline time.Time) ([]any, error) {
|
// c.free).
|
||||||
if err := c.ensureConnLocked(deadline); err != nil {
|
func (c *respClient) attemptPipelineOn(pc *respConn, cmds [][]string, deadline time.Time) ([]any, error) {
|
||||||
|
if err := c.ensureConnOn(pc, deadline); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -184,13 +341,13 @@ func (c *respClient) attemptPipelineLocked(cmds [][]string, deadline time.Time)
|
|||||||
for _, args := range cmds {
|
for _, args := range cmds {
|
||||||
buf.WriteString(encodeCommand(args))
|
buf.WriteString(encodeCommand(args))
|
||||||
}
|
}
|
||||||
if _, err := io.WriteString(c.conn, buf.String()); err != nil {
|
if _, err := io.WriteString(pc.conn, buf.String()); err != nil {
|
||||||
return nil, fmt.Errorf("resp: write: %w", err)
|
return nil, fmt.Errorf("resp: write: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
replies := make([]any, len(cmds))
|
replies := make([]any, len(cmds))
|
||||||
for i := range cmds {
|
for i := range cmds {
|
||||||
v, err := decodeReply(c.r, 0)
|
v, err := decodeReply(pc.r, 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("resp: read reply %d/%d: %w", i+1, len(cmds), err)
|
return nil, fmt.Errorf("resp: read reply %d/%d: %w", i+1, len(cmds), err)
|
||||||
}
|
}
|
||||||
@@ -199,14 +356,14 @@ func (c *respClient) attemptPipelineLocked(cmds [][]string, deadline time.Time)
|
|||||||
return replies, nil
|
return replies, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ensureConnLocked dials, authenticates, and selects the database if c has
|
// ensureConnOn dials, authenticates, and selects the database if pc has no
|
||||||
// no live connection, then (whether freshly dialled or reused) sets
|
// live connection, then (whether freshly dialled or reused) sets deadline
|
||||||
// deadline as the connection's single read/write deadline for the
|
// as the connection's single read/write deadline for the remainder of
|
||||||
// remainder of this call — including the handshake, when one runs.
|
// this call — including the handshake, when one runs. Callers own pc
|
||||||
// Callers must hold c.mu.
|
// exclusively for the duration of this call.
|
||||||
func (c *respClient) ensureConnLocked(deadline time.Time) error {
|
func (c *respClient) ensureConnOn(pc *respConn, deadline time.Time) error {
|
||||||
if c.conn != nil {
|
if pc.conn != nil {
|
||||||
if err := c.conn.SetDeadline(deadline); err != nil {
|
if err := pc.conn.SetDeadline(deadline); err != nil {
|
||||||
return fmt.Errorf("resp: set deadline: %w", err)
|
return fmt.Errorf("resp: set deadline: %w", err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -220,35 +377,35 @@ func (c *respClient) ensureConnLocked(deadline time.Time) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("resp: dial %q: %w", c.addr, err)
|
return fmt.Errorf("resp: dial %q: %w", c.addr, err)
|
||||||
}
|
}
|
||||||
c.conn = conn
|
pc.conn = conn
|
||||||
c.r = bufio.NewReader(conn)
|
pc.r = bufio.NewReader(conn)
|
||||||
|
|
||||||
if err := c.conn.SetDeadline(deadline); err != nil {
|
if err := pc.conn.SetDeadline(deadline); err != nil {
|
||||||
c.closeLocked()
|
closeConn(pc)
|
||||||
return fmt.Errorf("resp: set deadline: %w", err)
|
return fmt.Errorf("resp: set deadline: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if c.password != "" {
|
if c.password != "" {
|
||||||
if err := c.handshakeLocked("AUTH", c.password); err != nil {
|
if err := c.handshakeOn(pc, "AUTH", c.password); err != nil {
|
||||||
c.closeLocked()
|
closeConn(pc)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := c.handshakeLocked("SELECT", strconv.Itoa(c.db)); err != nil {
|
if err := c.handshakeOn(pc, "SELECT", strconv.Itoa(c.db)); err != nil {
|
||||||
c.closeLocked()
|
closeConn(pc)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// handshakeLocked sends one connection-setup command (AUTH or SELECT) and
|
// handshakeOn sends one connection-setup command (AUTH or SELECT) and
|
||||||
// requires a non-error reply. Callers must hold c.mu, have a live c.conn,
|
// requires a non-error reply. Callers own pc exclusively, with a live
|
||||||
// and have already set its deadline.
|
// pc.conn whose deadline is already set.
|
||||||
func (c *respClient) handshakeLocked(cmd, arg string) error {
|
func (c *respClient) handshakeOn(pc *respConn, cmd, arg string) error {
|
||||||
if _, err := io.WriteString(c.conn, encodeCommand([]string{cmd, arg})); err != nil {
|
if _, err := io.WriteString(pc.conn, encodeCommand([]string{cmd, arg})); err != nil {
|
||||||
return fmt.Errorf("resp: %s: write: %w", cmd, err)
|
return fmt.Errorf("resp: %s: write: %w", cmd, err)
|
||||||
}
|
}
|
||||||
v, err := decodeReply(c.r, 0)
|
v, err := decodeReply(pc.r, 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("resp: %s: read reply: %w", cmd, err)
|
return fmt.Errorf("resp: %s: read reply: %w", cmd, err)
|
||||||
}
|
}
|
||||||
@@ -258,15 +415,16 @@ func (c *respClient) handshakeLocked(cmd, arg string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// closeLocked closes and clears c's connection, if any. Callers must hold
|
// closeConn closes and clears pc's connection, if any, discarding it —
|
||||||
// c.mu.
|
// the connection itself is never reused after an error, only the slot pc
|
||||||
func (c *respClient) closeLocked() {
|
// occupies, which pipeline always returns to c.free regardless.
|
||||||
if c.conn == nil {
|
func closeConn(pc *respConn) {
|
||||||
|
if pc.conn == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
_ = c.conn.Close()
|
_ = pc.conn.Close()
|
||||||
c.conn = nil
|
pc.conn = nil
|
||||||
c.r = nil
|
pc.r = nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// encodeCommand renders args as a RESP2 command array:
|
// encodeCommand renders args as a RESP2 command array:
|
||||||
|
|||||||
+334
-53
@@ -9,6 +9,7 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -42,7 +43,7 @@ func TestEncodeCommand(t *testing.T) {
|
|||||||
// string, error, integer, bulk string (including the null bulk "$-1"), and
|
// string, error, integer, bulk string (including the null bulk "$-1"), and
|
||||||
// a single top-level array (including the null array "*-1"). Every call
|
// a single top-level array (including the null array "*-1"). Every call
|
||||||
// passes depth 0 — a top-level reply, matching every real call site in
|
// passes depth 0 — a top-level reply, matching every real call site in
|
||||||
// resp.go (attemptPipelineLocked, handshakeLocked, and decodeArray's own
|
// resp.go (attemptPipelineOn, handshakeOn, and decodeArray's own
|
||||||
// element loop, which passes depth+1).
|
// element loop, which passes depth+1).
|
||||||
func TestDecodeReply(t *testing.T) {
|
func TestDecodeReply(t *testing.T) {
|
||||||
t.Run("simple string", func(t *testing.T) {
|
t.Run("simple string", func(t *testing.T) {
|
||||||
@@ -340,7 +341,7 @@ func newFakeListener(t *testing.T) net.Listener {
|
|||||||
// refusing or resetting. Used by review item 1's tests to prove a call's
|
// refusing or resetting. Used by review item 1's tests to prove a call's
|
||||||
// latency is bounded by respCallTimeout rather than left to block
|
// latency is bounded by respCallTimeout rather than left to block
|
||||||
// indefinitely. Each accepted connection's reader goroutine exits once
|
// indefinitely. Each accepted connection's reader goroutine exits once
|
||||||
// the connection is closed — by the client's own closeLocked after its
|
// the connection is closed — by the client's own closeConn after its
|
||||||
// deadline fires, or by the listener's t.Cleanup at test end.
|
// deadline fires, or by the listener's t.Cleanup at test end.
|
||||||
func newHungListener(t *testing.T) net.Listener {
|
func newHungListener(t *testing.T) net.Listener {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
@@ -613,7 +614,7 @@ func (fakeAddr) String() string { return "fake" }
|
|||||||
// fakeConn implements net.Conn, returning writeErr from Write and
|
// fakeConn implements net.Conn, returning writeErr from Write and
|
||||||
// setDeadlineErr from SetDeadline when set, so a test can drive respClient
|
// setDeadlineErr from SetDeadline when set, so a test can drive respClient
|
||||||
// methods directly against a connection already known bad — bypassing
|
// methods directly against a connection already known bad — bypassing
|
||||||
// ensureConnLocked's real net.DialTimeout, which always succeeds against a
|
// ensureConnOn's real net.DialTimeout, which always succeeds against a
|
||||||
// live listener and so cannot itself be made to fail this way.
|
// live listener and so cannot itself be made to fail this way.
|
||||||
type fakeConn struct {
|
type fakeConn struct {
|
||||||
writeErr error
|
writeErr error
|
||||||
@@ -643,6 +644,19 @@ func (c fakeConn) SetDeadline(time.Time) error {
|
|||||||
func (fakeConn) SetReadDeadline(time.Time) error { return nil }
|
func (fakeConn) SetReadDeadline(time.Time) error { return nil }
|
||||||
func (fakeConn) SetWriteDeadline(time.Time) error { return nil }
|
func (fakeConn) SetWriteDeadline(time.Time) error { return nil }
|
||||||
|
|
||||||
|
// closeCountingConn wraps fakeConn, incrementing *count on every Close
|
||||||
|
// call — used to prove a method actually closed a connection, not just
|
||||||
|
// forgot about (or merely drained a channel referencing) it.
|
||||||
|
type closeCountingConn struct {
|
||||||
|
fakeConn
|
||||||
|
count *int
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *closeCountingConn) Close() error {
|
||||||
|
*c.count++
|
||||||
|
return c.fakeConn.Close()
|
||||||
|
}
|
||||||
|
|
||||||
// --- dialTimeoutFor / respDeadlineExceededErr ---
|
// --- dialTimeoutFor / respDeadlineExceededErr ---
|
||||||
|
|
||||||
// TestDialTimeoutFor covers dialTimeoutFor's clamp: a deadline farther away
|
// TestDialTimeoutFor covers dialTimeoutFor's clamp: a deadline farther away
|
||||||
@@ -688,45 +702,44 @@ func TestRespDeadlineExceededErr(t *testing.T) {
|
|||||||
assert.True(t, err.Temporary())
|
assert.True(t, err.Temporary())
|
||||||
}
|
}
|
||||||
|
|
||||||
// --- ensureConnLocked / attemptPipelineLocked / handshakeLocked error
|
// --- ensureConnOn / attemptPipelineOn / handshakeOn error branches only
|
||||||
// branches only reachable via a pre-set connection, not a real dial ---
|
// reachable via a pre-set connection, not a real dial ---
|
||||||
|
|
||||||
// TestRESPClient_EnsureConnLocked_Errors covers ensureConnLocked's two
|
// TestRESPClient_EnsureConnOn_Errors covers ensureConnOn's two error
|
||||||
// error returns that never touch the network: a reused connection whose
|
// returns that never touch the network: a reused connection whose
|
||||||
// SetDeadline itself fails, and a deadline that has already elapsed before
|
// SetDeadline itself fails, and a deadline that has already elapsed before
|
||||||
// a fresh dial would even start.
|
// a fresh dial would even start. Each case builds its own *respConn slot
|
||||||
func TestRESPClient_EnsureConnLocked_Errors(t *testing.T) {
|
// directly — under the pool, ensureConnOn takes the slot as a parameter
|
||||||
|
// rather than reading shared client state, so there is no c.mu precondition
|
||||||
|
// to honor here any more.
|
||||||
|
func TestRESPClient_EnsureConnOn_Errors(t *testing.T) {
|
||||||
cases := []struct {
|
cases := []struct {
|
||||||
name string
|
|
||||||
client func() *respClient
|
|
||||||
deadline time.Time
|
deadline time.Time
|
||||||
|
pc func() *respConn
|
||||||
|
name string
|
||||||
|
addr string
|
||||||
wantErrContains string
|
wantErrContains string
|
||||||
}{
|
}{
|
||||||
{
|
{
|
||||||
name: "reused connection: SetDeadline failure surfaces",
|
name: "reused connection: SetDeadline failure surfaces",
|
||||||
client: func() *respClient {
|
pc: func() *respConn {
|
||||||
return &respClient{conn: fakeConn{setDeadlineErr: errors.New("stub: set deadline failed")}}
|
return &respConn{conn: fakeConn{setDeadlineErr: errors.New("stub: set deadline failed")}}
|
||||||
},
|
},
|
||||||
deadline: time.Now().Add(time.Second),
|
deadline: time.Now().Add(time.Second),
|
||||||
wantErrContains: "set deadline",
|
wantErrContains: "set deadline",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "no connection yet: an already-elapsed deadline never dials",
|
name: "no connection yet: an already-elapsed deadline never dials",
|
||||||
client: func() *respClient {
|
pc: func() *respConn { return &respConn{} },
|
||||||
return &respClient{addr: "127.0.0.1:1"}
|
addr: "127.0.0.1:1",
|
||||||
},
|
|
||||||
deadline: time.Now().Add(-time.Second),
|
deadline: time.Now().Add(-time.Second),
|
||||||
wantErrContains: "call deadline already elapsed",
|
wantErrContains: "call deadline already elapsed",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
for _, c := range cases {
|
for _, c := range cases {
|
||||||
t.Run(c.name, func(t *testing.T) {
|
t.Run(c.name, func(t *testing.T) {
|
||||||
client := c.client()
|
client := &respClient{addr: c.addr}
|
||||||
// ensureConnLocked documents "callers must hold c.mu" — honor
|
err := client.ensureConnOn(c.pc(), c.deadline)
|
||||||
// that precondition even though this test is single-goroutine.
|
|
||||||
client.mu.Lock()
|
|
||||||
err := client.ensureConnLocked(c.deadline)
|
|
||||||
client.mu.Unlock()
|
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.Contains(t, err.Error(), c.wantErrContains)
|
assert.Contains(t, err.Error(), c.wantErrContains)
|
||||||
})
|
})
|
||||||
@@ -734,44 +747,40 @@ func TestRESPClient_EnsureConnLocked_Errors(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestRESPClient_WriteError_BrokenPipe covers the identical broken-pipe
|
// TestRESPClient_WriteError_BrokenPipe covers the identical broken-pipe
|
||||||
// shape shared by attemptPipelineLocked's command write and
|
// shape shared by attemptPipelineOn's command write and handshakeOn's own
|
||||||
// handshakeLocked's own write: both fail deterministically on an
|
// write: both fail deterministically on an already-connected socket that
|
||||||
// already-connected socket that refuses a write — distinct from a dial
|
// refuses a write — distinct from a dial failure, which never reaches
|
||||||
// failure, which never reaches either code path.
|
// either code path.
|
||||||
func TestRESPClient_WriteError_BrokenPipe(t *testing.T) {
|
func TestRESPClient_WriteError_BrokenPipe(t *testing.T) {
|
||||||
cases := []struct {
|
cases := []struct {
|
||||||
run func(c *respClient) error
|
run func(c *respClient, pc *respConn) error
|
||||||
name string
|
name string
|
||||||
wantErrContains string
|
wantErrContains string
|
||||||
}{
|
}{
|
||||||
{
|
{
|
||||||
name: "attemptPipelineLocked: command write fails",
|
name: "attemptPipelineOn: command write fails",
|
||||||
run: func(c *respClient) error {
|
run: func(c *respClient, pc *respConn) error {
|
||||||
_, err := c.attemptPipelineLocked([][]string{{"GET", "k"}}, time.Now().Add(time.Second))
|
_, err := c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, time.Now().Add(time.Second))
|
||||||
return err
|
return err
|
||||||
},
|
},
|
||||||
wantErrContains: "write",
|
wantErrContains: "write",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "handshakeLocked: AUTH write fails",
|
name: "handshakeOn: AUTH write fails",
|
||||||
run: func(c *respClient) error {
|
run: func(c *respClient, pc *respConn) error {
|
||||||
return c.handshakeLocked("AUTH", "pw")
|
return c.handshakeOn(pc, "AUTH", "pw")
|
||||||
},
|
},
|
||||||
wantErrContains: "AUTH",
|
wantErrContains: "AUTH",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
for _, c := range cases {
|
for _, c := range cases {
|
||||||
t.Run(c.name, func(t *testing.T) {
|
t.Run(c.name, func(t *testing.T) {
|
||||||
client := &respClient{
|
client := &respClient{}
|
||||||
|
pc := &respConn{
|
||||||
conn: fakeConn{writeErr: errors.New("stub: broken pipe")},
|
conn: fakeConn{writeErr: errors.New("stub: broken pipe")},
|
||||||
r: bufio.NewReader(strings.NewReader("")),
|
r: bufio.NewReader(strings.NewReader("")),
|
||||||
}
|
}
|
||||||
// attemptPipelineLocked/handshakeLocked document "callers must
|
err := c.run(client, pc)
|
||||||
// hold c.mu" — honor that precondition even though this test
|
|
||||||
// is single-goroutine.
|
|
||||||
client.mu.Lock()
|
|
||||||
err := c.run(client)
|
|
||||||
client.mu.Unlock()
|
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.Contains(t, err.Error(), c.wantErrContains)
|
assert.Contains(t, err.Error(), c.wantErrContains)
|
||||||
})
|
})
|
||||||
@@ -779,13 +788,14 @@ func TestRESPClient_WriteError_BrokenPipe(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces covers
|
// TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces covers
|
||||||
// ensureConnLocked's AUTH-failure branch and handshakeLocked's respErr
|
// ensureConnOn's AUTH-failure branch and handshakeOn's respErr branch
|
||||||
// branch together: a RESP error reply to AUTH must fail the call, not be
|
// together: a RESP error reply to AUTH must fail the call, not be
|
||||||
// silently treated as success. It calls attemptPipelineLocked directly
|
// silently treated as success. It acquires one pooled slot and calls
|
||||||
// (one attempt, one scripted connection) rather than the public do/
|
// attemptPipelineOn directly (one attempt, one scripted connection)
|
||||||
// pipeline, which would retry once more against a second connection the
|
// rather than the public do/pipeline, which would retry once more against
|
||||||
// fakeRESPServer fixture — built for a server-initiated closeConn, not a
|
// a second connection the fakeRESPServer fixture — built for a
|
||||||
// client-initiated close on handshake failure — cannot script cleanly.
|
// server-initiated closeConn, not a client-initiated close on handshake
|
||||||
|
// failure — cannot script cleanly.
|
||||||
func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) {
|
func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) {
|
||||||
ln := newFakeListener(t)
|
ln := newFakeListener(t)
|
||||||
runFakeRESPServer(t, ln, []respStep{
|
runFakeRESPServer(t, ln, []respStep{
|
||||||
@@ -793,11 +803,10 @@ func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
c := newRESPClient(ln.Addr().String(), "wrong-pw", 0)
|
c := newRESPClient(ln.Addr().String(), "wrong-pw", 0)
|
||||||
// attemptPipelineLocked documents "callers must hold c.mu" — honor that
|
deadline := time.Now().Add(2 * time.Second)
|
||||||
// precondition even though this test is single-goroutine.
|
pc, err := c.acquire(deadline)
|
||||||
c.mu.Lock()
|
require.NoError(t, err)
|
||||||
_, err := c.attemptPipelineLocked([][]string{{"GET", "k"}}, time.Now().Add(2*time.Second))
|
_, err = c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, deadline)
|
||||||
c.mu.Unlock()
|
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.Contains(t, err.Error(), "AUTH failed")
|
assert.Contains(t, err.Error(), "AUTH failed")
|
||||||
}
|
}
|
||||||
@@ -875,3 +884,275 @@ func TestRESPClient_GetBatch(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- connection pool (finding 1, perf audit 2026-08-2x): a single
|
||||||
|
// mutex-guarded connection was THE throughput ceiling under Yaegi
|
||||||
|
// interpretation; respClient now pools respPoolMin..respPoolMax
|
||||||
|
// connections (GOMAXPROCS-clamped by default, or an explicit
|
||||||
|
// RedisConfig.PoolSize override — llmgateway.go's buildRedisClient) ---
|
||||||
|
|
||||||
|
// TestDefaultRespPoolSize_ClampsToRange asserts defaultRespPoolSize never
|
||||||
|
// returns a value outside [respPoolMin, respPoolMax], regardless of the
|
||||||
|
// test host's actual GOMAXPROCS — the exact value is environment-
|
||||||
|
// dependent, but the clamp is not.
|
||||||
|
func TestDefaultRespPoolSize_ClampsToRange(t *testing.T) {
|
||||||
|
got := defaultRespPoolSize()
|
||||||
|
if got < respPoolMin || got > respPoolMax {
|
||||||
|
t.Errorf("defaultRespPoolSize() = %d, want in [%d, %d]", got, respPoolMin, respPoolMax)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestClampPoolSize covers every branch of the clamp defaultRespPoolSize
|
||||||
|
// applies to the host's real GOMAXPROCS: below respPoolMin, above
|
||||||
|
// respPoolMax, and already inside the range (passed through unchanged).
|
||||||
|
func TestClampPoolSize(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
n int
|
||||||
|
want int
|
||||||
|
}{
|
||||||
|
{name: "below min clamps up", n: 1, want: respPoolMin},
|
||||||
|
{name: "zero clamps up", n: 0, want: respPoolMin},
|
||||||
|
{name: "negative clamps up", n: -3, want: respPoolMin},
|
||||||
|
{name: "above max clamps down", n: 64, want: respPoolMax},
|
||||||
|
{name: "exactly min passes through", n: respPoolMin, want: respPoolMin},
|
||||||
|
{name: "exactly max passes through", n: respPoolMax, want: respPoolMax},
|
||||||
|
{name: "inside range passes through unchanged", n: 6, want: 6},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
t.Run(c.name, func(t *testing.T) {
|
||||||
|
if got := clampPoolSize(c.n); got != c.want {
|
||||||
|
t.Errorf("clampPoolSize(%d) = %d, want %d", c.n, got, c.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_Acquire_WaitsForASlotThenSucceeds exercises acquire's
|
||||||
|
// blocking-wait path (the pool is momentarily exhausted, not the
|
||||||
|
// non-blocking fast path a free slot satisfies immediately): with a
|
||||||
|
// pool of 1, a second acquire call blocks until the first caller's slot
|
||||||
|
// is released, then succeeds well within its deadline.
|
||||||
|
func TestRESPClient_Acquire_WaitsForASlotThenSucceeds(t *testing.T) {
|
||||||
|
c := newRESPClientPool("127.0.0.1:0", "", 0, 1)
|
||||||
|
held, err := c.acquire(time.Now().Add(time.Second))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("acquire (1st): %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
released := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
c.free <- held
|
||||||
|
close(released)
|
||||||
|
}()
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
pc, err := c.acquire(time.Now().Add(2 * time.Second))
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
<-released
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("acquire (2nd, blocking): %v", err)
|
||||||
|
}
|
||||||
|
if pc != held {
|
||||||
|
t.Errorf("acquire (2nd) returned a different slot than the one released, want the same *respConn")
|
||||||
|
}
|
||||||
|
if elapsed < 15*time.Millisecond {
|
||||||
|
t.Errorf("elapsed = %v, want it to have actually waited for the release (~20ms)", elapsed)
|
||||||
|
}
|
||||||
|
if elapsed >= 2*time.Second {
|
||||||
|
t.Errorf("elapsed = %v, want well under the 2s deadline", elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_Acquire_TimesOutWhenPoolExhausted covers acquire's other
|
||||||
|
// blocking-wait outcome: no slot is ever released before the deadline, so
|
||||||
|
// acquire returns respDeadlineExceededErr rather than blocking forever.
|
||||||
|
func TestRESPClient_Acquire_TimesOutWhenPoolExhausted(t *testing.T) {
|
||||||
|
c := newRESPClientPool("127.0.0.1:0", "", 0, 1)
|
||||||
|
if _, err := c.acquire(time.Now().Add(time.Second)); err != nil {
|
||||||
|
t.Fatalf("acquire (1st, drains the only slot): %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
_, err := c.acquire(time.Now().Add(50 * time.Millisecond))
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("want an error: the pool's only slot is held and never released")
|
||||||
|
}
|
||||||
|
var wantErr respDeadlineExceededErr
|
||||||
|
if !errors.As(err, &wantErr) {
|
||||||
|
t.Errorf("err = %v (%T), want respDeadlineExceededErr", err, err)
|
||||||
|
}
|
||||||
|
if elapsed < 40*time.Millisecond || elapsed >= time.Second {
|
||||||
|
t.Errorf("elapsed = %v, want roughly the 50ms deadline, not immediate and not unbounded", elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestNewRESPClientPool_SizesPoolExactly asserts newRESPClientPool builds
|
||||||
|
// exactly poolSize free slots (each an empty, not-yet-dialled *respConn),
|
||||||
|
// and that a poolSize of 0 or below is clamped up to 1 rather than
|
||||||
|
// producing a pool no acquire could ever succeed against.
|
||||||
|
func TestNewRESPClientPool_SizesPoolExactly(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
poolSize int
|
||||||
|
wantPoolSize int
|
||||||
|
}{
|
||||||
|
{name: "explicit size", poolSize: 3, wantPoolSize: 3},
|
||||||
|
{name: "zero clamps to 1", poolSize: 0, wantPoolSize: 1},
|
||||||
|
{name: "negative clamps to 1", poolSize: -5, wantPoolSize: 1},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
t.Run(c.name, func(t *testing.T) {
|
||||||
|
client := newRESPClientPool("127.0.0.1:0", "", 0, c.poolSize)
|
||||||
|
if got := len(client.free); got != c.wantPoolSize {
|
||||||
|
t.Errorf("len(free) = %d, want %d", got, c.wantPoolSize)
|
||||||
|
}
|
||||||
|
if got := cap(client.free); got != c.wantPoolSize {
|
||||||
|
t.Errorf("cap(free) = %d, want %d", got, c.wantPoolSize)
|
||||||
|
}
|
||||||
|
pc := <-client.free
|
||||||
|
if pc == nil || pc.conn != nil {
|
||||||
|
t.Errorf("pool slot = %+v, want a non-nil *respConn with a nil conn (not yet dialled)", pc)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_PoolSlotReturnedAfterError proves a slot always comes
|
||||||
|
// back onto c.free after a failed call — the physical connection is
|
||||||
|
// discarded (closeConn), never the SLOT itself — so a pool never shrinks
|
||||||
|
// after an error the way it would if a broken slot were dropped instead
|
||||||
|
// of returned.
|
||||||
|
func TestRESPClient_PoolSlotReturnedAfterError(t *testing.T) {
|
||||||
|
ln := newFakeListener(t)
|
||||||
|
deadAddr := ln.Addr().String()
|
||||||
|
require.NoError(t, ln.Close()) // nothing listens at deadAddr from here on
|
||||||
|
|
||||||
|
c := newRESPClientPool(deadAddr, "", 0, 1)
|
||||||
|
if _, err := c.do("GET", "k"); err == nil {
|
||||||
|
t.Fatal("want an error dialing a dead address")
|
||||||
|
}
|
||||||
|
if got := len(c.free); got != 1 {
|
||||||
|
t.Fatalf("len(free) after a failed call = %d, want 1 (the slot must return to the pool)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_PoolServesConcurrentCallersWithoutSerializing is the
|
||||||
|
// core regression for finding 1: a pool-of-2 client issues two concurrent
|
||||||
|
// GETs against a listener that withholds connection 1's reply until
|
||||||
|
// connection 2 has been fully accepted and served. A single shared
|
||||||
|
// connection (the pre-fix design) would deadlock here — the second
|
||||||
|
// caller could never even dial while do() held the only connection open
|
||||||
|
// — so both calls succeeding proves they ran on two separate connections
|
||||||
|
// at once, not serialized behind one mutex.
|
||||||
|
func TestRESPClient_PoolServesConcurrentCallersWithoutSerializing(t *testing.T) {
|
||||||
|
ln := newFakeListener(t)
|
||||||
|
conn1Read := make(chan struct{})
|
||||||
|
|
||||||
|
// Any read/write error below is deliberately swallowed rather than
|
||||||
|
// failing the test directly from this goroutine (t.Fatal/t.Errorf
|
||||||
|
// from a non-test goroutine after the test could return is unsafe): a
|
||||||
|
// script mismatch here instead surfaces as a do() error the two
|
||||||
|
// client goroutines below report via t.Errorf, or as the test timing
|
||||||
|
// out waiting on conn1Read/wg — either way, a real protocol failure
|
||||||
|
// still fails the test, just one step later.
|
||||||
|
go func() {
|
||||||
|
c1, acceptErr := ln.Accept()
|
||||||
|
if acceptErr != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
r1 := bufio.NewReader(c1)
|
||||||
|
_, _ = readRESPCommand(r1) // SELECT
|
||||||
|
_, _ = c1.Write([]byte("+OK\r\n"))
|
||||||
|
_, _ = readRESPCommand(r1) // GET
|
||||||
|
close(conn1Read) // conn 1's GET has been read; its reply is withheld
|
||||||
|
|
||||||
|
c2, acceptErr2 := ln.Accept()
|
||||||
|
if acceptErr2 != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
r2 := bufio.NewReader(c2)
|
||||||
|
_, _ = readRESPCommand(r2) // SELECT
|
||||||
|
_, _ = c2.Write([]byte("+OK\r\n"))
|
||||||
|
_, _ = readRESPCommand(r2) // GET
|
||||||
|
_, _ = c2.Write([]byte("$1\r\n2\r\n"))
|
||||||
|
|
||||||
|
_, _ = c1.Write([]byte("$1\r\n1\r\n")) // release conn 1's reply last
|
||||||
|
}()
|
||||||
|
|
||||||
|
c := newRESPClientPool(ln.Addr().String(), "", 0, 2)
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
results := make([]string, 2)
|
||||||
|
wg.Add(2)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
v, err := c.do("GET", "a")
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("do(a): %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
results[0] = string(v.([]byte)) //nolint:forcetypeassert // test-controlled reply, always a bulk string
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
<-conn1Read // wait until conn 1's GET has been read server-side
|
||||||
|
v, err := c.do("GET", "b")
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("do(b): %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
results[1] = string(v.([]byte)) //nolint:forcetypeassert // test-controlled reply, always a bulk string
|
||||||
|
}()
|
||||||
|
wg.Wait()
|
||||||
|
// assert.Equal, not ElementsMatch: results[0]/[1] are written by
|
||||||
|
// INDEX (which goroutine, not completion order), so the expected
|
||||||
|
// values are positionally deterministic — "a" always reads conn 1's
|
||||||
|
// scripted reply ("1"), "b" always reads conn 2's ("2"). An
|
||||||
|
// order-insensitive match would still pass under the exact cross-talk
|
||||||
|
// bug this test targets (e.g. two callers corrupting/swapping each
|
||||||
|
// other's reply via a shared connection), since {"1","2"} and
|
||||||
|
// {"2","1"} compare equal under ElementsMatch but not under Equal.
|
||||||
|
assert.Equal(t, []string{"1", "2"}, results)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_Close_DrainsAndClosesIdleConnections is the review fix
|
||||||
|
// (Should-Fix 4, 2026-08-2x): Close must actually close every idle
|
||||||
|
// pooled connection (not merely forget about it) and leave the pool
|
||||||
|
// empty.
|
||||||
|
func TestRESPClient_Close_DrainsAndClosesIdleConnections(t *testing.T) {
|
||||||
|
closed := 0
|
||||||
|
c := newRESPClientPool("127.0.0.1:0", "", 0, 3)
|
||||||
|
// Replace each of the 3 idle slots' connection with one that records
|
||||||
|
// whether Close was called on it.
|
||||||
|
for i := 0; i < 3; i++ {
|
||||||
|
pc := <-c.free
|
||||||
|
pc.conn = &closeCountingConn{count: &closed}
|
||||||
|
c.free <- pc
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.Close(); err != nil {
|
||||||
|
t.Fatalf("Close: %v", err)
|
||||||
|
}
|
||||||
|
if got := len(c.free); got != 0 {
|
||||||
|
t.Errorf("len(free) after Close = %d, want 0 (every slot drained)", got)
|
||||||
|
}
|
||||||
|
if closed != 3 {
|
||||||
|
t.Errorf("closed connections = %d, want 3 (every idle slot's connection actually closed)", closed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRESPClient_Close_EmptyPool_ReturnsNilImmediately asserts Close on a
|
||||||
|
// pool with nothing yet dialled (every slot's conn is nil) is a no-op
|
||||||
|
// that returns promptly, not a hang — closeConn already tolerates a nil
|
||||||
|
// conn.
|
||||||
|
func TestRESPClient_Close_EmptyPool_ReturnsNilImmediately(t *testing.T) {
|
||||||
|
c := newRESPClientPool("127.0.0.1:0", "", 0, 2)
|
||||||
|
if err := c.Close(); err != nil {
|
||||||
|
t.Fatalf("Close: %v", err)
|
||||||
|
}
|
||||||
|
if got := len(c.free); got != 0 {
|
||||||
|
t.Errorf("len(free) after Close = %d, want 0", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user