From 035aa6c201fb440323b8c79a67109155be0fe5e4 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 3 Sep 2026 06:52:33 +0800 Subject: [PATCH] test: cover the S3 Select message writer ordering Exercise the writer directly and decode the event stream with the client parser: a buffered plus a queued record before an error, an error with nothing queued, a record larger than the staging buffer on both the success and the error path, and the unchanged success ordering of records, Stats and End. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01QwirCh4nsrJCQp2WaXoVLK Signed-off-by: Feng Ruohang --- internal/s3select/message_test.go | 145 ++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/internal/s3select/message_test.go b/internal/s3select/message_test.go index 010fa93a2..e406ee04d 100644 --- a/internal/s3select/message_test.go +++ b/internal/s3select/message_test.go @@ -19,9 +19,54 @@ package s3select import ( "bytes" + "io" + "net/http" + "strings" "testing" + + "github.com/minio/minio-go/v7" ) +// decodeEvents parses a recorded event stream with the minio-go client and +// returns the concatenated record payloads and the terminal error, if any. +func decodeEvents(t *testing.T, response []byte) ([]byte, error) { + t.Helper() + body := &testResponseBody{Reader: bytes.NewReader(response), closed: make(chan struct{})} + res, err := minio.NewSelectResults(&http.Response{StatusCode: http.StatusOK, Body: body, ContentLength: int64(len(response))}, "testbucket") + if err != nil { + t.Fatal(err) + } + records, readErr := io.ReadAll(res) + <-body.closed + return records, readErr +} + +func newTestRecord(content string) *bytes.Buffer { + payload := bufPool.Get() + payload.Reset() + payload.WriteString(content) + return payload +} + +// eventOrder returns the byte offsets of the given markers in the response, +// or -1 for a marker that is absent. +func eventOrder(response []byte, markers ...string) []int { + offsets := make([]int, len(markers)) + for i, marker := range markers { + offsets[i] = bytes.Index(response, []byte(marker)) + } + return offsets +} + +func ascending(offsets []int) bool { + for i, offset := range offsets { + if offset < 0 || (i > 0 && offset <= offsets[i-1]) { + return false + } + } + return true +} + // 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 @@ -46,3 +91,103 @@ func TestMessageWriterFlushesQueuedRecordBeforeError(t *testing.T) { } } } + +// TestMessageWriterFlushesBufferedAndQueuedRecordsBeforeError: one record has +// already been staged into the buffer and a second one is still queued when +// the error arrives. Both must precede the error, in order. +func TestMessageWriterFlushesBufferedAndQueuedRecordsBeforeError(t *testing.T) { + for i := range 100 { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + if err := writer.SendRecord(newTestRecord(`{"id":1}` + "\n")); err != nil { + t.Fatal(err) + } + // Give the writer a chance to stage the first record; whether it + // did or not, the outcome must be the same. + if i%2 == 0 { + for range 100 { + if len(writer.payloadCh) == 0 { + break + } + } + } + if err := writer.SendRecord(newTestRecord(`{"id":2}` + "\n")); err != nil { + t.Fatal(err) + } + if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil { + t.Fatal(err) + } + if got := eventOrder(w.response, `{"id":1}`, `{"id":2}`, "OverMaxRecordSize"); !ascending(got) { + t.Fatalf("run %d: offsets %v: records must precede the error in order", i, got) + } + } +} + +// TestMessageWriterErrorWithoutRecords: an error with nothing queued writes +// only the error message, no empty Records event. +func TestMessageWriterErrorWithoutRecords(t *testing.T) { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + if err := writer.FinishWithError("InternalError", "boom"); err != nil { + t.Fatal(err) + } + if !bytes.Contains(w.response, []byte("InternalError")) || bytes.Contains(w.response, []byte("Records")) { + t.Fatalf("unexpected response: %q", w.response) + } +} + +// TestMessageWriterStagesRecordsLargerThanTheBuffer: a record larger than the +// staging buffer is split across several Records events and nothing is lost +// or reordered, whether the stream ends with success or with an error. +func TestMessageWriterStagesRecordsLargerThanTheBuffer(t *testing.T) { + big := strings.Repeat("x", bufLength+bufLength/2) + want := "head\n" + big + "\n" + "tail\n" + for _, withError := range []bool{false, true} { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + for _, rec := range []string{"head\n", big + "\n", "tail\n"} { + if err := writer.SendRecord(newTestRecord(rec)); err != nil { + t.Fatal(err) + } + } + if withError { + if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil { + t.Fatal(err) + } + } else if err := writer.Finish(10, 10); err != nil { + t.Fatal(err) + } + records, err := decodeEvents(t, w.response) + if string(records) != want { + t.Fatalf("withError=%v: got %d record bytes, want %d; head=%q", withError, len(records), len(want), string(records[:min(len(records), 8)])) + } + if withError && (err == nil || !strings.Contains(err.Error(), "OverMaxRecordSize")) { + t.Fatalf("expected OverMaxRecordSize after the records, got %v", err) + } + if !withError && err != nil { + t.Fatalf("unexpected error on the success path: %v", err) + } + } +} + +// TestMessageWriterSuccessOrder: the success path is unchanged: every record, +// then Stats, then End, and the client sees no error. +func TestMessageWriterSuccessOrder(t *testing.T) { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + for _, rec := range []string{`{"id":1}`, `{"id":2}`} { + if err := writer.SendRecord(newTestRecord(rec + "\n")); err != nil { + t.Fatal(err) + } + } + if err := writer.Finish(20, 20); err != nil { + t.Fatal(err) + } + if got := eventOrder(w.response, `{"id":1}`, `{"id":2}`, "Stats", "End"); !ascending(got) { + t.Fatalf("offsets %v", got) + } + records, err := decodeEvents(t, w.response) + if err != nil || string(records) != "{\"id\":1}\n{\"id\":2}\n" { + t.Fatalf("records %q err %v", records, err) + } +}