From 4e2bfeb14204019c3fa3fb70aa28bd9157cd6853 Mon Sep 17 00:00:00 2001 From: mhx Date: Fri, 4 Sep 2026 11:07:18 +0200 Subject: [PATCH] =?UTF-8?q?=E2=AC=86=EF=B8=8F=20bump=20simpleredis=20to=20?= =?UTF-8?q?the=20pooled=20build,=20teach=20the=20e2e=20mock=20RESP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pooled simpleredis sends RESP arrays instead of inline commands, so the e2e redis stand-in answered +OK to the "*2" header and the plugin read "OK" as a cached verdict, blocking a request that should pass. The mock now parses RESP arrays, still accepts inline, and handles MGET. Points at pool-redis-connections until maxlerebourg/simpleredis#8 is tagged. Co-Authored-By: Claude Opus 5 (1M context) --- go.mod | 2 +- go.sum | 4 +- tests/e2e/mock/mocklapi/main.go | 76 +++- .../maxlerebourg/simpleredis/README.md | 9 +- .../maxlerebourg/simpleredis/simpleredis.go | 398 ++++++++++++------ vendor/modules.txt | 2 +- 6 files changed, 353 insertions(+), 138 deletions(-) diff --git a/go.mod b/go.mod index c0bed0f..b8feafe 100644 --- a/go.mod +++ b/go.mod @@ -4,5 +4,5 @@ go 1.22.12 require ( github.com/leprosus/golang-ttl-map v1.1.7 - github.com/maxlerebourg/simpleredis v1.0.12 + github.com/maxlerebourg/simpleredis v1.0.13-0.20260904085131-f8801cc098d2 ) diff --git a/go.sum b/go.sum index 9aacc76..e854550 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,4 @@ github.com/leprosus/golang-ttl-map v1.1.7 h1:cF4AAFDDnJTFSV+/42sKLhmMluvLdRlCGS2UaifH6UM= github.com/leprosus/golang-ttl-map v1.1.7/go.mod h1:4QWHJPeVBbrkhOhXdhCv9IEiyj/YzkO04/iexy4vSe0= -github.com/maxlerebourg/simpleredis v1.0.12 h1:VsJpk2l8U9QqxOWbYnEXbrs7iSNZFm2fNmFvlxYlQNk= -github.com/maxlerebourg/simpleredis v1.0.12/go.mod h1:lT4LX02SOsE9PxUcSrz1QW5ZnO86gPbaiYBxmtcXEls= +github.com/maxlerebourg/simpleredis v1.0.13-0.20260904085131-f8801cc098d2 h1:fSYQSdirObi0BVp+0BS8QTuY4SmHwpm588jz6bzsF20= +github.com/maxlerebourg/simpleredis v1.0.13-0.20260904085131-f8801cc098d2/go.mod h1:lT4LX02SOsE9PxUcSrz1QW5ZnO86gPbaiYBxmtcXEls= diff --git a/tests/e2e/mock/mocklapi/main.go b/tests/e2e/mock/mocklapi/main.go index 65f5ad6..26a5d36 100644 --- a/tests/e2e/mock/mocklapi/main.go +++ b/tests/e2e/mock/mocklapi/main.go @@ -15,10 +15,12 @@ import ( "bufio" "encoding/json" "flag" + "fmt" "io" "log" "net" "net/http" + "strconv" "strings" "sync" ) @@ -49,10 +51,10 @@ func list(m map[string]Decision) []Decision { return out } -// --- Redis mock (inline-command wire format, as spoken by simpleredis) --- +// --- Redis mock (RESP arrays, as spoken by simpleredis; inline still accepted) --- // serveRedis is a hardcoded stand-in. When verdicts is true it plays a replica -// that holds decisions: every line is scanned for known IPs, 1.2.3.4 → "f" +// that holds decisions: keys are scanned for known IPs, 1.2.3.4 → "f" // (clean), 1.2.3.5 → "t" (banned); any other GET is a miss ($-1). When verdicts // is false it plays the primary and answers every GET with a miss, so a // scenario can prove reads are served from the replica and not the primary. @@ -73,18 +75,25 @@ func serveRedis(addr string, verdicts bool) { defer conn.Close() rd := bufio.NewReader(conn) for { - line, _, err := rd.ReadLine() + args, err := readRedisCommand(rd) if err != nil { return } - s := string(line) - switch { - case verdicts && strings.Contains(s, "1.2.3.4"): - conn.Write([]byte("$1\r\nf\r\n")) - case verdicts && strings.Contains(s, "1.2.3.5"): - conn.Write([]byte("$1\r\nt\r\n")) - case strings.HasPrefix(strings.ToUpper(s), "GET "): - conn.Write([]byte("$-1\r\n")) + if len(args) == 0 { + continue + } + switch strings.ToUpper(args[0]) { + case "GET": + if len(args) < 2 { + conn.Write([]byte("$-1\r\n")) + continue + } + conn.Write([]byte(redisValue(verdicts, args[1]))) + case "MGET": + fmt.Fprintf(conn, "*%d\r\n", len(args)-1) + for _, key := range args[1:] { + conn.Write([]byte(redisValue(verdicts, key))) + } default: conn.Write([]byte("+OK\r\n")) } @@ -93,6 +102,51 @@ func serveRedis(addr string, verdicts bool) { } } +func redisValue(verdicts bool, key string) string { + switch { + case verdicts && strings.Contains(key, "1.2.3.4"): + return "$1\r\nf\r\n" + case verdicts && strings.Contains(key, "1.2.3.5"): + return "$1\r\nt\r\n" + default: + return "$-1\r\n" + } +} + +// readRedisCommand reads one RESP array, falling back to a whitespace split for +// the inline commands older simpleredis releases sent. +func readRedisCommand(rd *bufio.Reader) ([]string, error) { + header, err := rd.ReadString('\n') + if err != nil { + return nil, err + } + header = strings.TrimRight(header, "\r\n") + if !strings.HasPrefix(header, "*") { + return strings.Fields(header), nil + } + count, err := strconv.Atoi(header[1:]) + if err != nil { + return nil, err + } + args := make([]string, 0, count) + for i := 0; i < count; i++ { + sizeLine, err := rd.ReadString('\n') + if err != nil { + return nil, err + } + size, err := strconv.Atoi(strings.TrimRight(sizeLine, "\r\n")[1:]) + if err != nil { + return nil, err + } + buf := make([]byte, size+2) + if _, err := io.ReadFull(rd, buf); err != nil { + return nil, err + } + args = append(args, string(buf[:size])) + } + return args, nil +} + func main() { lapiAddr := flag.String("lapi-addr", "127.0.0.1:8090", "address for the LAPI mock") // The stub upstream Traefik proxies allowed requests to — the binary-suite diff --git a/vendor/github.com/maxlerebourg/simpleredis/README.md b/vendor/github.com/maxlerebourg/simpleredis/README.md index b2d8e19..e51d3fb 100644 --- a/vendor/github.com/maxlerebourg/simpleredis/README.md +++ b/vendor/github.com/maxlerebourg/simpleredis/README.md @@ -1,8 +1,13 @@ # simpleredis -Minimal go redis with only `get`, `set` and `delete` operation. +Minimal go redis with only `get`, `mget`, `set` and `delete` operation. It supports password authentication with redis. With **NO** external dependencies. +Connections are pooled: a command reuses an already authenticated connection when +one is idle, so `AUTH` and `SELECT` are paid once per connection instead of once +per command. A `SimpleRedis` is safe for concurrent use and must not be copied +once initialized. + ## Example ```go import simpleredis "github.com/maxlerebourg/simpleredis" @@ -11,7 +16,7 @@ var redis simpleredis.SimpleRedis redis.Init("redis:6379", "", "") // redisHost, redisPass, redisDatabase -err := redis.Set("test", []bytes("whatever"), 60), // Set key "test" with "whatever" for 60 seconds +err := redis.Set("test", []byte("whatever"), 60) // Set key "test" with "whatever" for 60 seconds if err != nil { ... } diff --git a/vendor/github.com/maxlerebourg/simpleredis/simpleredis.go b/vendor/github.com/maxlerebourg/simpleredis/simpleredis.go index c939edb..917faa6 100644 --- a/vendor/github.com/maxlerebourg/simpleredis/simpleredis.go +++ b/vendor/github.com/maxlerebourg/simpleredis/simpleredis.go @@ -1,14 +1,16 @@ // Package simpleredis implements utility routines for interacting. -// It supports currently the following operations: GET, SET, DELETE, +// It supports currently the following operations: GET, MGET, SET, DELETE, // and support timetoleave for keys. package simpleredis import ( "bufio" - "fmt" + "errors" + "io" "net" - "net/textproto" + "strconv" "strings" + "sync" "time" ) @@ -21,13 +23,30 @@ const ( RedisIssue = "redis:issue?" ) -// A redisCmd is used to communicate with redis at low level using commands. -type redisCmd struct { - Command string - Name string - Data []byte - Duration int64 - Error error +const ( + maxIdleConns = 8 + idleTimeout = 30 * time.Second + dialTimeout = 2 * time.Second + ioTimeout = 1 * time.Second +) + +var ( + errUnreachable = errors.New(RedisUnreachable) + errMiss = errors.New(RedisMiss) + errTimeout = errors.New(RedisTimeout) + errNoAuth = errors.New(RedisNoAuth) + errIssue = errors.New(RedisIssue) +) + +type pooledConn struct { + netConn net.Conn + reader *bufio.Reader + writer *bufio.Writer + lastUsed time.Time +} + +func (c *pooledConn) close() { + _ = c.netConn.Close() } // A SimpleRedis is used to communicate with redis. @@ -35,97 +54,9 @@ type SimpleRedis struct { host string pass string database string -} -func genRedisArray(params ...[]byte) []byte { - MSG := "" - for cntr := 0; cntr < len(params); cntr++ { - MSG = strings.Join([]string{MSG, string(params[cntr])}, " ") - } - MSG = strings.Trim(MSG, " ") - MSG = strings.Join([]string{MSG, "\r\n"}, "") - return []byte(MSG) -} - -func send(wr *textproto.Writer, method string, data []byte) { - if err := wr.PrintfLine(string(data)); err != nil { - fmt.Printf("redis:%s %s", method, err.Error()) - } -} - -func (sr *SimpleRedis) waitRedis(reader *textproto.Reader, channel chan redisCmd) { - for { - select { - case <-time.After(time.Second * 1): - channel <- redisCmd{Error: fmt.Errorf(RedisTimeout)} - return - default: - read, _ := reader.ReadLineBytes() - if string(read) != "+OK" { - channel <- redisCmd{Error: fmt.Errorf(RedisNoAuth)} - return - } - } - // breaks out of for - break - } -} - -func (sr *SimpleRedis) askRedis(cmd redisCmd, channel chan redisCmd) redisCmd { - dialer := net.Dialer{Timeout: 2 * time.Second} - conn, err := dialer.Dial("tcp", sr.host) - if err != nil { - return redisCmd{Error: fmt.Errorf(RedisUnreachable)} - } - defer func() { - if err := conn.Close(); err != nil { - fmt.Printf("redis:connClose %s", err.Error()) - } - }() - - writer := textproto.NewWriter(bufio.NewWriter(conn)) - reader := textproto.NewReader(bufio.NewReader(conn)) - - if sr.pass != "" { - data := genRedisArray([]byte("AUTH"), []byte(sr.pass)) - send(writer, "auth", data) - sr.waitRedis(reader, channel) - } - - if sr.database != "" { - data := genRedisArray([]byte("SELECT"), []byte(sr.database)) - send(writer, "select", data) - sr.waitRedis(reader, channel) - } - - switch cmd.Command { - case "SET": - data := genRedisArray([]byte("SET"), []byte(cmd.Name), cmd.Data, []byte("EX"), []byte(fmt.Sprintf("%d", cmd.Duration))) - send(writer, "set", data) - case "DEL": - data := genRedisArray([]byte("DEL"), []byte(cmd.Name)) - send(writer, "del", data) - case "GET": - data := genRedisArray([]byte("GET"), []byte(cmd.Name)) - send(writer, "get", data) - for { - select { - case <-time.After(time.Second * 1): - return redisCmd{Error: fmt.Errorf(RedisTimeout)} - default: - read, _ := reader.ReadLineBytes() - str := string(read) - if strings.Contains(str, "-NOAUTH") { - return redisCmd{Error: fmt.Errorf(RedisNoAuth)} - } else if str == "$-1" { - return redisCmd{Error: fmt.Errorf(RedisMiss)} - } - read, _ = reader.ReadLineBytes() - return redisCmd{Data: read} - } - } - } - return redisCmd{Error: fmt.Errorf(RedisIssue)} + mu sync.Mutex + idle []*pooledConn } // Init sets the redisHost used to connect to redis. @@ -137,36 +68,261 @@ func (sr *SimpleRedis) Init(host, pass, database string) { // Get fetches the value for key name in redis. func (sr *SimpleRedis) Get(name string) ([]byte, error) { - cmd := redisCmd{ - Command: "GET", - Name: name, + values, err := sr.exec([]byte("GET"), []byte(name)) + if err != nil { + return nil, err } - channel := make(chan redisCmd) - resp := sr.askRedis(cmd, channel) - if resp.Error != nil { - return nil, resp.Error + if len(values) != 1 { + return nil, errIssue } - return resp.Data, nil + return values[0], nil +} + +// MGet fetches the values for keys names in redis, nil where a key is missing. +func (sr *SimpleRedis) MGet(names []string) ([][]byte, error) { + if len(names) == 0 { + return nil, nil + } + args := make([][]byte, 0, len(names)+1) + args = append(args, []byte("MGET")) + for _, name := range names { + args = append(args, []byte(name)) + } + values, err := sr.exec(args...) + if err != nil { + return nil, err + } + if len(values) != len(names) { + return nil, errIssue + } + return values, nil } // Set updates the value for key name in redis with value data for duration. func (sr *SimpleRedis) Set(name string, data []byte, duration int64) error { - cmd := redisCmd{ - Command: "SET", - Name: name, - Data: data, - Duration: duration, - } - sr.askRedis(cmd, nil) - return nil + _, err := sr.exec([]byte("SET"), []byte(name), data, []byte("EX"), []byte(strconv.FormatInt(duration, 10))) + return err } // Del removes the key name in redis. func (sr *SimpleRedis) Del(name string) error { - cmd := redisCmd{ - Command: "DEL", - Name: name, + _, err := sr.exec([]byte("DEL"), []byte(name)) + return err +} + +func (sr *SimpleRedis) exec(args ...[]byte) ([][]byte, error) { + conn, reused, err := sr.borrow() + if err != nil { + return nil, err } - sr.askRedis(cmd, nil) - return nil + values, reusable, err := sr.do(conn, args) + sr.release(conn, reusable) + if err == nil || reusable || !reused { + return values, err + } + conn, err = sr.dial() + if err != nil { + return nil, err + } + values, reusable, err = sr.do(conn, args) + sr.release(conn, reusable) + return values, err +} + +func (sr *SimpleRedis) borrow() (*pooledConn, bool, error) { + var reused *pooledConn + var stale []*pooledConn + now := time.Now() + + sr.mu.Lock() + for len(sr.idle) > 0 { + conn := sr.idle[len(sr.idle)-1] + sr.idle = sr.idle[:len(sr.idle)-1] + if now.Sub(conn.lastUsed) < idleTimeout { + reused = conn + break + } + stale = append(stale, conn) + } + sr.mu.Unlock() + + for _, conn := range stale { + conn.close() + } + if reused != nil { + return reused, true, nil + } + conn, err := sr.dial() + return conn, false, err +} + +func (sr *SimpleRedis) release(conn *pooledConn, reusable bool) { + if !reusable { + conn.close() + return + } + conn.lastUsed = time.Now() + + sr.mu.Lock() + if len(sr.idle) >= maxIdleConns { + sr.mu.Unlock() + conn.close() + return + } + sr.idle = append(sr.idle, conn) + sr.mu.Unlock() +} + +func (sr *SimpleRedis) dial() (*pooledConn, error) { + dialer := net.Dialer{Timeout: dialTimeout} + netConn, err := dialer.Dial("tcp", sr.host) + if err != nil { + return nil, errUnreachable + } + conn := &pooledConn{ + netConn: netConn, + reader: bufio.NewReader(netConn), + writer: bufio.NewWriter(netConn), + } + + if sr.pass != "" { + if _, _, err = sr.do(conn, [][]byte{[]byte("AUTH"), []byte(sr.pass)}); err != nil { + conn.close() + return nil, err + } + } + if sr.database != "" { + if _, _, err = sr.do(conn, [][]byte{[]byte("SELECT"), []byte(sr.database)}); err != nil { + conn.close() + return nil, err + } + } + return conn, nil +} + +func (sr *SimpleRedis) do(conn *pooledConn, args [][]byte) ([][]byte, bool, error) { + if err := conn.netConn.SetDeadline(time.Now().Add(ioTimeout)); err != nil { + return nil, false, errUnreachable + } + if err := writeCommand(conn.writer, args); err != nil { + return nil, false, ioError(err) + } + values, clean, err := readReply(conn.reader) + if err != nil && !clean { + return nil, false, ioError(err) + } + return values, true, err +} + +func writeCommand(writer *bufio.Writer, args [][]byte) error { + if _, err := writer.WriteString("*" + strconv.Itoa(len(args)) + "\r\n"); err != nil { + return err + } + for _, arg := range args { + if _, err := writer.WriteString("$" + strconv.Itoa(len(arg)) + "\r\n"); err != nil { + return err + } + if _, err := writer.Write(arg); err != nil { + return err + } + if _, err := writer.WriteString("\r\n"); err != nil { + return err + } + } + return writer.Flush() +} + +func readReply(reader *bufio.Reader) ([][]byte, bool, error) { + line, err := readLine(reader) + if err != nil { + return nil, false, err + } + if len(line) == 0 { + return nil, false, errIssue + } + + switch line[0] { + case '+', ':': + return [][]byte{line[1:]}, true, nil + case '-': + return nil, true, replyError(line[1:]) + case '$': + data, bulkErr := readBulk(reader, line) + if bulkErr == errMiss { + return nil, true, errMiss + } + if bulkErr != nil { + return nil, false, bulkErr + } + return [][]byte{data}, true, nil + case '*': + count, convErr := strconv.Atoi(string(line[1:])) + if convErr != nil || count < 0 { + return nil, false, errIssue + } + values := make([][]byte, count) + for i := 0; i < count; i++ { + head, headErr := readLine(reader) + if headErr != nil { + return nil, false, headErr + } + data, bulkErr := readBulk(reader, head) + if bulkErr == errMiss { + continue + } + if bulkErr != nil { + return nil, false, bulkErr + } + values[i] = data + } + return values, true, nil + default: + return nil, false, errIssue + } +} + +func readBulk(reader *bufio.Reader, head []byte) ([]byte, error) { + if len(head) == 0 || head[0] != '$' { + return nil, errIssue + } + length, err := strconv.Atoi(string(head[1:])) + if err != nil { + return nil, errIssue + } + if length < 0 { + return nil, errMiss + } + data := make([]byte, length+2) + if _, err = io.ReadFull(reader, data); err != nil { + return nil, err + } + return data[:length], nil +} + +func readLine(reader *bufio.Reader) ([]byte, error) { + line, err := reader.ReadBytes('\n') + if err != nil { + return nil, err + } + if len(line) < 2 || line[len(line)-2] != '\r' { + return nil, errIssue + } + return line[:len(line)-2], nil +} + +func replyError(message []byte) error { + text := string(message) + for _, prefix := range []string{"NOAUTH", "WRONGPASS", "NOPERM", "ERR Client sent AUTH"} { + if strings.HasPrefix(text, prefix) { + return errNoAuth + } + } + return errors.New(text) +} + +func ioError(err error) error { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + return errTimeout + } + return errUnreachable } diff --git a/vendor/modules.txt b/vendor/modules.txt index af43e44..cde6ca9 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,6 +1,6 @@ # github.com/leprosus/golang-ttl-map v1.1.7 ## explicit; go 1.15 github.com/leprosus/golang-ttl-map -# github.com/maxlerebourg/simpleredis v1.0.12 +# github.com/maxlerebourg/simpleredis v1.0.13-0.20260904085131-f8801cc098d2 ## explicit; go 1.22 github.com/maxlerebourg/simpleredis