From 5fcd9d81e909bf724018c6f1e3fd89ec35c5da88 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 16:51:26 -0800 Subject: [PATCH] mount: merge a peer's content change into a dirty handle A subscription event carrying a peer's content change replaced a dirty handle's entry and discarded its unflushed pages, so the handle's writes were lost and its close still returned OK. Fold the peer's entry into the handle with the same merge a refused flush uses, keep the dirty pages, and take the peer's entry as the new base. A replaced file still invalidates. Co-Authored-By: Claude Opus 5.5 --- weed/mount/filehandle.go | 14 ++++++ weed/mount/weedfs.go | 5 ++ .../weedfs_invalidate_open_handle_test.go | 50 +++++++++++++++++-- 3 files changed, 66 insertions(+), 3 deletions(-) diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 5b40622502..fc4112a3b9 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -198,6 +198,20 @@ func (fh *FileHandle) rebaseEntry(merged *filer_pb.Entry, sent []*filer_pb.FileC }) } +// mergePeerContent folds a peer's content change into a dirty handle, so its +// own writes stay on top of the peer's instead of being discarded. +func (fh *FileHandle) mergePeerContent(peer, base *filer_pb.Entry) bool { + var local *filer_pb.Entry + fh.UpdateEntry(func(e *filer_pb.Entry) { + local = proto.Clone(e).(*filer_pb.Entry) + }) + merged, ok := mergeContentEntry(peer, base, local, local) + if ok { + fh.rebaseEntry(merged, local.Chunks) + } + return ok +} + // setAuthoritativeBase installs the base snapshot a local ack acknowledged. func (fh *FileHandle) setAuthoritativeBase(base *filer_pb.Entry) { fh.baseEntry.Store(base) diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 5cd89e448a..b59f9937f0 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -1049,6 +1049,11 @@ 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) + return + } if contentChanged { fh.dirtyPages.Destroy() fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit) diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 7cddb221de..b3d0a9d434 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -1241,7 +1241,8 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) { t.Fatalf("handle version = %d, want 1500", got) } - // The adoption is one-shot: a genuinely foreign event still invalidates. + // The adoption is one-shot: a genuinely foreign event still installs, + // merged under the handle's own writes. if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateEventFor("file", 300, 1600), meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { t.Fatalf("apply foreign event: %v", err) } @@ -1249,8 +1250,11 @@ func TestCommittedCopyWithFailedReadbackAdoptsItsEvent(t *testing.T) { if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 300 { t.Fatalf("live entry file size = %d, want 300 (foreign event after adoption must install)", size) } - if fh.dirtyPages == pagesBefore { - t.Fatal("foreign event after adoption must invalidate dirty pages") + if base := fh.baseEntry.Load(); base.GetAttributes().GetFileSize() != 300 { + t.Fatalf("base %v, want the foreign entry", base) + } + if fh.dirtyPages != pagesBefore { + t.Fatal("dirty pages destroyed by a mergeable foreign event") } } @@ -2075,3 +2079,43 @@ func TestAckFromAnotherFilerIsNotDroppedBehindForeignFence(t *testing.T) { t.Fatal("an older ack from the fencing filer was installed; within one clock domain it must be refused") } } + +// A peer's append reaching a dirty handle is merged under the handle's own +// writes: the uploaded chunk and the unflushed pages both survive. +func TestPeerAppendMergesIntoDirtyHandle(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(1, 2) + peer.Attributes.FileSize = 4 + peer.Chunks[1].Offset, peer.Chunks[1].Size = 3, 1 + appendResp := &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(), appendResp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply append: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + if fh.dirtyPages != pagesBefore { + t.Fatal("dirty pages destroyed by a peer append") + } + entry := fh.GetEntry().GetEntry() + if got := len(entry.Chunks); got != 3 || entry.Attributes.FileSize != 4 { + t.Fatalf("handle has %d chunks, size %d; want the peer's and its own", got, entry.Attributes.FileSize) + } + if base := fh.baseEntry.Load(); len(base.GetChunks()) != 2 { + t.Fatalf("base %v, want the peer's entry", base) + } +}