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