Fix s3 cancelled read relookup (#12130)

* filer: keep cached volume locations when the reader cancels

* wdclient: do not count a caller's cancelled lookup as a filer failure

weed/filer/stream.go

@@ -119,12 +119,15 @@ type VolumeServerJwtFunction func(fileId string) string
 // refreshUrls lets a fetch loop relearn a chunk's locations inside a single
 // read: drop the cached entry and look it up again, whether every location
 // failed or one did while another answered. Nil when there is nothing to
-// invalidate against.
+// invalidate against. A cancelled read keeps the cached locations.
 func refreshUrls(ctx context.Context, invalidator CacheInvalidator, lookupFn wdclient.LookupFileIdFunctionType, fileId string) util_http.RefreshUrlsFunc {
 	if invalidator == nil || lookupFn == nil {
 		return nil
 	}
 	return func() []string {
+		if ctx.Err() != nil {
+			return nil
+		}
 		invalidator.InvalidateCache(fileId)
 		urls, err := lookupFn(ctx, fileId)
 		if err != nil {

weed/filer/stream_test.go

@@ -1,9 +1,15 @@
 package filer
 
 import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -26,3 +32,27 @@ func TestIsSameDataLeavesChunkOrderAlone(t *testing.T) {
 	assert.Equal(t, "3,ccc", b.Chunks[0].FileId)
 	assert.Equal(t, "4,ddd", b.Chunks[1].FileId)
 }
+
+// A client hanging up mid-read must not evict or re-look-up the volume's locations.
+func TestCancelledFetchKeepsCachedLocations(t *testing.T) {
+	volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-r.Context().Done()
+	}))
+	t.Cleanup(volume.Close)
+
+	inv := &countingInvalidator{}
+	var lookups atomic.Int32
+	lookup := func(ctx context.Context, fileId string) ([]string, error) {
+		lookups.Add(1)
+		return nil, ctx.Err()
+	}
+
+	fileId := "4,12c8f6c18f"
+	ctx, cancel := context.WithCancel(context.Background())
+	time.AfterFunc(50*time.Millisecond, cancel)
+	_, err := util_http.RetriedFetchChunkData(ctx, make([]byte, 8), []string{volume.URL + "/" + fileId}, nil, false, true, 0, fileId, refreshUrls(ctx, inv, lookup, fileId))
+
+	assert.ErrorIs(t, err, context.Canceled)
+	assert.Equal(t, int32(0), inv.invalidations.Load(), "a cancelled read must not evict the volume's locations")
+	assert.Equal(t, int32(0), lookups.Load(), "a cancelled read must not look the volume up again")
+}

weed/wdclient/filer_client.go

@@ -788,6 +788,13 @@ func (p *filerVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []s
 			}()
 
 			if err != nil {
+				// a caller that gave up says nothing about this filer
+				if ctxErr := ctx.Err(); ctxErr != nil {
+					if lastErr != nil {
+						return nil, fmt.Errorf("%w (earlier filer error: %v)", ctxErr, lastErr)
+					}
+					return nil, ctxErr
+				}
 				glog.V(1).Infof("FilerClient: filer %s lookup failed (attempt %d/%d, retry %d/%d): %v", filerAddress, x+1, n, retry+1, maxRetries, err)
 				fc.recordFilerFailureWithHealth(health)
 				lastErr = err

weed/wdclient/filer_client_test.go

@@ -1,13 +1,20 @@
 package wdclient
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func newTestFilerClient(addrs ...pb.ServerAddress) *FilerClient {
@@ -276,3 +283,92 @@ func TestOnPeerUpdateRejoinBumpsGeneration(t *testing.T) {
 		t.Fatalf("rejoined filer was pruned by an in-flight snapshot: %v", got)
 	}
 }
+
+// stallingFiler answers LookupVolume only when the call is abandoned.
+type stallingFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+}
+
+func (s *stallingFiler) LookupVolume(ctx context.Context, _ *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func startStallingFiler(t *testing.T) pb.ServerAddress {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	grpcServer := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(grpcServer, &stallingFiler{})
+	go func() { _ = grpcServer.Serve(lis) }()
+	t.Cleanup(grpcServer.Stop)
+
+	_, port, _ := net.SplitHostPort(lis.Addr().String())
+	return pb.ServerAddress(fmt.Sprintf("127.0.0.1:0.%s", port))
+}
+
+// A caller giving up must not count as a filer failure, the filer's own timeout must.
+func TestLookupVolumeIdsCountsOnlyFilerFailures(t *testing.T) {
+	tests := []struct {
+		name         string
+		callerCtx    func() (context.Context, context.CancelFunc)
+		grpcTimeout  time.Duration
+		wantErr      error
+		wantFailures int32
+	}{
+		{
+			name: "caller cancelled",
+			callerCtx: func() (context.Context, context.CancelFunc) {
+				ctx, cancel := context.WithCancel(context.Background())
+				time.AfterFunc(50*time.Millisecond, cancel)
+				return ctx, cancel
+			},
+			grpcTimeout:  5 * time.Second,
+			wantErr:      context.Canceled,
+			wantFailures: 0,
+		},
+		{
+			name: "caller deadline",
+			callerCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(context.Background(), 50*time.Millisecond)
+			},
+			grpcTimeout:  5 * time.Second,
+			wantErr:      context.DeadlineExceeded,
+			wantFailures: 0,
+		},
+		{
+			name: "filer timeout",
+			callerCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(context.Background())
+			},
+			grpcTimeout:  50 * time.Millisecond,
+			wantFailures: 1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fc := newTestFilerClient(startStallingFiler(t))
+			fc.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials())
+			fc.grpcTimeout = tt.grpcTimeout
+			fc.maxRetries = 1
+			fc.failureThreshold = 3
+			fc.resetTimeout = time.Minute
+
+			ctx, cancel := tt.callerCtx()
+			defer cancel()
+			_, err := (&filerVolumeProvider{filerClient: fc}).LookupVolumeIds(ctx, []string{"4"})
+
+			if err == nil {
+				t.Fatal("lookup against a stalling filer succeeded")
+			}
+			if tt.wantErr != nil && !errors.Is(err, tt.wantErr) {
+				t.Errorf("err = %v, want %v", err, tt.wantErr)
+			}
+			if got := atomic.LoadInt32(&fc.filerHealth[0].failureCount); got != tt.wantFailures {
+				t.Errorf("failureCount = %d, want %d", got, tt.wantFailures)
+			}
+		})
+	}
+}