From 000baf8dfa1299ebed9b8e51e2af2afdafea780f Mon Sep 17 00:00:00 2001 From: gflatters <56309061+gflatters@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:45:20 +0000 Subject: [PATCH] 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 * 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 * 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 * s3: check the cancel test reached the chunk lookup Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Chris Lu --- weed/s3api/s3api_object_handlers.go | 76 ++++++++------ weed/s3api/s3api_stream_error_test.go | 145 ++++++++++++++++++-------- 2 files changed, 146 insertions(+), 75 deletions(-) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index f618fd569a..18f7dc56b5 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/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 } diff --git a/weed/s3api/s3api_stream_error_test.go b/weed/s3api/s3api_stream_error_test.go index 35bcbf6859..2f8da45f84 100644 --- a/weed/s3api/s3api_stream_error_test.go +++ b/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) + } +}