diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index b4ddf00612..f62d798bc7 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -7,6 +7,7 @@ import ( "slices" "sync" "testing" + "time" "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -315,6 +316,19 @@ func TestMergeContentEntry(t *testing.T) { t.Fatalf("truncate then write kept %v, size %d", got, merged.Attributes.FileSize) } + // A peer truncate stands when the handle only overwrote inside it. + peerTruncated := proto.Clone(base).(*filer_pb.Entry) + peerTruncated.Chunks, peerTruncated.Attributes.FileSize = peerTruncated.Chunks[:1], 1 + overwrote := proto.Clone(base).(*filer_pb.Entry) + overwrote.Chunks = append(overwrote.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 6, Cookie: 1}, Size: 1, ModifiedTsNs: 6}) + merged, ok = mergeContentEntry(peerTruncated, base, overwrote, overwrote, -1) + if !ok { + t.Fatal("overwrite over a peer truncate not merged") + } + if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 6}) || merged.Attributes.FileSize != 1 { + t.Fatalf("overwrite over a peer truncate kept %v, size %d", got, merged.Attributes.FileSize) + } + stored.Attributes.Inode = 99 if _, ok := mergeContentEntry(stored, base, local, local, -1); ok { t.Fatal("merged into a replaced file") @@ -382,3 +396,25 @@ func TestContentFlushRepublishesUnchangedEntry(t *testing.T) { }) } } + +// Rebasing keeps a size the handle reached after the flush was taken, not the +// size it sent: a peer's truncate merged in must stand. +func TestRebaseEntryKeepsMergedSize(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, 2), 0, 0) + sent := proto.Clone(fh.GetEntry().GetEntry()).(*filer_pb.Entry) + merged := flushConditionEntry(1) + merged.Attributes.FileSize = 1 + + fh.rebaseEntry(merged, sent) + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 1 { + t.Fatalf("size %d after rebasing onto a truncate, want 1", size) + } + + fh.GetEntry().GetEntry().Attributes.FileSize = 5 + fh.rebaseEntry(merged, sent) + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 5 { + t.Fatalf("size %d after a write grew the handle, want 5", size) + } +} diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index fbf80cf410..70d312eb44 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -179,10 +179,11 @@ func (fh *FileHandle) installAckedEntry(entry *filer_pb.Entry, versionTsNs int64 } // rebaseEntry installs an entry merged with a peer's changes, keeping the -// chunks uploaders attached after sent was taken, and refreshes the read view. -func (fh *FileHandle) rebaseEntry(merged *filer_pb.Entry, sent []*filer_pb.FileChunk) { - inSent := make(map[string]bool, len(sent)) - for _, c := range sent { +// chunks and growth writers added after sent was taken, and refreshes the +// read view. +func (fh *FileHandle) rebaseEntry(merged, sent *filer_pb.Entry) { + inSent := make(map[string]bool, len(sent.Chunks)) + for _, c := range sent.Chunks { inSent[contentChunkKey(c)] = true } merged = proto.Clone(merged).(*filer_pb.Entry) @@ -194,7 +195,7 @@ func (fh *FileHandle) rebaseEntry(merged *filer_pb.Entry, sent []*filer_pb.FileC } } attrs := merged.Attributes - if size := e.GetAttributes().GetFileSize(); size > attrs.FileSize { + if size := e.GetAttributes().GetFileSize(); size > sent.GetAttributes().GetFileSize() && size > attrs.FileSize { attrs.FileSize = size } e.Attributes, e.Extended, e.Content, e.Chunks = attrs, merged.Extended, merged.Content, chunks @@ -211,7 +212,7 @@ func (fh *FileHandle) mergePeerContent(peer, base *filer_pb.Entry) bool { }) merged, ok := mergeContentEntry(peer, base, local, local, fh.truncatedSince(base)) if ok { - fh.rebaseEntry(merged, local.Chunks) + fh.rebaseEntry(merged, local) } return ok } diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 1b53171fd5..df106fba45 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -287,7 +287,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n return fmt.Errorf("fh flush create %s: %w", fileFullPath, err) } if merged != nil { - fh.rebaseEntry(merged, requestEntry.Chunks) + fh.rebaseEntry(merged, requestEntry) baseSnapshot = merged } @@ -373,7 +373,8 @@ const maxContentFlushMerges = 3 // content it changed, taking those values from sent (local in stored's id // space). Everything else keeps the stored value. A local truncation keeps // the handle's size and clips stored chunks at the lowest size truncated to -// (truncatedTo, or -1); otherwise the larger size wins. +// (truncatedTo, or -1); otherwise the stored size stands unless the handle +// changed its size or wrote past it. // A different file at the path, inline content changed on both sides, // inline content next to chunks (reads stop at the inline bytes), or a local // whole-file rewrite (maybeMergeChunks, unstamped chunks) over a peer's chunk @@ -420,12 +421,17 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in } merged.Chunks = append(merged.Chunks, c) } + size := int64(filer.FileSize(stored)) + if localSize != filer.FileSize(base) { + size = max(size, int64(localSize)) + } for _, c := range local.Chunks { if !inBase[contentChunkKey(c)] { if peerChanged && c.ModifiedTsNs == 0 && !c.IsChunkManifest { return nil, false } merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk)) + size = max(size, c.Offset+int64(c.Size)) } } if len(merged.Content) > 0 && len(merged.Chunks) > 0 { @@ -472,10 +478,9 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in } if truncatedTo >= 0 { - merged.Attributes.FileSize = localSize - } else { - merged.Attributes.FileSize = uint64(max(int64(localSize), int64(filer.FileSize(stored)))) + size = int64(localSize) } + merged.Attributes.FileSize = uint64(size) return merged, true } diff --git a/weed/mount/weedfs_metadata_flush.go b/weed/mount/weedfs_metadata_flush.go index 636a47c8d2..1b5fe38cd9 100644 --- a/weed/mount/weedfs_metadata_flush.go +++ b/weed/mount/weedfs_metadata_flush.go @@ -184,7 +184,7 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error { return err } if merged != nil { - fh.rebaseEntry(merged, requestEntry.Chunks) + fh.rebaseEntry(merged, requestEntry) baseSnapshot = merged }