fix: flush records queued before an S3 Select error message

The writer goroutine selects between the record channel and the error
channel at random. When Evaluate had just queued a record and then reported
an error, the error case could win first; it flushed only the staging
buffer, so the queued record was dropped in the exit drain and the client
saw the error without the records that preceded it. Stage whatever is queued
before flushing and writing the error. The CVE-2026-39414 regression test
asserted this contract and failed once under the race detector in CI; a new
unit test exercises the ordering directly.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QwirCh4nsrJCQp2WaXoVLK
Signed-off-by: Feng Ruohang <rh@vonng.com>

internal/s3select/message.go

@@ -295,7 +295,15 @@ func (writer *messageWriter) start() {
 		select {
 		case data := <-writer.errCh:
 			quitFlag = true
-			// Flush collected records before sending error message
+			// A record accepted by SendRecord may still be queued when the
+			// error arrives, because select picks between the two channels
+			// at random. Stage it first so every record produced before the
+			// error precedes the error message instead of being dropped.
+			for len(writer.payloadCh) > 0 {
+				if !writer.stageRecord(<-writer.payloadCh) {
+					break
+				}
+			}
 			if !writer.flushRecords() {
 				break
 			}
@@ -316,23 +324,8 @@ func (writer *messageWriter) start() {
 					break
 				}
 				writer.write(endMessage)
-			} else {
+			} else if !writer.stageRecord(payload) {
-				for payload.Len() > 0 {
+				quitFlag = true
-					copiedLen := copy(writer.payloadBuffer[writer.payloadBufferIndex:], payload.Bytes())
-					writer.payloadBufferIndex += copiedLen
-					payload.Next(copiedLen)
-
-					// If buffer is filled, flush it now!
-					freeSpace := bufLength - writer.payloadBufferIndex
-					if freeSpace == 0 {
-						if !writer.flushRecords() {
-							quitFlag = true
-							break
-						}
-					}
-				}
-
-				bufPool.Put(payload)
 			}
 
 		case <-recordStagingTicker.C:
@@ -368,6 +361,26 @@ func (writer *messageWriter) start() {
 	}
 }
 
+// stageRecord copies a record into the payload buffer, flushing whenever
+// the buffer fills, and returns the buffer to the pool. It reports false when
+// a flush failed.
+func (writer *messageWriter) stageRecord(payload *bytes.Buffer) bool {
+	defer bufPool.Put(payload)
+	for payload.Len() > 0 {
+		copiedLen := copy(writer.payloadBuffer[writer.payloadBufferIndex:], payload.Bytes())
+		writer.payloadBufferIndex += copiedLen
+		payload.Next(copiedLen)
+
+		// If buffer is filled, flush it now!
+		if bufLength-writer.payloadBufferIndex == 0 {
+			if !writer.flushRecords() {
+				return false
+			}
+		}
+	}
+	return true
+}
+
 // Sends a single whole record.
 func (writer *messageWriter) SendRecord(payload *bytes.Buffer) error {
 	select {

internal/s3select/message_test.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package s3select
+
+import (
+	"bytes"
+	"testing"
+)
+
+// TestMessageWriterFlushesQueuedRecordBeforeError sends a record and an error
+// back to back, so the writer goroutine sees both channels ready and picks
+// one at random. The record must appear in the response before the error
+// message every time.
+func TestMessageWriterFlushesQueuedRecordBeforeError(t *testing.T) {
+	for i := range 200 {
+		w := &testResponseWriter{}
+		writer := newMessageWriter(w, nil)
+		payload := bufPool.Get()
+		payload.Reset()
+		payload.WriteString(`{"id":1}` + "\n")
+		if err := writer.SendRecord(payload); err != nil {
+			t.Fatalf("run %d: SendRecord: %v", i, err)
+		}
+		if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil {
+			t.Fatalf("run %d: FinishWithError: %v", i, err)
+		}
+		record := bytes.Index(w.response, []byte(`{"id":1}`))
+		errMsg := bytes.Index(w.response, []byte("OverMaxRecordSize"))
+		if record < 0 || errMsg < 0 || record > errMsg {
+			t.Fatalf("run %d: record at %d, error at %d: queued record was dropped or reordered", i, record, errMsg)
+		}
+	}
+}