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