From f7be2a75f5b9046cca5e18412e75e3944903640a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 17:13:32 -0800 Subject: [PATCH] mount: clip peer chunks at the lowest truncate, not the final size A handle that truncated and then wrote past the old end ends up larger than its base, so the merge did not see the truncate and kept the peer's chunks inside the range the truncate cleared. SetAttr now records the lowest size truncated to over the current base. Co-Authored-By: Claude Opus 5.5 --- weed/mount/content_flush_condition_test.go | 26 ++++++++++++----- weed/mount/filehandle.go | 28 +++++++++++++++++- weed/mount/weedfs_attr.go | 3 ++ weed/mount/weedfs_file_sync.go | 34 ++++++++++------------ 4 files changed, 65 insertions(+), 26 deletions(-) diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index 565fce30df..b4ddf00612 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -278,7 +278,7 @@ func TestMergeContentEntry(t *testing.T) { delete(local.Extended, "user.b") local.Chunks = append(local.Chunks, &filer_pb.FileChunk{FileId: "1,0300000001", Offset: 2, Size: 1, ModifiedTsNs: 3}) - merged, ok := mergeContentEntry(stored, base, local, local) + merged, ok := mergeContentEntry(stored, base, local, local, -1) if !ok { t.Fatal("not merged") } @@ -295,7 +295,7 @@ func TestMergeContentEntry(t *testing.T) { truncated := proto.Clone(base).(*filer_pb.Entry) truncated.Chunks = truncated.Chunks[:1] truncated.Attributes.FileSize = 1 - merged, ok = mergeContentEntry(stored, base, truncated, truncated) + merged, ok = mergeContentEntry(stored, base, truncated, truncated, -1) if !ok { t.Fatal("truncation not merged") } @@ -303,17 +303,29 @@ func TestMergeContentEntry(t *testing.T) { t.Fatalf("truncated merge kept %v, size %d", got, merged.Attributes.FileSize) } + // Truncated to 1, then written at 7: the peer chunk at 2 stays cut. + regrown := proto.Clone(truncated).(*filer_pb.Entry) + regrown.Chunks = append(regrown.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 7, Cookie: 1}, Offset: 7, Size: 1, ModifiedTsNs: 7}) + regrown.Attributes.FileSize = 8 + merged, ok = mergeContentEntry(stored, base, regrown, regrown, 1) + if !ok { + t.Fatal("truncate then write not merged") + } + if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1, 7}) || merged.Attributes.FileSize != 8 { + t.Fatalf("truncate then write kept %v, size %d", got, merged.Attributes.FileSize) + } + stored.Attributes.Inode = 99 - if _, ok := mergeContentEntry(stored, base, local, local); ok { + if _, ok := mergeContentEntry(stored, base, local, local, -1); ok { t.Fatal("merged into a replaced file") } rewritten := proto.Clone(base).(*filer_pb.Entry) rewritten.Chunks = []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 8, Cookie: 1}, Size: 2}} - if _, ok := mergeContentEntry(stored, base, rewritten, rewritten); ok { + if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1); ok { t.Fatal("merged a stale whole-file rewrite over a peer append") } - if _, ok := mergeContentEntry(proto.Clone(base).(*filer_pb.Entry), base, rewritten, rewritten); !ok { + if _, ok := mergeContentEntry(proto.Clone(base).(*filer_pb.Entry), base, rewritten, rewritten, -1); !ok { t.Fatal("refused a rewrite with no peer chunk change") } @@ -322,10 +334,10 @@ func TestMergeContentEntry(t *testing.T) { inline.Content = []byte("peer") appended := proto.Clone(empty).(*filer_pb.Entry) appended.Chunks = flushConditionEntry(5).Chunks - if _, ok := mergeContentEntry(inline, empty, appended, appended); ok { + if _, ok := mergeContentEntry(inline, empty, appended, appended, -1); ok { t.Fatal("merged chunks next to inline content") } - if _, ok := mergeContentEntry(appended, empty, inline, inline); ok { + if _, ok := mergeContentEntry(appended, empty, inline, inline, -1); ok { t.Fatal("merged inline content next to chunks") } } diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index fc4112a3b9..fbf80cf410 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -66,6 +66,10 @@ type FileHandle struct { // must be judged here, not against the live entry. Always store a clone. baseEntry atomic.Pointer[filer_pb.Entry] + // truncatedTo is the lowest size a truncate set over baseEntry. Later + // writes can grow the file past it; a merge still clips peer chunks there. + truncatedTo atomic.Pointer[truncateMark] + // dlmLock holds the distributed lock for cross-mount write coordination. // Non-nil only when -dlm is enabled and the file was opened for writing. // Acquired in AcquireHandle, released in ReleaseHandle. @@ -205,13 +209,35 @@ func (fh *FileHandle) mergePeerContent(peer, base *filer_pb.Entry) bool { fh.UpdateEntry(func(e *filer_pb.Entry) { local = proto.Clone(e).(*filer_pb.Entry) }) - merged, ok := mergeContentEntry(peer, base, local, local) + merged, ok := mergeContentEntry(peer, base, local, local, fh.truncatedSince(base)) if ok { fh.rebaseEntry(merged, local.Chunks) } return ok } +type truncateMark struct { + base *filer_pb.Entry + size int64 +} + +// noteTruncate records a truncate to size over the current base. +func (fh *FileHandle) noteTruncate(size uint64) { + base := fh.baseEntry.Load() + if m := fh.truncatedTo.Load(); m != nil && m.base == base && m.size <= int64(size) { + return + } + fh.truncatedTo.Store(&truncateMark{base: base, size: int64(size)}) +} + +// truncatedSince is the lowest size truncated to over base, or -1. +func (fh *FileHandle) truncatedSince(base *filer_pb.Entry) int64 { + if m := fh.truncatedTo.Load(); m != nil && m.base == base { + return m.size + } + return -1 +} + // 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_attr.go b/weed/mount/weedfs_attr.go index b394912d9e..c3f610bf41 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -134,6 +134,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse wfs.invalidateOpenMtimeCache(input.NodeId) oldFileSize := filer.FileSize(entry) if size < oldFileSize { + if fh != nil { + fh.noteTruncate(size) + } // fmt.Printf("truncate %v \n", fullPath) var chunks []*filer_pb.FileChunk var truncatedChunks []*filer_pb.FileChunk diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 9b468d45a9..1b53171fd5 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -355,7 +355,7 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request if err == nil || stored == nil { return resp, merged, err } - next, ok := mergeContentEntry(stored, base, local, sent) + next, ok := mergeContentEntry(stored, base, local, sent, fh.truncatedSince(base)) if !ok { return resp, nil, err } @@ -372,12 +372,13 @@ const maxContentFlushMerges = 3 // chunks it added or dropped, and the attributes, extended keys and inline // content it changed, taking those values from sent (local in stored's id // space). Everything else keeps the stored value. A local truncation keeps -// its size and clips stored chunks to it; otherwise the larger size wins. +// the handle's size and clips stored chunks at the lowest size truncated to +// (truncatedTo, or -1); otherwise the larger size wins. // 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 // change cannot be merged: the rewrite carries the handle's stale view. -func mergeContentEntry(stored, base, local, sent *filer_pb.Entry) (*filer_pb.Entry, bool) { +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 } @@ -401,14 +402,23 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry) (*filer_pb.Ent for _, c := range local.Chunks { inLocal[contentChunkKey(c)] = true } + localSize := filer.FileSize(local) + if truncatedTo < 0 && localSize < filer.FileSize(base) { + truncatedTo = int64(localSize) + } peerChanged := len(stored.Chunks) != len(base.Chunks) merged.Chunks = merged.Chunks[:0] for _, c := range stored.Chunks { key := contentChunkKey(c) peerChanged = peerChanged || !inBase[key] - if !inBase[key] || inLocal[key] { - merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk)) + if inBase[key] && !inLocal[key] || truncatedTo >= 0 && c.Offset >= truncatedTo { + continue } + c = proto.Clone(c).(*filer_pb.FileChunk) + if truncatedTo >= 0 && c.Offset+int64(c.Size) > truncatedTo { + c.Size = uint64(truncatedTo - c.Offset) + } + merged.Chunks = append(merged.Chunks, c) } for _, c := range local.Chunks { if !inBase[contentChunkKey(c)] { @@ -461,19 +471,7 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry) (*filer_pb.Ent } } - localSize := filer.FileSize(local) - if localSize < filer.FileSize(base) { - clipped := merged.Chunks[:0] - for _, c := range merged.Chunks { - if c.Offset >= int64(localSize) { - continue - } - if c.Offset+int64(c.Size) > int64(localSize) { - c.Size = localSize - uint64(c.Offset) - } - clipped = append(clipped, c) - } - merged.Chunks = clipped + if truncatedTo >= 0 { merged.Attributes.FileSize = localSize } else { merged.Attributes.FileSize = uint64(max(int64(localSize), int64(filer.FileSize(stored))))