diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 260c6f5c24..d31918a115 100644 --- a/weed/filer/stream.go +++ b/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 { diff --git a/weed/filer/stream_test.go b/weed/filer/stream_test.go index ec3aba337e..c3b0a02cac 100644 --- a/weed/filer/stream_test.go +++ b/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") +} diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go index f41cfaa7fa..b09bc303fc 100644 --- a/weed/wdclient/filer_client.go +++ b/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 diff --git a/weed/wdclient/filer_client_test.go b/weed/wdclient/filer_client_test.go index b7c589ecc7..f1ccc59ec4 100644 --- a/weed/wdclient/filer_client_test.go +++ b/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) + } + }) + } +}