From 249999a3bb936808e40c5d44107593d86fa7445f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 10 Oct 2026 23:44:54 +0800 Subject: [PATCH] mount: count only wire silence against the rename timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- weed/mount/weedfs_rename.go | 22 ++++++++++++++++++---- weed/mount/weedfs_stream_mutate.go | 12 +++++++++++- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go index bb9ae8b99e..6285fe7fb6 100644 --- a/weed/mount/weedfs_rename.go +++ b/weed/mount/weedfs_rename.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "strings" + "sync/atomic" "syscall" "time" @@ -33,10 +34,21 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr // 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. + // 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() - silence := time.AfterFunc(wfs.filerRPCWait(), 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) @@ -54,8 +66,10 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr } return fmt.Errorf("dir Rename %s => %s receive: %v", oldPath, newPath, recvErr) } - silence.Reset(wfs.filerRPCWait()) - 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 } } diff --git a/weed/mount/weedfs_stream_mutate.go b/weed/mount/weedfs_stream_mutate.go index ad9f02e0d2..87d9ac811d 100644 --- a/weed/mount/weedfs_stream_mutate.go +++ b/weed/mount/weedfs_stream_mutate.go @@ -209,7 +209,16 @@ func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRename if !ok { return fmt.Errorf("rename recv: %w: stream closed", ErrStreamTransport) } - respTimer.Reset(m.wfs.filerRPCWait()) + // 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 { @@ -226,6 +235,7 @@ 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)