mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 00:09:00 +01:00
Chris LuandClaude Opus 5.5
f7be2a75f5
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 <noreply@anthropic.com>
385 lines
14 KiB
Go
385 lines
14 KiB
Go
//go:build !freebsd
|
|
|
|
package mount
|
|
|
|
import (
|
|
"context"
|
|
"slices"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/go-fuse/v2/fuse"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type contentFlushConditionFiler struct {
|
|
filer_pb.UnimplementedSeaweedFilerServer
|
|
mu sync.Mutex
|
|
entry *filer_pb.Entry
|
|
calls int
|
|
}
|
|
|
|
func (s *contentFlushConditionFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.calls++
|
|
for _, c := range r.GetCondition().GetClauses() {
|
|
matches := false
|
|
switch c.Kind {
|
|
case filer_pb.WriteCondition_IF_ENTRY_EQUAL:
|
|
matches = proto.Equal(c.ExpectedEntry, s.entry)
|
|
case filer_pb.WriteCondition_IF_NOT_EXISTS:
|
|
matches = s.entry == nil
|
|
}
|
|
if !matches {
|
|
return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil
|
|
}
|
|
}
|
|
// Like the filer, stamp the stored version and announce what was stored.
|
|
old := s.entry
|
|
s.entry = proto.Clone(r.Entry).(*filer_pb.Entry)
|
|
s.entry.CommitTsNs = 1000 + int64(s.calls)
|
|
return &filer_pb.CreateEntryResponse{LogTsNs: 2000, MetadataEvent: &filer_pb.SubscribeMetadataResponse{
|
|
Directory: r.Directory,
|
|
TsNs: s.entry.CommitTsNs,
|
|
EventNotification: &filer_pb.EventNotification{
|
|
OldEntry: old,
|
|
NewEntry: proto.Clone(s.entry).(*filer_pb.Entry),
|
|
},
|
|
}}, nil
|
|
}
|
|
func (s *contentFlushConditionFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.entry == nil {
|
|
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
|
|
}
|
|
return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(s.entry).(*filer_pb.Entry)}, nil
|
|
}
|
|
|
|
func flushConditionEntry(ids ...uint64) *filer_pb.Entry {
|
|
e := &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{Inode: 42, FileMode: 0644, Uid: 2000, Gid: 20, FileSize: 3}}
|
|
for i, id := range ids {
|
|
e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: id, Cookie: 1}, Offset: int64(i), Size: 1, ModifiedTsNs: int64(id)})
|
|
}
|
|
return e
|
|
}
|
|
|
|
// Exercise the actual mount content-publication caller. A peer's append is
|
|
// merged with ours; a replaced or deleted file is not ours to merge into, so
|
|
// the refused flush keeps our dirty entry instead of overwriting the peer.
|
|
func TestContentFlushPreservesPeerChunks(t *testing.T) {
|
|
for _, name := range []string{"single-writer", "peer-append", "replaced", "deleted", "deferred-create", "deferred-create-race"} {
|
|
t.Run(name, func(t *testing.T) {
|
|
wfs := newInvalidateTestWFS(t)
|
|
observed := flushConditionEntry(1)
|
|
live := flushConditionEntry(1, 3)
|
|
live.Chunks[1].Offset = 2
|
|
current := flushConditionEntry(1)
|
|
wantError, wantCalls := false, 1
|
|
switch name {
|
|
case "peer-append":
|
|
current = flushConditionEntry(1, 2)
|
|
wantCalls = 2
|
|
case "replaced":
|
|
current = flushConditionEntry(4)
|
|
current.Attributes.Inode = 99
|
|
wantError = true
|
|
case "deleted":
|
|
current = nil
|
|
wantError = true
|
|
case "deferred-create":
|
|
observed = nil
|
|
current = nil
|
|
case "deferred-create-race":
|
|
observed = nil
|
|
current = flushConditionEntry(4)
|
|
current.Attributes.Inode = 99
|
|
wantError = true
|
|
}
|
|
if observed != nil {
|
|
wfs.mapPbIdFromFilerToLocal(observed)
|
|
}
|
|
wfs.mapPbIdFromFilerToLocal(live)
|
|
fake := &contentFlushConditionFiler{entry: current}
|
|
startFakeFiler(t, wfs, fake)
|
|
fh := newFileHandle(wfs, 1, 42, live)
|
|
t.Cleanup(func() { fh.dirtyPages.Destroy() })
|
|
fh.baseEntry.Store(observed)
|
|
fh.dirtyMetadata = true
|
|
fh.RememberPath(util.FullPath("/dir/file"))
|
|
before := proto.Clone(current)
|
|
code := wfs.doFlush(context.Background(), fh, 1000, 20, false)
|
|
fake.mu.Lock()
|
|
defer fake.mu.Unlock()
|
|
if wantError {
|
|
if code == fuse.OK {
|
|
t.Fatal("stale content flush acknowledged after a peer changed the entry")
|
|
}
|
|
if !proto.Equal(fake.entry, before) {
|
|
t.Fatalf("peer entry changed: %v", fake.entry)
|
|
}
|
|
if !fh.dirtyMetadata {
|
|
t.Fatal("refused flush lost dirty metadata")
|
|
}
|
|
if len(fh.entry.Entry.Chunks) != 2 || fh.entry.Entry.Chunks[1].GetFileIdString() != live.Chunks[1].GetFileIdString() {
|
|
t.Fatal("refused flush lost pending content")
|
|
}
|
|
} else {
|
|
if code != fuse.OK {
|
|
t.Fatal(code)
|
|
}
|
|
if fh.dirtyMetadata {
|
|
t.Fatal("successful flush remains dirty")
|
|
}
|
|
want := []uint64{1, 3}
|
|
if name == "peer-append" {
|
|
want = []uint64{1, 2, 3}
|
|
}
|
|
if got := chunkFileKeys(fake.entry); !slices.Equal(got, want) {
|
|
t.Fatalf("stored chunks %v, want %v", got, want)
|
|
}
|
|
if got := chunkFileKeys(fh.GetEntry().GetEntry()); !slices.Equal(got, want) {
|
|
t.Fatalf("handle chunks %v, want %v", got, want)
|
|
}
|
|
}
|
|
if fake.calls != wantCalls {
|
|
t.Fatalf("unexpected replay: %d calls", fake.calls)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// Each acknowledged flush must leave the handle a base the next flush on the
|
|
// same handle can match: the stored entry, with the filer's own commit stamp.
|
|
func TestContentFlushRepeatsOnOneHandle(t *testing.T) {
|
|
wfs := newInvalidateTestWFS(t)
|
|
stored := flushConditionEntry(1)
|
|
stored.CommitTsNs = 7
|
|
fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)}
|
|
startFakeFiler(t, wfs, fake)
|
|
observed := proto.Clone(stored).(*filer_pb.Entry)
|
|
wfs.mapPbIdFromFilerToLocal(observed)
|
|
fh := newFileHandle(wfs, 1, 42, proto.Clone(observed).(*filer_pb.Entry))
|
|
t.Cleanup(func() { fh.dirtyPages.Destroy() })
|
|
fh.RememberPath(util.FullPath("/dir/file"))
|
|
for i := uint64(2); i <= 3; i++ {
|
|
fh.UpdateEntry(func(e *filer_pb.Entry) {
|
|
e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: i, Cookie: 1}, Offset: int64(i - 1), Size: 1, ModifiedTsNs: int64(i)})
|
|
e.Attributes.FileSize = i
|
|
})
|
|
fh.dirtyMetadata = true
|
|
if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
|
|
t.Fatalf("flush %d: %v", i-1, code)
|
|
}
|
|
}
|
|
fake.mu.Lock()
|
|
defer fake.mu.Unlock()
|
|
if len(fake.entry.Chunks) != 3 || fake.calls != 2 {
|
|
t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls)
|
|
}
|
|
}
|
|
|
|
func chunkFileKeys(e *filer_pb.Entry) (keys []uint64) {
|
|
for _, c := range e.GetChunks() {
|
|
keys = append(keys, c.GetFid().GetFileKey())
|
|
}
|
|
return keys
|
|
}
|
|
|
|
// The periodic metadata flush publishes the same whole entry, so it must not
|
|
// overwrite a peer's append either.
|
|
func TestPeriodicMetadataFlushPreservesPeerChunks(t *testing.T) {
|
|
wfs := newInvalidateTestWFS(t)
|
|
current := flushConditionEntry(1, 2)
|
|
fake := &contentFlushConditionFiler{entry: proto.Clone(current).(*filer_pb.Entry)}
|
|
startFakeFiler(t, wfs, fake)
|
|
observed := flushConditionEntry(1)
|
|
wfs.mapPbIdFromFilerToLocal(observed)
|
|
live := flushConditionEntry(1, 3)
|
|
live.Chunks[1].Offset = 2
|
|
wfs.mapPbIdFromFilerToLocal(live)
|
|
fh := newFileHandle(wfs, 1, 42, live)
|
|
t.Cleanup(func() { fh.dirtyPages.Destroy() })
|
|
fh.baseEntry.Store(observed)
|
|
fh.dirtyMetadata = true
|
|
fh.RememberPath(util.FullPath("/dir/file"))
|
|
if err := wfs.flushFileMetadata(fh); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
fake.mu.Lock()
|
|
defer fake.mu.Unlock()
|
|
if got := chunkFileKeys(fake.entry); !slices.Equal(got, []uint64{1, 2, 3}) || fake.calls != 2 {
|
|
t.Fatalf("stored chunks %v after %d calls", got, fake.calls)
|
|
}
|
|
if got := chunkFileKeys(fh.GetEntry().GetEntry()); !slices.Equal(got, []uint64{1, 2, 3}) {
|
|
t.Fatalf("handle chunks %v", got)
|
|
}
|
|
if !fh.dirtyMetadata {
|
|
t.Fatal("periodic flush cleared dirty metadata")
|
|
}
|
|
}
|
|
|
|
// Every merged flush is followed by a flush that matches what it stored, so
|
|
// the peer's chunk is not dropped by the next publication from this handle.
|
|
func TestContentFlushMergeThenRepeat(t *testing.T) {
|
|
wfs := newInvalidateTestWFS(t)
|
|
fake := &contentFlushConditionFiler{entry: flushConditionEntry(1, 2)}
|
|
startFakeFiler(t, wfs, fake)
|
|
observed := flushConditionEntry(1)
|
|
wfs.mapPbIdFromFilerToLocal(observed)
|
|
live := flushConditionEntry(1, 3)
|
|
live.Chunks[1].Offset = 2
|
|
wfs.mapPbIdFromFilerToLocal(live)
|
|
fh := newFileHandle(wfs, 1, 42, live)
|
|
t.Cleanup(func() { fh.dirtyPages.Destroy() })
|
|
fh.baseEntry.Store(observed)
|
|
fh.RememberPath(util.FullPath("/dir/file"))
|
|
for i := uint64(4); i <= 5; i++ {
|
|
fh.dirtyMetadata = true
|
|
if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
|
|
t.Fatalf("flush before chunk %d: %v", i, code)
|
|
}
|
|
fh.UpdateEntry(func(e *filer_pb.Entry) {
|
|
e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: i, Cookie: 1}, Offset: int64(i - 1), Size: 1, ModifiedTsNs: int64(i)})
|
|
e.Attributes.FileSize = i
|
|
})
|
|
}
|
|
fh.dirtyMetadata = true
|
|
if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
|
|
t.Fatal(code)
|
|
}
|
|
fake.mu.Lock()
|
|
defer fake.mu.Unlock()
|
|
if got := chunkFileKeys(fake.entry); !slices.Equal(got, []uint64{1, 2, 3, 4, 5}) || fake.calls != 4 {
|
|
t.Fatalf("stored chunks %v after %d calls", got, fake.calls)
|
|
}
|
|
}
|
|
|
|
// The merge keeps a peer's attribute and extended changes the handle did not
|
|
// make, applies the ones it did, and lets a local truncation clip the peer's
|
|
// chunks past the new end.
|
|
func TestMergeContentEntry(t *testing.T) {
|
|
base := flushConditionEntry(1, 2)
|
|
base.Extended = map[string][]byte{"user.a": []byte("1"), "user.b": []byte("1")}
|
|
stored := proto.Clone(base).(*filer_pb.Entry)
|
|
stored.Chunks = append(stored.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 9, Cookie: 1}, Offset: 2, Size: 4})
|
|
stored.Attributes.FileMode, stored.Attributes.FileSize = 0600, 6
|
|
stored.Extended["user.b"] = []byte("peer")
|
|
stored.CommitTsNs = 77
|
|
|
|
local := proto.Clone(base).(*filer_pb.Entry)
|
|
local.Attributes.Mtime = 123
|
|
local.Extended["user.a"] = []byte("mine")
|
|
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)
|
|
if !ok {
|
|
t.Fatal("not merged")
|
|
}
|
|
if got := len(merged.Chunks); got != 4 || merged.Attributes.FileSize != 6 {
|
|
t.Fatalf("merged %d chunks, size %d", got, merged.Attributes.FileSize)
|
|
}
|
|
if merged.Attributes.FileMode != 0600 || merged.Attributes.Mtime != 123 || merged.CommitTsNs != 77 {
|
|
t.Fatalf("attributes %v", merged.Attributes)
|
|
}
|
|
if string(merged.Extended["user.a"]) != "mine" || merged.Extended["user.b"] != nil {
|
|
t.Fatalf("extended %v", merged.Extended)
|
|
}
|
|
|
|
truncated := proto.Clone(base).(*filer_pb.Entry)
|
|
truncated.Chunks = truncated.Chunks[:1]
|
|
truncated.Attributes.FileSize = 1
|
|
merged, ok = mergeContentEntry(stored, base, truncated, truncated, -1)
|
|
if !ok {
|
|
t.Fatal("truncation not merged")
|
|
}
|
|
if got := chunkFileKeys(merged); !slices.Equal(got, []uint64{1}) || merged.Attributes.FileSize != 1 {
|
|
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, -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, -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, -1); !ok {
|
|
t.Fatal("refused a rewrite with no peer chunk change")
|
|
}
|
|
|
|
empty := flushConditionEntry()
|
|
inline := proto.Clone(empty).(*filer_pb.Entry)
|
|
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 {
|
|
t.Fatal("merged chunks next to inline content")
|
|
}
|
|
if _, ok := mergeContentEntry(appended, empty, inline, inline, -1); ok {
|
|
t.Fatal("merged inline content next to chunks")
|
|
}
|
|
}
|
|
|
|
// A miss on state no peer changed is republished once against the stored
|
|
// entry: a rename while open, or a filer uid the local id mapping cannot
|
|
// reproduce (newInvalidateTestWFS maps local 1000 to filer 2000).
|
|
func TestContentFlushRepublishesUnchangedEntry(t *testing.T) {
|
|
for _, name := range []string{"renamed", "colliding-uid"} {
|
|
t.Run(name, func(t *testing.T) {
|
|
wfs := newInvalidateTestWFS(t)
|
|
stored := flushConditionEntry(1)
|
|
stored.CommitTsNs = 7
|
|
observed := proto.Clone(stored).(*filer_pb.Entry)
|
|
path := util.FullPath("/dir/file")
|
|
switch name {
|
|
case "renamed":
|
|
wfs.mapPbIdFromFilerToLocal(observed)
|
|
stored.Name, stored.CommitTsNs = "moved", 8
|
|
path = "/dir/moved"
|
|
case "colliding-uid":
|
|
stored.Attributes.Uid, observed.Attributes.Uid = 1000, 1000
|
|
}
|
|
fake := &contentFlushConditionFiler{entry: proto.Clone(stored).(*filer_pb.Entry)}
|
|
startFakeFiler(t, wfs, fake)
|
|
live := proto.Clone(observed).(*filer_pb.Entry)
|
|
live.Name = stored.Name
|
|
live.Chunks = append(live.Chunks, &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: 3, Cookie: 1}, Offset: 1, Size: 1, ModifiedTsNs: 3})
|
|
fh := newFileHandle(wfs, 1, 42, live)
|
|
t.Cleanup(func() { fh.dirtyPages.Destroy() })
|
|
fh.baseEntry.Store(observed)
|
|
fh.dirtyMetadata = true
|
|
fh.RememberPath(path)
|
|
if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK {
|
|
t.Fatal(code)
|
|
}
|
|
fake.mu.Lock()
|
|
defer fake.mu.Unlock()
|
|
if len(fake.entry.Chunks) != 2 || fake.calls != 2 {
|
|
t.Fatalf("stored %d chunks after %d calls", len(fake.entry.Chunks), fake.calls)
|
|
}
|
|
})
|
|
}
|
|
}
|