8000 [pull] master from minio:master by pull[bot] · Pull Request #472 · kokizzu/minio · GitHub
[go: up one dir, main page]

Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions cmd/api-response.go
Original file line number Diff line number Diff line change
Expand Up @@ -889,6 +889,12 @@ func generateMultiDeleteResponse(quiet bool, deletedObjects []DeletedObject, err
}

func writeResponse(w http.ResponseWriter, statusCode int, response []byte, mType mimeType) {
// Don't write a response if one has already been written.
// Fixes https://github.com/minio/minio/issues/21633
if headersAlreadyWritten(w) {
return
}

if statusCode == 0 {
statusCode = 200
}
Expand Down Expand Up @@ -1015,3 +1021,45 @@ func writeCustomErrorResponseJSON(ctx context.Context, w http.ResponseWriter, er
encodedErrorResponse := encodeResponseJSON(errorResponse)
writeResponse(w, err.HTTPStatusCode, encodedErrorResponse, mimeJSON)
}

type unwrapper interface {
Unwrap() http.ResponseWriter
}

// headersAlreadyWritten returns true if the headers have already been written
// to this response writer. It will unwrap the ResponseWriter if possible to try
// and find a trackingResponseWriter.
func headersAlreadyWritten(w http.ResponseWriter) bool {
for {
if trw, ok := w.(*trackingResponseWriter); ok {
return trw.headerWritten
} else if uw, ok := w.(unwrapper); ok {
w = uw.Unwrap()
} else {
return false
}
}
}

// trackingResponseWriter wraps a ResponseWriter and notes when WriterHeader has
// been called. This allows high level request handlers to check if something
// has already sent the header.
type trackingResponseWriter struct {
http.ResponseWriter
headerWritten bool
}

func (w *trackingResponseWriter) WriteHeader(statusCode int) {
if !w.headerWritten {
w.headerWritten = true
w.ResponseWriter.WriteHeader(statusCode)
}
}

func (w *trackingResponseWriter) Write(b []byte) (int, error) {
return w.ResponseWriter.Write(b)
}

func (w *trackingResponseWriter) Unwrap() http.ResponseWriter {
return w.ResponseWriter
}
90 changes: 90 additions & 0 deletions cmd/api-response_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,12 @@
package cmd

import (
"io"
"net/http"
"net/http/httptest"
"testing"

"github.com/klauspost/compress/gzhttp"
)

// Tests object location.
Expand Down Expand Up @@ -122,3 +126,89 @@ func TestGetURLScheme(t *testing.T) {
t.Errorf("Expected %s, got %s", httpsScheme, gotScheme)
}
}

func TestTrackingResponseWriter(t *testing.T) {
rw := httptest.NewRecorder()
trw := &trackingResponseWriter{ResponseWriter: rw}
trw.WriteHeader(123)
if !trw.headerWritten {
t.Fatal("headerWritten was not set by WriteHeader call")
}

_, err := trw.Write([]byte("hello"))
if err != nil {
t.Fatalf("Write unexpectedly failed: %v", err)
}

// Check that WriteHeader and Write were called on the underlying response writer
resp := rw.Result()
if resp.StatusCode != 123 {
t.Fatalf("unexpected status: %v", resp.StatusCode)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("reading response body failed: %v", err)
}
if string(body) != "hello" {
t.Fatalf("response body incorrect: %v", string(body))
}

// Check that Unwrap works
if trw.Unwrap() != rw {
t.Fatalf("Unwrap returned wrong result: %v", trw.Unwrap())
}
}

func TestHeadersAlreadyWritten(t *testing.T) {
rw := httptest.NewRecorder()
trw := &trackingResponseWriter{ResponseWriter: rw}

if headersAlreadyWritten(trw) {
t.Fatal("headers have not been written yet")
}

trw.WriteHeader(123)
if !headersAlreadyWritten(trw) {
t.Fatal("headers were written")
}
}

func TestHeadersAlreadyWrittenWrapped(t *testing.T) {
rw := httptest.NewRecorder()
trw := &trackingResponseWriter{ResponseWriter: rw}
wrap1 := &gzhttp.NoGzipResponseWriter{ResponseWriter: trw}
wrap2 := &gzhttp.NoGzipResponseWriter{ResponseWriter: wrap1}

if headersAlreadyWritten(wrap2) {
t.Fatal("headers have not been written yet")
}

wrap2.WriteHeader(123)
if !headersAlreadyWritten(wrap2) {
t.Fatal("headers were written")
}
}

func TestWriteResponseHeadersNotWritten(t *testing.T) {
rw := httptest.NewRecorder()
trw := &trackingResponseWriter{ResponseWriter: rw}

writeResponse(trw, 299, []byte("hello"), "application/foo")

resp := rw.Result()
if resp.StatusCode != 299 {
t.Fatal("response wasn't written")
}
}

func TestWriteResponseHeadersWritten(t *testing.T) {
rw := httptest.NewRecorder()
rw.Code = -1
trw := &trackingResponseWriter{ResponseWriter: rw, headerWritten: true}

writeResponse(trw, 200, []byte("hello"), "application/foo")

if rw.Code != -1 {
t.Fatalf("response was written when it shouldn't have been (Code=%v)", rw.Code)
}
}
2 changes: 2 additions & 0 deletions cmd/api-router.go
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,8 @@ func s3APIMiddleware(f http.HandlerFunc, flags ...s3HFlag) http.HandlerFunc {
handlerName := getHandlerName(f, "objectAPIHandlers")

var handler http.HandlerFunc = func(w http.ResponseWriter, r *http.Request) {
w = &trackingResponseWriter{ResponseWriter: w}

// Wrap the actual handler with the appropriate tracing middleware.
var tracedHandler http.HandlerFunc
if handlerFlags.has(traceHdrsS3HFlag) {
Expand Down
225 changes: 225 additions & 0 deletions cmd/erasure-multipart-conditional_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,225 @@
// Copyright (c) 2015-2025 MinIO, Inc.
//
// 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 cmd

import (
"bytes"
"context"
"testing"

"github.com/dustin/go-humanize"
xhttp "github.com/minio/minio/internal/http"
)

// TestNewMultipartUploadConditionalWithReadQuorumFailure tests that conditional
// multipart uploads (with if-match/if-none-match) behave correctly when read quorum
// cannot be reached.
//
// Related to: https://github.com/minio/minio/issues/21603
//
// Should return an error when read quorum cannot
// be reached, as we cannot reliably determine if the precondition is met.
func TestNewMultipartUploadConditionalWithReadQuorumFailure(t *testing.T) {
ctx := context.Background()

obj, fsDirs, err := prepareErasure16(ctx)
if err != nil {
t.Fatal(err)
}
defer obj.Shutdown(context.Background())
defer removeRoots(fsDirs)

z := obj.(*erasureServerPools)
xl := z.serverPools[0].sets[0]

bucket := "test-bucket"
object := "test-object"

err = obj.MakeBucket(ctx, bucket, MakeBucketOptions{})
if err != nil {
t.Fatal(err)
}

// Put an initial object so it exists
_, err = obj.PutObject(ctx, bucket, object,
mustGetPutObjReader(t, bytes.NewReader([]byte("initial-value")),
int64(len("initial-value")), "", ""), ObjectOptions{})
if err != nil {
t.Fatal(err)
}

// Get object info to capture the ETag
objInfo, err := obj.GetObjectInfo(ctx, bucket, object, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
existingETag := objInfo.ETag

// Simulate read quorum failure by taking enough disks offline
// With 16 disks (EC 8+8), read quorum is 9. Taking 8 disks offline leaves only 8,
// which is below read quorum.
erasureDisks := xl.getDisks()
z.serverPools[0].erasureDisksMu.Lock()
xl.getDisks = func() []StorageAPI {
for i := range erasureDisks[:8] {
erasureDisks[i] = nil
}
return erasureDisks
}
z.serverPools[0].erasureDisksMu.Unlock()

t.Run("if-none-match with read quorum failure", func(t *testing.T) {
// Test Case 1: if-none-match (create only if doesn't exist)
// With if-none-match: *, this should only succeed if object doesn't exist.
// Since read quorum fails, we can't determine if object exists.
opts := ObjectOptions{
UserDefined: map[string]string{
xhttp.IfNoneMatch: "*",
},
CheckPrecondFn: func(oi ObjectInfo) bool {
// Precondition fails if object exists (ETag is not empty)
return oi.ETag != ""
},
}

_, err := obj.NewMultipartUpload(ctx, bucket, object, opts)
if !isErrReadQuorum(err) {
t.Errorf("Expected read quorum error when if-none-match is used with quorum failure, got: %v", err)
}
})

t.Run("if-match with wrong ETag and read quorum failure", func(t *testing.T) {
// Test Case 2: if-match with WRONG ETag
// This should fail even without quorum issues, but with quorum failure
// we can't verify the ETag at all.
opts := ObjectOptions{
UserDefined: map[string]string{
xhttp.IfMatch: "wrong-etag-12345",
},
HasIfMatch: true,
CheckPrecondFn: func(oi ObjectInfo) bool {
// Precondition fails if ETags don't match
return oi.ETag != "wrong-etag-12345"
},
}

_, err := obj.NewMultipartUpload(ctx, bucket, object, opts)
if !isErrReadQuorum(err) {
t.Logf("Got error (as expected): %v", err)
t.Logf("But expected read quorum error, not object-not-found error")
}
})

t.Run("if-match with correct ETag and read quorum failure", func(t *testing.T) {
// Test Case 3: if-match with CORRECT ETag but read quorum failure
// Even with the correct ETag, we shouldn't proceed if we can't verify it.
opts := ObjectOptions{
UserDefined: map[string]string{
xhttp.IfMatch: existingETag,
},
HasIfMatch: true,
CheckPrecondFn: func(oi ObjectInfo) bool {
// Precondition fails if ETags don't match
return oi.ETag != existingETag
},
}

_, err := obj.NewMultipartUpload(ctx, bucket, object, opts)
if !isErrReadQuorum(err) {
t.Errorf("Expected read quorum error when if-match is used with quorum failure, got: %v", err)
}
})
}

// TestCompleteMultipartUploadConditionalWithReadQuorumFailure tests that conditional
// complete multipart upload operations behave correctly when read quorum cannot be reached.
func TestCompleteMultipartUploadConditionalWithReadQuorumFailure(t *testing.T) {
ctx := context.Background()

obj, fsDirs, err := prepareErasure16(ctx)
if err != nil {
t.Fatal(err)
}
defer obj.Shutdown(context.Background())
defer removeRoots(fsDirs)

z := obj.(*erasureServerPools)
xl := z.serverPools[0].sets[0]

bucket := "test-bucket"
object := "test-object"

err = obj.MakeBucket(ctx, bucket, MakeBucketOptions{})
if err != nil {
t.Fatal(err)
}

// Put an initial object
_, err = obj.PutObject(ctx, bucket, object,
mustGetPutObjReader(t, bytes.NewReader([]byte("initial-value")),
int64(len("initial-value")), "", ""), ObjectOptions{})
if err != nil {
t.Fatal(err)
}

// Start a multipart upload WITHOUT conditional checks (this should work)
res, err := obj.NewMultipartUpload(ctx, bucket, object, ObjectOptions{})
if err != nil {
t.Fatal(err)
}

// Upload a part
partData := bytes.Repeat([]byte("a"), 5*humanize.MiByte)
md5Hex := getMD5Hash(partData)
_, err = obj.PutObjectPart(ctx, bucket, object, res.UploadID, 1,
mustGetPutObjReader(t, bytes.NewReader(partData), int64(len(partData)), md5Hex, ""),
ObjectOptions{})
if err != nil {
t.Fatal(err)
}

// Now simulate read quorum failure
erasureDisks := xl.getDisks()
z.serverPools[0].erasureDisksMu.Lock()
xl.getDisks = func() []StorageAPI {
for i := range erasureDisks[:8] {
erasureDisks[i] = nil
}
return erasureDisks
}
z.serverPools[0].erasureDisksMu.Unlock()

t.Run("complete multipart with if-none-match and read quorum failure", func(t *testing.T) {
// Try to complete the multipart upload with if-none-match
// This should fail because we can't verify the condition due to read quorum failure
opts := ObjectOptions{
UserDefined: map[string]string{
xhttp.IfNoneMatch: "*",
},
CheckPrecondFn: func(oi ObjectInfo) bool {
return oi.ETag != ""
},
}

parts := []CompletePart{{PartNumber: 1, ETag: md5Hex}}
_, err := obj.CompleteMultipartUpload(ctx, bucket, object, res.UploadID, parts, opts)
if !isErrReadQuorum(err) {
t.Errorf("Expected read quorum error, got: %v", err)
}
})
}
Loading
Loading
0