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 <noreply@anthropic.com>

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 {

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.

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