mount: exercise the unary fallback for the timed-out mutation

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.

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

@@ -84,6 +84,7 @@ type fakeFilerServer struct {
 	lookupCalls             atomic.Int32
 	lookupStarted           chan struct{} // closed when the first lookup arrives
 	lookupGate              chan struct{} // first lookup waits here when non-nil
+	createCalls             atomic.Int32
 }
 
 func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
@@ -125,6 +126,7 @@ func (s *fakeFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream f
 }
 
 func (s *fakeFilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	s.createCalls.Add(1)
 	return &filer_pb.CreateEntryResponse{
 		MetadataEvent: &filer_pb.SubscribeMetadataResponse{
 			Directory: req.Directory,
@@ -672,7 +674,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 +1508,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 +1619,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 +1646,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 +1803,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 +1933,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 +1968,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,217 @@
+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: the sticky index moves off the dead filer and
+// streamCreateEntry's unary fallback lands the same mutation on the live one.
+func TestStreamMutateTimesOutAndRotatesFiler(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	wfs.filerCallTimeout = 200 * time.Millisecond
+
+	wedged := listenFiler(t, &wedgedFiler{})
+	live := &fakeFilerServer{}
+	liveAddress := listenFiler(t, live)
+	wfs.option.FilerAddresses = []pb.ServerAddress{wedged, liveAddress}
+	atomic.StoreInt32(&wfs.option.filerIndex, 0)
+
+	wfs.streamMutate = newStreamMutateMux(wfs)
+	defer wfs.streamMutate.Close()
+
+	resp, err := wfs.streamCreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
+		Directory: "/dir",
+		Entry:     &filer_pb.Entry{Name: "file"},
+	})
+	if err != nil {
+		t.Fatalf("streamCreateEntry: %v, want the unary fallback to succeed", err)
+	}
+	if resp == nil {
+		t.Fatal("streamCreateEntry returned no response")
+	}
+	if got := atomic.LoadInt32(&wfs.option.filerIndex); got != 1 {
+		t.Fatalf("filerIndex = %d, want 1 after teardown of the wedged stream", got)
+	}
+	if got := live.createCalls.Load(); got != 1 {
+		t.Fatalf("live filer createCalls = %d, want 1 (unary fallback ran)", 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)
+	}
+}