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