⬆️ bump simpleredis to the pooled build, teach the e2e mock RESP

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) <noreply@anthropic.com>
This commit is contained in:
mhx
2026-09-04 11:07:18 +02:00
co-authored by Claude Opus 5
parent 3ceb9617f2
commit 4e2bfeb142
6 changed files with 353 additions and 138 deletions
+1 -1
View File
@@ -4,5 +4,5 @@ go 1.22.12
require ( require (
github.com/leprosus/golang-ttl-map v1.1.7 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
) )
+2 -2
View File
@@ -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 h1:cF4AAFDDnJTFSV+/42sKLhmMluvLdRlCGS2UaifH6UM=
github.com/leprosus/golang-ttl-map v1.1.7/go.mod h1:4QWHJPeVBbrkhOhXdhCv9IEiyj/YzkO04/iexy4vSe0= 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.13-0.20260904085131-f8801cc098d2 h1:fSYQSdirObi0BVp+0BS8QTuY4SmHwpm588jz6bzsF20=
github.com/maxlerebourg/simpleredis v1.0.12/go.mod h1:lT4LX02SOsE9PxUcSrz1QW5ZnO86gPbaiYBxmtcXEls= github.com/maxlerebourg/simpleredis v1.0.13-0.20260904085131-f8801cc098d2/go.mod h1:lT4LX02SOsE9PxUcSrz1QW5ZnO86gPbaiYBxmtcXEls=
+65 -11
View File
@@ -15,10 +15,12 @@ import (
"bufio" "bufio"
"encoding/json" "encoding/json"
"flag" "flag"
"fmt"
"io" "io"
"log" "log"
"net" "net"
"net/http" "net/http"
"strconv"
"strings" "strings"
"sync" "sync"
) )
@@ -49,10 +51,10 @@ func list(m map[string]Decision) []Decision {
return out 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 // 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 // (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 // 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. // 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() defer conn.Close()
rd := bufio.NewReader(conn) rd := bufio.NewReader(conn)
for { for {
line, _, err := rd.ReadLine() args, err := readRedisCommand(rd)
if err != nil { if err != nil {
return return
} }
s := string(line) if len(args) == 0 {
switch { continue
case verdicts && strings.Contains(s, "1.2.3.4"): }
conn.Write([]byte("$1\r\nf\r\n")) switch strings.ToUpper(args[0]) {
case verdicts && strings.Contains(s, "1.2.3.5"): case "GET":
conn.Write([]byte("$1\r\nt\r\n")) if len(args) < 2 {
case strings.HasPrefix(strings.ToUpper(s), "GET "): conn.Write([]byte("$-1\r\n"))
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: default:
conn.Write([]byte("+OK\r\n")) 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() { func main() {
lapiAddr := flag.String("lapi-addr", "127.0.0.1:8090", "address for the LAPI mock") 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 // The stub upstream Traefik proxies allowed requests to — the binary-suite
+7 -2
View File
@@ -1,8 +1,13 @@
# simpleredis # 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. It supports password authentication with redis.
With **NO** external dependencies. 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 ## Example
```go ```go
import simpleredis "github.com/maxlerebourg/simpleredis" import simpleredis "github.com/maxlerebourg/simpleredis"
@@ -11,7 +16,7 @@ var redis simpleredis.SimpleRedis
redis.Init("redis:6379", "", "") // redisHost, redisPass, redisDatabase 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 { if err != nil {
... ...
} }
+277 -121
View File
@@ -1,14 +1,16 @@
// Package simpleredis implements utility routines for interacting. // 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. // and support timetoleave for keys.
package simpleredis package simpleredis
import ( import (
"bufio" "bufio"
"fmt" "errors"
"io"
"net" "net"
"net/textproto" "strconv"
"strings" "strings"
"sync"
"time" "time"
) )
@@ -21,13 +23,30 @@ const (
RedisIssue = "redis:issue?" RedisIssue = "redis:issue?"
) )
// A redisCmd is used to communicate with redis at low level using commands. const (
type redisCmd struct { maxIdleConns = 8
Command string idleTimeout = 30 * time.Second
Name string dialTimeout = 2 * time.Second
Data []byte ioTimeout = 1 * time.Second
Duration int64 )
Error error
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. // A SimpleRedis is used to communicate with redis.
@@ -35,97 +54,9 @@ type SimpleRedis struct {
host string host string
pass string pass string
database string database string
}
func genRedisArray(params ...[]byte) []byte { mu sync.Mutex
MSG := "" idle []*pooledConn
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)}
} }
// Init sets the redisHost used to connect to redis. // 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. // Get fetches the value for key name in redis.
func (sr *SimpleRedis) Get(name string) ([]byte, error) { func (sr *SimpleRedis) Get(name string) ([]byte, error) {
cmd := redisCmd{ values, err := sr.exec([]byte("GET"), []byte(name))
Command: "GET", if err != nil {
Name: name, return nil, err
} }
channel := make(chan redisCmd) if len(values) != 1 {
resp := sr.askRedis(cmd, channel) return nil, errIssue
if resp.Error != nil {
return nil, resp.Error
} }
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. // 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 { func (sr *SimpleRedis) Set(name string, data []byte, duration int64) error {
cmd := redisCmd{ _, err := sr.exec([]byte("SET"), []byte(name), data, []byte("EX"), []byte(strconv.FormatInt(duration, 10)))
Command: "SET", return err
Name: name,
Data: data,
Duration: duration,
}
sr.askRedis(cmd, nil)
return nil
} }
// Del removes the key name in redis. // Del removes the key name in redis.
func (sr *SimpleRedis) Del(name string) error { func (sr *SimpleRedis) Del(name string) error {
cmd := redisCmd{ _, err := sr.exec([]byte("DEL"), []byte(name))
Command: "DEL", return err
Name: name, }
func (sr *SimpleRedis) exec(args ...[]byte) ([][]byte, error) {
conn, reused, err := sr.borrow()
if err != nil {
return nil, err
} }
sr.askRedis(cmd, nil) values, reusable, err := sr.do(conn, args)
return nil 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
} }
+1 -1
View File
@@ -1,6 +1,6 @@
# github.com/leprosus/golang-ttl-map v1.1.7 # github.com/leprosus/golang-ttl-map v1.1.7
## explicit; go 1.15 ## explicit; go 1.15
github.com/leprosus/golang-ttl-map 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 ## explicit; go 1.22
github.com/maxlerebourg/simpleredis github.com/maxlerebourg/simpleredis