mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-13 23:39:33 +01:00
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
3f915905ec
commit
88323d9db0
4 files changed
+82
-49
No files matched your search
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
+48
-20
@@ -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.
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user