From 2309a59b40e215034f9af8f8cc279e654472cefb Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sat, 22 Aug 2026 13:52:00 +0100 Subject: [PATCH] perf(resp): pool redis connections instead of one mutex-guarded conn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- cache_test.go | 17 +- llmgateway.go | 64 +++++++- llmgateway_test.go | 116 ++++++++++++++ resp.go | 318 +++++++++++++++++++++++++++---------- resp_test.go | 387 ++++++++++++++++++++++++++++++++++++++------- 5 files changed, 766 insertions(+), 136 deletions(-) diff --git a/cache_test.go b/cache_test.go index 716184c..01fc530 100644 --- a/cache_test.go +++ b/cache_test.go @@ -301,9 +301,19 @@ func TestBuildResponseCache_EnabledWithRedis_ReturnsWorkingCache(t *testing.T) { // newTestResponseCache builds a responseCache over ln's fake server, with // nowFn pinned to a fixed instant so logCacheError's rate-limit window is // 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) { t.Helper() - client := newRESPClient(ln.Addr().String(), "", 0) + client := newRESPClientPool(ln.Addr().String(), "", 0, 1) msgs := []string{} spy := func(format string, args ...any) { msgs = append(msgs, fmt.Sprintf(format, args...)) } 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-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) {} c := newResponseCache(client, time.Minute, defaultCacheMaxBodyBytes, noop, noop) // global ttl = 60s c.nowFn = time.Now diff --git a/llmgateway.go b/llmgateway.go index 5f43ab5..6991b7a 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -196,6 +196,19 @@ type RedisConfig struct { Address string `json:"address"` Password string `json:"password,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 @@ -309,7 +322,14 @@ type Gateway struct { // before using it, rather than responseCache having its own // always-disabled zero value. 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 @@ -322,6 +342,35 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h 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) { if config == nil || len(config.Providers) == 0 { 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 { return nil, err } + g.redisClient = redisClient lim := newConfiguredLimiter(config, redisClient) lim.logf = g.errorf g.limiter = lim @@ -410,12 +460,24 @@ func buildRedisClient(rc *RedisConfig) (*respClient, error) { if rc.DB < 0 { 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) if err != nil { 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 } diff --git a/llmgateway_test.go b/llmgateway_test.go index a5744f3..3b7a830 100644 --- a/llmgateway_test.go +++ b/llmgateway_test.go @@ -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 // block with FailOpen left nil defaults to true, matching the no-Redis // 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 // that newGateway resolves Config.Redis.Password through resolveSecret // (env:/file:/literal) before handing it to respClient: an "env:" password diff --git a/resp.go b/resp.go index 200b4d8..5afdec2 100644 --- a/resp.go +++ b/resp.go @@ -6,9 +6,9 @@ import ( "fmt" "io" "net" + "runtime" "strconv" "strings" - "sync" "time" ) @@ -20,9 +20,7 @@ import ( // conn.SetDeadline, not reset before each individual read or write — a // server that accepts the connection and then never answers must cost // this one call ~respCallTimeout, not respCallTimeout multiplied by every -// handshake and command step it happens to perform (which, with the -// mutex serializing every caller behind one connection, previously turned -// one hung server into a many-times-respCallTimeout stall per request). +// handshake and command step it happens to perform. const ( respDialTimeout = 2 * time.Second respCallTimeout = 2 * time.Second @@ -48,6 +46,63 @@ const respMaxArrayLen = 1024 // bytes before erroring. 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 // caller inspecting a decoded reply can type-assert for it the same way as // any other error. @@ -56,7 +111,7 @@ const respMaxLineLen = 64 << 10 // 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 // 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 // has no such ambiguity: verified empirically under real Traefik (Task // 15's integration suite, cross-replica rate limiting via Redis). @@ -65,33 +120,98 @@ type respErr struct{ msg string } // Error implements the error interface. 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- -// compatible server: one mutex-guarded TCP connection, lazily dialled on -// first use and reconnected at most once when a command round trip fails -// for a reason other than a timeout (see pipeline). It exists because the -// plugin runs interpreted under Yaegi with only the Go standard library -// available, so a full Redis client library is not an option. +// compatible server: a pool of independently usable TCP connections, each +// lazily dialled on first use and reconnected at most once when a command +// round trip fails for a reason other than a timeout (see pipeline). It +// exists because the plugin runs interpreted under Yaegi with only the Go +// 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 -// round trip, including any reconnect — since RESP is not multiplexed: -// interleaving two callers' writes on one connection would corrupt both -// replies. +// free holds every currently-idle connection slot; a caller acquires one +// (blocking, bounded by its call deadline, if the pool is momentarily +// exhausted) for its whole round trip and returns it via a deferred send +// 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 { - conn net.Conn - r *bufio.Reader + free chan *respConn addr string password string - mu sync.Mutex db int } -// newRESPClient returns a respClient for addr. It does not connect until -// the first do or pipeline call. password, if non-empty, is sent via AUTH -// on every 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. +// newRESPClient returns a respClient for addr, sized by defaultRespPoolSize +// — the self-tuned pool size used everywhere an explicit override +// (RedisConfig.PoolSize) is absent. It does not connect until the first +// do or pipeline call. password, if non-empty, is sent via AUTH on every +// 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 { - 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 @@ -104,42 +224,77 @@ func (c *respClient) do(args ...string) (any, error) { 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. // // 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 -// respCallTimeout for each. On a non-timeout I/O error (connection reset, -// EOF from a peer that closed mid-command) it closes the connection and -// retries the whole pipeline once against the same deadline. On a timeout -// specifically, it does not retry: a server that accepted the connection -// and then went silent would just be given the same non-answer a second -// time, doubling the caller's wait for nothing. Either way, a failure -// leaves the connection closed so the next call lazily reconnects fresh -// rather than retrying against a connection already known bad. +// acquiring a connection, the first attempt, and (if it happens) the one +// retry — not a fresh respCallTimeout for each. On a non-timeout I/O error +// (connection reset, EOF from a peer that closed mid-command) it closes +// the connection and retries the whole pipeline once against the same +// deadline, on the SAME pooled slot (a fresh dial reusing that slot, not a +// second slot acquired from the pool). On a timeout specifically, it does +// not retry: a server that accepted the connection and then went silent +// would just be given the same non-answer a second time, doubling the +// 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) { deadline := time.Now().Add(respCallTimeout) - c.mu.Lock() - defer c.mu.Unlock() + pc, err := c.acquire(deadline) + 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 { return replies, nil } - c.closeLocked() + closeConn(pc) if isTimeout(err) { return nil, err } - replies, err = c.attemptPipelineLocked(cmds, deadline) + replies, err = c.attemptPipelineOn(pc, cmds, deadline) if err != nil { - c.closeLocked() + closeConn(pc) return nil, err } 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. func isTimeout(err error) bool { var netErr net.Error @@ -161,22 +316,24 @@ func dialTimeoutFor(deadline time.Time) time.Duration { } // respDeadlineExceededErr signals that a call's deadline had already -// elapsed before an attempt could even dial — e.g. this call queued -// behind respClient's mutex for the whole budget while an earlier call -// was stuck. It implements net.Error so pipeline's isTimeout check treats -// it the same as any other timed-out I/O: no retry, since retrying an -// already-elapsed deadline can only fail the same way again. +// elapsed before an attempt could even acquire a pool slot or dial — e.g. +// this call queued behind every one of respClient's pool slots being busy +// for the whole budget while earlier calls were stuck. It implements +// net.Error so pipeline's isTimeout check treats it the same as any other +// timed-out I/O: no retry, since retrying an already-elapsed deadline can +// only fail the same way again. type respDeadlineExceededErr struct{} func (respDeadlineExceededErr) Error() string { return "resp: call deadline already elapsed" } func (respDeadlineExceededErr) Timeout() bool { return true } func (respDeadlineExceededErr) Temporary() bool { return true } -// attemptPipelineLocked runs one full attempt of cmds over c's connection, -// connecting first if needed, with every read and write bound by -// deadline. Callers must hold c.mu. -func (c *respClient) attemptPipelineLocked(cmds [][]string, deadline time.Time) ([]any, error) { - if err := c.ensureConnLocked(deadline); err != nil { +// attemptPipelineOn runs one full attempt of cmds over pc, connecting +// first if needed, with every read and write bound by deadline. Callers +// own pc exclusively for the duration of this call (acquired from +// c.free). +func (c *respClient) attemptPipelineOn(pc *respConn, cmds [][]string, deadline time.Time) ([]any, error) { + if err := c.ensureConnOn(pc, deadline); err != nil { return nil, err } @@ -184,13 +341,13 @@ func (c *respClient) attemptPipelineLocked(cmds [][]string, deadline time.Time) for _, args := range cmds { 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) } replies := make([]any, len(cmds)) for i := range cmds { - v, err := decodeReply(c.r, 0) + v, err := decodeReply(pc.r, 0) if err != nil { 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 } -// ensureConnLocked dials, authenticates, and selects the database if c has -// no live connection, then (whether freshly dialled or reused) sets -// deadline as the connection's single read/write deadline for the -// remainder of this call — including the handshake, when one runs. -// Callers must hold c.mu. -func (c *respClient) ensureConnLocked(deadline time.Time) error { - if c.conn != nil { - if err := c.conn.SetDeadline(deadline); err != nil { +// ensureConnOn dials, authenticates, and selects the database if pc has no +// live connection, then (whether freshly dialled or reused) sets deadline +// as the connection's single read/write deadline for the remainder of +// this call — including the handshake, when one runs. Callers own pc +// exclusively for the duration of this call. +func (c *respClient) ensureConnOn(pc *respConn, deadline time.Time) error { + if pc.conn != nil { + if err := pc.conn.SetDeadline(deadline); err != nil { return fmt.Errorf("resp: set deadline: %w", err) } return nil @@ -220,35 +377,35 @@ func (c *respClient) ensureConnLocked(deadline time.Time) error { if err != nil { return fmt.Errorf("resp: dial %q: %w", c.addr, err) } - c.conn = conn - c.r = bufio.NewReader(conn) + pc.conn = conn + pc.r = bufio.NewReader(conn) - if err := c.conn.SetDeadline(deadline); err != nil { - c.closeLocked() + if err := pc.conn.SetDeadline(deadline); err != nil { + closeConn(pc) return fmt.Errorf("resp: set deadline: %w", err) } if c.password != "" { - if err := c.handshakeLocked("AUTH", c.password); err != nil { - c.closeLocked() + if err := c.handshakeOn(pc, "AUTH", c.password); err != nil { + closeConn(pc) return err } } - if err := c.handshakeLocked("SELECT", strconv.Itoa(c.db)); err != nil { - c.closeLocked() + if err := c.handshakeOn(pc, "SELECT", strconv.Itoa(c.db)); err != nil { + closeConn(pc) return err } return nil } -// handshakeLocked sends one connection-setup command (AUTH or SELECT) and -// requires a non-error reply. Callers must hold c.mu, have a live c.conn, -// and have already set its deadline. -func (c *respClient) handshakeLocked(cmd, arg string) error { - if _, err := io.WriteString(c.conn, encodeCommand([]string{cmd, arg})); err != nil { +// handshakeOn sends one connection-setup command (AUTH or SELECT) and +// requires a non-error reply. Callers own pc exclusively, with a live +// pc.conn whose deadline is already set. +func (c *respClient) handshakeOn(pc *respConn, cmd, arg string) error { + if _, err := io.WriteString(pc.conn, encodeCommand([]string{cmd, arg})); err != nil { return fmt.Errorf("resp: %s: write: %w", cmd, err) } - v, err := decodeReply(c.r, 0) + v, err := decodeReply(pc.r, 0) if err != nil { return fmt.Errorf("resp: %s: read reply: %w", cmd, err) } @@ -258,15 +415,16 @@ func (c *respClient) handshakeLocked(cmd, arg string) error { return nil } -// closeLocked closes and clears c's connection, if any. Callers must hold -// c.mu. -func (c *respClient) closeLocked() { - if c.conn == nil { +// closeConn closes and clears pc's connection, if any, discarding it — +// the connection itself is never reused after an error, only the slot pc +// occupies, which pipeline always returns to c.free regardless. +func closeConn(pc *respConn) { + if pc.conn == nil { return } - _ = c.conn.Close() - c.conn = nil - c.r = nil + _ = pc.conn.Close() + pc.conn = nil + pc.r = nil } // encodeCommand renders args as a RESP2 command array: diff --git a/resp_test.go b/resp_test.go index 07de6cd..1b761f9 100644 --- a/resp_test.go +++ b/resp_test.go @@ -9,6 +9,7 @@ import ( "net" "strconv" "strings" + "sync" "testing" "time" @@ -42,7 +43,7 @@ func TestEncodeCommand(t *testing.T) { // string, error, integer, bulk string (including the null bulk "$-1"), and // 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 -// resp.go (attemptPipelineLocked, handshakeLocked, and decodeArray's own +// resp.go (attemptPipelineOn, handshakeOn, and decodeArray's own // element loop, which passes depth+1). func TestDecodeReply(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 // latency is bounded by respCallTimeout rather than left to block // 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. func newHungListener(t *testing.T) net.Listener { t.Helper() @@ -613,7 +614,7 @@ func (fakeAddr) String() string { return "fake" } // fakeConn implements net.Conn, returning writeErr from Write and // setDeadlineErr from SetDeadline when set, so a test can drive respClient // 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. type fakeConn struct { writeErr error @@ -643,6 +644,19 @@ func (c fakeConn) SetDeadline(time.Time) error { func (fakeConn) SetReadDeadline(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 --- // TestDialTimeoutFor covers dialTimeoutFor's clamp: a deadline farther away @@ -688,45 +702,44 @@ func TestRespDeadlineExceededErr(t *testing.T) { assert.True(t, err.Temporary()) } -// --- ensureConnLocked / attemptPipelineLocked / handshakeLocked error -// branches only reachable via a pre-set connection, not a real dial --- +// --- ensureConnOn / attemptPipelineOn / handshakeOn error branches only +// reachable via a pre-set connection, not a real dial --- -// TestRESPClient_EnsureConnLocked_Errors covers ensureConnLocked's two -// error returns that never touch the network: a reused connection whose +// TestRESPClient_EnsureConnOn_Errors covers ensureConnOn's two error +// returns that never touch the network: a reused connection whose // SetDeadline itself fails, and a deadline that has already elapsed before -// a fresh dial would even start. -func TestRESPClient_EnsureConnLocked_Errors(t *testing.T) { +// a fresh dial would even start. Each case builds its own *respConn slot +// 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 { - name string - client func() *respClient deadline time.Time + pc func() *respConn + name string + addr string wantErrContains string }{ { name: "reused connection: SetDeadline failure surfaces", - client: func() *respClient { - return &respClient{conn: fakeConn{setDeadlineErr: errors.New("stub: set deadline failed")}} + pc: func() *respConn { + return &respConn{conn: fakeConn{setDeadlineErr: errors.New("stub: set deadline failed")}} }, deadline: time.Now().Add(time.Second), wantErrContains: "set deadline", }, { - name: "no connection yet: an already-elapsed deadline never dials", - client: func() *respClient { - return &respClient{addr: "127.0.0.1:1"} - }, + name: "no connection yet: an already-elapsed deadline never dials", + pc: func() *respConn { return &respConn{} }, + addr: "127.0.0.1:1", deadline: time.Now().Add(-time.Second), wantErrContains: "call deadline already elapsed", }, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - client := c.client() - // ensureConnLocked documents "callers must hold c.mu" — honor - // that precondition even though this test is single-goroutine. - client.mu.Lock() - err := client.ensureConnLocked(c.deadline) - client.mu.Unlock() + client := &respClient{addr: c.addr} + err := client.ensureConnOn(c.pc(), c.deadline) require.Error(t, err) 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 -// shape shared by attemptPipelineLocked's command write and -// handshakeLocked's own write: both fail deterministically on an -// already-connected socket that refuses a write — distinct from a dial -// failure, which never reaches either code path. +// shape shared by attemptPipelineOn's command write and handshakeOn's own +// write: both fail deterministically on an already-connected socket that +// refuses a write — distinct from a dial failure, which never reaches +// either code path. func TestRESPClient_WriteError_BrokenPipe(t *testing.T) { cases := []struct { - run func(c *respClient) error + run func(c *respClient, pc *respConn) error name string wantErrContains string }{ { - name: "attemptPipelineLocked: command write fails", - run: func(c *respClient) error { - _, err := c.attemptPipelineLocked([][]string{{"GET", "k"}}, time.Now().Add(time.Second)) + name: "attemptPipelineOn: command write fails", + run: func(c *respClient, pc *respConn) error { + _, err := c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, time.Now().Add(time.Second)) return err }, wantErrContains: "write", }, { - name: "handshakeLocked: AUTH write fails", - run: func(c *respClient) error { - return c.handshakeLocked("AUTH", "pw") + name: "handshakeOn: AUTH write fails", + run: func(c *respClient, pc *respConn) error { + return c.handshakeOn(pc, "AUTH", "pw") }, wantErrContains: "AUTH", }, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - client := &respClient{ + client := &respClient{} + pc := &respConn{ conn: fakeConn{writeErr: errors.New("stub: broken pipe")}, r: bufio.NewReader(strings.NewReader("")), } - // attemptPipelineLocked/handshakeLocked document "callers must - // hold c.mu" — honor that precondition even though this test - // is single-goroutine. - client.mu.Lock() - err := c.run(client) - client.mu.Unlock() + err := c.run(client, pc) require.Error(t, err) assert.Contains(t, err.Error(), c.wantErrContains) }) @@ -779,13 +788,14 @@ func TestRESPClient_WriteError_BrokenPipe(t *testing.T) { } // TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces covers -// ensureConnLocked's AUTH-failure branch and handshakeLocked's respErr -// branch together: a RESP error reply to AUTH must fail the call, not be -// silently treated as success. It calls attemptPipelineLocked directly -// (one attempt, one scripted connection) rather than the public do/ -// pipeline, which would retry once more against a second connection the -// fakeRESPServer fixture — built for a server-initiated closeConn, not a -// client-initiated close on handshake failure — cannot script cleanly. +// ensureConnOn's AUTH-failure branch and handshakeOn's respErr branch +// together: a RESP error reply to AUTH must fail the call, not be +// silently treated as success. It acquires one pooled slot and calls +// attemptPipelineOn directly (one attempt, one scripted connection) +// rather than the public do/pipeline, which would retry once more against +// a second connection the fakeRESPServer fixture — built for a +// server-initiated closeConn, not a client-initiated close on handshake +// failure — cannot script cleanly. func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) { ln := newFakeListener(t) runFakeRESPServer(t, ln, []respStep{ @@ -793,11 +803,10 @@ func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) { }) c := newRESPClient(ln.Addr().String(), "wrong-pw", 0) - // attemptPipelineLocked documents "callers must hold c.mu" — honor that - // precondition even though this test is single-goroutine. - c.mu.Lock() - _, err := c.attemptPipelineLocked([][]string{{"GET", "k"}}, time.Now().Add(2*time.Second)) - c.mu.Unlock() + deadline := time.Now().Add(2 * time.Second) + pc, err := c.acquire(deadline) + require.NoError(t, err) + _, err = c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, deadline) require.Error(t, err) 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) + } +}