From 0c83825b23d2d3b349701a32e3f3be1ef233ec59 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 13 Nov 2026 01:09:31 -0800 Subject: [PATCH] mount: test an open handle across link and alias unlink Co-Authored-By: Claude Opus 5.5 --- weed/mount/weedfs_link_open_handle_test.go | 284 +++++++++++++++++++++ 1 file changed, 284 insertions(+) create mode 100644 weed/mount/weedfs_link_open_handle_test.go diff --git a/weed/mount/weedfs_link_open_handle_test.go b/weed/mount/weedfs_link_open_handle_test.go new file mode 100644 index 0000000000..96aa8c78be --- /dev/null +++ b/weed/mount/weedfs_link_open_handle_test.go @@ -0,0 +1,284 @@ +//go:build !freebsd + +package mount + +import ( + "context" + "sync" + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// hardLinkFiler stores the entries of "/" the way the filer stores hard +// links: a name record pointing at a shared blob that holds the content and +// the link count. Removing a link lowers the count and stamps the ctime. +type hardLinkFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + mu sync.Mutex + records map[string]*filer_pb.Entry + blobs map[string]*filer_pb.Entry + stamp int64 +} + +func (s *hardLinkFiler) resolve(name string) *filer_pb.Entry { + record := s.records[name] + if record == nil { + return nil + } + entry := record + if len(record.HardLinkId) > 0 { + entry = s.blobs[string(record.HardLinkId)] + } + entry = proto.Clone(entry).(*filer_pb.Entry) + entry.Name = name + return entry +} + +func (s *hardLinkFiler) store(entry *filer_pb.Entry) *filer_pb.Entry { + s.stamp++ + stored := filer.FromPbEntry("/", entry).ToProtoEntry() + stored.CommitTsNs = s.stamp + if old := s.records[entry.Name]; old != nil && len(old.HardLinkId) > 0 && string(old.HardLinkId) != string(entry.HardLinkId) { + s.unlinkBlob(old.HardLinkId) + } + if len(entry.HardLinkId) > 0 { + s.blobs[string(entry.HardLinkId)] = stored + s.records[entry.Name] = &filer_pb.Entry{Name: entry.Name, HardLinkId: entry.HardLinkId} + } else { + s.records[entry.Name] = stored + } + return s.resolve(entry.Name) +} + +func (s *hardLinkFiler) unlinkBlob(id []byte) { + blob := s.blobs[string(id)] + if blob.HardLinkCounter--; blob.HardLinkCounter <= 0 { + delete(s.blobs, string(id)) + return + } + blob.Attributes.Ctime = 777 +} + +func (s *hardLinkFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + entry := s.resolve(r.Name) + if entry == nil { + return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil +} + +func (s *hardLinkFiler) CreateEntry(ctx context.Context, r *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + current := s.resolve(r.Entry.Name) + for _, c := range r.GetCondition().GetClauses() { + matches := false + switch c.Kind { + case filer_pb.WriteCondition_IF_ENTRY_EQUAL: + expected := proto.Clone(c.ExpectedEntry).(*filer_pb.Entry) + filer_pb.BeforeEntrySerialization(expected.Chunks) + if current != nil { + filer_pb.BeforeEntrySerialization(current.Chunks) + } + matches = current != nil && proto.Equal(expected, current) + case filer_pb.WriteCondition_IF_NOT_EXISTS: + matches = current == nil + } + if !matches { + return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } + } + stored := s.store(r.Entry) + return &filer_pb.CreateEntryResponse{LogTsNs: stored.CommitTsNs, MetadataEvent: &filer_pb.SubscribeMetadataResponse{ + Directory: r.Directory, + TsNs: stored.CommitTsNs, + EventNotification: &filer_pb.EventNotification{OldEntry: current, NewEntry: stored, NewParentPath: r.Directory}, + }}, nil +} + +func (s *hardLinkFiler) UpdateEntry(ctx context.Context, r *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + current := s.resolve(r.Entry.Name) + stored := s.store(r.Entry) + return &filer_pb.UpdateEntryResponse{MetadataEvent: &filer_pb.SubscribeMetadataResponse{ + Directory: r.Directory, + TsNs: stored.CommitTsNs, + EventNotification: &filer_pb.EventNotification{OldEntry: current, NewEntry: stored, NewParentPath: r.Directory}, + }}, nil +} + +func (s *hardLinkFiler) DeleteEntry(ctx context.Context, r *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + record := s.records[r.Name] + if record == nil { + return &filer_pb.DeleteEntryResponse{}, nil + } + delete(s.records, r.Name) + if len(record.HardLinkId) > 0 { + s.unlinkBlob(record.HardLinkId) + } + return &filer_pb.DeleteEntryResponse{}, nil +} + +func (s *hardLinkFiler) stored(name string) *filer_pb.Entry { + s.mu.Lock() + defer s.mu.Unlock() + return s.resolve(name) +} + +func linkHandleChunk(key uint64, offset int64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 1, FileKey: key, Cookie: 1}, Offset: offset, Size: 100, ModifiedTsNs: int64(key)} +} + +// openLinkTestFile stores "/tmp" as the filer holds a flushed file and opens +// a clean handle on it, the way a create, write and fsync leave one. +func openLinkTestFile(t *testing.T) (*WFS, *hardLinkFiler, *FileHandle, uint64) { + t.Helper() + wfs := newInvalidateTestWFS(t) + wfs.option.FilerMountRootPath = "/" + fake := &hardLinkFiler{records: map[string]*filer_pb.Entry{}, blobs: map[string]*filer_pb.Entry{}} + startFakeFiler(t, wfs, fake) + stored := fake.store(&filer_pb.Entry{ + Name: "tmp", + Chunks: []*filer_pb.FileChunk{linkHandleChunk(1, 0)}, + Attributes: &filer_pb.FuseAttributes{Inode: 42, FileMode: 0o600, Uid: 2000, Gid: 20, FileSize: 100, Crtime: 1, Mtime: 1, Ctime: 1}, + }) + inode := wfs.inodeToPath.Lookup("/tmp", 1, false, false, 42, true) + wfs.mapPbIdFromFilerToLocal(stored) + fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, stored, stored.CommitTsNs, 0) + t.Cleanup(func() { fh.dirtyPages.Destroy() }) + fh.RememberPath("/tmp") + return wfs, fake, fh, inode +} + +func appendAndFlush(t *testing.T, wfs *WFS, fh *FileHandle, key uint64) fuse.Status { + t.Helper() + fh.UpdateEntry(func(e *filer_pb.Entry) { + e.Chunks = append(e.Chunks, linkHandleChunk(key, int64(len(e.Chunks))*100)) + e.Attributes.FileSize = uint64(len(e.Chunks)) * 100 + }) + fh.dirtyMetadata = true + return wfs.doFlush(context.Background(), fh, 1000, 20, false) +} + +func linkInRoot(t *testing.T, wfs *WFS, inode uint64, name string) { + t.Helper() + in := &fuse.LinkIn{InHeader: fuse.InHeader{NodeId: 1, Caller: fuse.Caller{Owner: fuse.Owner{Uid: 1000, Gid: 20}}}, Oldnodeid: inode} + if code := wfs.Link(make(chan struct{}), in, name, &fuse.EntryOut{}); code != fuse.OK { + t.Fatalf("link %s: %v", name, code) + } +} + +func unlinkInRoot(t *testing.T, wfs *WFS, name string) { + t.Helper() + header := &fuse.InHeader{NodeId: 1, Caller: fuse.Caller{Owner: fuse.Owner{Uid: 1000, Gid: 20}}} + if code := wfs.Unlink(make(chan struct{}), header, name); code != fuse.OK { + t.Fatalf("unlink %s: %v", name, code) + } +} + +func wantStored(t *testing.T, fake *hardLinkFiler, name string, chunks int, links int32) { + t.Helper() + entry := fake.stored(name) + if entry == nil { + t.Fatalf("%s is gone", name) + } + if len(entry.Chunks) != chunks || entry.HardLinkCounter != links { + t.Fatalf("%s holds %d chunks and %d links, want %d and %d", name, len(entry.Chunks), entry.HardLinkCounter, chunks, links) + } +} + +// A file still open from its creation gets a second name; writes through the +// open handle must reach both names. +func TestLinkKeepsOpenHandlePublishing(t *testing.T) { + wfs, fake, fh, inode := openLinkTestFile(t) + linkInRoot(t, wfs, inode, "final") + if code := appendAndFlush(t, wfs, fh, 2); code != fuse.OK { + t.Fatalf("flush after link: %v", code) + } + wantStored(t, fake, "tmp", 2, 2) + wantStored(t, fake, "final", 2, 2) +} + +// The temp-file publish pattern: write tmp, link it to its final name, unlink +// tmp, keep appending through the handle. Every fsync must reach the filer +// under the surviving name, and the removed name must stay removed. +func TestUnlinkedAliasKeepsOpenHandlePublishing(t *testing.T) { + for _, dirty := range []bool{false, true} { + t.Run(map[bool]string{false: "clean", true: "dirty"}[dirty], func(t *testing.T) { + wfs, fake, fh, inode := openLinkTestFile(t) + linkInRoot(t, wfs, inode, "final") + if dirty { + fh.UpdateEntry(func(e *filer_pb.Entry) { + e.Chunks = append(e.Chunks, linkHandleChunk(9, 100)) + e.Attributes.FileSize = 200 + }) + fh.dirtyMetadata = true + } + unlinkInRoot(t, wfs, "tmp") + if fh.isDeleted { + t.Fatal("the handle of a file that still has a name is marked deleted") + } + if code := appendAndFlush(t, wfs, fh, 2); code != fuse.OK { + t.Fatalf("flush after unlinking the alias: %v", code) + } + chunks := 2 + if dirty { + chunks = 3 + } + wantStored(t, fake, "final", chunks, 1) + if fake.stored("tmp") != nil { + t.Fatal("the flush recreated the unlinked name") + } + }) + } +} + +// Unlinking the last name still stops the handle from writing it back. +func TestUnlinkLastLinkKeepsOpenHandleDeleted(t *testing.T) { + wfs, fake, fh, inode := openLinkTestFile(t) + linkInRoot(t, wfs, inode, "final") + unlinkInRoot(t, wfs, "tmp") + unlinkInRoot(t, wfs, "final") + if !fh.isDeleted { + t.Fatal("the handle of a file with no names left is not marked deleted") + } + if code := appendAndFlush(t, wfs, fh, 2); code != fuse.OK { + t.Fatalf("flush after the last unlink: %v", code) + } + for _, name := range []string{"tmp", "final"} { + if fake.stored(name) != nil { + t.Fatalf("the flush recreated %s", name) + } + } +} + +// adoptLinkChange must not paper over a peer's write: a stored entry that +// differs from the base in more than the link fields leaves the base alone. +func TestAdoptLinkChangeKeepsBaseOnPeerWrite(t *testing.T) { + wfs, _, fh, _ := openLinkTestFile(t) + base := fh.baseEntry.Load() + stored := proto.Clone(base).(*filer_pb.Entry) + wfs.mapPbIdFromLocalToFiler(stored) + stored.HardLinkId, stored.HardLinkCounter = []byte("link"), 2 + stored.Chunks = append(stored.Chunks, linkHandleChunk(7, 100)) + fh.adoptLinkChange(stored, 99, 0) + if fh.baseEntry.Load() != base || len(fh.GetEntry().GetEntry().HardLinkId) != 0 { + t.Fatal("a peer's chunk change was adopted as the handle's base") + } + stored.Chunks = stored.Chunks[:1] + fh.adoptLinkChange(stored, 99, 0) + if got := fh.GetEntry().GetEntry(); string(got.HardLinkId) != "link" || got.HardLinkCounter != 2 { + t.Fatalf("handle did not adopt the link: %v", got) + } +}