fix(mount): serialize hard-link mutations on HardLinkId (#9064)

* fix(mount): serialize hard-link mutations on HardLinkId

syncHardLinkSiblings stamps every sibling of a hard-link to
authoritativeEntry.HardLinkCounter, and the caller computes that value
as entry.HardLinkCounter - 1 (Unlink) or entry.HardLinkCounter + 1
(Link) from a cached entry read before the filer mutation. With
concurrent Unlinks on different links of the same file, both callers
observe the same pre-decrement counter, the filer's atomic blob
decrement lands correctly, but both then stamp their siblings to
counter-1 — leaving the mount metacache one higher than the authoritative
blob.

Serialize Link and Unlink on string(HardLinkId) via a new
hardLinkLockTable on WFS, and re-load the entry under the lock so the
second caller sees the updated sibling counter its predecessor just
wrote before computing its own delta. First-link races (empty
HardLinkId on the source) are a separate pre-existing issue and are
not addressed here.

Full pjdfstest suite still passes (235 files, 8803 tests).

* fix(mount): abort on stale pre-lock entry after HardLinkId lock

Review follow-up: if maybeLoadEntry fails after acquiring the
hardLinkLockTable lock, the prior revision silently fell back to the
pre-lock snapshot, reintroducing the stale-base update the lock is
meant to prevent.

- Unlink: treat fuse.ENOENT as success (the file was already removed by
  the thread that held the lock before us) and propagate any other
  error.
- Link: abort with the returned status so we never derive the next
  HardLinkCounter from a stale source entry.

* fix(mount): re-resolve Link source alias under HardLinkId lock

Review follow-up: Link resolved oldEntryPath from in.Oldnodeid before
waiting on the HardLinkId lock. A concurrent Unlink that held the same
lock could remove the specific alias we picked pre-lock while leaving
other sibling hard links for the same inode intact. The post-lock
maybeLoadEntry then returned ENOENT even though the source inode was
still reachable.

Call GetPath(in.Oldnodeid) again under the lock to pick whichever
alias is still active, refresh oldParentPath, and only return ENOENT
if no sibling survived.

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),

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)

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)

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"),

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