From 574f170e16a06e2fc5403d72b543699849b99d9c Mon Sep 17 00:00:00 2001 From: Sean Chittenden Date: Mon, 15 Jun 2026 07:43:08 -0700 Subject: [PATCH] pgconn: drain socket in asyncClose to avoid RST on context cancel When a context is canceled while the server is still streaming rows, asyncClose() sends CancelRequest, writes Terminate, and closes the underlying net.Conn. At that point the kernel receive buffer typically still holds DataRow messages that were already in flight before the cancel landed. Closing a TCP socket with unread data causes the kernel to discard the receive queue and send RST instead of FIN (RFC 1122 4.2.2.13), which surfaces on the server or proxy as "connection reset by peer". Drain the socket with io.Copy(io.Discard, conn) after sending Terminate. The existing 15s deadline bounds the read; in the normal case the server stops producing rows after CancelRequest, processes Terminate, and closes its side, so the drain returns on io.EOF and the deferred Close() emits a clean FIN. Add TestAsyncCloseDrainsBeforeClose, which floods a mock client with DataRows, cancels mid-stream, and asserts the server observes Terminate followed by io.EOF rather than ECONNRESET. The test reproduces the reported "connection reset by peer" without the fix. Fixes #2584 Signed-off-by: Sean Chittenden --- CHANGELOG.md | 6 ++ pgconn/pgconn.go | 10 +++ pgconn/pgconn_test.go | 147 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12e633a4b..fa21d95e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +# Unreleased + +## Fixes + +* pgconn: drain socket before close in `asyncClose` so context cancellation produces a TCP FIN instead of RST, avoiding "connection reset by peer" on the server / proxy (Sean Chittenden at CrowdStrike, Inc.) + # 5.10.0 (June 3, 2026) This release includes a significant amount of hardening against malicious or compromised PostgreSQL servers, diff --git a/pgconn/pgconn.go b/pgconn/pgconn.go index d181f7f5f..882131759 100644 --- a/pgconn/pgconn.go +++ b/pgconn/pgconn.go @@ -790,6 +790,16 @@ func (pgConn *PgConn) asyncClose() { pgConn.frontend.Send(&pgproto3.Terminate{}) pgConn.flushWithPotentialWriteReadDeadlock() + + // Drain any data already in flight from the server (DataRows that were sent before the + // CancelRequest landed, the resulting ErrorResponse, ReadyForQuery, and finally the server's + // own close after it processes Terminate). Closing a TCP socket while unread data remains in + // the kernel receive buffer causes the OS to send RST instead of FIN, which surfaces on the + // server or proxy as "connection reset by peer". The deadline set above bounds how long this + // will block; on timeout we fall through to Close() and accept the abortive close. + // + // See https://github.com/jackc/pgx/issues/2584 + io.Copy(io.Discard, pgConn.conn) }() } diff --git a/pgconn/pgconn_test.go b/pgconn/pgconn_test.go index ee7377d3c..abe6df20e 100644 --- a/pgconn/pgconn_test.go +++ b/pgconn/pgconn_test.go @@ -4704,6 +4704,153 @@ func mustEncode(buf []byte, err error) []byte { return buf } +// https://github.com/jackc/pgx/issues/2584 +// +// When a context is canceled while the server is mid-stream, asyncClose must drain the socket +// before closing it. Closing a TCP socket with unread data in the kernel receive buffer causes +// the OS to send RST instead of FIN, which surfaces on the server / proxy as "connection reset by +// peer". This test floods the client with DataRows, cancels the context, and asserts the mock +// server sees Terminate followed by a clean EOF rather than ECONNRESET. +func TestAsyncCloseDrainsBeforeClose(t *testing.T) { + t.Parallel() + + ln, err := net.Listen("tcp", "127.0.0.1:") + require.NoError(t, err) + defer ln.Close() + + type serverResult struct { + gotTerminate bool + readErr error + } + serverDone := make(chan serverResult, 1) + + go func() { + var res serverResult + defer func() { serverDone <- res }() + + conn, err := ln.Accept() + if err != nil { + res.readErr = err + return + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(30 * time.Second)) + + // asyncClose dials a second connection to deliver CancelRequest and then blocks on a Read + // until the server closes (libpq does the same). Accept it, consume the request, and close + // promptly so asyncClose proceeds to Terminate. Spawned after the primary Accept() so there + // is no race over which goroutine gets the first connection. + go func() { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() + c.SetDeadline(time.Now().Add(5 * time.Second)) + io.ReadFull(c, make([]byte, 16)) // len + code + pid + 4-byte secret + }() + + backend := pgproto3.NewBackend(conn, conn) + + if _, err := backend.ReceiveStartupMessage(); err != nil { + res.readErr = err + return + } + backend.Send(&pgproto3.AuthenticationOk{}) + backend.Send(&pgproto3.BackendKeyData{ProcessID: 0, SecretKey: []byte{0, 0, 0, 0}}) + backend.Send(&pgproto3.ReadyForQuery{TxStatus: 'I'}) + if err := backend.Flush(); err != nil { + res.readErr = err + return + } + + // Read the simple Query message. + if _, err := backend.Receive(); err != nil { + res.readErr = err + return + } + + backend.Send(&pgproto3.RowDescription{Fields: []pgproto3.FieldDescription{{Name: []byte("n")}}}) + if err := backend.Flush(); err != nil { + res.readErr = err + return + } + + // Flood DataRows. Use a payload large enough that the client's kernel receive buffer holds + // unread data at the moment of cancellation; without the drain in asyncClose, the client's + // Close() would emit RST and the server's subsequent Receive() would see ECONNRESET. Stop + // when the client closes its write side or the socket buffers are full enough that we have + // blocked past the cancel point; either way fall through to read Terminate. + row := &pgproto3.DataRow{Values: [][]byte{bytes.Repeat([]byte("x"), 1024)}} + conn.SetWriteDeadline(time.Now().Add(2 * time.Second)) + for i := 0; i < 16*1024; i++ { + backend.Send(row) + if err := backend.Flush(); err != nil { + break + } + } + conn.SetWriteDeadline(time.Time{}) + + // The client's asyncClose sends Terminate and then drains until we close. Read until we see + // Terminate, then half-close our write side so the client's drain observes EOF and proceeds + // to Close() its end. + conn.SetReadDeadline(time.Now().Add(20 * time.Second)) + for { + msg, err := backend.Receive() + if err != nil { + res.readErr = err + return + } + if _, ok := msg.(*pgproto3.Terminate); ok { + res.gotTerminate = true + break + } + } + + // Half-close: send FIN to the client so its io.Copy(io.Discard, ...) returns io.EOF, while + // keeping our read side open to observe how the client closes. A clean client close yields + // io.EOF here; an abortive close (RST due to unread data) yields ECONNRESET. + conn.(*net.TCPConn).CloseWrite() + buf := make([]byte, 1) + _, res.readErr = conn.Read(buf) + }() + + host, port, _ := strings.Cut(ln.Addr().String(), ":") + connStr := fmt.Sprintf("sslmode=disable host=%s port=%s", host, port) + + conn, err := pgconn.Connect(context.Background(), connStr) + require.NoError(t, err) + + queryCtx, queryCancel := context.WithCancel(context.Background()) + mrr := conn.Exec(queryCtx, "select n from generate_series(1,1000000) n") + require.True(t, mrr.NextResult()) + rr := mrr.ResultReader() + + // Read a handful of rows so the server is definitely streaming, then cancel. + for i := 0; i < 50 && rr.NextRow(); i++ { + } + queryCancel() + for rr.NextRow() { + } + rr.Close() + mrr.Close() + + // asyncClose runs in a background goroutine; wait for it (and the drain) to finish. + select { + case <-conn.CleanupDone(): + case <-time.After(30 * time.Second): + t.Fatal("timed out waiting for client cleanup") + } + + select { + case res := <-serverDone: + require.True(t, res.gotTerminate, "server never received Terminate message") + require.ErrorIs(t, res.readErr, io.EOF, "server saw %v, expected clean EOF (FIN); RST indicates the client closed with unread data", res.readErr) + case <-time.After(30 * time.Second): + t.Fatal("timed out waiting for server") + } +} + func TestDeadlineContextWatcherHandler(t *testing.T) { t.Parallel()