diff --git a/resp.go b/resp.go index 5afdec2..82594b3 100644 --- a/resp.go +++ b/resp.go @@ -224,7 +224,15 @@ func (c *respClient) do(args ...string) (any, error) { return replies[0], nil } -// pipeline sends every command in cmds on one round trip, over one pooled +// pipeline sends every command in cmds on one round trip and returns their +// decoded replies in the same order. It encodes cmds once and hands off to +// pipelineEncoded, which owns the connection/retry logic. +func (c *respClient) pipeline(cmds [][]string) ([]any, error) { + return c.pipelineEncoded(encodeCommands(cmds), len(cmds)) +} + +// pipelineEncoded runs pipeline's connection/retry logic over an +// already-encoded RESP2 payload representing n commands, over one pooled // connection acquired for the duration of the call, and returns their // decoded replies in the same order. // @@ -241,8 +249,10 @@ func (c *respClient) do(args ...string) (any, error) { // 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) { +// whether or not its connection survived the call. payload is encoded once +// by the caller, so the retry re-sends the identical bytes without +// re-encoding cmds a second time. +func (c *respClient) pipelineEncoded(payload []byte, n int) ([]any, error) { deadline := time.Now().Add(respCallTimeout) pc, err := c.acquire(deadline) @@ -251,7 +261,7 @@ func (c *respClient) pipeline(cmds [][]string) ([]any, error) { } defer func() { c.free <- pc }() - replies, err := c.attemptPipelineOn(pc, cmds, deadline) + replies, err := c.attemptEncodedOn(pc, payload, n, deadline) if err == nil { return replies, nil } @@ -260,7 +270,7 @@ func (c *respClient) pipeline(cmds [][]string) ([]any, error) { return nil, err } - replies, err = c.attemptPipelineOn(pc, cmds, deadline) + replies, err = c.attemptEncodedOn(pc, payload, n, deadline) if err != nil { closeConn(pc) return nil, err @@ -328,28 +338,24 @@ func (respDeadlineExceededErr) Error() string { return "resp: call deadline al func (respDeadlineExceededErr) Timeout() bool { return true } func (respDeadlineExceededErr) Temporary() bool { return true } -// 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) { +// attemptEncodedOn runs one full attempt of an already-encoded RESP2 +// payload representing n commands 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) attemptEncodedOn(pc *respConn, payload []byte, n int, deadline time.Time) ([]any, error) { if err := c.ensureConnOn(pc, deadline); err != nil { return nil, err } - var buf strings.Builder - for _, args := range cmds { - buf.WriteString(encodeCommand(args)) - } - if _, err := io.WriteString(pc.conn, buf.String()); err != nil { + if _, err := pc.conn.Write(payload); err != nil { return nil, fmt.Errorf("resp: write: %w", err) } - replies := make([]any, len(cmds)) - for i := range cmds { + replies := make([]any, n) + for i := 0; i < n; i++ { v, err := decodeReply(pc.r, 0) 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, n, err) } replies[i] = v } @@ -402,7 +408,7 @@ func (c *respClient) ensureConnOn(pc *respConn, deadline time.Time) error { // 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 { + if _, err := pc.conn.Write(encodeCommands([][]string{{cmd, arg}})); err != nil { return fmt.Errorf("resp: %s: write: %w", cmd, err) } v, err := decodeReply(pc.r, 0) @@ -427,15 +433,27 @@ func closeConn(pc *respConn) { pc.r = nil } +// encodeCommands renders every command in cmds as consecutive RESP2 arrays, +// one fmt.Appendf call per header/argument appending into a single growing +// slice — no strings.Builder and no string/[]byte conversion. Measured +// under yaegi v0.16.1: cheaper than either presizing (the interpreted size +// loop cost more than native slice regrowth) or appending raw bytes +// directly (append with a string operand copies it under yaegi). +func encodeCommands(cmds [][]string) []byte { + var dst []byte + for _, args := range cmds { + dst = fmt.Appendf(dst, "*%d\r\n", len(args)) + for _, a := range args { + dst = fmt.Appendf(dst, "$%d\r\n%s\r\n", len(a), a) + } + } + return dst +} + // encodeCommand renders args as a RESP2 command array: // "*N\r\n$len\r\narg\r\n..." for each arg. func encodeCommand(args []string) string { - var b strings.Builder - fmt.Fprintf(&b, "*%d\r\n", len(args)) - for _, a := range args { - fmt.Fprintf(&b, "$%d\r\n%s\r\n", len(a), a) - } - return b.String() + return string(encodeCommands([][]string{args})) } // decodeReply reads and decodes one RESP2 reply from r. depth is 0 for a @@ -561,10 +579,11 @@ func decodeArray(r *bufio.Reader, lenField string, depth int) (any, error) { // seconds", used by the response cache (cache.go) to store a cached // response. ttl is rounded up to whole seconds and floored at 1s — Redis's // EX argument is whole seconds only — matching redisStore.incrBy's EXPIRE -// rounding exactly. val is converted to a string via a bare string(val): a -// Go string is just a byte sequence, so this conversion, and -// encodeCommand's own len(a)-based framing, are lossless for arbitrary -// binary data, not just UTF-8 text. +// rounding exactly. The payload is built directly with fmt.Appendf rather +// than going through do/encodeCommands: val (a cached response, often tens +// of KB) is framed straight from its own bytes — "%s" on a []byte writes +// it verbatim — avoiding both the string(val) copy the old []string-args +// call required and a second encoding pass. func (c *respClient) setEx(key string, val []byte, ttl time.Duration) error { ttlSeconds := int64(ttl / time.Second) if ttl%time.Second != 0 { @@ -574,10 +593,13 @@ func (c *respClient) setEx(key string, val []byte, ttl time.Duration) error { ttlSeconds = 1 } - reply, err := c.do("SET", key, string(val), "EX", strconv.FormatInt(ttlSeconds, 10)) + ttlArg := strconv.FormatInt(ttlSeconds, 10) + payload := fmt.Appendf(nil, "*5\r\n$3\r\nSET\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n$2\r\nEX\r\n$%d\r\n%s\r\n", len(key), key, len(val), val, len(ttlArg), ttlArg) + replies, err := c.pipelineEncoded(payload, 1) if err != nil { return fmt.Errorf("resp: setEx %q: %w", key, err) } + reply := replies[0] if e, ok := reply.(respErr); ok { return fmt.Errorf("resp: setEx %q: %w", key, e) } diff --git a/resp_test.go b/resp_test.go index 1b761f9..4f32a4e 100644 --- a/resp_test.go +++ b/resp_test.go @@ -39,11 +39,65 @@ func TestEncodeCommand(t *testing.T) { } } +// referenceEncodeCommand is encodeCommand's pre-optimisation implementation +// (fmt.Fprintf into a strings.Builder), kept verbatim as the byte-for-byte +// oracle encodeCommands must match. +func referenceEncodeCommand(args []string) string { + var b strings.Builder + fmt.Fprintf(&b, "*%d\r\n", len(args)) + for _, a := range args { + fmt.Fprintf(&b, "$%d\r\n%s\r\n", len(a), a) + } + return b.String() +} + +// TestEncodeCommands_MatchesReference checks encodeCommands against +// referenceEncodeCommand byte-for-byte across a range of pipelines. +func TestEncodeCommands_MatchesReference(t *testing.T) { + hundredK := strings.Repeat("x", 100000) + cases := []struct { + name string + cmds [][]string + }{ + {name: "one command", cmds: [][]string{{"GET", "k"}}}, + {name: "pipeline of several commands", cmds: [][]string{ + {"SET", "k", "v"}, + {"GET", "k"}, + {"EXPIRE", "k", "10"}, + }}, + {name: "empty string arg", cmds: [][]string{{"SET", "k", ""}}}, + {name: "command with zero args", cmds: [][]string{{}}}, + {name: "zero commands", cmds: nil}, + {name: "binary arg with CRLF, NUL and invalid UTF-8 bytes", cmds: [][]string{{"SET", "k", "a\r\nb\x00c\xff\x80"}}}, + {name: "100000 byte arg (multi-digit length)", cmds: [][]string{{"SET", "k", hundredK}}}, + {name: "arg length crosses 9 to 10 bytes", cmds: [][]string{ + {"SET", "k", strings.Repeat("a", 9)}, + {"SET", "k", strings.Repeat("a", 10)}, + }}, + {name: "arg length crosses 99 to 100 bytes", cmds: [][]string{ + {"SET", "k", strings.Repeat("a", 99)}, + {"SET", "k", strings.Repeat("a", 100)}, + }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var want strings.Builder + for _, args := range c.cmds { + want.WriteString(referenceEncodeCommand(args)) + } + + got := encodeCommands(c.cmds) + + assert.Equal(t, want.String(), string(got)) + }) + } +} + // TestDecodeReply covers every RESP2 type respClient must decode: simple // 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 (attemptPipelineOn, handshakeOn, and decodeArray's own +// resp.go (attemptEncodedOn, handshakeOn, and decodeArray's own // element loop, which passes depth+1). func TestDecodeReply(t *testing.T) { t.Run("simple string", func(t *testing.T) { @@ -524,6 +578,53 @@ func TestRESPClient_SetEx_ServerErrorReplyIsReturned(t *testing.T) { } } +// TestRESPClient_SetEx_WireBytesMatchReference asserts setEx's hand-built +// fmt.Appendf payload is byte-for-byte identical to referenceEncodeCommand's +// framing of the equivalent "SET key val EX seconds" command — catching a +// mis-sized literal segment or wrong length field that a respStep.wantArgs +// check (parsed back through readRESPCommand) would not distinguish from a +// correct encoding. It presets the pooled connection's conn directly +// (fakeConn, recording writes) so ensureConnOn skips the AUTH/SELECT +// handshake entirely, leaving only the SET payload on the wire. +func TestRESPClient_SetEx_WireBytesMatchReference(t *testing.T) { + cases := []struct { + name string + key string + val []byte + ttl time.Duration + }{ + {name: "small ASCII value", key: "k", val: []byte("hello"), ttl: 5 * time.Second}, + {name: "binary value with CRLF, NUL and non-UTF8 bytes", key: "k", val: []byte("a\r\nb\x00c\xff\x80"), ttl: 5 * time.Second}, + {name: "100000 byte value", key: "k", val: bytes.Repeat([]byte("x"), 100000), ttl: 5 * time.Second}, + {name: "ttl rounds 1500ms up to 2s", key: "k", val: []byte("v"), ttl: 1500 * time.Millisecond}, + {name: "ttl floors 0 up to 1s", key: "k", val: []byte("v"), ttl: 0}, + {name: "multi-digit ttl (default 5m cache TTL)", key: "k", val: []byte("v"), ttl: 300 * time.Second}, + {name: "key with CRLF, NUL, 0xff and a literal %s", key: "k\r\n\x00\xff%s", val: []byte("v"), ttl: 5 * time.Second}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + ttlSeconds := int64(c.ttl / time.Second) + if c.ttl%time.Second != 0 { + ttlSeconds++ + } + if ttlSeconds < 1 { + ttlSeconds = 1 + } + want := referenceEncodeCommand([]string{"SET", c.key, string(c.val), "EX", strconv.FormatInt(ttlSeconds, 10)}) + + var written []byte + client := &respClient{free: make(chan *respConn, 1)} + client.free <- &respConn{ + conn: fakeConn{written: &written}, + r: bufio.NewReader(strings.NewReader("+OK\r\n")), + } + + require.NoError(t, client.setEx(c.key, c.val, c.ttl)) + assert.Equal(t, want, string(written)) + }) + } +} + // TestRESPClient_GetBytes_HitReturnsBody covers the cache-hit case: a // bulk reply decodes to the stored bytes with found=true. func TestRESPClient_GetBytes_HitReturnsBody(t *testing.T) { @@ -603,7 +704,8 @@ func TestRESPClient_GetBytes_DownServer_ReturnsError(t *testing.T) { // --- fakeConn: a net.Conn whose Write/SetDeadline fail on command, for // the handful of respClient error branches a real TCP connection cannot be // coaxed into deterministically (a write failing on an otherwise-live -// connection, SetDeadline failing on a reused connection) --- +// connection, SetDeadline failing on a reused connection), and which can +// optionally record what was written for a wire-format assertion --- // fakeAddr is a trivial net.Addr for fakeConn's LocalAddr/RemoteAddr. type fakeAddr struct{} @@ -615,10 +717,14 @@ func (fakeAddr) String() string { return "fake" } // setDeadlineErr from SetDeadline when set, so a test can drive respClient // methods directly against a connection already known bad — bypassing // 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. When +// written is non-nil, every Write appends its argument to *written instead +// of discarding it, letting a test assert on the exact bytes a method put +// on the wire. type fakeConn struct { writeErr error setDeadlineErr error + written *[]byte } func (fakeConn) Read([]byte) (int, error) { return 0, io.EOF } @@ -627,6 +733,9 @@ func (c fakeConn) Write(b []byte) (int, error) { if c.writeErr != nil { return 0, c.writeErr } + if c.written != nil { + *c.written = append(*c.written, b...) + } return len(b), nil } @@ -702,7 +811,7 @@ func TestRespDeadlineExceededErr(t *testing.T) { assert.True(t, err.Temporary()) } -// --- ensureConnOn / attemptPipelineOn / handshakeOn error branches only +// --- ensureConnOn / attemptEncodedOn / handshakeOn error branches only // reachable via a pre-set connection, not a real dial --- // TestRESPClient_EnsureConnOn_Errors covers ensureConnOn's two error @@ -747,7 +856,7 @@ func TestRESPClient_EnsureConnOn_Errors(t *testing.T) { } // TestRESPClient_WriteError_BrokenPipe covers the identical broken-pipe -// shape shared by attemptPipelineOn's command write and handshakeOn's own +// shape shared by attemptEncodedOn's payload 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. @@ -758,9 +867,9 @@ func TestRESPClient_WriteError_BrokenPipe(t *testing.T) { wantErrContains string }{ { - name: "attemptPipelineOn: command write fails", + name: "attemptEncodedOn: payload write fails", run: func(c *respClient, pc *respConn) error { - _, err := c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, time.Now().Add(time.Second)) + _, err := c.attemptEncodedOn(pc, encodeCommands([][]string{{"GET", "k"}}), 1, time.Now().Add(time.Second)) return err }, wantErrContains: "write", @@ -791,7 +900,7 @@ func TestRESPClient_WriteError_BrokenPipe(t *testing.T) { // 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) +// attemptEncodedOn 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 @@ -806,7 +915,7 @@ func TestRESPClient_AuthHandshakeFailure_ErrorReplySurfaces(t *testing.T) { deadline := time.Now().Add(2 * time.Second) pc, err := c.acquire(deadline) require.NoError(t, err) - _, err = c.attemptPipelineOn(pc, [][]string{{"GET", "k"}}, deadline) + _, err = c.attemptEncodedOn(pc, encodeCommands([][]string{{"GET", "k"}}), 1, deadline) require.Error(t, err) assert.Contains(t, err.Error(), "AUTH failed") }