From 6d692ab0136fa15ada493a4c38d709ce7f9e27d3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 11 Oct 2026 20:18:22 +0800 Subject: [PATCH] mount: fail writes over to the next filer instead of hanging on a dead one (#11696) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * mount: bound each filer call so a wedged filer cannot hold the failover walk WithFilerClient callers run their RPCs on context.Background(), so a filer whose transport stalls without refusing (stopped peer, black-holed connection) blocks the call forever and the walk never reaches the remaining filer addresses. Run each attempt under a timeout; on expiry drop the cached channel so the abandoned call is unblocked and the next address is tried. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: time out wedged mutation streams and rotate to the next filer doUnary and Rename wait on stream Send/Recv under context.Background(); a dead transport never delivers an error, so writes hang without ever reaching the unary fallback. Bound the waits and tear the stream down on timeout; teardown also advances the sticky filer index so the next stream opens against a different filer instead of redialing the dead one. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: cover filer failover past a wedged filer Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: cancel filer attempts by context and unwind them before retry Review follow-ups on the filer failover: - callFiler derives a per-attempt context instead of only abandoning the wait, so a timed-out attempt's RPCs die on their own without tearing down the shared cached channel. The walk now also waits for the attempt's callback to unwind before trying the next filer, so two runs of the same closure cannot race on the caller's captured results. Callbacks that ignore the attempt context are still forced out by dropping the channel, as a last resort after a grace period. - teardownStream captures the dying stream's filer index and rotates it while holding m.mu, so it can never rotate a successor generation's filer by mistake. - The unary rename fallback bounds silence on its dedicated stream the way the mux does, instead of trusting Recv on a wedged filer. - downloadRemoteEntry opts out of the metadata-call bound: a healthy whole-file download can legitimately outlast it. * mount: count only wire silence against the rename timeout Applying a rename event can outlast the silence bound on a healthy filer — a directory rename migrates an open child's DLM lock under another writer. Stop the mux's response timer for the local work and re-arm it before the next wait; the fallback stream's watchdog defers while the handler runs instead of cancelling a live stream. * mount: restart the rename fallback's silence bound after each event The bound covers the wait for the next response, not the rename's whole lifetime — a healthy rename that keeps sending events can outlast it. Re-arm after each handled event; the defer-while-handling stays for local work that outlasts the interval on its own. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/mount/filehandle_read.go | 11 +- weed/mount/filer_conf.go | 4 +- weed/mount/weedfs.go | 5 +- .../weedfs_invalidate_open_handle_test.go | 14 +- weed/mount/weedfs_posix_lock_routed.go | 4 +- weed/mount/weedfs_quota.go | 4 +- weed/mount/weedfs_rename.go | 36 ++- weed/mount/weedfs_stats.go | 4 +- weed/mount/weedfs_stream_helpers.go | 6 +- weed/mount/weedfs_stream_mutate.go | 52 ++++- weed/mount/wfs_filer_client.go | 91 +++++++- weed/mount/wfs_filer_failover_test.go | 212 ++++++++++++++++++ 12 files changed, 412 insertions(+), 31 deletions(-) create mode 100644 weed/mount/wfs_filer_failover_test.go 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) + } +}