pgconn: guard receiveMessage against closed connection

Add a status check at the top of PgConn.receiveMessage() to return an
error immediately when the connection is already closed. This prevents
pipeline mode from calling receiveMessage() on a dead connection when
multiple FATAL errors are buffered together.

Previously, the fix guarded close(cleanupDone) in the ErrorResponse
handler to prevent double-close. This approach is more fundamental: it
stops the second FATAL from even being processed by receiveMessage(),
which is consistent with how lock() already rejects calls on closed
connections.

The nil-results check in Pipeline.Close() remains as a safety net for
the case where FATAL errors exhaust the request queue without the server
ever sending ReadyForQuery.

Author: veeceey

pgconn/pgconn.go

@@ -584,6 +584,10 @@ func (pgConn *PgConn) peekMessage() (pgproto3.BackendMessage, error) {
 
 // receiveMessage receives a message without setting up context cancellation
 func (pgConn *PgConn) receiveMessage() (pgproto3.BackendMessage, error) {
+	if pgConn.status == connStatusClosed {
+		return nil, &connLockError{status: "conn closed"}
+	}
+
 	msg, err := pgConn.peekMessage()
 	if err != nil {
 		return nil, err
@@ -598,13 +602,9 @@ func (pgConn *PgConn) receiveMessage() (pgproto3.BackendMessage, error) {
 	case *pgproto3.ErrorResponse:
 		err := ErrorResponseToPgError(msg)
 		if pgConn.config.OnPgError != nil && !pgConn.config.OnPgError(pgConn, err) {
-			// If the connection is already being closed (e.g. by asyncClose), don't close cleanupDone again to avoid
-			// a panic from closing an already-closed channel.
-			if pgConn.status != connStatusClosed {
-				pgConn.status = connStatusClosed
-				pgConn.conn.Close() // Ignore error as the connection is already broken and there is already an error to return.
-				close(pgConn.cleanupDone)
-			}
+			pgConn.status = connStatusClosed
+			pgConn.conn.Close() // Ignore error as the connection is already broken and there is already an error to return.
+			close(pgConn.cleanupDone)
 			return nil, err
 		}
 	case *pgproto3.NoticeResponse:
@@ -1415,11 +1415,9 @@ func (pgConn *PgConn) CopyFrom(ctx context.Context, r io.Reader, sql string) (Co
 			// the goroutine. So instead check pgConn.bufferingReceiveErr which will have been set by the signalMessage. If an
 			// error is found then forcibly close the connection without sending the Terminate message.
 			if err := pgConn.bufferingReceiveErr; err != nil {
-				if pgConn.status != connStatusClosed {
-					pgConn.status = connStatusClosed
-					pgConn.conn.Close()
-					close(pgConn.cleanupDone)
-				}
+				pgConn.status = connStatusClosed
+				pgConn.conn.Close()
+				close(pgConn.cleanupDone)
 				return CommandTag{}, normalizeTimeoutError(ctx, err)
 			}
 			// peekMessage never returns err in the bufferingReceive mode - it only forwards the bufferingReceive variables.
@@ -2836,11 +2834,9 @@ func (p *Pipeline) Close() error {
 				break
 			}
 		} else if results == nil {
-			// getResults returns (nil, nil) when the request queue is exhausted (pipelineNil) but
-			// ExpectedReadyForQuery is still > 0. This happens when the server sends FATAL errors that
-			// consume queued request slots (e.g. Prepare, Sync) without ever sending ReadyForQuery --
-			// so expectedReadyForQueryCount is never decremented. Without this check the loop would spin
-			// indefinitely because there are no more requests to issue and no ReadyForQuery to receive.
+			// getResults returns (nil, nil) when the request queue is exhausted but
+			// ExpectedReadyForQuery is still > 0. This can happen when FATAL errors consume
+			// queued request slots without the server ever sending ReadyForQuery.
 			p.conn.asyncClose()
 			if p.err == nil {
 				p.err = errors.New("pipeline: no more results but expected ReadyForQuery")