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