diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 627453fa17..2d331cf130 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -49,6 +49,9 @@ type FileHandle struct { // nothing to delete after all can tell its own mark from a later one. deleteEpoch uint64 isRenamed bool // set by Rename before waiting for async flush; skips old-path metadata flush + // linkRefresh is the hard link id of a link change whose stored entry + // could not be read; the next flush reads it before publishing. + linkRefresh []byte // entryVersionTsNs is the filer log position the handle's entry reflects. // State at or below it must not replace the entry — that rolls it back. diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index f191d1ddc8..548eefcd3d 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -1,7 +1,6 @@ package mount import ( - "bytes" "context" "fmt" "syscall" @@ -363,9 +362,11 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin if isHardLink && markedInode != 0 { if survivor, status := wfs.inodeToPath.GetPath(markedInode); status == fuse.OK { if fh, found := wfs.fhMap.FindFileHandle(markedInode); found { - if stored, tsNs, signature, err := filer_pb.GetEntry(ctx, wfs, survivor); err == nil && bytes.Equal(stored.GetHardLinkId(), entry.HardLinkId) { - fh.adoptLinkChange(stored, tsNs, signature) + lock := wfs.fhLockTable.AcquireLock("Unlink", fh.fh, util.ExclusiveLock) + if err := wfs.adoptStoredLink(ctx, fh, survivor, entry.HardLinkId); err != nil { + glog.V(1).Infof("unlink %s: read %s again at the next flush: %v", entryFullPath, survivor, err) } + wfs.fhLockTable.ReleaseLock(fh.fh, lock) } wfs.clearHandleDeleted(markedInode, deleteEpoch) } diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 89f5652858..0d16be633d 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -223,6 +223,11 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n glog.V(3).Infof("flushMetadataToFiler %s fh %d: file was unlinked, skipping", fileFullPath, fh.fh) return nil } + if fh.linkRefresh != nil { + if err := wfs.adoptStoredLink(ctx, fh, util.NewFullPath(dir, name), fh.linkRefresh); err != nil { + return err + } + } entry := fh.GetEntry() entry.Name = name // this flush may be just after a rename operation diff --git a/weed/mount/weedfs_link.go b/weed/mount/weedfs_link.go index 85d7e6dcd9..d26a486914 100644 --- a/weed/mount/weedfs_link.go +++ b/weed/mount/weedfs_link.go @@ -3,6 +3,7 @@ package mount import ( "bytes" "context" + "errors" "syscall" "time" @@ -227,11 +228,15 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * // fields and times is left alone, so a peer's write is still refused. Pending // content stays pending. func (fh *FileHandle) adoptLinkChange(stored *filer_pb.Entry, versionTsNs int64, signature int32) { + fhActiveLock := fh.wfs.fhLockTable.AcquireLock("adoptLinkChange", fh.fh, util.ExclusiveLock) + defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) + fh.adoptLinkChangeLocked(stored, versionTsNs, signature) +} + +func (fh *FileHandle) adoptLinkChangeLocked(stored *filer_pb.Entry, versionTsNs int64, signature int32) { if stored == nil { return } - fhActiveLock := fh.wfs.fhLockTable.AcquireLock("adoptLinkChange", fh.fh, util.ExclusiveLock) - defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) if base := fh.baseEntry.Load(); base != nil && !fh.wfs.sameApartFromLink(stored, base) { return } @@ -249,6 +254,23 @@ func (fh *FileHandle) adoptLinkChange(stored *filer_pb.Entry, versionTsNs int64, fh.advanceEntryVersion(versionTsNs, signature) } +// adoptStoredLink reads the entry stored at path and adopts it as a change of +// hardLinkId. A failed read leaves the id on the handle, so its next flush +// reads again instead of publishing over a base the change made stale. The +// caller holds the handle's lock. +func (wfs *WFS) adoptStoredLink(ctx context.Context, fh *FileHandle, path util.FullPath, hardLinkId []byte) error { + stored, tsNs, signature, err := filer_pb.GetEntry(ctx, wfs, path) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + fh.linkRefresh = hardLinkId + return err + } + fh.linkRefresh = nil + if bytes.Equal(stored.GetHardLinkId(), hardLinkId) { + fh.adoptLinkChangeLocked(stored, tsNs, signature) + } + return nil +} + // sameApartFromLink reports whether stored, in filer ids, still holds base // apart from what a link or unlink rewrites: the link fields, the times, the // name, the commit stamp and the derived size. diff --git a/weed/mount/weedfs_link_open_handle_test.go b/weed/mount/weedfs_link_open_handle_test.go index 96aa8c78be..ec24176fd9 100644 --- a/weed/mount/weedfs_link_open_handle_test.go +++ b/weed/mount/weedfs_link_open_handle_test.go @@ -24,6 +24,10 @@ type hardLinkFiler struct { records map[string]*filer_pb.Entry blobs map[string]*filer_pb.Entry stamp int64 + // failLookups fails that many lookups of failName once a delete ran. + failName string + failLookups int + deleted bool } func (s *hardLinkFiler) resolve(name string) *filer_pb.Entry { @@ -68,6 +72,10 @@ func (s *hardLinkFiler) unlinkBlob(id []byte) { func (s *hardLinkFiler) LookupDirectoryEntry(ctx context.Context, r *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { s.mu.Lock() defer s.mu.Unlock() + if s.deleted && r.Name == s.failName && s.failLookups > 0 { + s.failLookups-- + return nil, status.Error(codes.Internal, "store read failed") + } entry := s.resolve(r.Name) if entry == nil { return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) @@ -124,6 +132,7 @@ func (s *hardLinkFiler) DeleteEntry(ctx context.Context, r *filer_pb.DeleteEntry return &filer_pb.DeleteEntryResponse{}, nil } delete(s.records, r.Name) + s.deleted = true if len(record.HardLinkId) > 0 { s.unlinkBlob(record.HardLinkId) } @@ -244,6 +253,27 @@ func TestUnlinkedAliasKeepsOpenHandlePublishing(t *testing.T) { } } +// The read of the surviving entry right after the delete can fail; the next +// flush has to read it again rather than publish over the stale base forever. +func TestUnlinkedAliasRereadsLostEntryAtFlush(t *testing.T) { + wfs, fake, fh, inode := openLinkTestFile(t) + linkInRoot(t, wfs, inode, "final") + fake.mu.Lock() + fake.failName, fake.failLookups = "final", 1 + fake.mu.Unlock() + unlinkInRoot(t, wfs, "tmp") + if fh.linkRefresh == nil { + t.Fatal("a failed read of the surviving entry was not left for the next flush") + } + if code := appendAndFlush(t, wfs, fh, 2); code != fuse.OK { + t.Fatalf("flush after the lost read: %v", code) + } + wantStored(t, fake, "final", 2, 1) + if fh.linkRefresh != nil { + t.Fatal("the pending read survived a successful flush") + } +} + // Unlinking the last name still stops the handle from writing it back. func TestUnlinkLastLinkKeepsOpenHandleDeleted(t *testing.T) { wfs, fake, fh, inode := openLinkTestFile(t)