diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index af12cb41f..42f3348e4 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -123,6 +123,7 @@ type WFS struct { fuseServer *fuse.Server IsOverQuota bool fhLockTable *util.LockTable[FileHandleId] + hardLinkLockTable *util.LockTable[string] posixLocks *PosixLockTable rdmaClient *RDMAMountClient FilerConf *filer.FilerConf @@ -213,6 +214,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { filerClient: filerClient, // nil for proxy mode, initialized for direct access pendingAsyncFlush: make(map[uint64]chan struct{}), fhLockTable: util.NewLockTable[FileHandleId](), + hardLinkLockTable: util.NewLockTable[string](), posixLocks: NewPosixLockTable(), refreshingDirs: make(map[util.FullPath]struct{}), atimeMap: make(map[uint64]time.Time, 8192), diff --git a/weed/mount/weedfs_file_copy_range_test.go b/weed/mount/weedfs_file_copy_range_test.go index 5142f3f50..02c34409b 100644 --- a/weed/mount/weedfs_file_copy_range_test.go +++ b/weed/mount/weedfs_file_copy_range_test.go @@ -314,9 +314,10 @@ func newCopyRangeTestWFS() *WFS { VolumeServerAccess: "filerProxy", FilerAddresses: []pb.ServerAddress{"127.0.0.1:8888"}, }, - inodeToPath: NewInodeToPath(util.FullPath("/"), 0), - fhMap: NewFileHandleToInode(), - fhLockTable: util.NewLockTable[FileHandleId](), + inodeToPath: NewInodeToPath(util.FullPath("/"), 0), + fhMap: NewFileHandleToInode(), + fhLockTable: util.NewLockTable[FileHandleId](), + hardLinkLockTable: util.NewLockTable[string](), } wfs.copyBufferPool.New = func() any { return make([]byte, 1024) diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index 039702c0f..c50702da3 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -188,6 +188,33 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin return fuse.EPERM } + // For hard-linked files, serialize all concurrent mutations on the + // same HardLinkId so the filer-side blob decrement and the mount-side + // sibling cache sync are observed atomically by other unlinkers. + // Without this, two concurrent unlinks on different links of the same + // file both read the same pre-decrement counter and both stamp their + // siblings to counter-1, leaving the cache one higher than the blob. + // Re-load the entry under the lock so we see any sibling update a + // prior holder just applied. + if entry != nil && len(entry.HardLinkId) > 0 { + hlKey := string(entry.HardLinkId) + lock := wfs.hardLinkLockTable.AcquireLock("unlink", hlKey, util.ExclusiveLock) + defer wfs.hardLinkLockTable.ReleaseLock(hlKey, lock) + // If another thread unlinked the entry while we waited for the + // lock, the file is already gone — return OK like the initial + // maybeLoadEntry path above. Do not fall back to the stale + // pre-lock snapshot: proceeding with its HardLinkCounter would + // reintroduce the stale-base update the lock is meant to prevent. + fresh, freshCode := wfs.maybeLoadEntry(entryFullPath) + if freshCode == fuse.ENOENT { + return fuse.OK + } + if freshCode != fuse.OK { + return freshCode + } + entry = fresh + } + // POSIX: enforce sticky bit on the parent directory. if dirEntry, dirCode := wfs.maybeLoadEntry(dirFullPath); dirCode == fuse.OK && dirEntry != nil && dirEntry.Attributes != nil { targetUid := uint32(0) diff --git a/weed/mount/weedfs_file_mkrm_test.go b/weed/mount/weedfs_file_mkrm_test.go index abfb93d20..f7d7600c2 100644 --- a/weed/mount/weedfs_file_mkrm_test.go +++ b/weed/mount/weedfs_file_mkrm_test.go @@ -114,7 +114,8 @@ func newCreateTestWFS(t *testing.T) (*WFS, *createEntryTestServer) { signature: 1, inodeToPath: NewInodeToPath(root, 0), fhMap: NewFileHandleToInode(), - fhLockTable: util.NewLockTable[FileHandleId](), + fhLockTable: util.NewLockTable[FileHandleId](), + hardLinkLockTable: util.NewLockTable[string](), } wfs.metaCache = meta_cache.NewMetaCache( filepath.Join(t.TempDir(), "meta"), diff --git a/weed/mount/weedfs_link.go b/weed/mount/weedfs_link.go index 5dd20f6ed..9413705c9 100644 --- a/weed/mount/weedfs_link.go +++ b/weed/mount/weedfs_link.go @@ -55,6 +55,35 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * return fuse.EPERM } + // If the source is already a hard link, serialize on its HardLinkId + // so concurrent Link/Unlink operations on different siblings cannot + // both compute a new counter from a stale base. Re-load the entry + // under the lock to pick up any prior holder's sibling update. + if len(oldEntry.HardLinkId) > 0 { + hlKey := string(oldEntry.HardLinkId) + lock := wfs.hardLinkLockTable.AcquireLock("link", hlKey, util.ExclusiveLock) + defer wfs.hardLinkLockTable.ReleaseLock(hlKey, lock) + // Under the lock, re-resolve the source alias from the inode. + // A concurrent Unlink that held this same lock may have removed + // the specific alias we picked pre-lock even though other + // sibling hard links for the same inode are still around; + // GetPath(Oldnodeid) returns whichever alias is still active. + refreshedPath, refreshedStatus := wfs.inodeToPath.GetPath(in.Oldnodeid) + if refreshedStatus != fuse.OK { + return refreshedStatus + } + oldEntryPath = refreshedPath + oldParentPath, _ = oldEntryPath.DirAndName() + // Do not fall back to the pre-lock snapshot: if every alias + // was deleted while we waited, abort instead of deriving the + // next counter from a stale entry. + fresh, freshStatus := wfs.maybeLoadEntry(oldEntryPath) + if freshStatus != fuse.OK { + return freshStatus + } + oldEntry = fresh + } + // update old file to hardlink mode origHardLinkId := oldEntry.HardLinkId origHardLinkCounter := oldEntry.HardLinkCounter