From 797c2366c55c12ffc1e1660be9e5f317884668fd Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 30 Mar 2026 11:15:57 -0700 Subject: [PATCH] s3api: handle canceled stream requests cleanly --- weed/filer/stream.go | 4 +++ weed/filer/stream_failover_test.go | 35 +++++++++++++++++++ weed/s3api/s3api_object_handlers.go | 36 +++++++++++++++++--- weed/s3api/s3api_stream_error_test.go | 48 +++++++++++++++++++++++++++ 4 files changed, 119 insertions(+), 4 deletions(-) create mode 100644 weed/s3api/s3api_stream_error_test.go diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 511deb696..8b5cae85f 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -179,6 +179,10 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien jwt := jwtFunc(chunkView.FileId) written, err := retriedStreamFetchChunkData(ctx, writer, urlStrings, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize)) + if err != nil && ctx.Err() != nil { + return ctx.Err() + } + // If read failed, try to invalidate cache and re-lookup if err != nil && written == 0 { if invalidator, ok := masterClient.(CacheInvalidator); ok { diff --git a/weed/filer/stream_failover_test.go b/weed/filer/stream_failover_test.go index fcb4917d8..5425d9c21 100644 --- a/weed/filer/stream_failover_test.go +++ b/weed/filer/stream_failover_test.go @@ -1,6 +1,7 @@ package filer import ( + "bytes" "context" "testing" @@ -173,3 +174,37 @@ func TestRetryLogicSkipsSameUrls(t *testing.T) { t.Error("Expected different URLs to not be equal") } } + +func TestCanceledStreamSkipsCacheInvalidation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + fileId := "3,canceled" + + mock := &mockMasterClient{ + lookupFunc: func(ctx context.Context, fid string) ([]string, error) { + return []string{"http://server:8080"}, nil + }, + } + + chunks := []*filer_pb.FileChunk{ + { + FileId: fileId, + Offset: 0, + Size: 10, + }, + } + + streamFn, err := PrepareStreamContentWithThrottler(ctx, mock, noJwtFunc, chunks, 0, 10, 0) + if err != nil { + t.Fatalf("PrepareStreamContentWithThrottler failed: %v", err) + } + + cancel() + + err = streamFn(&bytes.Buffer{}) + if err != context.Canceled { + t.Fatalf("expected context.Canceled, got %v", err) + } + if len(mock.invalidatedFileIds) != 0 { + t.Fatalf("expected no cache invalidation on cancellation, got %v", mock.invalidatedFileIds) + } +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 3d08b39da..41bc4af57 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -248,6 +248,10 @@ func newStreamErrorWithResponse(err error) *StreamError { return &StreamError{Err: err, ResponseWritten: true} } +func shouldWriteStreamingErrorResponse(err error) bool { + return err != nil && !errors.Is(err, context.Canceled) +} + func mimeDetect(r *http.Request, dataReader io.Reader) io.ReadCloser { mimeBuffer := make([]byte, 512) size, _ := dataReader.Read(mimeBuffer) @@ -879,7 +883,15 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) err = s3a.streamFromVolumeServersWithSSE(w, r, objectEntryForSSE, primarySSEType, bucket, object, versionId) streamTime = time.Since(tStream) if err != nil { - glog.Errorf("GetObjectHandler: failed to stream %s/%s from volume servers: %v", bucket, object, err) + switch { + case errors.Is(err, context.Canceled): + glog.V(3).Infof("GetObjectHandler: client disconnected while streaming %s/%s: %v", bucket, object, err) + return + 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 { @@ -891,7 +903,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) // 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 { + } else if shouldWriteStreamingErrorResponse(err) { s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) } return @@ -1027,7 +1039,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size) chunkResolveTime = time.Since(tChunkResolve) if err != nil { - glog.Errorf("streamFromVolumeServers: failed to resolve chunks: %v", err) + if errors.Is(err, context.Canceled) { + glog.V(3).Infof("streamFromVolumeServers: request canceled while resolving chunks: %v", err) + return err + } + if errors.Is(err, context.DeadlineExceeded) { + glog.Warningf("streamFromVolumeServers: request deadline exceeded while resolving chunks: %v", err) + } else { + glog.Errorf("streamFromVolumeServers: failed to resolve chunks: %v", err) + } // Write S3-compliant XML error response s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) return newStreamErrorWithResponse(fmt.Errorf("failed to resolve chunks: %v", err)) @@ -1047,7 +1067,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R ) streamPrepTime = time.Since(tStreamPrep) if err != nil { - glog.Errorf("streamFromVolumeServers: failed to prepare stream: %v", err) + if errors.Is(err, context.Canceled) { + glog.V(3).Infof("streamFromVolumeServers: request canceled while preparing stream: %v", err) + return err + } + if errors.Is(err, context.DeadlineExceeded) { + glog.Warningf("streamFromVolumeServers: request deadline exceeded while preparing stream: %v", err) + } else { + glog.Errorf("streamFromVolumeServers: failed to prepare stream: %v", err) + } // Write S3-compliant XML error response s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) return newStreamErrorWithResponse(fmt.Errorf("failed to prepare stream: %v", err)) diff --git a/weed/s3api/s3api_stream_error_test.go b/weed/s3api/s3api_stream_error_test.go new file mode 100644 index 000000000..90eb9f425 --- /dev/null +++ b/weed/s3api/s3api_stream_error_test.go @@ -0,0 +1,48 @@ +package s3api + +import ( + "context" + "testing" +) + +func TestShouldWriteStreamingErrorResponse(t *testing.T) { + tests := []struct { + name string + err error + expected bool + }{ + { + 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: "deadline exceeded", + err: context.DeadlineExceeded, + expected: true, + }, + { + name: "wrapped deadline exceeded", + err: &StreamError{Err: context.DeadlineExceeded}, + expected: true, + }, + } + + 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) + } + }) + } +}