diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 014b1947c..862a6e63d 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -43,7 +43,7 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte entry.RUnlock() if remoteOnly { glog.V(4).Infof("download remote entry %s", fileFullPath) - err := fh.downloadRemoteEntry(entry) + err := fh.downloadRemoteEntry(ctx, entry) if err != nil { glog.V(1).Infof("download remote entry %s: %v", fileFullPath, err) return 0, 0, err @@ -111,12 +111,15 @@ func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte return int64(totalRead), ts, err } -func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { +func (fh *FileHandle) downloadRemoteEntry(ctx context.Context, entry *LockedEntry) error { fileFullPath := fh.FullPath() dir, _ := fileFullPath.DirAndName() - err := fh.wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + // callWait 0: this RPC streams a whole remote object filer-side, so a + // healthy download can legitimately outlast any metadata-call bound. The + // attempt stays bound to the caller's context instead. + err := fh.wfs.withFilerClient(ctx, false, 0, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { request := &filer_pb.CacheRemoteObjectToLocalClusterRequest{ Directory: string(dir), @@ -124,7 +127,7 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { } glog.V(4).Infof("download entry: %v", request) - resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request) + resp, err := client.CacheRemoteObjectToLocalCluster(ctx, request) if err != nil { return fmt.Errorf("CacheRemoteObjectToLocalCluster file %s: %v", fileFullPath, err) } diff --git a/weed/mount/filer_conf.go b/weed/mount/filer_conf.go index fde7dc215..c35bdde24 100644 --- a/weed/mount/filer_conf.go +++ b/weed/mount/filer_conf.go @@ -20,8 +20,8 @@ func (wfs *WFS) subscribeFilerConfEvents() (*meta_cache.MetadataFollower, error) confFullName := filepath.Join(filer.DirectoryEtcSeaweedFS, filer.FilerConfName) // read current conf - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - content, err := filer.ReadInsideFiler(context.Background(), client, confDir, confName) + err := wfs.withFilerClient(context.Background(), false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + content, err := filer.ReadInsideFiler(ctx, client, confDir, confName) if err != nil { return err } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index a636ded60..1c7e2eb47 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -145,6 +145,7 @@ type WFS struct { manifestCache *filer.ChunkManifestCache writeBufferAccountant *page_writer.WriteBufferAccountant signature int32 + filerCallTimeout time.Duration // bounds one filer call; tests only, 0 = filerRPCTimeout concurrentWriters *util.LimitedConcurrentExecutor copyBufferPool sync.Pool concurrentCopiersSem chan struct{} @@ -657,8 +658,8 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, var entry *filer_pb.Entry var lookupVersion entryVersion lookupDir, lookupName := fullpath.DirAndName() - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ + err := wfs.withFilerClient(context.Background(), false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + resp, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ Directory: lookupDir, Name: lookupName, }) diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 07c4ca895..f53d23aee 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -672,7 +672,7 @@ func TestRemoteCacheResponseVersionFencesUndeliveredEvents(t *testing.T) { Attributes: &filer_pb.FuseAttributes{FileSize: 88}, }, 0, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 { @@ -1506,7 +1506,7 @@ func TestRemoteDownloadBaseMappedToLocal(t *testing.T) { RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 200}, }, 0, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } // The base — and the live entry — must be in local uid form. @@ -1617,7 +1617,7 @@ func TestStaleRemoteDownloadDoesNotRollBack(t *testing.T) { }) fh.advanceEntryVersion(3000, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } @@ -1644,7 +1644,7 @@ func TestRemoteOnlyHandleTakesUnversionedDownload(t *testing.T) { }, 0, 0) fh.advanceEntryVersion(3000, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } @@ -1801,7 +1801,7 @@ func TestUnversionedRemoteDownloadDoesNotOverwriteVersioned(t *testing.T) { }) fh.advanceEntryVersion(3000, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } @@ -1931,7 +1931,7 @@ func TestRejectedDownloadDoesNotRollBackCache(t *testing.T) { }, 0, 0) fh.advanceEntryVersion(3000, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } // The download publishes asynchronously; a synchronous apply behind it @@ -1966,7 +1966,7 @@ func TestRemoteOnlyHandleRefusesKnownOlderDownload(t *testing.T) { }, 0, 0) fh.advanceEntryVersion(3000, 0) - if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil { + if err := fh.downloadRemoteEntry(context.Background(), fh.GetEntry()); err != nil { t.Fatalf("downloadRemoteEntry: %v", err) } diff --git a/weed/mount/weedfs_posix_lock_routed.go b/weed/mount/weedfs_posix_lock_routed.go index 17c47b4b3..89d311ca6 100644 --- a/weed/mount/weedfs_posix_lock_routed.go +++ b/weed/mount/weedfs_posix_lock_routed.go @@ -153,7 +153,7 @@ func posixLockContext(cancel <-chan struct{}) (context.Context, context.CancelFu func (wfs *WFS) callPosixLock(ctx context.Context, key string, op filer_pb.PosixLockOp, lk posixlock.Range) (*filer_pb.PosixLockResponse, error) { var resp *filer_pb.PosixLockResponse - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var e error resp, e = client.PosixLock(ctx, &filer_pb.PosixLockRequest{ Key: key, @@ -377,7 +377,7 @@ func (wfs *WFS) callPosixReassert(ctx context.Context, key string, locks []posix Sid: l.Sid, Owner: l.Owner, Pid: l.Pid, IsFlock: l.IsFlock, }) } - return wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { _, e := client.PosixLock(ctx, &filer_pb.PosixLockRequest{ Key: key, Op: filer_pb.PosixLockOp_KEEP_ALIVE, diff --git a/weed/mount/weedfs_quota.go b/weed/mount/weedfs_quota.go index c09ddbaf2..96cee6fab 100644 --- a/weed/mount/weedfs_quota.go +++ b/weed/mount/weedfs_quota.go @@ -110,7 +110,7 @@ func (wfs *WFS) getQuotaCheckInterval() time.Duration { } func (wfs *WFS) checkQuotaOnce() { - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(context.Background(), false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { request := &filer_pb.StatisticsRequest{ Collection: wfs.option.Collection, @@ -119,7 +119,7 @@ func (wfs *WFS) checkQuotaOnce() { DiskType: string(wfs.option.DiskType), } - resp, err := client.Statistics(context.Background(), request) + resp, err := client.Statistics(ctx, request) if err != nil { glog.V(0).Infof("reading quota usage %v: %v", request, err) return err diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go index 6a000f3a2..cd0ae361b 100644 --- a/weed/mount/weedfs_rename.go +++ b/weed/mount/weedfs_rename.go @@ -6,7 +6,9 @@ import ( "fmt" "io" "strings" + "sync/atomic" "syscall" + "time" "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/cluster" @@ -28,7 +30,28 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr glog.V(1).Infof("Rename %s => %s: stream failed, falling back to unary: %v", oldPath, newPath, err) } return wfs.WithFilerClient(true, func(client filer_pb.SeaweedFilerClient) error { - stream, streamErr := client.StreamRenameEntry(ctx, request) + // The stream is dedicated to this attempt, so cancelling its context + // abandons just this rename. Bound the silence between events the way + // the mutation mux does: a filer that accepts the stream then stalls + // must not hold the failover walk hostage, but a long rename that + // keeps sending events may take as long as it needs. Local event + // work is not wire silence — a directory rename can spend longer + // than the bound migrating a child's DLM lock — so the timer defers + // while the handler runs instead of cancelling a healthy stream. + streamCtx, cancel := context.WithCancel(ctx) + defer cancel() + var inHandler atomic.Bool + var silence *time.Timer + silence = time.AfterFunc(wfs.filerRPCWait(), func() { + if inHandler.Load() { + silence.Reset(wfs.filerRPCWait()) + return + } + cancel() + }) + defer silence.Stop() + + stream, streamErr := client.StreamRenameEntry(streamCtx, request) if streamErr != nil { return fmt.Errorf("dir AtomicRenameEntry %s => %s : %v", oldPath, newPath, streamErr) } @@ -38,11 +61,20 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr if recvErr == io.EOF { break } + if ctx.Err() == nil && streamCtx.Err() != nil { + return fmt.Errorf("dir Rename %s => %s: no event for %s", oldPath, newPath, wfs.filerRPCWait()) + } return fmt.Errorf("dir Rename %s => %s receive: %v", oldPath, newPath, recvErr) } - if err := wfs.handleRenameResponse(ctx, resp, newPath, newPathLock); err != nil { + inHandler.Store(true) + err := wfs.handleRenameResponse(ctx, resp, newPath, newPathLock) + inHandler.Store(false) + if err != nil { return err } + // The event just proved the filer alive; the bound covers the + // wait for the next one, not the whole rename. + silence.Reset(wfs.filerRPCWait()) } return nil }) diff --git a/weed/mount/weedfs_stats.go b/weed/mount/weedfs_stats.go index 701dc5619..be2c01ce8 100644 --- a/weed/mount/weedfs_stats.go +++ b/weed/mount/weedfs_stats.go @@ -37,7 +37,7 @@ func (wfs *WFS) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.Stat if wfs.stats.lastChecked < time.Now().Unix()-20 { - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(context.Background(), false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { request := &filer_pb.StatisticsRequest{ Collection: wfs.option.Collection, @@ -47,7 +47,7 @@ func (wfs *WFS) StatFs(cancel <-chan struct{}, in *fuse.InHeader, out *fuse.Stat } glog.V(4).Infof("reading filer stats: %+v", request) - resp, err := client.Statistics(context.Background(), request) + resp, err := client.Statistics(ctx, request) if err != nil { glog.V(0).Infof("reading filer stats %v: %v", request, err) return err diff --git a/weed/mount/weedfs_stream_helpers.go b/weed/mount/weedfs_stream_helpers.go index 1498df4de..2a1858b63 100644 --- a/weed/mount/weedfs_stream_helpers.go +++ b/weed/mount/weedfs_stream_helpers.go @@ -20,7 +20,7 @@ func (wfs *WFS) streamCreateEntry(ctx context.Context, req *filer_pb.CreateEntry glog.V(1).Infof("streamCreateEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Entry.Name, err) } var resp *filer_pb.CreateEntryResponse - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var err error resp, err = filer_pb.CreateEntryWithResponse(ctx, client, req) return err @@ -39,7 +39,7 @@ func (wfs *WFS) streamUpdateEntry(ctx context.Context, req *filer_pb.UpdateEntry glog.V(1).Infof("streamUpdateEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Entry.Name, err) } var resp *filer_pb.UpdateEntryResponse - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var err error resp, err = client.UpdateEntry(ctx, req) return err @@ -58,7 +58,7 @@ func (wfs *WFS) streamDeleteEntry(ctx context.Context, req *filer_pb.DeleteEntry glog.V(1).Infof("streamDeleteEntry %s/%s: stream failed, falling back to unary: %v", req.Directory, req.Name, err) } var resp *filer_pb.DeleteEntryResponse - err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var err error resp, err = client.DeleteEntry(ctx, req) if err != nil { diff --git a/weed/mount/weedfs_stream_mutate.go b/weed/mount/weedfs_stream_mutate.go index 79406c402..87d9ac811 100644 --- a/weed/mount/weedfs_stream_mutate.go +++ b/weed/mount/weedfs_stream_mutate.go @@ -8,6 +8,7 @@ import ( "sync" "sync/atomic" "syscall" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -53,6 +54,7 @@ type streamMutateMux struct { stream filer_pb.SeaweedFiler_StreamMutateEntryClient cancel context.CancelFunc grpcConn *grpc.ClientConn // dedicated connection, closed on stream teardown + filerIdx int32 // FilerAddresses index the current stream dialed closed bool disabled bool // permanently disabled if filer doesn't support the RPC stopSend chan struct{} // closed to signal the current sendLoop to exit @@ -182,22 +184,41 @@ func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRename case <-ctx.Done(): return ctx.Err() } + sendTimer := time.NewTimer(m.wfs.filerRPCWait()) select { case err := <-sendReq.errCh: + sendTimer.Stop() if err != nil { return fmt.Errorf("rename send: %w: %v", ErrStreamTransport, err) } + case <-sendTimer.C: + m.teardownStream(gen) + return fmt.Errorf("rename send: %w: timed out", ErrStreamTransport) case <-ctx.Done(): + sendTimer.Stop() return ctx.Err() } - // Collect rename events until is_last=true. + // Collect rename events until is_last=true. The timer bounds silence + // between events, not the whole rename. + respTimer := time.NewTimer(m.wfs.filerRPCWait()) + defer respTimer.Stop() for { select { case resp, ok := <-ch: if !ok { return fmt.Errorf("rename recv: %w: stream closed", ErrStreamTransport) } + // Applying an event can legitimately outlast the silence bound — + // a directory rename migrates an open child's DLM lock under + // another writer. The timer bounds wire silence only, so stop it + // for the local work and re-arm it before the next wait. + if !respTimer.Stop() { + select { + case <-respTimer.C: + default: + } + } if r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_RenameResponse); ok { if r.RenameResponse != nil && r.RenameResponse.EventNotification != nil { if err := onEvent(r.RenameResponse); err != nil { @@ -214,6 +235,10 @@ func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRename } return nil } + respTimer.Reset(m.wfs.filerRPCWait()) + case <-respTimer.C: + m.teardownStream(gen) + return fmt.Errorf("rename recv: %w: timed out", ErrStreamTransport) case <-ctx.Done(): return ctx.Err() } @@ -243,17 +268,29 @@ func (m *streamMutateMux) doUnary(ctx context.Context, req *filer_pb.StreamMutat case <-ctx.Done(): return nil, ctx.Err() } + // Most callers pass context.Background(), so a stream blocked on Send or + // Recv over a dead transport would never surface an error here. Bound the + // wait and tear the stream down so the call can fall back to a unary RPC + // on another filer. + sendTimer := time.NewTimer(m.wfs.filerRPCWait()) select { case err := <-sendReq.errCh: + sendTimer.Stop() if err != nil { return nil, fmt.Errorf("%w: %v", ErrStreamTransport, err) } + case <-sendTimer.C: + m.teardownStream(gen) + return nil, fmt.Errorf("%w: send timed out", ErrStreamTransport) case <-ctx.Done(): + sendTimer.Stop() return nil, ctx.Err() } + respTimer := time.NewTimer(m.wfs.filerRPCWait()) select { case resp, ok := <-ch: + respTimer.Stop() if !ok { return nil, fmt.Errorf("%w: stream closed", ErrStreamTransport) } @@ -268,7 +305,11 @@ func (m *streamMutateMux) doUnary(ctx context.Context, req *filer_pb.StreamMutat } } return resp, nil + case <-respTimer.C: + m.teardownStream(gen) + return nil, fmt.Errorf("%w: response timed out", ErrStreamTransport) case <-ctx.Done(): + respTimer.Stop() return nil, ctx.Err() } } @@ -364,6 +405,7 @@ func (m *streamMutateMux) openStream(out *filer_pb.SeaweedFiler_StreamMutateEntr } atomic.StoreInt32(&m.wfs.option.filerIndex, idx) + m.filerIdx = idx m.cancel = cancel m.grpcConn = grpcConn *out = stream @@ -428,6 +470,7 @@ func (m *streamMutateMux) teardownStream(gen uint64) { m.mu.Unlock() return } + filerIdx := m.filerIdx m.stream = nil if m.stopSend != nil { close(m.stopSend) @@ -439,6 +482,13 @@ func (m *streamMutateMux) teardownStream(gen uint64) { } conn := m.grpcConn m.grpcConn = nil + // Rotate while still holding mu: a new generation cannot open until + // teardown releases the lock, so the index advanced here always belongs + // to the stream that just died — never to a successor ensureStream may + // have opened in between. + if n := int32(len(m.wfs.option.FilerAddresses)); n > 0 { + atomic.CompareAndSwapInt32(&m.wfs.option.filerIndex, filerIdx, (filerIdx+1)%n) + } m.mu.Unlock() // Do NOT call failAllPending here — recvLoop is the sole owner of diff --git a/weed/mount/wfs_filer_client.go b/weed/mount/wfs_filer_client.go index d3e8da546..4f5bf0e66 100644 --- a/weed/mount/wfs_filer_client.go +++ b/weed/mount/wfs_filer_client.go @@ -2,9 +2,13 @@ package mount import ( "context" + "fmt" "sync/atomic" + "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -14,6 +18,20 @@ import ( var _ = filer_pb.FilerClient(&WFS{}) func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) (err error) { + return wfs.withFilerClient(context.Background(), streamingMode, wfs.filerRPCWait(), func(_ context.Context, client filer_pb.SeaweedFilerClient) error { + return fn(client) + }) +} + +// withFilerClient walks the filer addresses until one answers fn. callWait +// bounds one attempt; callWait <= 0 leaves the attempt to ctx alone, for +// calls whose duration the metadata bound cannot anticipate (a whole-file +// remote download). Streaming calls ignore callWait — their callbacks bound +// themselves (the rename fallback arms its own silence timer). The callback +// must run its RPCs under the supplied context: that is what lets a +// timed-out attempt be cancelled without disturbing the shared cached +// channel's other callers. +func (wfs *WFS) withFilerClient(ctx context.Context, streamingMode bool, callWait time.Duration, fn func(ctx context.Context, client filer_pb.SeaweedFilerClient) error) (err error) { return util.Retry("filer grpc", func() error { @@ -22,10 +40,7 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile for x := 0; x < n; x++ { filerGrpcAddress := wfs.option.FilerAddresses[i].ToGrpcAddress() - err = pb.WithGrpcClient(context.Background(), streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error { - client := filer_pb.NewSeaweedFilerClient(grpcConnection) - return fn(client) - }, filerGrpcAddress, false, wfs.option.GrpcDialOption) + err = wfs.callFiler(ctx, streamingMode, callWait, filerGrpcAddress, fn) if err != nil { // glog.V(0).Infof("WithFilerClient %d %v: %v", x, filerGrpcAddress, err) @@ -45,6 +60,74 @@ func (wfs *WFS) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFile } +// filerRPCWait bounds one filer call when the callers' RPCs carry no deadline +// of their own. filerCallTimeout is only settable on the WFS for tests. +func (wfs *WFS) filerRPCWait() time.Duration { + if wfs.filerCallTimeout > 0 { + return wfs.filerCallTimeout + } + return filerRPCTimeout +} + +// callFiler runs fn against one filer under a per-attempt context derived +// from ctx, so a filer whose transport silently stalls (stopped peer, +// black-holed connection) cannot block forever and keep the walk from +// reaching the remaining filer addresses. Cancelling the attempt aborts its +// RPCs on the shared cached channel without tearing that channel down. +// +// On timeout the call waits for the attempt's callback to unwind before +// returning: a slow attempt must not overlap the next filer's run of the +// same closure and clobber the caller's shared result variables. A callback +// that ignored the attempt context is forced out by dropping the channel. +func (wfs *WFS) callFiler(ctx context.Context, streamingMode bool, callWait time.Duration, filerGrpcAddress string, fn func(ctx context.Context, client filer_pb.SeaweedFilerClient) error) error { + if streamingMode { + return pb.WithGrpcClient(ctx, streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error { + return fn(ctx, filer_pb.NewSeaweedFilerClient(grpcConnection)) + }, filerGrpcAddress, false, wfs.option.GrpcDialOption) + } + + attemptCtx, cancel := context.WithCancel(ctx) + if callWait > 0 { + attemptCtx, cancel = context.WithTimeout(ctx, callWait) + } + defer cancel() + + done := make(chan error, 1) + go func() { + done <- pb.WithGrpcClient(attemptCtx, streamingMode, wfs.signature, func(grpcConnection *grpc.ClientConn) error { + return fn(attemptCtx, filer_pb.NewSeaweedFilerClient(grpcConnection)) + }, filerGrpcAddress, false, wfs.option.GrpcDialOption) + }() + + select { + case err := <-done: + if err != nil && ctx.Err() == nil && attemptCtx.Err() == context.DeadlineExceeded { + // fn surfaced the attempt deadline itself; report it like the + // timer path so the failure reads as a filer timeout, not as a + // caller cancellation (which IsTransientError would not retry). + return status.Error(codes.Unavailable, fmt.Sprintf("filer %s: call timed out", filerGrpcAddress)) + } + return err + case <-attemptCtx.Done(): + // Give the callback one RPC's grace to unwind on its own; only if it + // ignored the attempt context does the cached channel get dropped — + // the collateral is borne by calls to the same unresponsive filer, + // which the walk is abandoning anyway. + drain := time.NewTimer(wfs.filerRPCWait()) + defer drain.Stop() + select { + case <-done: + case <-drain.C: + pb.InvalidateGrpcConnection(filerGrpcAddress) + <-done + } + if ctx.Err() != nil { + return ctx.Err() + } + return status.Error(codes.Unavailable, fmt.Sprintf("filer %s: call timed out", filerGrpcAddress)) + } +} + func (wfs *WFS) AdjustedUrl(location *filer_pb.Location) string { if wfs.option.VolumeServerAccess == "publicUrl" { return location.PublicUrl diff --git a/weed/mount/wfs_filer_failover_test.go b/weed/mount/wfs_filer_failover_test.go new file mode 100644 index 000000000..e5b05ddaf --- /dev/null +++ b/weed/mount/wfs_filer_failover_test.go @@ -0,0 +1,212 @@ +package mount + +import ( + "context" + "errors" + "net" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + + "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/util" +) + +// wedgedFiler accepts connections but never answers: LookupDirectoryEntry and +// StreamMutateEntry block until the transport dies. It stands in for a filer +// whose TCP connection black-holes instead of refusing. +type wedgedFiler struct { + filer_pb.UnimplementedSeaweedFilerServer +} + +func (s *wedgedFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + <-ctx.Done() + return nil, ctx.Err() +} + +func (s *wedgedFiler) StreamMutateEntry(stream filer_pb.SeaweedFiler_StreamMutateEntryServer) error { + <-stream.Context().Done() + return stream.Context().Err() +} + +func (s *wedgedFiler) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, stream filer_pb.SeaweedFiler_StreamRenameEntryServer) error { + <-stream.Context().Done() + return stream.Context().Err() +} + +// StreamRenameEntry ends cleanly with no events: enough for a failover test +// to confirm the walk reached this filer. +func (s *fakeFilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, stream filer_pb.SeaweedFiler_StreamRenameEntryServer) error { + return nil +} + +func listenFiler(t *testing.T, fake filer_pb.SeaweedFilerServer) pb.ServerAddress { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = listener.Close() }) + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, fake) + go server.Serve(listener) + t.Cleanup(server.Stop) + return pb.NewServerAddressWithGrpcPort("127.0.0.1:1", listener.Addr().(*net.TCPAddr).Port) +} + +// A wedged filer must not hold the walk hostage: the call is bounded and the +// walk reaches the next address. +func TestWithFilerClientFailsOverPastWedgedFiler(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 200 * time.Millisecond + + wedged := listenFiler(t, &wedgedFiler{}) + live := listenFiler(t, &fakeFilerServer{lookupSize: 42, lookupLogTsNs: 1000}) + wfs.option.FilerAddresses = []pb.ServerAddress{wedged, live} + atomic.StoreInt32(&wfs.option.filerIndex, 0) + + var entry *filer_pb.Entry + err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: "/dir", + Name: "file", + }) + if err != nil { + return err + } + entry = resp.Entry + return nil + }) + if err != nil { + t.Fatalf("WithFilerClient: %v", err) + } + if entry == nil || entry.Attributes.FileSize != 42 { + t.Fatalf("entry = %+v, want lookup served by the live filer", entry) + } + if got := atomic.LoadInt32(&wfs.option.filerIndex); got != 1 { + t.Fatalf("filerIndex = %d, want 1 (the live filer)", got) + } +} + +// A mutation stream whose peer never answers must surface as a transport +// error instead of hanging, and the sticky index must move off the dead filer +// so the next stream opens elsewhere. +func TestStreamMutateTimesOutAndRotatesFiler(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 200 * time.Millisecond + + wedged := listenFiler(t, &wedgedFiler{}) + wfs.option.FilerAddresses = []pb.ServerAddress{ + wedged, + pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1), + } + atomic.StoreInt32(&wfs.option.filerIndex, 0) + + mux := newStreamMutateMux(wfs) + defer mux.Close() + + _, err := mux.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{ + Directory: "/dir", + Entry: &filer_pb.Entry{Name: "file"}, + }) + if !errors.Is(err, ErrStreamTransport) { + t.Fatalf("err = %v, want ErrStreamTransport", err) + } + if got := atomic.LoadInt32(&wfs.option.filerIndex); got != 1 { + t.Fatalf("filerIndex = %d, want 1 after teardown of the wedged stream", got) + } +} + +// A timed-out attempt must unwind before the walk runs the next filer's +// attempt: every attempt runs the same closure, so two in flight at once +// race on the caller's captured result variables. +func TestWithFilerClientDoesNotOverlapAttempts(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 200 * time.Millisecond + + wedged := listenFiler(t, &wedgedFiler{}) + live := listenFiler(t, &fakeFilerServer{lookupSize: 42}) + wfs.option.FilerAddresses = []pb.ServerAddress{wedged, live} + atomic.StoreInt32(&wfs.option.filerIndex, 0) + + var attempts, inside, overlap atomic.Int32 + release := make(chan struct{}) + + done := make(chan error, 1) + go func() { + done <- wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + n := attempts.Add(1) + if !inside.CompareAndSwap(0, 1) { + overlap.Add(1) + } + defer inside.Store(0) + if n == 1 { + // Outlive the attempt deadline on purpose: the walk must + // wait for this callback instead of racing ahead. + <-release + return errors.New("first attempt released") + } + return nil + }) + }() + + deadline := time.Now().Add(10 * wfs.filerCallTimeout) + for attempts.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if attempts.Load() == 0 { + t.Fatal("first attempt never started") + } + + // Past the deadline plus its unwind grace, the blocked callback must + // still be the only attempt: no second run may start alongside it. + time.Sleep(3 * wfs.filerCallTimeout) + if got := attempts.Load(); got != 1 { + t.Fatalf("attempts = %d, want 1 while the timed-out attempt is still unwinding", got) + } + + close(release) + if err := <-done; err != nil { + t.Fatalf("WithFilerClient: %v", err) + } + if got := overlap.Load(); got != 0 { + t.Fatalf("attempts overlapped %d times", got) + } + if got := attempts.Load(); got != 2 { + t.Fatalf("attempts = %d, want 2", got) + } +} + +// The unary fallback for a streamed rename must bound silence too: a filer +// that accepts StreamRenameEntry then never answers cannot hold the walk. +func TestDoRenameFailsOverPastWedgedFiler(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 200 * time.Millisecond + + wedged := listenFiler(t, &wedgedFiler{}) + live := listenFiler(t, &fakeFilerServer{}) + wfs.option.FilerAddresses = []pb.ServerAddress{wedged, live} + atomic.StoreInt32(&wfs.option.filerIndex, 0) + + var newPathLock *cluster.LiveLock + start := time.Now() + err := wfs.doRename(context.Background(), &filer_pb.StreamRenameEntryRequest{ + OldDirectory: "/a", + OldName: "f", + NewDirectory: "/b", + NewName: "f", + }, util.FullPath("/a/f"), util.FullPath("/b/f"), &newPathLock) + if err != nil { + t.Fatalf("doRename: %v", err) + } + if elapsed := time.Since(start); elapsed > 10*wfs.filerCallTimeout { + t.Fatalf("doRename took %v, wedged fallback stream was not bounded", elapsed) + } + if got := atomic.LoadInt32(&wfs.option.filerIndex); got != 1 { + t.Fatalf("filerIndex = %d, want 1 (the live filer)", got) + } +}