s3api: handle canceled stream requests cleanly

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 {

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)
+	}
+}

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))

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)
+			}
+		})
+	}
+}