s3: return an error for GETs canceled before the first byte (#12502)

* s3: answer GETs canceled before the first byte

A stream that fails before writing anything now always gets an S3 error
response, canceled errors included. The idle timeout and a shared lookup
canceled by another request both cancel the request context while the
client is still connected, and returning silently sent that client a
200 with Content-Length: 0. A client that has gone away never sees the
write.

Refs #12500

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: mark inline GET write errors as written; bound the idle-cancel test

An inline object's write error comes after the status line, so the
caller must not write an S3 error on top of it. The idle-cancel test
caps the client wait and gives the response write its own deadline,
keeping only the read side on the 200 ms idle timeout.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: test a GET whose request is canceled before the first byte

Drives GetObjectHandler through a chunk lookup that cancels the request
context, so a handler that returns before answering shows up as the empty
200 again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3: check the cancel test reached the chunk lookup

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

weed/s3api/s3api_object_handlers.go

@@ -289,8 +289,32 @@ func isCanceledStreamingError(err error) bool {
 	return errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled
 }
 
-func shouldWriteStreamingErrorResponse(err error) bool {
-	return err != nil && !isCanceledStreamingError(err)
+// writeStreamErrorResponse answers a stream failure unless the stream already
+// wrote a response. A canceled error is answered too: the idle timeout and a
+// shared lookup canceled by another request both cancel it while the client is
+// still connected, and returning silently would send that client an empty 200.
+// A client that has gone away never sees the write.
+func writeStreamErrorResponse(w http.ResponseWriter, r *http.Request, err error) {
+	var streamErr *StreamError
+	if errors.As(err, &streamErr) && streamErr.ResponseWritten {
+		return
+	}
+	if errors.Is(err, util_http.ErrTooManyRequests) {
+		s3err.WriteErrorResponse(w, r, s3err.ErrRequestBytesExceed)
+		return
+	}
+	s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+}
+
+// logCanceledStream logs a canceled stream at V(3) when the request context is
+// canceled too, since that is usually the client leaving, and as a warning when
+// only shared work was canceled under a live request.
+func logCanceledStream(r *http.Request, caller, bucket, object string, err error) {
+	if r.Context().Err() != nil {
+		glog.V(3).Infof("%s: request canceled while streaming %s/%s: %v", caller, bucket, object, err)
+		return
+	}
+	glog.Warningf("%s: streaming %s/%s canceled under a live request: %v", caller, bucket, object, err)
 }
 
 func mimeDetect(r *http.Request, dataReader io.Reader) io.ReadCloser {
@@ -480,20 +504,12 @@ func (s3a *S3ApiServer) serveDirectoryContent(w http.ResponseWriter, r *http.Req
 	if r.Method != http.MethodHead && len(entry.Content) == 0 && len(entry.Chunks) > 0 {
 		sseType := s3a.detectPrimarySSEType(entry)
 		if err := s3a.streamFromVolumeServersWithSSE(w, r, entry, sseType, bucket, object, ""); err != nil {
-			var streamErr *StreamError
-			if errors.As(err, &streamErr) && streamErr.ResponseWritten {
-				return
-			}
 			if isCanceledStreamingError(err) {
-				glog.V(3).Infof("serveDirectoryContent: client disconnected while streaming %s/%s: %v", bucket, object, err)
-				return
-			}
-			glog.Errorf("serveDirectoryContent: failed to stream %s/%s: %v", bucket, object, err)
-			if errors.Is(err, util_http.ErrTooManyRequests) {
-				s3err.WriteErrorResponse(w, r, s3err.ErrRequestBytesExceed)
-			} else if shouldWriteStreamingErrorResponse(err) {
-				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+				logCanceledStream(r, "serveDirectoryContent", bucket, object, err)
+			} else {
+				glog.Errorf("serveDirectoryContent: failed to stream %s/%s: %v", bucket, object, err)
 			}
+			writeStreamErrorResponse(w, r, err)
 		}
 		return
 	}
@@ -1127,27 +1143,13 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 	if err != nil {
 		switch {
 		case isCanceledStreamingError(err):
-			glog.V(3).Infof("GetObjectHandler: client disconnected while streaming %s/%s: %v", bucket, object, err)
-			return
+			logCanceledStream(r, "GetObjectHandler", bucket, object, err)
 		case errors.Is(err, context.DeadlineExceeded):
 			glog.Warningf("GetObjectHandler: deadline exceeded while streaming %s/%s: %v", bucket, object, err)
 		default:
 			glog.Errorf("GetObjectHandler: failed to stream %s/%s from volume servers: %v", bucket, object, err)
 		}
-		// Check if the streaming function already wrote an HTTP response
-		var streamErr *StreamError
-		if errors.As(err, &streamErr) && streamErr.ResponseWritten {
-			// Response already written (headers + status code), don't write again
-			// to avoid "superfluous response.WriteHeader call" and malformed S3 error bodies
-			return
-		}
-		// Response not yet written - safe to write S3 error response
-		// Check if error is due to volume server rate limiting (HTTP 429)
-		if errors.Is(err, util_http.ErrTooManyRequests) {
-			s3err.WriteErrorResponse(w, r, s3err.ErrRequestBytesExceed)
-		} else if shouldWriteStreamingErrorResponse(err) {
-			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
-		}
+		writeStreamErrorResponse(w, r, err)
 		return
 	}
 }
@@ -1216,7 +1218,10 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 			if written > 0 {
 				BucketTrafficSent(int64(written), r)
 			}
-			return err
+			if err != nil {
+				return newStreamErrorWithResponse(err)
+			}
+			return nil
 		}
 		// Non-range request for inline content
 		s3a.setResponseHeaders(w, r, entry, totalSize)
@@ -1225,7 +1230,10 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		if written > 0 {
 			BucketTrafficSent(int64(written), r)
 		}
-		return err
+		if err != nil {
+			return newStreamErrorWithResponse(err)
+		}
+		return nil
 	}
 
 	// Get chunks and validate BEFORE setting headers
@@ -1262,8 +1270,8 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 				// Authoritative: the entry vanished or its remote object is gone
 				return respondObjectGone(w, r, bucket, object, cacheErr)
 			} else {
-				// Client disconnected during the cache wait: report cancellation, not an
-				// error response, so we don't write to a closed connection.
+				// Request canceled during the cache wait: skip the origin and let the
+				// caller answer the cancellation.
 				if ctxErr := r.Context().Err(); ctxErr != nil {
 					return ctxErr
 				}

weed/s3api/s3api_stream_error_test.go

@@ -2,60 +2,123 @@ package s3api
 
 import (
 	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"testing"
+	"time"
 
+	"github.com/gorilla/mux"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
-func TestShouldWriteStreamingErrorResponse(t *testing.T) {
+func TestWriteStreamErrorResponse(t *testing.T) {
 	tests := []struct {
-		name     string
-		err      error
-		expected bool
+		name       string
+		err        error
+		wantStatus int
 	}{
-		{
-			name:     "nil error",
-			err:      nil,
-			expected: false,
-		},
-		{
-			name:     "context canceled",
-			err:      context.Canceled,
-			expected: false,
-		},
-		{
-			name:     "wrapped context canceled",
-			err:      &StreamError{Err: context.Canceled},
-			expected: false,
-		},
-		{
-			name:     "grpc canceled",
-			err:      status.Error(codes.Canceled, "client connection is closing"),
-			expected: false,
-		},
-		{
-			name:     "wrapped grpc canceled",
-			err:      &StreamError{Err: status.Error(codes.Canceled, "client connection is closing")},
-			expected: false,
-		},
-		{
-			name:     "deadline exceeded",
-			err:      context.DeadlineExceeded,
-			expected: true,
-		},
-		{
-			name:     "wrapped deadline exceeded",
-			err:      &StreamError{Err: context.DeadlineExceeded},
-			expected: true,
-		},
+		{"context canceled", context.Canceled, http.StatusInternalServerError},
+		{"wrapped context canceled", fmt.Errorf("resolve chunks: %w", context.Canceled), http.StatusInternalServerError},
+		{"grpc canceled", status.Error(codes.Canceled, "client connection is closing"), http.StatusInternalServerError},
+		{"deadline exceeded", context.DeadlineExceeded, http.StatusInternalServerError},
+		{"unwritten stream error", &StreamError{Err: context.Canceled}, http.StatusInternalServerError},
+		{"too many requests", util_http.ErrTooManyRequests, http.StatusServiceUnavailable},
+		// the recorder's default code stands for "nothing written"
+		{"response already written", newStreamErrorWithResponse(context.Canceled), http.StatusOK},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := shouldWriteStreamingErrorResponse(tt.err); got != tt.expected {
-				t.Fatalf("shouldWriteStreamingErrorResponse(%v) = %v, want %v", tt.err, got, tt.expected)
+			rec := httptest.NewRecorder()
+			writeStreamErrorResponse(rec, httptest.NewRequest(http.MethodGet, "/bucket/object", nil), tt.err)
+			if rec.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d", rec.Code, tt.wantStatus)
 			}
 		})
 	}
 }
+
+// TestIdleTimeoutCancelStillAnswersConnectedClient checks that a request whose
+// context the idle timeout canceled can still answer its connected client, so
+// a canceled stream must not return without a response.
+func TestIdleTimeoutCancelStillAnswersConnectedClient(t *testing.T) {
+	listener, err := util.NewListener("127.0.0.1:0", 200*time.Millisecond)
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// a first read stalled past the idle timeout
+		<-r.Context().Done()
+		// only the read side should be short; a write timing out would fail the
+		// test on a slow runner
+		if err := http.NewResponseController(w).SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+			t.Error(err)
+		}
+		writeStreamErrorResponse(w, r, r.Context().Err())
+	})}
+	go server.Serve(listener)
+	defer server.Close()
+
+	// bounded so a broken idle cancel fails here instead of hanging the package
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+listener.Addr().String()+"/bucket/object", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusInternalServerError {
+		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
+	}
+}
+
+// cancelingLookupFiler cancels the request context while the first chunk
+// lookup is in flight, as the idle timeout does to a client still connected.
+type cancelingLookupFiler struct {
+	anonymousReadFiler
+	cancel context.CancelFunc
+}
+
+func (f *cancelingLookupFiler) LookupVolume(context.Context, *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
+	f.cancel()
+	return nil, status.Error(codes.Canceled, "lookup canceled")
+}
+
+func TestGetObjectAnswersRequestCanceledBeforeFirstByte(t *testing.T) {
+	f := &cancelingLookupFiler{anonymousReadFiler: anonymousReadFiler{entries: map[string]*filer_pb.Entry{
+		"/buckets/b/o": {
+			Name:       "o",
+			Attributes: &filer_pb.FuseAttributes{FileSize: 11, Mtime: 1700000000},
+			Chunks:     []*filer_pb.FileChunk{{FileId: "3,01637037d6", Size: 11, ModifiedTsNs: 1}},
+		},
+	}}}
+	s3a := newPutTestServer(t, startFakeFiler(t, f))
+	s3a.readerCache = filer.NewReaderCache(8, (*chunk_cache.TieredChunkCache)(nil), s3a.filerClient.GetLookupFileIdFunction(), s3a.filerClient)
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b"})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	f.cancel = cancel
+	r := mux.SetURLVars(httptest.NewRequestWithContext(ctx, http.MethodGet, "/b/o", nil), map[string]string{"bucket": "b", "object": "o"})
+	w := httptest.NewRecorder()
+	s3a.GetObjectHandler(w, r)
+	if err := ctx.Err(); err != context.Canceled {
+		t.Fatalf("context error = %v, want %v: the lookup was never reached", err, context.Canceled)
+	}
+	if w.Code != http.StatusInternalServerError {
+		t.Fatalf("status = %d with %d body bytes, want %d", w.Code, w.Body.Len(), http.StatusInternalServerError)
+	}
+}