mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-14 07:48:41 +01:00
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
bb7e4692c0
commit
5fcd9d81e9
3 files changed
+66
-3
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user