From 7ee635cd27bbfc803e270eee6d64eb1ec30aa733 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 20 Aug 2026 02:16:07 +0100 Subject: [PATCH] fix(resp): bound store-call latency, cap wire lengths, cover fail-closed budget path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bound respClient's worst-case latency to one respCallTimeout per do/pipeline call instead of per read/write/handshake step. Compute a single absolute deadline at call entry, apply it once via conn.SetDeadline after connecting (covering the AUTH/SELECT handshake and the call's own command I/O), and skip the reconnect-retry when the failing attempt was a timeout (errors.As -> net.Error, Timeout()) — a peer that accepted the connection and went silent must not be asked the same non-question twice. Previously each read, write, and dial got its own fresh 2s budget and every caller queued behind the client's single mutex, so a hung (not merely refused) Redis-compatible server could stall one checkAndCount call for 16-32s+. Cap decodeBulk/decodeArray's wire-declared lengths (respMaxBulkLen 64MiB, respMaxArrayLen 1<<20) before allocating: an unguarded "$9223372036854775807\r\n" overflows n+2 into a negative slice length and panics make([]byte, n+2); a merely huge-but-valid length forces an outsized allocation either way. decodeBulk also now verifies its two trailing bytes are CRLF. Bound readLine to respMaxLineLen (64KiB), reading byte-by-byte so it can bail before ever buffering an unbounded line from a peer that never sends '\n'. Fix TestLimiter_FailClosed_BudgetReadReturnsStoreDownViolation: the erroringStore it used fails incrBy too, so checkAndCount's own increment loop already returned a storeDown violation before evaluateScope (and therefore budgetViolation) ever ran, leaving budgetViolation's storeDown branch uncovered. Replace it with a stub that only fails get, so the test's TokensPerDay limit actually drives evaluateScope into budgetViolation's fail-closed path. Document redis_store.go's incrBy as deliberately at-least-once: the reconnect-retry can re-send INCRBY after a lost reply (possible over-count), and fail-open can additionally count in the fallback — both fail-safe (more restrictive), never fail-open in the unsafe direction. Validate Redis.DB >= 0 in newConfiguredLimiter alongside Address, so a negative db is a constructor error rather than a SELECT -1 on the wire. New tests: a hung (accepts, never replies) fake RESP server proving a respClient call and a failOpen limiter call both stay bounded rather than stacking timeouts; decodeReply subtests for an oversized bulk length, a length just over the cap, a huge array length, a bad bulk trailer, and an oversized line; a negative-DB constructor-rejection test. --- limits_test.go | 26 ++++++- llmgateway.go | 3 + llmgateway_test.go | 13 ++++ redis_store.go | 12 +++ redis_store_test.go | 27 +++++++ resp.go | 183 +++++++++++++++++++++++++++++++------------- resp_test.go | 97 +++++++++++++++++++++++ 7 files changed, 303 insertions(+), 58 deletions(-) diff --git a/limits_test.go b/limits_test.go index b8d4226..7537f49 100644 --- a/limits_test.go +++ b/limits_test.go @@ -433,18 +433,38 @@ func TestLimiter_FailClosed_StoreErrorReturnsStoreDownViolation(t *testing.T) { } } +// succeedIncrFailGetStore is a counterStore stub whose incrBy always +// succeeds and whose get always errors. It isolates budgetViolation's own +// storeDown branch (limits.go) from checkAndCount's earlier incrCounter +// one: an erroringStore that fails both methods makes checkAndCount's +// initial req:min/req:day increments fail closed and return before +// evaluateScope — and therefore budgetViolation — ever runs, so a test +// built on it cannot actually prove budgetViolation's own branch works. +type succeedIncrFailGetStore struct { + getErr error +} + +func (s *succeedIncrFailGetStore) incrBy(string, int64, time.Duration) (int64, error) { + return 1, nil +} +func (s *succeedIncrFailGetStore) get(string) (int64, error) { return 0, s.getErr } + // TestLimiter_FailClosed_BudgetReadReturnsStoreDownViolation covers the // budgetViolation fail-closed path specifically (as opposed to the -// request-counter incrCounter path covered above): a store error on a +// request-counter incrCounter path TestLimiter_FailClosed_ +// StoreErrorReturnsStoreDownViolation covers above): a store error on a // token/cost budget read also refuses the request when failOpen is false. +// The store's incrBy succeeds so checkAndCount's req:min/req:day +// increments pass and evaluateScope actually reaches budgetViolation; a +// TokensPerDay limit makes evaluateScope call it. func TestLimiter_FailClosed_BudgetReadReturnsStoreDownViolation(t *testing.T) { - store := &erroringStore{err: errors.New("boom")} + store := &succeedIncrFailGetStore{getErr: errors.New("boom")} l := newLimiter(store, false) scopes := []limitScope{{kind: "user", id: "u", limits: &LimitsConfig{TokensPerDay: 100}}} v := l.checkAndCount(scopes) if v == nil { - t.Fatal("want a storeDown violation on a store error with failOpen=false") + t.Fatal("want a storeDown violation from the budget-read path with failOpen=false") } if !v.storeDown { t.Error("v.storeDown = false, want true") diff --git a/llmgateway.go b/llmgateway.go index e1b18e4..46294f5 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -168,6 +168,9 @@ func newConfiguredLimiter(config *Config) (*limiter, error) { if config.Redis.Address == "" { return nil, errors.New("llmgateway: redis: address must not be empty") } + if config.Redis.DB < 0 { + return nil, fmt.Errorf("llmgateway: redis: db must not be negative, got %d", config.Redis.DB) + } password, err := resolveSecret(config.Redis.Password) if err != nil { diff --git a/llmgateway_test.go b/llmgateway_test.go index 0076295..05fdcbf 100644 --- a/llmgateway_test.go +++ b/llmgateway_test.go @@ -547,6 +547,19 @@ func TestNewGateway_RedisConfigured_EmptyAddress_ReturnsConstructorError(t *test } } +// TestNewGateway_RedisConfigured_NegativeDB_ReturnsConstructorError is +// review item 5: a negative Redis.DB must fail construction immediately, +// same as an empty address, rather than reaching SELECT -1 on the wire. +func TestNewGateway_RedisConfigured_NegativeDB_ReturnsConstructorError(t *testing.T) { + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{"openai": {Type: "openai", APIKey: "k"}} + cfg.Redis = &RedisConfig{Address: "127.0.0.1:0", DB: -1} + next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) + if _, err := New(context.Background(), next, cfg, "llmgw"); err == nil { + t.Fatal("want constructor error for a negative Redis db") + } +} + // 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/redis_store.go b/redis_store.go index c1b6866..d9e7214 100644 --- a/redis_store.go +++ b/redis_store.go @@ -26,6 +26,18 @@ func newRedisStore(client *respClient) *redisStore { // the count. ttl is rounded up to whole seconds (Redis EXPIRE's unit), // with a floor of 1s so a sub-second ttl never turns into EXPIRE 0 (an // immediate delete). +// +// Semantics are deliberately at-least-once, not exactly-once, both in the +// conservative direction (never under-counts): respClient.pipeline's +// reconnect-once retry can re-send this same INCRBY if the first attempt's +// reply was lost after the server already applied it (e.g. the connection +// dropped between the server processing INCRBY and the client reading its +// reply), which can over-count by n on that key; and the limiter's +// failOpen path can additionally count the same request in its in-process +// fallback store when a call to this method errors out after a partial +// success upstream. Both are accepted trade-offs — a rate/budget counter +// that occasionally over-counts by one request's worth is fail-safe (more +// restrictive than reality), never fail-open in the unsafe direction. func (s *redisStore) incrBy(key string, n int64, ttl time.Duration) (int64, error) { ttlSeconds := int64(ttl / time.Second) if ttl%time.Second != 0 { diff --git a/redis_store_test.go b/redis_store_test.go index 447330b..3dfce1b 100644 --- a/redis_store_test.go +++ b/redis_store_test.go @@ -136,3 +136,30 @@ func TestLimiter_FailOpenTrue_DeadRedisAddress_FallsBackAndAllows(t *testing.T) t.Errorf("fallback req:day counter = %d, want 1 (the request must have been counted in the fallback)", got) } } + +// TestLimiter_FailOpenTrue_HungRedisServer_FallsBackQuickly is review item +// 1's limiter-level proof: a hung (accepts the connection, never replies) +// Redis-compatible server must not make a failOpen=true request wait out +// several stacked multi-second timeouts (the pre-fix behavior for a +// checkAndCount call that performs several store operations) — each of +// checkAndCount's two store ops here (req:min, req:day) is bounded by one +// respCallTimeout, not respCallTimeout times every read/write step it +// happens to perform, so the whole call stays well under the old +// multi-times-respCallTimeout stall. +func TestLimiter_FailOpenTrue_HungRedisServer_FallsBackQuickly(t *testing.T) { + ln := newHungListener(t) + store := newRedisStore(newRESPClient(ln.Addr().String(), "", 0)) + l := newLimiter(store, true) + scopes := []limitScope{{kind: "user", id: "u", limits: &LimitsConfig{RequestsPerMinute: 100}}} + + start := time.Now() + v := l.checkAndCount(scopes) + elapsed := time.Since(start) + + if v != nil { + t.Fatalf("want no violation with failOpen=true and a hung store, got %+v", v) + } + if elapsed >= 6*time.Second { + t.Fatalf("elapsed = %v, want well under the old worst case (two bounded store ops, not many stacked timeouts)", elapsed) + } +} diff --git a/resp.go b/resp.go index 5ec5b15..14324d7 100644 --- a/resp.go +++ b/resp.go @@ -2,6 +2,7 @@ package traefikllmgateway import ( "bufio" + "errors" "fmt" "io" "net" @@ -11,18 +12,42 @@ import ( "time" ) -// respDialTimeout, respReadTimeout, and respWriteTimeout bound every -// network operation respClient performs: connecting, writing a pipeline, -// and reading each reply. A hung or unreachable Redis-compatible server -// must never block a request indefinitely — the limiter's fail-open/ -// fail-closed handling (limits.go) depends on a store call returning -// within a bounded time. +// respDialTimeout bounds connecting to the server when no connection +// exists yet. respCallTimeout bounds everything that happens on the +// connection during one do/pipeline call: the AUTH/SELECT handshake (on a +// freshly dialled connection) and the call's own command write and reply +// reads. It is computed ONCE per call and applied via a single +// 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). const ( - respDialTimeout = 2 * time.Second - respReadTimeout = 2 * time.Second - respWriteTimeout = 2 * time.Second + respDialTimeout = 2 * time.Second + respCallTimeout = 2 * time.Second ) +// respMaxBulkLen caps a RESP2 bulk string's declared length. Without a +// cap, a malformed or malicious reply — "$9223372036854775807\r\n" — makes +// n+2 overflow into a negative slice length and panic in make([]byte, +// n+2); a reply merely claiming a huge-but-valid length would otherwise +// force a huge allocation. 64MiB is generous for anything this client +// actually reads (integer counter values and short simple/error strings). +const respMaxBulkLen = 64 << 20 + +// respMaxArrayLen caps a RESP2 array's declared element count, for the +// same reason as respMaxBulkLen: this client only ever pipelines a +// handful of commands, so a reply claiming millions of elements is +// malformed or hostile, not legitimate. +const respMaxArrayLen = 1 << 20 + +// respMaxLineLen caps a single RESP2 header line (the "+", "-", ":", "$", +// or "*" line preceding any body) read by readLine. Without a cap, a peer +// that never sends '\n' makes readLine buffer an unbounded number of +// bytes before erroring. +const respMaxLineLen = 64 << 10 + // 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. @@ -33,10 +58,10 @@ func (e respErr) Error() string { return string(e) } // 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. -// 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. +// 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: @@ -71,23 +96,34 @@ func (c *respClient) do(args ...string) (any, error) { } // pipeline sends every command in cmds on one round trip and returns their -// decoded replies in the same order. On any I/O error — including a -// failure to (re)connect — it closes the current connection, reconnects -// once, and retries the whole pipeline from scratch; a second failure is -// returned to the caller with the connection left closed, so the next call -// lazily reconnects fresh rather than retrying against a connection -// already known bad. +// 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. func (c *respClient) pipeline(cmds [][]string) ([]any, error) { + deadline := time.Now().Add(respCallTimeout) + c.mu.Lock() defer c.mu.Unlock() - replies, err := c.attemptPipelineLocked(cmds) + replies, err := c.attemptPipelineLocked(cmds, deadline) if err == nil { return replies, nil } - c.closeLocked() - replies, err = c.attemptPipelineLocked(cmds) + if isTimeout(err) { + return nil, err + } + + replies, err = c.attemptPipelineLocked(cmds, deadline) if err != nil { c.closeLocked() return nil, err @@ -95,10 +131,17 @@ func (c *respClient) pipeline(cmds [][]string) ([]any, error) { return replies, nil } +// isTimeout reports whether err is (or wraps) a net.Error that timed out. +func isTimeout(err error) bool { + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} + // attemptPipelineLocked runs one full attempt of cmds over c's connection, -// connecting first if needed. Callers must hold c.mu. -func (c *respClient) attemptPipelineLocked(cmds [][]string) ([]any, error) { - if err := c.ensureConnLocked(); err != nil { +// 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 { return nil, err } @@ -106,18 +149,12 @@ func (c *respClient) attemptPipelineLocked(cmds [][]string) ([]any, error) { for _, args := range cmds { buf.WriteString(encodeCommand(args)) } - if err := c.conn.SetWriteDeadline(time.Now().Add(respWriteTimeout)); err != nil { - return nil, fmt.Errorf("resp: set write deadline: %w", err) - } if _, err := io.WriteString(c.conn, buf.String()); err != nil { return nil, fmt.Errorf("resp: write: %w", err) } replies := make([]any, len(cmds)) for i := range cmds { - if err := c.conn.SetReadDeadline(time.Now().Add(respReadTimeout)); err != nil { - return nil, fmt.Errorf("resp: set read deadline: %w", err) - } v, err := decodeReply(c.r) if err != nil { return nil, fmt.Errorf("resp: read reply %d/%d: %w", i+1, len(cmds), err) @@ -128,9 +165,15 @@ func (c *respClient) attemptPipelineLocked(cmds [][]string) ([]any, error) { } // ensureConnLocked dials, authenticates, and selects the database if c has -// no live connection. Callers must hold c.mu. -func (c *respClient) ensureConnLocked() error { +// 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 { + return fmt.Errorf("resp: set deadline: %w", err) + } return nil } @@ -141,6 +184,11 @@ func (c *respClient) ensureConnLocked() error { c.conn = conn c.r = bufio.NewReader(conn) + if err := c.conn.SetDeadline(deadline); err != nil { + c.closeLocked() + return fmt.Errorf("resp: set deadline: %w", err) + } + if c.password != "" { if err := c.handshakeLocked("AUTH", c.password); err != nil { c.closeLocked() @@ -155,18 +203,12 @@ func (c *respClient) ensureConnLocked() error { } // handshakeLocked sends one connection-setup command (AUTH or SELECT) and -// requires a non-error reply. Callers must hold c.mu and have a live -// c.conn. +// 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 := c.conn.SetWriteDeadline(time.Now().Add(respWriteTimeout)); err != nil { - return fmt.Errorf("resp: set write deadline: %w", err) - } if _, err := io.WriteString(c.conn, encodeCommand([]string{cmd, arg})); err != nil { return fmt.Errorf("resp: %s: write: %w", cmd, err) } - if err := c.conn.SetReadDeadline(time.Now().Add(respReadTimeout)); err != nil { - return fmt.Errorf("resp: set read deadline: %w", err) - } v, err := decodeReply(c.r) if err != nil { return fmt.Errorf("resp: %s: read reply: %w", cmd, err) @@ -209,10 +251,11 @@ func encodeCommand(args []string) string { // array ("*-1\r\n") // // The returned error is non-nil only for a transport or protocol failure -// (short read, malformed line, unknown type byte) — a well-formed RESP -// error reply decodes successfully to a respErr value, not to a non-nil -// error, so a caller pipelining several commands can still read every -// reply after one of them fails server-side. +// (short read, malformed line, unknown type byte, an out-of-range or +// mistrailed bulk/array) — a well-formed RESP error reply decodes +// successfully to a respErr value, not to a non-nil error, so a caller +// pipelining several commands can still read every reply after one of +// them fails server-side. func decodeReply(r *bufio.Reader) (any, error) { line, err := readLine(r) if err != nil { @@ -244,32 +287,50 @@ func decodeReply(r *bufio.Reader) (any, error) { } // decodeBulk reads a bulk string body given its already-parsed "$" length -// field lenField, returning nil for a null bulk ("$-1"). +// field lenField, returning nil for a null bulk ("$-1"). It rejects a +// negative length other than -1 and a length exceeding respMaxBulkLen +// before allocating — n+2 would otherwise be able to overflow into a +// negative slice length for a maliciously huge declared length, panicking +// make([]byte, n+2) — and verifies the two bytes following the payload +// are the expected CRLF trailer. func decodeBulk(r *bufio.Reader, lenField string) (any, error) { n, err := strconv.Atoi(lenField) if err != nil { return nil, fmt.Errorf("resp: malformed bulk length %q: %w", lenField, err) } - if n < 0 { + if n == -1 { return nil, nil } - buf := make([]byte, n+2) // payload plus trailing CRLF + if n < -1 || n > respMaxBulkLen { + return nil, fmt.Errorf("resp: bulk length %d out of range (max %d)", n, respMaxBulkLen) + } + + buf := make([]byte, n+2) // payload plus trailing CRLF; n is bounded above, so n+2 cannot overflow if _, err := io.ReadFull(r, buf); err != nil { return nil, fmt.Errorf("resp: read bulk body: %w", err) } + if buf[n] != '\r' || buf[n+1] != '\n' { + return nil, fmt.Errorf("resp: bulk reply missing CRLF trailer") + } return buf[:n], nil } // decodeArray reads n elements given the already-parsed "*" length field -// lenField, returning nil for a null array ("*-1"). +// lenField, returning nil for a null array ("*-1"). It rejects a negative +// length other than -1 and a length exceeding respMaxArrayLen before +// allocating, for the same reason as decodeBulk's cap. func decodeArray(r *bufio.Reader, lenField string) (any, error) { n, err := strconv.Atoi(lenField) if err != nil { return nil, fmt.Errorf("resp: malformed array length %q: %w", lenField, err) } - if n < 0 { + if n == -1 { return nil, nil } + if n < -1 || n > respMaxArrayLen { + return nil, fmt.Errorf("resp: array length %d out of range (max %d)", n, respMaxArrayLen) + } + arr := make([]any, n) for i := 0; i < n; i++ { v, err := decodeReply(r) @@ -282,11 +343,23 @@ func decodeArray(r *bufio.Reader, lenField string) (any, error) { } // readLine reads one CRLF-terminated line from r, with the trailing CRLF -// (or bare LF) stripped. +// (or bare LF) stripped. It reads a byte at a time so it can bail out +// after respMaxLineLen bytes without ever finding '\n' — a peer that +// never terminates a line must not make this buffer an unbounded number +// of bytes first and only reject the result afterward. func readLine(r *bufio.Reader) (string, error) { - line, err := r.ReadString('\n') - if err != nil { - return "", fmt.Errorf("resp: read line: %w", err) + buf := make([]byte, 0, 64) + for { + b, err := r.ReadByte() + if err != nil { + return "", fmt.Errorf("resp: read line: %w", err) + } + if b == '\n' { + return strings.TrimSuffix(string(buf), "\r"), nil + } + if len(buf) >= respMaxLineLen { + return "", fmt.Errorf("resp: reply line exceeds %d bytes", respMaxLineLen) + } + buf = append(buf, b) } - return strings.TrimRight(line, "\r\n"), nil } diff --git a/resp_test.go b/resp_test.go index 31e3aca..985b59d 100644 --- a/resp_test.go +++ b/resp_test.go @@ -3,11 +3,13 @@ package traefikllmgateway import ( "bufio" "bytes" + "fmt" "io" "net" "strconv" "strings" "testing" + "time" ) // TestEncodeCommand pins the exact RESP2 wire format every command must @@ -127,6 +129,48 @@ func TestDecodeReply(t *testing.T) { t.Fatal("want error for an unknown reply type byte") } }) + + // The following four subtests are review item 3: decodeBulk/ + // decodeArray must reject an out-of-range wire length before + // allocating, and decodeBulk must validate its CRLF trailer. + + t.Run("oversized bulk length is a protocol error, not a panic", func(t *testing.T) { + // n+2 (9223372036854775807+2) overflows int64 into a negative + // slice length; make([]byte, n+2) with that unguarded would + // panic. Reaching this line without panicking already proves the + // fix; the error check confirms it fails cleanly too. + if _, err := decodeReply(newReader("$9223372036854775807\r\n")); err == nil { + t.Fatal("want an error for a bulk length exceeding respMaxBulkLen") + } + }) + + t.Run("bulk length just over the cap is a protocol error", func(t *testing.T) { + if _, err := decodeReply(newReader(fmt.Sprintf("$%d\r\n", respMaxBulkLen+1))); err == nil { + t.Fatal("want an error for a bulk length one over respMaxBulkLen") + } + }) + + t.Run("huge array length is a protocol error", func(t *testing.T) { + if _, err := decodeReply(newReader("*1000000000\r\n")); err == nil { + t.Fatal("want an error for an array length exceeding respMaxArrayLen") + } + }) + + t.Run("bulk with wrong trailer is a protocol error", func(t *testing.T) { + if _, err := decodeReply(newReader("$5\r\nhelloXX")); err == nil { + t.Fatal("want an error when a bulk reply's trailing bytes are not CRLF") + } + }) + + t.Run("reply line exceeding the cap is a protocol error, not unbounded buffering", func(t *testing.T) { + // No trailing "\r\n" at all: without the byte-by-byte cap check, + // readLine would keep buffering every byte offered until the + // reader errors (e.g. EOF) rather than bailing out early. + huge := strings.Repeat("x", respMaxLineLen+10) + if _, err := decodeReply(newReader("+" + huge)); err == nil { + t.Fatal("want an error for a reply line exceeding respMaxLineLen") + } + }) } // newReader wraps s in a *bufio.Reader, matching what respClient reads @@ -259,6 +303,38 @@ func newFakeListener(t *testing.T) net.Listener { return ln } +// newHungListener returns a listener whose accepted connections read (and +// discard) anything the client sends but never reply — simulating a +// server that accepted the TCP connection and then went silent (a +// black-holed network path, a wedged process), as opposed to actively +// 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 +// deadline fires, or by the listener's t.Cleanup at test end. +func newHungListener(t *testing.T) net.Listener { + t.Helper() + ln := newFakeListener(t) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer func() { _ = c.Close() }() + buf := make([]byte, 4096) + for { + if _, err := c.Read(buf); err != nil { + return + } + } + }(conn) + } + }() + return ln +} + // TestRESPClient_ConnectSendsAuthThenSelect is the brief's Step-1 // connection-setup case: a configured password sends AUTH before SELECT, // both requiring a non-error reply, and SELECT is sent with the @@ -345,3 +421,24 @@ func TestRESPClient_ReconnectFailsTwice_ReturnsError(t *testing.T) { t.Fatal("want an error dialing a dead address") } } + +// TestRESPClient_HungServer_BoundedByCallDeadline is review item 1: a +// server that accepts the connection and then never replies must not +// block a call past respCallTimeout, and — because that failure is a +// timeout, not a reset/refusal — must not be retried, which would double +// the wait for the exact same non-answer. +func TestRESPClient_HungServer_BoundedByCallDeadline(t *testing.T) { + ln := newHungListener(t) + c := newRESPClient(ln.Addr().String(), "", 0) + + start := time.Now() + _, err := c.do("GET", "k") + elapsed := time.Since(start) + + if err == nil { + t.Fatal("want an error from a server that never replies") + } + if elapsed >= 3*time.Second { + t.Fatalf("elapsed = %v, want < 3s (one bounded call deadline, no retry-doubling on timeout)", elapsed) + } +}