From 53377bcb5de01894fd037100924b4c84808bbb69 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 14 Nov 2026 11:11:01 -0800 Subject: [PATCH] mount: keep a handle dirty when a write lands during its flush (#12652) A flush uploads the dirty pages without the handle lock, then commits the entry under it. A write in between grows the entry's size, but its pages are not in the committed chunks. Clearing dirtyMetadata then made the next flush skip the commit, leaving a file with the new size and no chunk for the write, which reads back as zeros. Count page writes on the handle and clear dirtyMetadata only when none landed since the flush started uploading. Co-authored-by: Claude Opus 5.5 --- weed/mount/content_gap.go | 3 ++ weed/mount/filehandle.go | 3 ++ weed/mount/flush_write_race_test.go | 72 ++++++++++++++++++++++++++++ weed/mount/weedfs_async_flush.go | 2 +- weed/mount/weedfs_file_copy_range.go | 1 + weed/mount/weedfs_file_sync.go | 6 +-- weed/mount/weedfs_file_write.go | 1 + 7 files changed, 84 insertions(+), 4 deletions(-) create mode 100644 weed/mount/flush_write_race_test.go diff --git a/weed/mount/content_gap.go b/weed/mount/content_gap.go index b9b5ba8c77..811676bc2b 100644 --- a/weed/mount/content_gap.go +++ b/weed/mount/content_gap.go @@ -14,6 +14,9 @@ import ( type contentPublication struct { refused bool failure *atomic.Bool + // writes is the handle's pageWrites before its pages were flushed. A + // later write is not in the published chunks, so it keeps the handle dirty. + writes uint64 } func (p *contentPublication) retire(err error) { diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 13e144b165..89f2f41d6c 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -58,6 +58,9 @@ type FileHandle struct { // could not be read; the next flush reads it before publishing. linkRefresh []byte + // pageWrites counts the pages added to dirtyPages, under the handle lock. + pageWrites atomic.Uint64 + // entryVersionTsNs is the filer log position the handle's entry reflects. // State at or below it must not replace the entry — that rolls it back. entryVersionTsNs atomic.Int64 diff --git a/weed/mount/flush_write_race_test.go b/weed/mount/flush_write_race_test.go new file mode 100644 index 0000000000..d73d707df2 --- /dev/null +++ b/weed/mount/flush_write_race_test.go @@ -0,0 +1,72 @@ +//go:build !freebsd + +package mount + +import ( + "bytes" + "io" + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" +) + +// A FLUSH from another fd of the open file uploads its pages, then a WRITE +// lands before it commits. The commit carries the new size but not the +// write's chunk, so the next flush must still publish that chunk. +func TestFlushRacingWriteKeepsHandleDirty(t *testing.T) { + base := attributeAppendEntry(1) + base.CommitTsNs = 100 + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + w, handle := g6AppendHandle(t, peer, base, 100) + fh := w.GetHandle(FileHandleId(handle)) + write := func(offset uint64, value byte) { + t.Helper() + n, code := w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle, Offset: offset, Size: 256}, bytes.Repeat([]byte{value}, 256)) + if n != 256 || code != fuse.OK { + t.Fatalf("write at %d: %d %v", offset, n, code) + } + } + flush := func() fuse.Status { + return w.Flush(nil, &fuse.FlushIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: handle}) + } + + uploading, resume := make(chan struct{}), make(chan struct{}) + upload := fh.dirtyPages.randomWriter.(*ChunkedDirtyPages).saveDataAsChunk + first := true + fh.dirtyPages.randomWriter.(*ChunkedDirtyPages).saveDataAsChunk = func(p util.FullPath) filer.SaveDataAsChunkFunctionType { + save := upload(p) + return func(r io.Reader, name string, offset, ts int64, ttl uint64) (*filer_pb.FileChunk, error) { + if first { + first = false + close(uploading) + <-resume + } + return save(r, name, offset, ts, ttl) + } + } + + write(256, 'A') + racing := make(chan fuse.Status, 1) + go func() { racing <- flush() }() + <-uploading + write(512, 'B') + close(resume) + if code := <-racing; code != fuse.OK { + t.Fatalf("racing flush: %v", code) + } + if code := flush(); code != fuse.OK { + t.Fatalf("owner flush: %v", code) + } + + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + want = append(want, bytes.Repeat([]byte{'B'}, 256)...) + if got := g6StoredBytes(t, peer); !bytes.Equal(got, want) { + t.Fatalf("stored size %d chunks %v: the racing write's bytes were not published", peer.entry.Attributes.FileSize, peer.entry.Chunks) + } +} diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index 8e81a14a7a..f86743fe7c 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -140,7 +140,7 @@ func isFilerUnreachable(err error) bool { // nothing else holds these references: like the writeback chunk uploads, it // keeps retrying for as long as the filer is unreachable. func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFullPath util.FullPath) error { - publication := &contentPublication{failure: fh.writebackContentError} + publication := &contentPublication{failure: fh.writebackContentError, writes: fh.pageWrites.Load()} attempts := 0 err := retryMetadataFlushUntil(context.Background(), func() error { attempts++ diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index e589ea82cb..0497d23b6d 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -188,6 +188,7 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn) glog.Errorf("AddPage error: %v", err) return 0, writeErrorToFuseStatus(err) } + fhOut.pageWrites.Add(1) // Accumulate for the next loop iteration totalCopied += numBytesRead diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index d61c5a6650..da68a4ca7a 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -191,6 +191,7 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al // Synchronous flush path (normal mode, fsync, or no dirty data) fh.asyncFlushPending = false + publication := &contentPublication{failure: fh.writebackContentError, writes: fh.pageWrites.Load()} // Check quota including uncommitted writes for real-time enforcement isOverQuota := wfs.IsOverQuotaWithUncommitted() @@ -222,7 +223,6 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al commit = &commitBudget{left: metadataTimeout} } - publication := &contentPublication{failure: fh.writebackContentError} if err := retryMetadataFlushIf(ctx, func() error { return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid, commit, publication) }, func(err error) bool { @@ -255,7 +255,7 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al // commit bounds the filer calls made here, over all attempts of one flush; nil // leaves them to the per-call filer timeouts. func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32, commit *commitBudget, publications ...*contentPublication) error { - publication := &contentPublication{} + publication := &contentPublication{writes: fh.pageWrites.Load()} if len(publications) > 0 { publication = publications[0] } @@ -391,7 +391,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n wfs.inodeToPath.InvalidateChildrenCache(util.FullPath(dir)) } - if err == nil { + if err == nil && fh.pageWrites.Load() == publication.writes { fh.pendingAppend = false fh.dirtyMetadata = false fh.dirtyContent = false diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index e7ee4187ef..a630b8984a 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -102,6 +102,7 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr glog.Errorf("AddPage error: %v", err) return 0, writeErrorToFuseStatus(err) } + fh.pageWrites.Add(1) written = uint32(len(data))