mount: fail writes over to the next filer instead of hanging on a dead one (#11696)

* 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

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

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
 		}

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

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

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,

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

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

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

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 {

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

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

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