From efbfa16f4f1c4541bf83544add68aadd837afb37 Mon Sep 17 00:00:00 2001 From: Mario Date: Tue, 28 Jul 2026 17:10:19 +0200 Subject: [PATCH 1/2] memcache: return ErrServerError for SERVER_ERROR responses SERVER_ERROR replies currently surface as generic 'unexpected response line' errors. Parse them into the exported (previously unreachable) ErrServerError, wrapped with the server's message, so callers can match them with errors.Is. Connection pooling behavior is unchanged: these errors still close the connection. --- memcache/memcache.go | 29 +++++++++ memcache/memcache_test.go | 132 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 161 insertions(+) diff --git a/memcache/memcache.go b/memcache/memcache.go index 6f48caa..ad80244 100644 --- a/memcache/memcache.go +++ b/memcache/memcache.go @@ -87,6 +87,16 @@ func resumableError(err error) bool { return false } +// serverErrorFromLine returns an error wrapping ErrServerError if line is a +// SERVER_ERROR response, or nil otherwise. +func serverErrorFromLine(line []byte) error { + if !bytes.HasPrefix(line, resultServerErrorPrefix) { + return nil + } + msg := bytes.TrimSuffix(line[len(resultServerErrorPrefix):], crlf) + return fmt.Errorf("%w: %s", ErrServerError, msg) +} + func legalKey(key string) bool { if len(key) > 250 { return false @@ -113,6 +123,7 @@ var ( resultTouched = []byte("TOUCHED\r\n") resultClientErrorPrefix = []byte("CLIENT_ERROR ") + resultServerErrorPrefix = []byte("SERVER_ERROR ") versionPrefix = []byte("VERSION") ) @@ -464,6 +475,9 @@ func (c *Client) touchFromAddr(addr net.Addr, keys []string, expiration int32) e case bytes.Equal(line, resultNotFound): return ErrCacheMiss default: + if err := serverErrorFromLine(line); err != nil { + return err + } return fmt.Errorf("memcache: unexpected response line from touch: %q", string(line)) } } @@ -530,6 +544,11 @@ func parseGetResponse(r *bufio.Reader, conn *conn, cb func(*Item)) error { it := new(Item) size, err := scanGetResponseLine(line, it) if err != nil { + // The line is not a VALUE line: check whether it's a protocol-level + // server error before reporting it as unexpected. + if serverErr := serverErrorFromLine(line); serverErr != nil { + return serverErr + } return err } it.Value = make([]byte, size+2) @@ -699,6 +718,10 @@ func (c *Client) populateOne(rw *bufio.ReadWriter, verb string, item *Item) erro return ErrCASConflict case bytes.Equal(line, resultNotFound): return ErrCacheMiss + default: + if err := serverErrorFromLine(line); err != nil { + return err + } } return fmt.Errorf("memcache: unexpected response line from %q: %q", verb, string(line)) } @@ -731,6 +754,10 @@ func writeExpectf(rw *bufio.ReadWriter, expect []byte, format string, args ...in return ErrCASConflict case bytes.Equal(line, resultNotFound): return ErrCacheMiss + default: + if err := serverErrorFromLine(line); err != nil { + return err + } } return fmt.Errorf("memcache: unexpected response line: %q", string(line)) } @@ -817,6 +844,8 @@ func (c *Client) incrDecr(verb, key string, delta uint64) (uint64, error) { case bytes.HasPrefix(line, resultClientErrorPrefix): errMsg := line[len(resultClientErrorPrefix) : len(line)-2] return errors.New("memcache: client error: " + string(errMsg)) + case bytes.HasPrefix(line, resultServerErrorPrefix): + return serverErrorFromLine(line) } val, err = strconv.ParseUint(string(line[:len(line)-2]), 10, 64) if err != nil { diff --git a/memcache/memcache_test.go b/memcache/memcache_test.go index a0fa746..bc80408 100644 --- a/memcache/memcache_test.go +++ b/memcache/memcache_test.go @@ -22,6 +22,7 @@ import ( "bytes" "context" "crypto/tls" + "errors" "flag" "fmt" "io" @@ -502,3 +503,134 @@ func TestScanGetResponseLine(t *testing.T) { }) } } + +func TestServerError_ReturnsErrServerError(t *testing.T) { + t.Run("Set", func(t *testing.T) { + c, srvw, srvReqs := getClientWithFakeServer(t) + + errCh := make(chan error) + go func() { + errCh <- c.Set(&Item{Key: "foo", Value: []byte("hello")}) + }() + + if req := <-srvReqs; req != "set foo 0 0 5\r\n" { + t.Fatalf("unexpected request: %q", req) + } + <-srvReqs // data block + if _, err := io.WriteString(srvw, "SERVER_ERROR object too large for cache\r\n"); err != nil { + t.Fatalf("write response: %v", err) + } + + err := <-errCh + if !errors.Is(err, ErrServerError) { + t.Fatalf("Set: got err=%v, want ErrServerError", err) + } + if want := "memcache: server error: object too large for cache"; err.Error() != want { + t.Fatalf("Set: got err=%q, want %q", err.Error(), want) + } + }) + + t.Run("Get", func(t *testing.T) { + c, srvw, srvReqs := getClientWithFakeServer(t) + + errCh := make(chan error) + go func() { + _, err := c.Get("foo") + errCh <- err + }() + + if req := <-srvReqs; req != "gets foo\r\n" { + t.Fatalf("unexpected request: %q", req) + } + if _, err := io.WriteString(srvw, "SERVER_ERROR out of memory writing get response\r\n"); err != nil { + t.Fatalf("write response: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrServerError) { + t.Fatalf("Get: got err=%v, want ErrServerError", err) + } + }) + + t.Run("Delete", func(t *testing.T) { + c, srvw, srvReqs := getClientWithFakeServer(t) + + errCh := make(chan error) + go func() { + errCh <- c.Delete("foo") + }() + + if req := <-srvReqs; req != "delete foo\r\n" { + t.Fatalf("unexpected request: %q", req) + } + if _, err := io.WriteString(srvw, "SERVER_ERROR temporary failure\r\n"); err != nil { + t.Fatalf("write response: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrServerError) { + t.Fatalf("Delete: got err=%v, want ErrServerError", err) + } + }) + + t.Run("Increment", func(t *testing.T) { + c, srvw, srvReqs := getClientWithFakeServer(t) + + errCh := make(chan error) + go func() { + _, err := c.Increment("foo", 1) + errCh <- err + }() + + if req := <-srvReqs; req != "incr foo 1\r\n" { + t.Fatalf("unexpected request: %q", req) + } + if _, err := io.WriteString(srvw, "SERVER_ERROR temporary failure\r\n"); err != nil { + t.Fatalf("write response: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrServerError) { + t.Fatalf("Increment: got err=%v, want ErrServerError", err) + } + }) +} + +// getClientWithFakeServer creates a new client, whose dial function is bound to an in-memory synchronous server. +func getClientWithFakeServer(t *testing.T) (*Client, io.Writer, <-chan string) { + cliConn, srvConn := net.Pipe() + t.Cleanup(func() { + _ = srvConn.Close() + _ = cliConn.Close() + }) + + c := NewFromSelector(dummySelector{}) + c.DialContext = func(context.Context, string, string) (net.Conn, error) { + return cliConn, nil + } + + srvReqs := make(chan string) + go func() { + br := bufio.NewReader(srvConn) + for { + // ignores error for simplicity + line, _ := br.ReadSlice('\n') + srvReqs <- string(line) + } + }() + + return c, srvConn, srvReqs +} + +type dummyAddr struct{} + +func (dummyAddr) Network() string { return "dummy" } + +func (dummyAddr) String() string { return "dummy" } + +type dummySelector struct{} + +func (dummySelector) PickServer(string) (net.Addr, error) { + return dummyAddr{}, nil +} + +func (dummySelector) Each(f func(net.Addr) error) error { + return f(dummyAddr{}) +} From 841f5a59cee8f0a32d1d752583fb92c905d909ba Mon Sep 17 00:00:00 2001 From: Mario Date: Tue, 28 Jul 2026 17:10:27 +0200 Subject: [PATCH 2/2] memcache: don't close connections on clean protocol-level SERVER_ERROR responses A SERVER_ERROR \r\n reply is a complete, well-formed protocol line: the server keeps the connection open and in sync (on storage errors it even swallows the data block). Closing our side causes needless reconnect churn, e.g. on 'object too large for cache' or OOM bursts, the worst possible time to add TCP handshake load. --- memcache/memcache.go | 19 ++++++++--- memcache/memcache_test.go | 67 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 81 insertions(+), 5 deletions(-) diff --git a/memcache/memcache.go b/memcache/memcache.go index ad80244..a9897b0 100644 --- a/memcache/memcache.go +++ b/memcache/memcache.go @@ -75,13 +75,24 @@ const ( const buffered = 8 // arbitrary buffered channel size, for readability -// resumableError returns true if err is only a protocol-level cache error. +// resumableError returns true if err is only a protocol-level error. // This is used to determine whether or not a server connection should // be re-used or not. If an error occurs, by default we don't reuse the -// connection, unless it was just a cache error. +// connection, unless it was just a protocol-level error. +// +// SERVER_ERROR replies are resumable: memcached sends them as complete, +// well-formed lines and keeps the connection in sync (on storage errors it +// even swallows the request's data block; see process_update_command in +// memcached's proto_text.c). The known exception is "out of memory reading +// request", which closes the connection server-side; reusing it costs one +// failed request, the same as any stale pooled connection. func resumableError(err error) bool { - switch err { - case ErrCacheMiss, ErrCASConflict, ErrNotStored, ErrMalformedKey: + switch { + case errors.Is(err, ErrCacheMiss), + errors.Is(err, ErrCASConflict), + errors.Is(err, ErrNotStored), + errors.Is(err, ErrMalformedKey), + errors.Is(err, ErrServerError): return true } return false diff --git a/memcache/memcache_test.go b/memcache/memcache_test.go index bc80408..fb49c2a 100644 --- a/memcache/memcache_test.go +++ b/memcache/memcache_test.go @@ -504,7 +504,41 @@ func TestScanGetResponseLine(t *testing.T) { } } -func TestServerError_ReturnsErrServerError(t *testing.T) { +func TestServerError_KeepsConnectionOpen(t *testing.T) { + // countFreeConns returns the number of connections in the client's free pool. + countFreeConns := func(c *Client) int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.freeconn[dummyAddr{}.String()]) + } + + // assertReusable verifies that the connection went back to the free pool and + // that a subsequent request over the very same connection still works. + assertReusable := func(t *testing.T, c *Client, srvw io.Writer, srvReqs <-chan string) { + t.Helper() + + if got := countFreeConns(c); got != 1 { + t.Fatalf("free conns after SERVER_ERROR: got %d, want 1", got) + } + + errCh := make(chan error) + go func() { + _, err := c.Get("foo") + errCh <- err + }() + + if req := <-srvReqs; req != "gets foo\r\n" { + t.Fatalf("unexpected follow-up request: %q", req) + } + if _, err := io.WriteString(srvw, "END\r\n"); err != nil { + t.Fatalf("write follow-up response: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrCacheMiss) { + t.Fatalf("follow-up Get on reused connection: got err=%v, want ErrCacheMiss", err) + } + } + t.Run("Set", func(t *testing.T) { c, srvw, srvReqs := getClientWithFakeServer(t) @@ -528,6 +562,8 @@ func TestServerError_ReturnsErrServerError(t *testing.T) { if want := "memcache: server error: object too large for cache"; err.Error() != want { t.Fatalf("Set: got err=%q, want %q", err.Error(), want) } + + assertReusable(t, c, srvw, srvReqs) }) t.Run("Get", func(t *testing.T) { @@ -549,6 +585,8 @@ func TestServerError_ReturnsErrServerError(t *testing.T) { if err := <-errCh; !errors.Is(err, ErrServerError) { t.Fatalf("Get: got err=%v, want ErrServerError", err) } + + assertReusable(t, c, srvw, srvReqs) }) t.Run("Delete", func(t *testing.T) { @@ -569,6 +607,8 @@ func TestServerError_ReturnsErrServerError(t *testing.T) { if err := <-errCh; !errors.Is(err, ErrServerError) { t.Fatalf("Delete: got err=%v, want ErrServerError", err) } + + assertReusable(t, c, srvw, srvReqs) }) t.Run("Increment", func(t *testing.T) { @@ -590,6 +630,31 @@ func TestServerError_ReturnsErrServerError(t *testing.T) { if err := <-errCh; !errors.Is(err, ErrServerError) { t.Fatalf("Increment: got err=%v, want ErrServerError", err) } + + assertReusable(t, c, srvw, srvReqs) + }) + + t.Run("non-protocol errors still close the connection", func(t *testing.T) { + c, srvw, srvReqs := getClientWithFakeServer(t) + + errCh := make(chan error) + go func() { + errCh <- c.Set(&Item{Key: "foo", Value: []byte("hello")}) + }() + + <-srvReqs // command line + <-srvReqs // data block + if _, err := io.WriteString(srvw, "BOGUS RESPONSE\r\n"); err != nil { + t.Fatalf("write response: %v", err) + } + + err := <-errCh + if err == nil || errors.Is(err, ErrServerError) { + t.Fatalf("Set: got err=%v, want a non-ErrServerError error", err) + } + if got := countFreeConns(c); got != 0 { + t.Fatalf("free conns after unexpected response: got %d, want 0", got) + } }) }