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