From 88323d9db032dd57b90fcc4a53f9452987a3a23a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 18:01:38 -0800 Subject: [PATCH] mount: track a whole-file chunk rewrite on the handle The merge told a rewrite by its unstamped chunks, which a manifest fold hides: manifests are unstamped either way. Record the rewrite against the base it was made from, carried across peer base changes like a pending truncate, and refuse the merge from that. Co-Authored-By: Claude Opus 5.5 --- weed/mount/content_flush_condition_test.go | 38 ++++++----- weed/mount/filehandle.go | 68 +++++++++++++------ weed/mount/weedfs_file_sync.go | 18 ++--- .../weedfs_invalidate_open_handle_test.go | 7 +- 4 files changed, 82 insertions(+), 49 deletions(-) diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index d19c60c78b..ba9d588651 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -279,7 +279,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, -1) + merged, ok := mergeContentEntry(stored, base, local, local, -1, false) if !ok { t.Fatal("not merged") } @@ -296,7 +296,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, -1) + merged, ok = mergeContentEntry(stored, base, truncated, truncated, -1, false) if !ok { t.Fatal("truncation not merged") } @@ -308,7 +308,7 @@ func TestMergeContentEntry(t *testing.T) { 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) + merged, ok = mergeContentEntry(stored, base, regrown, regrown, 1, false) if !ok { t.Fatal("truncate then write not merged") } @@ -321,7 +321,7 @@ func TestMergeContentEntry(t *testing.T) { 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) + merged, ok = mergeContentEntry(peerTruncated, base, overwrote, overwrote, -1, false) if !ok { t.Fatal("overwrite over a peer truncate not merged") } @@ -329,26 +329,30 @@ func TestMergeContentEntry(t *testing.T) { 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") - } - + // A rewrite folded into a manifest is still a rewrite. 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, -1); ok { + rewritten.Chunks = []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 8, Cookie: 1}, Size: 2, IsChunkManifest: true}} + if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, true); 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, -1); !ok { + if _, ok := mergeContentEntry(stored, base, rewritten, rewritten, -1, false); !ok { + t.Fatal("refused new chunks the handle did not rewrite") + } + if _, ok := mergeContentEntry(proto.Clone(base).(*filer_pb.Entry), base, rewritten, rewritten, -1, true); !ok { t.Fatal("refused a rewrite with no peer chunk change") } + + stored.Attributes.Inode = 99 + if _, ok := mergeContentEntry(stored, base, local, local, -1, false); ok { + t.Fatal("merged into a replaced file") + } unstamped := flushConditionEntry(4) unstamped.Chunks[0].Size, unstamped.Chunks[0].ModifiedTsNs = 3, 0 shortened := proto.Clone(unstamped).(*filer_pb.Entry) shortened.Chunks[0].Size, shortened.Attributes.FileSize = 2, 2 peerAppend := proto.Clone(unstamped).(*filer_pb.Entry) peerAppend.Chunks = append(peerAppend.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 9, Cookie: 1}, Offset: 3, Size: 1, ModifiedTsNs: 9}) - if _, ok := mergeContentEntry(peerAppend, unstamped, shortened, shortened, 2); !ok { + if _, ok := mergeContentEntry(peerAppend, unstamped, shortened, shortened, 2, false); !ok { t.Fatal("refused a truncate that shortened an unstamped base chunk") } @@ -356,11 +360,11 @@ func TestMergeContentEntry(t *testing.T) { noInode.Attributes.Inode, noInode.Attributes.Crtime = 0, 5 sameFile := proto.Clone(noInode).(*filer_pb.Entry) sameFile.Chunks = append(sameFile.Chunks, stored.Chunks[2]) - if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1); !ok { + if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); !ok { t.Fatal("refused an inode-less file with the same creation time") } sameFile.Attributes.Crtime = 6 - if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1); ok { + if _, ok := mergeContentEntry(sameFile, noInode, local, local, -1, false); ok { t.Fatal("merged into an inode-less replacement") } @@ -369,10 +373,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, -1); ok { + if _, ok := mergeContentEntry(inline, empty, appended, appended, -1, false); ok { t.Fatal("merged chunks next to inline content") } - if _, ok := mergeContentEntry(appended, empty, inline, inline, -1); ok { + if _, ok := mergeContentEntry(appended, empty, inline, inline, -1, false); ok { t.Fatal("merged inline content next to chunks") } } diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 2672a49151..e606a0a842 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -67,9 +67,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] + // sinceBase marks a truncate or chunk rewrite over baseEntry that a + // merge must honor: later writes can grow the file past a truncate, and + // rewritten chunks carry the handle's view of the base. + sinceBase atomic.Pointer[baseMark] // chunkViewStale marks a read view whose manifest resolution failed; the // next read or lseek resolves the entry's chunks again. @@ -230,7 +231,8 @@ 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, fh.truncatedSince(base)) + mark := fh.markSince(base) + merged, ok := mergeContentEntry(peer, base, local, local, mark.truncatedTo, mark.rewrote) ok = ok && mergeKeepsBuffered(merged, local) if ok { fh.rebaseEntry(merged, local) @@ -245,35 +247,61 @@ func mergeKeepsBuffered(merged, local *filer_pb.Entry) bool { return merged.Attributes.FileSize >= filer.FileSize(local) && bytes.Equal(merged.Content, local.Content) } -type truncateMark struct { - base *filer_pb.Entry - size int64 +// baseMark records what the handle did over base that its entry does not +// show: the lowest size a truncate set (-1 for none), and a whole-file +// rewrite of its chunks. +type baseMark struct { + base *filer_pb.Entry + truncatedTo int64 + rewrote bool +} + +func (fh *FileHandle) markBase(update func(m *baseMark)) { + for { + base, old := fh.baseEntry.Load(), fh.sinceBase.Load() + next := baseMark{base: base, truncatedTo: -1} + if old != nil && old.base == base { + next = *old + } + update(&next) + if fh.sinceBase.CompareAndSwap(old, &next) { + return + } + } } // 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)}) + fh.markBase(func(m *baseMark) { + if m.truncatedTo < 0 || int64(size) < m.truncatedTo { + m.truncatedTo = int64(size) + } + }) } -// adoptPeerBase moves a dirty handle's base to a peer's entry. A truncate the +// noteRewrite records that the handle's chunks were rewritten from its view +// of the current base. +func (fh *FileHandle) noteRewrite() { + fh.markBase(func(m *baseMark) { m.rewrote = true }) +} + +// adoptPeerBase moves a dirty handle's base to a peer's entry. What the // handle has not published yet still applies over the new base. func (fh *FileHandle) adoptPeerBase(base *filer_pb.Entry) { old := fh.baseEntry.Swap(base) - if m := fh.truncatedTo.Load(); m != nil && m.base == old { - fh.truncatedTo.Store(&truncateMark{base: base, size: m.size}) + if m := fh.sinceBase.Load(); m != nil && m.base == old { + moved := *m + moved.base = base + fh.sinceBase.CompareAndSwap(m, &moved) } } -// 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 +// markSince is what the handle did over base. +func (fh *FileHandle) markSince(base *filer_pb.Entry) baseMark { + if m := fh.sinceBase.Load(); m != nil && m.base == base { + return *m } - return -1 + return baseMark{base: base, truncatedTo: -1} } // setAuthoritativeBase installs the base snapshot a local ack acknowledged. diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index b77f877f52..533d4f1034 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -254,6 +254,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n } else if mergedChunks != nil { chunks = mergedChunks manifestChunks = nil + fh.noteRewrite() } // no chunk deleter here: a failed fold reports the blobs it saved @@ -356,7 +357,8 @@ 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, fh.truncatedSince(base)) + mark := fh.markSince(base) + next, ok := mergeContentEntry(stored, base, local, sent, mark.truncatedTo, mark.rewrote) if !ok || buffered && !mergeKeepsBuffered(next, local) { return resp, nil, err } @@ -378,9 +380,9 @@ const maxContentFlushMerges = 3 // 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 -// change cannot be merged: the rewrite carries the handle's stale view. -func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo int64) (*filer_pb.Entry, bool) { +// whole-file rewrite (rewrote) 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, truncatedTo int64, rewrote bool) (*filer_pb.Entry, bool) { if !sameContentFile(stored, base) { return nil, false } @@ -394,10 +396,8 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in } inBase := make(map[string]bool, len(base.Chunks)) - baseFileIds := make(map[string]bool, len(base.Chunks)) for _, c := range base.Chunks { inBase[contentChunkKey(c)] = true - baseFileIds[contentFileIdKey(c)] = true } inLocal := make(map[string]bool, len(local.Chunks)) for _, c := range local.Chunks { @@ -421,15 +421,15 @@ func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo in } merged.Chunks = append(merged.Chunks, c) } + if rewrote && peerChanged { + return nil, false + } 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 && !baseFileIds[contentFileIdKey(c)] { - return nil, false - } merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk)) size = max(size, c.Offset+int64(c.Size)) } diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 78b9bb1569..076ea0e25f 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -2161,7 +2161,7 @@ func TestUnmergeablePeerChangeKeepsDirtyHandle(t *testing.T) { } // A peer event that moves a dirty handle's base keeps its unpublished -// truncate, so a later merge still clips peer chunks at it. +// truncate and chunk rewrite, so a later merge still honors them. func TestPeerEventKeepsPendingTruncate(t *testing.T) { wfs := newInvalidateTestWFS(t) @@ -2169,6 +2169,7 @@ func TestPeerEventKeepsPendingTruncate(t *testing.T) { fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, flushConditionEntry(1, 2), 0, 0) fh.dirtyMetadata = true fh.noteTruncate(1) + fh.noteRewrite() chmod := flushConditionEntry(1, 2) chmod.Attributes.FileMode = 0600 @@ -2190,8 +2191,8 @@ func TestPeerEventKeepsPendingTruncate(t *testing.T) { if base.GetAttributes().GetFileMode() != 0600 { t.Fatalf("base %v did not move to the peer's entry", base) } - if got := fh.truncatedSince(base); got != 1 { - t.Fatalf("pending truncate %d after the base moved, want 1", got) + if got := fh.markSince(base); got.truncatedTo != 1 || !got.rewrote { + t.Fatalf("pending truncate %d, rewrite %v after the base moved", got.truncatedTo, got.rewrote) } }