From 6ad4b075984c2f280587d2d08072f20f84d14815 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 14 Nov 2026 10:47:33 -0800 Subject: [PATCH] mount: scope and order the final-release writeback fence (#12651) * mount: fence a dropped handle only when it holds unpublished content dirtyMetadata is also set by chmod, xattr and fallocate, so a failed metadata-only close refused every later write of the inode. Co-Authored-By: Claude Opus 5.5 * mount: fence the inode before its last handle leaves the handle map A reopen between the map removal and the fence could publish at the stale kernel EOF. Co-Authored-By: Claude Opus 5.5 * mount: fence a failed metadata flush only when the handle holds content A refused synchronous flush or an abandoned writeback metadata flush of a handle with only chmod, xattr or fallocate changes fenced the inode, so later writes were refused until FORGET. Co-Authored-By: Claude Opus 5.5 * mount: count a size change on an open handle as unpublished content A truncate or extend through an open handle changes the entry's size and chunks, but it set only dirtyMetadata. A failed close then skipped the final-release fence, so a reopen before FORGET could publish against the untruncated entry. Co-Authored-By: Claude Opus 5.5 * mount: mark copied content dirty before each page of a streaming copy A copy that failed or was cancelled after earlier pages were accepted left dirtyContent false, so the final-release fence skipped that content. Co-Authored-By: Claude Opus 5.5 * mount: mark copied content dirty only after a page is accepted A copy whose first page is rejected leaves the destination unchanged and the kernel EOF where it was, so fencing the inode on release only blocked later writes until FORGET. Co-Authored-By: Claude Opus 5.5 * mount: skip the server-side copy while the destination holds unpublished content A whole-file server copy clears dirtyContent when it commits, so a page a cancelled streaming copy left behind would lose its release fence. Fall back to the streaming copy instead. Co-Authored-By: Claude Opus 5.5 * mount: skip the server-side copy while the source holds unpublished content The filer copies the stored source, so pages still pending on the source handle would be missing from the destination. The streaming copy reads them through the dirty pages. Co-Authored-By: Claude Opus 5.5 * mount: check the publication fence before a server-side whole-file copy The filer copy publishes the destination without going through the flush-time guard, so a fenced writeback inode could still take new content before FORGET. Fall back to the streaming copy, whose flush is guarded. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/mount/filehandle.go | 2 + weed/mount/filehandle_map.go | 4 + weed/mount/weedfs_async_flush.go | 4 +- weed/mount/weedfs_async_flush_test.go | 171 ++++++++++++++-------- weed/mount/weedfs_attr.go | 3 + weed/mount/weedfs_file_copy_range.go | 10 +- weed/mount/weedfs_file_copy_range_test.go | 86 +++++++++++ weed/mount/weedfs_file_sync.go | 5 +- weed/mount/weedfs_file_write.go | 1 + weed/mount/weedfs_filehandle.go | 3 - 10 files changed, 223 insertions(+), 66 deletions(-) diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 736529b2e2..13e144b165 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -38,6 +38,7 @@ type FileHandle struct { pendingAppendOffset uint64 writebackContentError *atomic.Bool dirtyMetadata bool + dirtyContent bool dirtyPages *PageWriter reader *filer.ChunkReadAt contentType string @@ -340,6 +341,7 @@ func (fh *FileHandle) ResetDirtyPages() { fh.dirtyPages.Destroy() fh.dirtyPages = newPageWriter(fh, fh.wfs.option.ChunkSizeLimit) fh.dirtyMetadata = false + fh.dirtyContent = false fh.pendingAppend = false fh.contentType = "" } diff --git a/weed/mount/filehandle_map.go b/weed/mount/filehandle_map.go index 99159bf3cc..84831f94c1 100644 --- a/weed/mount/filehandle_map.go +++ b/weed/mount/filehandle_map.go @@ -142,6 +142,10 @@ func (i *FileHandleToInode) ReleaseByHandle(fh FileHandleId) *FileHandle { // via FindFileHandle during the background drain. return fhHandle } + // Fence before a reopen can see the inode without this handle. + if fhHandle.dirtyContent && !fhHandle.isDeleted { + (&contentPublication{failure: fhHandle.writebackContentError}).abandon() + } delete(i.inode2fh, inode) for openFh := range fhHandle.fhs { delete(i.fh2inode, openFh) diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index 1ed0d54beb..8e81a14a7a 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -155,7 +155,9 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul fileFullPath, nextAttempt, totalAttempts, backoff, err) }) if err != nil { - publication.abandon() + if fh.dirtyContent { + publication.abandon() + } glog.Errorf("completeAsyncFlush %s: metadata flush failed after %d attempts: %v - "+ "chunks are uploaded but NOT referenced in filer metadata; "+ "they will appear as orphans in volume.fsck", diff --git a/weed/mount/weedfs_async_flush_test.go b/weed/mount/weedfs_async_flush_test.go index 79772343ab..bbf26a5feb 100644 --- a/weed/mount/weedfs_async_flush_test.go +++ b/weed/mount/weedfs_async_flush_test.go @@ -89,37 +89,43 @@ func TestAbandonedWritebackFlushFencesInode(t *testing.T) { }) metadataFlushSleep = func(ctx context.Context, d time.Duration) {} - const name = "abandoned-writeback.log" - wfs, _ := newCreateTestWFS(t) - fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { - return nil, status.Error(codes.PermissionDenied, "denied") - }} - startFakeFiler(t, wfs, fake) + for _, content := range []bool{true, false} { + const name = "abandoned-writeback.log" + wfs, _ := newCreateTestWFS(t) + fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + }} + startFakeFiler(t, wfs, fake) - out := &fuse.CreateOut{} - if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ - InHeader: fuse.InHeader{NodeId: 1}, - Flags: syscall.O_WRONLY | syscall.O_CREAT, - Mode: 0o644, - }, name, out); code != fuse.OK { - t.Fatalf("Create status = %v, want OK", code) - } - fh := wfs.GetHandle(FileHandleId(out.Fh)) - if fh == nil { - t.Fatal("Create did not register an open file handle") - } - wfs.option.WritebackCache = true - fh.writebackContentError = new(atomic.Bool) - fh.dirtyMetadata = true + out := &fuse.CreateOut{} + if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ + InHeader: fuse.InHeader{NodeId: 1}, + Flags: syscall.O_WRONLY | syscall.O_CREAT, + Mode: 0o644, + }, name, out); code != fuse.OK { + t.Fatalf("Create status = %v, want OK", code) + } + fh := wfs.GetHandle(FileHandleId(out.Fh)) + if fh == nil { + t.Fatal("Create did not register an open file handle") + } + wfs.option.WritebackCache = true + fh.writebackContentError = new(atomic.Bool) + fh.dirtyMetadata = true + fh.dirtyContent = content - if err := wfs.flushMetadataWithRetry(fh, "/", name, util.FullPath("/").Child(name)); err == nil { - t.Fatal("flush against a refusing filer succeeded") - } - if !fh.writebackContentError.Load() { - t.Fatal("abandoned writeback flush left the inode unfenced") - } - if err := wfs.checkContentPublication(fh, nil); !errors.Is(err, filer_pb.ErrPreconditionFailed) { - t.Fatalf("checkContentPublication after abandonment = %v, want a precondition refusal", err) + if err := wfs.flushMetadataWithRetry(fh, "/", name, util.FullPath("/").Child(name)); err == nil { + t.Fatal("flush against a refusing filer succeeded") + } + if got := fh.writebackContentError.Load(); got != content { + t.Fatalf("content=%v: fenced after abandoned writeback flush = %v, want %v", content, got, content) + } + if !content { + continue + } + if err := wfs.checkContentPublication(fh, nil); !errors.Is(err, filer_pb.ErrPreconditionFailed) { + t.Fatalf("checkContentPublication after abandonment = %v, want a precondition refusal", err) + } } } @@ -130,42 +136,89 @@ func TestFailedFlockReleaseFencesInode(t *testing.T) { }) metadataFlushSleep = func(ctx context.Context, d time.Duration) {} - const name = "flock-release.log" + for _, c := range []struct { + name string + content bool + truncate bool + wantFenced bool + }{ + {"content", true, false, true}, + {"metadata only", false, false, false}, + {"truncate", false, true, true}, + } { + t.Run(c.name, func(t *testing.T) { + const name = "flock-release.log" + wfs, _ := newCreateTestWFS(t) + wfs.posixLocks = NewPosixLockTable() + fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + }} + startFakeFiler(t, wfs, fake) + + out := &fuse.CreateOut{} + if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ + InHeader: fuse.InHeader{NodeId: 1}, + Flags: syscall.O_WRONLY | syscall.O_CREAT, + Mode: 0o644, + }, name, out); code != fuse.OK { + t.Fatalf("Create status = %v, want OK", code) + } + fh := wfs.GetHandle(FileHandleId(out.Fh)) + if fh == nil { + t.Fatal("Create did not register an open file handle") + } + wfs.option.WritebackCache = true + fenced := new(atomic.Bool) + fh.writebackContentError = fenced + fh.dirtyMetadata = true + fh.dirtyContent = c.content + if c.truncate { + if code := wfs.SetAttr(make(chan struct{}), &fuse.SetAttrIn{SetAttrInCommon: fuse.SetAttrInCommon{ + InHeader: fuse.InHeader{NodeId: out.NodeId}, + Valid: fuse.FATTR_SIZE | fuse.FATTR_FH, + Fh: out.Fh, + Size: 64, + }}, &fuse.AttrOut{}); code != fuse.OK { + t.Fatalf("SetAttr size = %v, want OK", code) + } + } + + wfs.Release(make(chan struct{}), &fuse.ReleaseIn{ + InHeader: fuse.InHeader{NodeId: out.NodeId}, + Fh: out.Fh, + ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK, + }) + if fake.calls.Load() == 0 { + t.Fatal("synchronous release did not attempt the flush") + } + if got := fenced.Load(); got != c.wantFenced { + t.Fatalf("fenced after failed release = %v, want %v", got, c.wantFenced) + } + }) + } +} + +func TestReleaseFencesBeforeReopenSeesInode(t *testing.T) { wfs, _ := newCreateTestWFS(t) - wfs.posixLocks = NewPosixLockTable() - fake := &rejectingCreateFiler{rejections: 1 << 20, refuse: func(req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { - return nil, status.Error(codes.PermissionDenied, "denied") - }} - startFakeFiler(t, wfs, fake) - - out := &fuse.CreateOut{} - if code := wfs.Create(make(chan struct{}), &fuse.CreateIn{ - InHeader: fuse.InHeader{NodeId: 1}, - Flags: syscall.O_WRONLY | syscall.O_CREAT, - Mode: 0o644, - }, name, out); code != fuse.OK { - t.Fatalf("Create status = %v, want OK", code) - } - fh := wfs.GetHandle(FileHandleId(out.Fh)) - if fh == nil { - t.Fatal("Create did not register an open file handle") - } wfs.option.WritebackCache = true - fenced := new(atomic.Bool) - fh.writebackContentError = fenced + const inode = 42 + entry := &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{}} + fh, id, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, entry, 1, 0) fh.dirtyMetadata = true + fh.dirtyContent = true - wfs.Release(make(chan struct{}), &fuse.ReleaseIn{ - InHeader: fuse.InHeader{NodeId: out.NodeId}, - Fh: out.Fh, - ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK, - }) - if fake.calls.Load() == 0 { - t.Fatal("synchronous release did not attempt the flush") + released := wfs.fhMap.ReleaseByHandle(id) + if released != fh { + t.Fatal("last release did not hand back the handle") } - if !fenced.Load() { - t.Fatal("release dropped uncommitted writeback content without fencing the inode") + reopened, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, entry, 1, 0) + if reopened == fh { + t.Fatal("reopen reused the released handle") } + if err := wfs.checkContentPublication(reopened, nil); !errors.Is(err, filer_pb.ErrPreconditionFailed) { + t.Fatalf("publication on a reopen after an abandoned release = %v, want a precondition refusal", err) + } + released.ReleaseHandle() } func TestIsFilerUnreachable(t *testing.T) { diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index c60cff7e8f..467a4b88d4 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -191,6 +191,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse if fh != nil { fh.dirtyMetadata = true + if includeSize { + fh.dirtyContent = true + } return fuse.OK } if path == "" { diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index 3ce3c4ad7b..e589ea82cb 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -191,6 +191,7 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn) // Accumulate for the next loop iteration totalCopied += numBytesRead + fhOut.dirtyContent = true } if totalCopied == 0 { @@ -209,6 +210,10 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn) } func (wfs *WFS) tryServerSideWholeFileCopy(cancel <-chan struct{}, in *fuse.CopyFileRangeIn, fhIn, fhOut *FileHandle) (written uint32, handled bool, code fuse.Status) { + if err := wfs.checkContentPublication(fhOut, nil); err != nil { + glog.V(4).Infof("server-side copy: skipped (%v)", err) + return 0, false, fuse.OK + } copyRequest, ok := wholeFileServerCopyCandidate(fhIn, fhOut, in) if !ok { return 0, false, fuse.OK @@ -256,6 +261,7 @@ func (wfs *WFS) applyServerSideWholeFileCopyResult(fhIn, fhOut *FileHandle, dstP fhOut.contentType = entry.Attributes.Mime } fhOut.dirtyMetadata = false + fhOut.dirtyContent = false wfs.updateServerSideWholeFileCopyMetaCache(dstPath, entry) // Note: we intentionally skip fuseServer.InodeNotify/EntryNotify here. // This runs inside the CopyFileRange request handler; those notifies @@ -316,8 +322,8 @@ func wholeFileServerCopyCandidate(fhIn, fhOut *FileHandle, in *fuse.CopyFileRang glog.V(4).Infof("server-side copy: skipped (same file handle)") return wholeFileServerCopyRequest{}, false } - if fhIn.dirtyMetadata || fhOut.dirtyMetadata { - glog.V(4).Infof("server-side copy: skipped (dirty metadata: in=%v out=%v)", fhIn.dirtyMetadata, fhOut.dirtyMetadata) + if fhIn.dirtyMetadata || fhIn.dirtyContent || fhOut.dirtyMetadata || fhOut.dirtyContent { + glog.V(4).Infof("server-side copy: skipped (dirty handle: in=%v/%v out=%v/%v)", fhIn.dirtyMetadata, fhIn.dirtyContent, fhOut.dirtyMetadata, fhOut.dirtyContent) return wholeFileServerCopyRequest{}, false } if in.OffIn != 0 || in.OffOut != 0 { diff --git a/weed/mount/weedfs_file_copy_range_test.go b/weed/mount/weedfs_file_copy_range_test.go index 6224d91648..6ea5471c89 100644 --- a/weed/mount/weedfs_file_copy_range_test.go +++ b/weed/mount/weedfs_file_copy_range_test.go @@ -72,6 +72,14 @@ func TestWholeFileServerCopyCandidate(t *testing.T) { } srcHandle.dirtyMetadata = false + for _, h := range []*FileHandle{srcHandle, dstHandle} { + h.dirtyContent = true + if _, ok := wholeFileServerCopyCandidate(srcHandle, dstHandle, in); ok { + t.Fatal("handle with unpublished content should disable whole-file server copy") + } + h.dirtyContent = false + } + in.Len = 4 if _, ok := wholeFileServerCopyCandidate(srcHandle, dstHandle, in); ok { t.Fatal("short copy request should disable whole-file server copy") @@ -355,3 +363,81 @@ func newCopyRangeTestWFSWithMetaCache(t *testing.T) *WFS { return wfs } + +func TestStreamingCopyMarksOnlyAcceptedContentDirty(t *testing.T) { + for _, c := range []struct { + name string + addFails bool + wantOK bool + wantDirty bool + }{ + {"accepted", false, true, true}, + {"rejected first page", true, false, false}, + } { + t.Run(c.name, func(t *testing.T) { + wfs := newCopyRangeTestWFS() + srcInode := wfs.inodeToPath.Lookup(util.FullPath("/src.txt"), 1, false, false, 0, true) + dstInode := wfs.inodeToPath.Lookup(util.FullPath("/dst.txt"), 1, false, false, 0, true) + srcHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{ + Name: "src.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, FileSize: 5, Inode: srcInode}, + Content: []byte("hello"), + }, 0, 0) + dstHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{ + Name: "dst.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, Inode: dstInode}, + }, 0, 0) + if c.addFails { + dstHandle.dirtyPages.randomWriter.(*ChunkedDirtyPages).setLastError(errors.New("swap file unavailable")) + } + + _, status := wfs.CopyFileRange(make(chan struct{}), &fuse.CopyFileRangeIn{ + FhIn: uint64(srcHandle.fh), + FhOut: uint64(dstHandle.fh), + OffIn: 1, + OffOut: 0, + Len: 4, + }) + if (status == fuse.OK) != c.wantOK { + t.Fatalf("CopyFileRange status = %v, want OK = %v", status, c.wantOK) + } + if dstHandle.dirtyContent != c.wantDirty { + t.Fatalf("dirtyContent = %v, want %v", dstHandle.dirtyContent, c.wantDirty) + } + }) + } +} + +func TestServerSideCopySkipsFencedDestination(t *testing.T) { + wfs := newCopyRangeTestWFS() + wfs.option.WritebackCache = true + srcInode := wfs.inodeToPath.Lookup(util.FullPath("/src.txt"), 1, false, false, 0, true) + dstInode := wfs.inodeToPath.Lookup(util.FullPath("/dst.txt"), 1, false, false, 0, true) + srcHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{ + Name: "src.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, FileSize: 5, Inode: srcInode}, + Content: []byte("hello"), + }, 0, 0) + dstHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{ + Name: "dst.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, Inode: dstInode}, + }, 0, 0) + srcHandle.RememberPath("/src.txt") + dstHandle.RememberPath("/dst.txt") + dstHandle.writebackContentError.Store(true) + + originalCopy := performServerSideWholeFileCopy + t.Cleanup(func() { performServerSideWholeFileCopy = originalCopy }) + performServerSideWholeFileCopy = func(<-chan struct{}, *WFS, wholeFileServerCopyRequest) (*filer_pb.Entry, entryVersion, serverSideWholeFileCopyOutcome, error) { + t.Fatal("fenced destination must not publish through the filer copy") + return nil, entryVersion{}, serverSideWholeFileCopyCommitted, nil + } + + if _, handled, _ := wfs.tryServerSideWholeFileCopy(make(chan struct{}), &fuse.CopyFileRangeIn{ + FhIn: uint64(srcHandle.fh), + FhOut: uint64(dstHandle.fh), + Len: 8, + }, srcHandle, dstHandle); handled { + t.Fatal("fenced destination should fall back to the streaming copy") + } +} diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 8fbd26ef3a..d61c5a6650 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -231,7 +231,9 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err) }); err != nil { - publication.retire(err) + if fh.dirtyContent { + publication.retire(err) + } glog.Errorf("%v fh %d flush: %v", fileFullPath, fh.fh, err) return grpcErrorToFuseStatus(err) } @@ -392,6 +394,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n if err == nil { fh.pendingAppend = false fh.dirtyMetadata = false + fh.dirtyContent = false } return err diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index 38866dc803..e7ee4187ef 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -111,6 +111,7 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr } fh.dirtyMetadata = true + fh.dirtyContent = true // Invalidate the mtime cache so the next Open will not set FOPEN_KEEP_CACHE. wfs.invalidateOpenMtimeCache(in.NodeId) diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 16af2b9e40..288c8d7133 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -183,9 +183,6 @@ func (wfs *WFS) ReleaseHandle(handleId FileHandleId) { wfs.pendingAsyncFlushMu.Unlock() if fhToRelease != nil { - if fhToRelease.dirtyMetadata && !fhToRelease.isDeleted { - (&contentPublication{failure: fhToRelease.writebackContentError}).abandon() - } fhToRelease.ReleaseHandle() } }