From b0cc4abb873547ae794f70cc9af13ba41f9fd193 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 17:17:53 -0800 Subject: [PATCH] mount: keep a dirty handle's writes when a peer change cannot merge The new merge refusals sent a peer event on the same file down the old path, which discarded the dirty pages and adopted the peer's entry, so close returned OK without the writes. The handle now keeps its writes and base, and the flush meets the conflict and fails. Co-Authored-By: Claude Opus 5.5 --- weed/mount/weedfs.go | 9 +++-- weed/mount/weedfs_file_sync.go | 14 +++++-- .../weedfs_invalidate_open_handle_test.go | 40 +++++++++++++++++++ 3 files changed, 56 insertions(+), 7 deletions(-) diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index b59f9937f0..3a6633c96f 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -1049,9 +1049,12 @@ func (wfs *WFS) invalidateOpenFileHandle(invalidation meta_cache.EntryInvalidati // against an approximate base) leaves them valid. A dirty handle likewise // keeps its diverged entry unless foreign content supersedes it. contentChanged := base == nil || !sameEntryContent(candidate, base) - if contentChanged && fh.dirtyMetadata && fh.mergePeerContent(candidate, base) { - fh.baseEntry.Store(proto.Clone(candidate).(*filer_pb.Entry)) - fh.advanceEntryVersion(candidateTsNs, 0) + if contentChanged && fh.dirtyMetadata && sameContentFile(candidate, base) { + // An unmergeable change keeps the writes; the flush reports it. + if fh.mergePeerContent(candidate, base) { + fh.baseEntry.Store(proto.Clone(candidate).(*filer_pb.Entry)) + fh.advanceEntryVersion(candidateTsNs, 0) + } return } if contentChanged { diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index df106fba45..a6c21d6f69 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -380,10 +380,7 @@ const maxContentFlushMerges = 3 // whole-file rewrite (maybeMergeChunks, unstamped chunks) over a peer's chunk // change cannot be merged: the rewrite carries the handle's stale view. func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo int64) (*filer_pb.Entry, bool) { - if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) { - return nil, false - } - if inode := base.GetAttributes().GetInode(); inode != 0 && stored.GetAttributes().GetInode() != inode { + if !sameContentFile(stored, base) { return nil, false } merged := proto.Clone(stored).(*filer_pb.Entry) @@ -484,6 +481,15 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in return merged, true } +// sameContentFile reports whether stored is still the regular file base was. +func sameContentFile(stored, base *filer_pb.Entry) bool { + if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) { + return false + } + inode := base.GetAttributes().GetInode() + return inode == 0 || stored.GetAttributes().GetInode() == inode +} + // contentChunkKey identifies a chunk reference: truncation shrinks a chunk in // place, so the same file id at another size is a different reference. Both // file id encodings give one key, and the chunk is left as is. diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index b3d0a9d434..a13aff42b4 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -2119,3 +2119,43 @@ func TestPeerAppendMergesIntoDirtyHandle(t *testing.T) { t.Fatalf("base %v, want the peer's entry", base) } } + +// A peer change that cannot be merged (inline content over the handle's new +// chunk) leaves the dirty handle alone, so its flush fails instead of the +// event silently dropping the writes. +func TestUnmergeablePeerChangeKeepsDirtyHandle(t *testing.T) { + wfs := newInvalidateTestWFS(t) + + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1), 0, 0) + fh.AddChunks([]*filer_pb.FileChunk{{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}}) + fh.dirtyMetadata = true + pagesBefore := fh.dirtyPages + + peer := flushConditionEntry() + peer.Content = []byte("peer") + peer.Attributes.FileSize = 4 + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: flushConditionEntry(1), + NewEntry: peer, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + if fh.dirtyPages != pagesBefore { + t.Fatal("dirty pages destroyed by an unmergeable peer change") + } + if got := len(fh.GetEntry().GetEntry().Chunks); got != 2 { + t.Fatalf("handle has %d chunks, want its own 2", got) + } + if base := fh.baseEntry.Load(); len(base.GetChunks()) != 1 || len(base.GetContent()) != 0 { + t.Fatalf("base %v moved to the peer's entry", base) + } +}