Only treat oversized ErrorResponse as plain text during startup

conn.go

@@ -1101,7 +1101,14 @@ func (cn *conn) saveMessage(typ proto.ResponseCode, buf *readBuf) error {
 
 // recvMessage receives any message from the backend, or returns an error if
 // a problem occurred while reading the message.
-func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) {
+//
+// startup must only be true for callers reading messages during the
+// connection-startup sequence (see [conn.recv]). It is never appropriate for
+// steady-state message parsing: a v3 ErrorResponse is a normal, properly
+// length-prefixed message post-handshake, and treating a large one as plain
+// text (see the comment below) desynchronizes the wire and leaves inProgress
+// stuck, poisoning the connection.
+func (cn *conn) recvMessage(r *readBuf, startup bool) (proto.ResponseCode, error) {
 	// workaround for a QueryRow bug, see exec
 	if cn.saveMessageType != 0 {
 		t := cn.saveMessageType
@@ -1127,11 +1134,16 @@ func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) {
 
 	// When PostgreSQL cannot start a backend (e.g., an external process limit),
 	// it sends plain text like "Ecould not fork new process [..]", which
-	// doesn't use the standard encoding for the Error message.
+	// doesn't use the standard encoding for the Error message. This can only
+	// happen before the protocol handshake completes: libpq only applies this
+	// heuristic while awaiting the startup response (fe-connect.c,
+	// PQconnectPoll), and once the protocol is established, an ErrorResponse
+	// is explicitly allowed to exceed MaxErrlen (fe-protocol3.c,
+	// VALID_LONG_MESSAGE_TYPE) since it's a normal, properly framed message.
 	//
 	// libpq checks "if ErrorResponse && (msgLength < 8 || msgLength > MAX_ERRLEN)",
 	// but check < 4 since n represents bytes remaining to be read after length.
-	if t == proto.ErrorResponse && (n < 4 || n > proto.MaxErrlen) {
+	if startup && t == proto.ErrorResponse && (n < 4 || n > proto.MaxErrlen) {
 		msg, _ := cn.buf.ReadString('\x00')
 		return 0, fmt.Errorf("pq: server error: %s%s", string(x[1:]), strings.TrimSuffix(msg, "\x00"))
 	}
@@ -1160,7 +1172,7 @@ func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) {
 func (cn *conn) recv() (proto.ResponseCode, *readBuf, error) {
 	for {
 		r := new(readBuf)
-		t, err := cn.recvMessage(r)
+		t, err := cn.recvMessage(r, true)
 		if err != nil {
 			return 0, nil, err
 		}
@@ -1185,7 +1197,7 @@ func (cn *conn) recv() (proto.ResponseCode, *readBuf, error) {
 // the caller to avoid an allocation.
 func (cn *conn) recv1Buf(r *readBuf) (proto.ResponseCode, error) {
 	for {
-		t, err := cn.recvMessage(r)
+		t, err := cn.recvMessage(r, false)
 		if err != nil {
 			return 0, err
 		}

copy.go

@@ -133,7 +133,7 @@ func (ci *copyin) flush(buf []byte) error {
 func (ci *copyin) resploop() {
 	for {
 		var r readBuf
-		t, err := ci.cn.recvMessage(&r)
+		t, err := ci.cn.recvMessage(&r, false)
 		if err != nil {
 			ci.setBad(driver.ErrBadConn)
 			ci.setError(err)

issues_test.go

@@ -3,10 +3,14 @@ package pq
 import (
 	"context"
 	"database/sql"
+	"errors"
+	"net"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/lib/pq/internal/pqtest"
+	"github.com/lib/pq/internal/proto"
 	"github.com/lib/pq/pqerror"
 )
 
@@ -133,3 +137,69 @@ func TestQueryCancelledReused(t *testing.T) {
 	// get a connection: it must be valid
 	connIsValid(t, db)
 }
+
+// #1324: an ErrorResponse larger than proto.MaxErrlen sent after the startup
+// handshake was misparsed as a pre-protocol plain-text error (garbling the
+// message) and left the connection desynchronized: the ErrorResponse body and
+// the trailing ReadyForQuery were never drained, so inProgress stayed true
+// and the poisoned connection was handed back out by the pool.
+func TestOversizedErrorResponse(t *testing.T) {
+	t.Parallel()
+
+	wantMsg := strings.Repeat("x", proto.MaxErrlen+10000)
+
+	f := pqtest.NewFake(t, func(f pqtest.Fake, cn net.Conn) {
+		f.Startup(cn, nil)
+		for {
+			code, msg, ok := f.ReadMsg(cn)
+			if !ok {
+				return
+			}
+			switch code {
+			case proto.Terminate:
+				cn.Close()
+				return
+			case proto.Query:
+				switch strings.TrimRight(string(msg), "\x00") {
+				case ";":
+					// MustDB's Ping.
+					f.WriteMsg(cn, proto.EmptyQueryResponse, "")
+					f.WriteMsg(cn, proto.ReadyForQuery, "I")
+				case "select 1":
+					f.WriteMsg(cn, proto.CommandComplete, "SELECT 1\x00")
+					f.WriteMsg(cn, proto.ReadyForQuery, "I")
+				default:
+					f.WriteMsg(cn, proto.ErrorResponse, "SERROR\x00C58030\x00M"+wantMsg+"\x00\x00")
+					f.WriteMsg(cn, proto.ReadyForQuery, "I")
+				}
+			}
+		}
+	})
+	defer f.Close()
+
+	db := pqtest.MustDB(t, f.DSN())
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
+
+	_, err := db.Exec(`DO $$ BEGIN RAISE EXCEPTION 'x'; END $$;`)
+	if err == nil {
+		t.Fatal("first Exec: want non-nil error, got nil")
+	}
+	var pqErr *Error
+	if !errors.As(err, &pqErr) {
+		t.Fatalf("first Exec: want *pq.Error, got %T: %v", err, err)
+	}
+	if pqErr.Message != wantMsg {
+		t.Errorf("first Exec: Message mangled: got %d bytes, want %d bytes", len(pqErr.Message), len(wantMsg))
+	}
+	if pqErr.Code != "58030" {
+		t.Errorf("first Exec: Code = %q, want 58030", pqErr.Code)
+	}
+
+	// The connection must not be poisoned: a second query on the same pooled
+	// connection must not fail with errQueryInProgress.
+	_, err = db.Exec("select 1")
+	if err != nil {
+		t.Fatalf("second Exec: connection left poisoned: %v", err)
+	}
+}

notify.go

@@ -180,7 +180,7 @@ func (l *ListenerConn) setState(newState int32) bool {
 func (l *ListenerConn) listenerConnLoop() (err error) {
 	r := &readBuf{}
 	for {
-		t, err := l.cn.recvMessage(r)
+		t, err := l.cn.recvMessage(r, false)
 		if err != nil {
 			return err
 		}