mount: keep publishing an open file through its surviving hard link (#12609)

* mount: keep flushing an open file whose other hard link survives an unlink

Unlink marks the inode's open handle deleted so no flush recreates the
removed name. When the entry was one of several hard links the inode lives
on under another name, yet the mark stayed and every later fsync and close
returned OK without publishing metadata. Lift the unlink's own mark once
the name is gone and the inode still has a path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: carry a link's stored entry into the file's open handle

Link rewrites the source entry with the new hard link id, count and times,
but an open handle kept its old base and entry. Its next conditional
content flush was refused (EIO), and an unconditional one would have
dropped the hard link id. Move the handle onto the stored entry when only
the link fields and times changed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: carry an alias unlink's stored entry into the open handle

Removing one of several hard links lowers the stored link count and
stamps a new ctime, and the handle may still hold the removed name. Look up
the entry at the name the handle flushes through and adopt it, so the next
content flush matches the stored entry, writes the right count, and
attribute saves address the surviving name.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: test an open handle across link and alias unlink

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: read an alias unlink's lost entry again at the next flush

When the read of the surviving entry right after an alias unlink fails,
the handle kept a base with the old link count and ctime, and every later
flush was refused even after the filer recovered. Leave the hard link id
on the handle and have the next flush read the stored entry and adopt it
before publishing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: publish a queued close before unlinking a name another link keeps

With writeback caching, close only queues the flush. Unlink marked the
handle deleted before waiting for that flush, so the flush skipped its
metadata and the surviving link kept the content from before the close.
When another link keeps the file, wait for the queued flush first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: let a queued close reach the filer before linking the file

With writeback caching a new file is created on the filer by its queued
close, and a listed directory serves the source from the cache, so a link
right after close updated a name the filer did not have yet and failed
with EIO.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: skip a link change no newer than the handle's entry

A link change carried into the handle could replace a base already moved
past it by a later acknowledgment. Use the same clock-domain check as
installAckedEntry.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: keep a link change's ctime in the dirty entry

The flush does not restamp times, so a dirty entry carrying an earlier
write's ctime moved the filer's ctime back from the link or unlink.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: flush through the file's current name when the resolved one is gone

doFlush resolves the name before uploading the data. An unlink of that
alias, or a rename, in the meantime sent the publish to a name the file no
longer has.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: keep the pending link read when its name is already gone

A flush could read the removed alias name, find nothing, and drop the
pending read of the surviving link, leaving the stale base in place.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: test link and alias unlink across queued closes, flush races, versions and ctime

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: pick a flush's name again once its locks are held

An unlink or rename that lands between the name pick and the locks would
otherwise send the flush to the removed name, where the create guard fails.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: test a flush waiting on its lock through an alias unlink

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: check the flush target after every lock acquisition

A flush whose name kept changing published to the name chosen before its
last lock wait. Check after each acquisition and return a retryable error
once the attempts run out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* mount: test a flush whose name changes during every lock wait

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/mount/filehandle.go

@@ -50,6 +50,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.

weed/mount/weedfs_file_mkrm.go

@@ -286,13 +286,20 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin
 	// before Unlink sets the flag).  By waiting here, any in-flight flush
 	// finishes first; even if it recreated the entry, the filer delete below
 	// will remove it again.
+	isHardLink := entry != nil && entry.Attributes != nil && len(entry.HardLinkId) > 0 && entry.HardLinkCounter > 1
+	var markedInode, deleteEpoch uint64
 	if inode, found := wfs.inodeToPath.GetInode(entryFullPath); found {
-		wfs.markHandleDeleted(inode)
-		wfs.waitForPendingAsyncFlush(context.Background(), inode)
+		markedInode = inode
 	} else if entry != nil && entry.Attributes != nil && entry.Attributes.Inode != 0 {
-		inodeFromEntry := entry.Attributes.Inode
-		wfs.markHandleDeleted(inodeFromEntry)
-		wfs.waitForPendingAsyncFlush(context.Background(), inodeFromEntry)
+		markedInode = entry.Attributes.Inode
+	}
+	if markedInode != 0 {
+		if isHardLink {
+			// Another link keeps the content, so a queued close publishes it.
+			wfs.waitForPendingAsyncFlush(context.Background(), markedInode)
+		}
+		deleteEpoch = wfs.markHandleDeleted(markedInode)
+		wfs.waitForPendingAsyncFlush(context.Background(), markedInode)
 	}
 
 	// first, ensure the filer store can correctly delete
@@ -333,7 +340,6 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin
 	// to every sibling cache entry and invalidate the kernel attr cache
 	// so subsequent lstats on other links see the new nlink — pjdfstest
 	// link/00.t asserts this after `unlink n0` leaves n1/n2 behind.
-	isHardLink := entry != nil && entry.Attributes != nil && len(entry.HardLinkId) > 0 && entry.HardLinkCounter > 1
 	var sharedInode uint64
 	if isHardLink {
 		sharedInode = entry.Attributes.Inode
@@ -355,6 +361,21 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin
 		wfs.syncHardLinkSiblings(ctx, sharedInode, decremented, event.GetTsNs(), entryFullPath)
 	}
 
+	// Another link still names the inode, so its open handle keeps flushing
+	// through that name, over the count and ctime the filer stored there.
+	if isHardLink && markedInode != 0 {
+		if survivor, status := wfs.inodeToPath.GetPath(markedInode); status == fuse.OK {
+			if fh, found := wfs.fhMap.FindFileHandle(markedInode); found {
+				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)
+		}
+	}
+
 	return fuse.OK
 
 }

weed/mount/weedfs_file_sync.go

@@ -205,17 +205,33 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al
 // from open-for-write through close, so no additional distributed lock is
 // needed here. The local fhLockTable lock below serializes within this mount.
 func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32) error {
-	fileFullPath := fh.FullPath()
-	// Attribute acknowledgement and local retention must complete before a
-	// whole-entry flush can publish. Take this before the handle's active lock.
-	unlock := wfs.lockMetadataPublication(fileFullPath)
-	defer unlock()
+	fileFullPath, dir, name := wfs.flushTarget(fh, dir, name)
+	var unlock func()
+	var fhActiveLock *util.ActiveLock
+	// An unlink or rename can land while the flush waits for its locks.
+	for attempt := 0; ; attempt++ {
+		// Attribute acknowledgement and local retention must complete before a
+		// whole-entry flush can publish. Take this before the handle's active lock.
+		unlock = wfs.lockMetadataPublication(fileFullPath)
+		fhActiveLock = fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock)
+		nowPath, nowDir, nowName := wfs.flushTarget(fh, dir, name)
+		if nowPath == fileFullPath && nowDir == dir && nowName == name {
+			break
+		}
+		fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
+		unlock()
+		if attempt == 3 {
+			return fmt.Errorf("flush %s: name changed while waiting for its locks, now %s", fileFullPath, nowPath)
+		}
+		fileFullPath, dir, name = nowPath, nowDir, nowName
+	}
+	defer func() {
+		fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
+		unlock()
+	}()
 
 	glog.V(4).Infof("flushMetadataToFiler %s/%s inode %d fh %d", dir, name, fh.inode, fh.fh)
 
-	fhActiveLock := fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock)
-	defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
-
 	// Re-check under the lock: Unlink sets the flag under it, so a flush that
 	// was already past the earlier check cannot write the entry back after
 	// the delete removed it.
@@ -223,6 +239,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
@@ -316,6 +337,19 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 	return err
 }
 
+// flushTarget returns the lock path and the name a flush of fh publishes
+// through: the caller's name while it still names the file, else the name the
+// file has now.
+func (wfs *WFS) flushTarget(fh *FileHandle, dir, name string) (util.FullPath, string, string) {
+	fileFullPath := fh.FullPath()
+	if resolved := util.FullPath(dir).Child(name); fileFullPath != "" && fileFullPath != resolved {
+		if inode, found := wfs.inodeToPath.GetInode(resolved); !found || inode != fh.inode {
+			dir, name = fileFullPath.DirAndName()
+		}
+	}
+	return fileFullPath, dir, name
+}
+
 // createContentEntry publishes a handle's whole entry only over the filer
 // state the handle observed. Deferred creates expect no entry; existing
 // handles compare the full base, including inode identity. The unary call

weed/mount/weedfs_link.go

@@ -3,10 +3,12 @@ package mount
 import (
 	"bytes"
 	"context"
+	"errors"
 	"syscall"
 	"time"
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -59,6 +61,10 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 
 	ctx, cancelFunc := ctxFromCancel(cancel)
 	defer cancelFunc()
+	// A queued close may not have created or updated the source on the filer.
+	if !wfs.waitForPendingAsyncFlush(ctx, sourceInode) {
+		return fuse.EINTR
+	}
 	oldEntry, _, status := wfs.maybeLoadEntry(ctx, oldEntryPath)
 	if status != fuse.OK {
 		return status
@@ -154,6 +160,8 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 		}
 	}
 	var linkVersionTsNs int64
+	var linkSignature int32
+	var linkedEntry *filer_pb.Entry
 	if err == nil {
 		var createResp *filer_pb.CreateEntryResponse
 		createResp, err = wfs.streamCreateEntry(context.Background(), request)
@@ -171,6 +179,7 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 			}
 		} else {
 			linkVersionTsNs = ackVersionTsNs(createResp)
+			linkSignature = createResp.GetLogSignature()
 			createEvent := createResp.GetMetadataEvent()
 			if createEvent == nil {
 				createEvent = metadataCreateEvent(string(newParentPath), request.Entry)
@@ -178,6 +187,10 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 					createEvent.TsNs = linkVersionTsNs
 				}
 			}
+			if stored := createEvent.GetEventNotification().GetNewEntry(); stored != nil {
+				linkedEntry = proto.Clone(stored).(*filer_pb.Entry)
+				linkedEntry.Name = oldEntry.Name
+			}
 			if applyErr := wfs.applyLocalMetadataEvent(context.Background(), createEvent); applyErr != nil {
 				glog.Warningf("link %s: best-effort metadata apply failed: %v", newParentPath.Child(name), applyErr)
 				wfs.inodeToPath.InvalidateChildrenCache(newParentPath)
@@ -197,6 +210,9 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 	}
 
 	wfs.inodeToPath.AddPath(sourceInode, newEntryPath)
+	if fh, found := wfs.fhMap.FindFileHandle(sourceInode); found {
+		fh.adoptLinkChange(linkedEntry, linkVersionTsNs, linkSignature)
+	}
 
 	// Propagate the new HardLinkCounter to sibling cache entries and
 	// invalidate the kernel's inode attr cache. Without this, `stat` on any
@@ -210,6 +226,86 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 	return fuse.OK
 }
 
+// adoptLinkChange moves the handle past a hard link change this mount made,
+// stored being the entry the filer now holds at the name the handle flushes
+// through, in filer ids. A base that differs from it in more than the link
+// 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
+	}
+	handleSignature := fh.entryVersionSignature.Load()
+	provablyOtherClock := signature != 0 && handleSignature != 0 && signature != handleSignature
+	if versionTsNs != 0 && !provablyOtherClock && versionTsNs <= fh.entryVersionTsNs.Load() {
+		return
+	}
+	if base := fh.baseEntry.Load(); base != nil && !fh.wfs.sameApartFromLink(stored, base) {
+		return
+	}
+	linked := proto.Clone(stored).(*filer_pb.Entry)
+	fh.wfs.mapPbIdFromFilerToLocal(linked)
+	if fh.dirtyMetadata {
+		fh.UpdateEntry(func(entry *filer_pb.Entry) {
+			entry.Name = linked.Name
+			entry.HardLinkId, entry.HardLinkCounter = linked.HardLinkId, linked.HardLinkCounter
+			if a, b := entry.Attributes, linked.Attributes; a != nil && b != nil && (b.Ctime > a.Ctime || b.Ctime == a.Ctime && b.CtimeNs > a.CtimeNs) {
+				a.Ctime, a.CtimeNs = b.Ctime, b.CtimeNs
+			}
+		})
+	} else {
+		fh.SetEntry(proto.Clone(linked).(*filer_pb.Entry))
+	}
+	fh.setAuthoritativeBase(linked)
+	fh.advanceEntryVersion(versionTsNs, signature)
+}
+
+// adoptStoredLink reads the entry stored at path and adopts it as a change of
+// hardLinkId. A failed read, or a path already gone, 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
+	}
+	if stored == nil {
+		fh.linkRefresh = hardLinkId
+		return nil
+	}
+	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.
+func (wfs *WFS) sameApartFromLink(stored, base *filer_pb.Entry) bool {
+	current, observed := proto.Clone(stored).(*filer_pb.Entry), proto.Clone(base).(*filer_pb.Entry)
+	wfs.mapPbIdFromFilerToLocal(current)
+	for _, e := range []*filer_pb.Entry{current, observed} {
+		e.Name, e.CommitTsNs = "", 0
+		e.HardLinkId, e.HardLinkCounter = nil, 0
+		if e.Attributes != nil {
+			e.Attributes.Mtime, e.Attributes.MtimeNs = 0, 0
+			e.Attributes.Ctime, e.Attributes.CtimeNs = 0, 0
+			e.Attributes.FileSize = filer.FileSize(e)
+		}
+		filer_pb.BeforeEntrySerialization(e.Chunks)
+	}
+	return proto.Equal(current, observed)
+}
+
 // syncHardLinkSiblings refreshes every sibling link of the given inode from
 // the filer, so their cached HardLinkCounter (and any other fields) reflect
 // the mutation. `authoritativeEntry` identifies the hard-link group;

weed/mount/weedfs_link_open_handle_test.go

@@ -0,0 +1,534 @@
+//go:build !freebsd
+
+package mount
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"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"
+)
+
+// 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
+	// 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 {
+	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
+	}
+	now := time.Now()
+	blob.Attributes.Ctime, blob.Attributes.CtimeNs = now.Unix(), int32(now.Nanosecond())
+}
+
+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())
+	}
+	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)
+	if current == nil {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	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)
+	s.deleted = true
+	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)
+}
+
+// failSurvivorReads fails every lookup of "final" once a delete ran.
+func failSurvivorReads(fake *hardLinkFiler, fail bool) {
+	fake.mu.Lock()
+	defer fake.mu.Unlock()
+	fake.failName, fake.failLookups = "final", 0
+	if fail {
+		fake.failLookups = 1 << 30
+	}
+}
+
+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)
+}
+
+// queueClose closes fh the way a writeback mount does. Its flush starts once
+// the handle is marked deleted, or a moment later, so callers that do not wait
+// for it run first.
+func queueClose(wfs *WFS, fh *FileHandle) {
+	fh.asyncFlushPending = true
+	wfs.pendingAsyncFlush = map[uint64]chan struct{}{}
+	wfs.asyncFlushCh = make(chan *asyncFlushItem, 1)
+	wfs.ReleaseHandle(fh.fh)
+	markedDeleted := func() bool {
+		lock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.SharedLock)
+		defer wfs.fhLockTable.ReleaseLock(fh.fh, lock)
+		return fh.isDeleted
+	}
+	go func() {
+		item := <-wfs.asyncFlushCh
+		for deadline := time.Now().Add(200 * time.Millisecond); !markedDeleted() && time.Now().Before(deadline); {
+			time.Sleep(time.Millisecond)
+		}
+		wfs.processAsyncFlushItem(item)
+	}()
+}
+
+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")
+			}
+			unlinked := fake.stored("final").Attributes
+			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 got := fake.stored("final").Attributes; got.Ctime != unlinked.Ctime || got.CtimeNs != unlinked.CtimeNs {
+				t.Fatalf("the flush moved the ctime from the unlink's %d.%09d back to %d.%09d", unlinked.Ctime, unlinked.CtimeNs, got.Ctime, got.CtimeNs)
+			}
+			if fake.stored("tmp") != nil {
+				t.Fatal("the flush recreated the unlinked name")
+			}
+		})
+	}
+}
+
+// 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")
+	failSurvivorReads(fake, true)
+	unlinkInRoot(t, wfs, "tmp")
+	failSurvivorReads(fake, false)
+	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")
+	}
+}
+
+// A flush that resolved the alias before the unlink removed it reaches the
+// surviving name, and still reads it again when the unlink could not.
+func TestFlushResolvedBeforeAliasUnlinkReachesSurvivor(t *testing.T) {
+	wfs, fake, fh, inode := openLinkTestFile(t)
+	linkInRoot(t, wfs, inode, "final")
+	failSurvivorReads(fake, true)
+	unlinkInRoot(t, wfs, "tmp")
+	failSurvivorReads(fake, false)
+	lock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	err := wfs.adoptStoredLink(context.Background(), fh, "/tmp", fh.linkRefresh)
+	wfs.fhLockTable.ReleaseLock(fh.fh, lock)
+	if err != nil || fh.linkRefresh == nil {
+		t.Fatalf("reading the removed name dropped the pending read: %v", err)
+	}
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		e.Chunks = append(e.Chunks, linkHandleChunk(2, 100))
+		e.Attributes.FileSize = 200
+	})
+	fh.dirtyMetadata = true
+	if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20); err != nil {
+		t.Fatalf("flush resolved before the unlink: %v", err)
+	}
+	wantStored(t, fake, "final", 2, 1)
+	if fake.stored("tmp") != nil || fh.linkRefresh != nil {
+		t.Fatal("the flush recreated the unlinked name or left the read pending")
+	}
+}
+
+// A flush that picked the alias and waits for its publication lock while the
+// alias is unlinked publishes through the surviving name.
+func TestFlushWaitingThroughAliasUnlinkReachesSurvivor(t *testing.T) {
+	wfs, fake, fh, inode := openLinkTestFile(t)
+	linkInRoot(t, wfs, inode, "final")
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		e.Chunks = append(e.Chunks, linkHandleChunk(2, 100))
+		e.Attributes.FileSize = 200
+	})
+	fh.dirtyMetadata = true
+	table := wfs.metadataPublicationTable()
+	held := table.AcquireLock("test", "/tmp", util.SharedLock)
+	done := make(chan error, 1)
+	go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20) }()
+	waitForPublicationWaiter(t, table, "/tmp")
+	unlinkInRoot(t, wfs, "tmp")
+	table.ReleaseLock("/tmp", held)
+	if err := <-done; err != nil {
+		t.Fatalf("flush that waited through the unlink: %v", err)
+	}
+	wantStored(t, fake, "final", 2, 1)
+	if fake.stored("tmp") != nil {
+		t.Fatal("the flush recreated the unlinked name")
+	}
+}
+
+// A flush whose name is unlinked during every lock wait gives up with a
+// retryable error instead of publishing to a name it has not checked.
+func TestFlushWhoseNameKeepsChangingRetries(t *testing.T) {
+	wfs, fake, fh, inode := openLinkTestFile(t)
+	for _, name := range []string{"a", "b", "c", "final"} {
+		linkInRoot(t, wfs, inode, name)
+	}
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		e.Chunks = append(e.Chunks, linkHandleChunk(2, 100))
+		e.Attributes.FileSize = 200
+	})
+	fh.dirtyMetadata = true
+	table := wfs.metadataPublicationTable()
+	current := util.FullPath("/tmp")
+	held := table.AcquireLock("test", current, util.SharedLock)
+	done := make(chan error, 1)
+	go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20) }()
+	var removed []string
+	for i := 0; i < 4; i++ {
+		waitForPublicationWaiter(t, table, current)
+		_, name := current.DirAndName()
+		unlinkInRoot(t, wfs, name)
+		removed = append(removed, name)
+		next := fh.FullPath()
+		nextHeld := table.AcquireLock("test", next, util.SharedLock)
+		table.ReleaseLock(current, held)
+		current, held = next, nextHeld
+	}
+	err := <-done
+	table.ReleaseLock(current, held)
+	if current != "/final" {
+		t.Fatalf("the surviving name is %s, want /final", current)
+	}
+	if err == nil || !isRetryableFilerError(err) {
+		t.Fatalf("flush whose name changed during every lock wait: %v, want a retryable error", err)
+	}
+	if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20); err != nil {
+		t.Fatalf("retried flush: %v", err)
+	}
+	wantStored(t, fake, "final", 2, 1)
+	for _, name := range removed {
+		if fake.stored(name) != nil {
+			t.Fatalf("the flush recreated the unlinked name %s", name)
+		}
+	}
+}
+
+// waitForPublicationWaiter returns once a writer is queued behind the shared
+// publication lock the test holds on path.
+func waitForPublicationWaiter(t *testing.T, table *util.LockTable[util.FullPath], path util.FullPath) {
+	t.Helper()
+	for deadline := time.Now().Add(5 * time.Second); ; time.Sleep(time.Millisecond) {
+		probe, free := table.TryAcquireLock("probe", path, util.SharedLock)
+		if !free {
+			return
+		}
+		table.ReleaseLock(path, probe)
+		if time.Now().After(deadline) {
+			t.Fatalf("the flush never waited for the publication lock on %s", path)
+		}
+	}
+}
+
+// With writeback caching close only queues the flush. Unlinking the alias,
+// found in a listed directory, before that flush runs must not drop the
+// content the other link keeps.
+func TestUnlinkedAliasPublishesQueuedClose(t *testing.T) {
+	wfs, fake, fh, inode := openLinkTestFile(t)
+	wfs.inodeToPath.MarkChildrenCached("/")
+	linkInRoot(t, wfs, inode, "final")
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		e.Chunks = append(e.Chunks, linkHandleChunk(2, 100))
+		e.Attributes.FileSize = 200
+	})
+	fh.dirtyMetadata = true
+	queueClose(wfs, fh)
+	unlinkInRoot(t, wfs, "tmp")
+	wantStored(t, fake, "final", 2, 1)
+	if fake.stored("tmp") != nil {
+		t.Fatal("the queued flush recreated the unlinked name")
+	}
+}
+
+// With writeback caching the filer create of a new file waits for its queued
+// close, so a link made right after the close has to wait for it too.
+func TestLinkWaitsForQueuedClose(t *testing.T) {
+	wfs, fake, fh, inode := openLinkTestFile(t)
+	wfs.inodeToPath.MarkChildrenCached("/")
+	fake.mu.Lock()
+	delete(fake.records, "tmp")
+	fake.mu.Unlock()
+	if err := wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/", fh.GetEntry().GetEntry()), 0); err != nil {
+		t.Fatalf("cache the local entry: %v", err)
+	}
+	fh.baseEntry.Store(nil)
+	fh.UpdateEntry(func(e *filer_pb.Entry) {
+		e.Chunks = append(e.Chunks, linkHandleChunk(2, 100))
+		e.Attributes.FileSize = 200
+	})
+	fh.dirtyMetadata = true
+	queueClose(wfs, fh)
+	linkInRoot(t, wfs, inode, "final")
+	wantStored(t, fake, "tmp", 2, 2)
+	wantStored(t, fake, "final", 2, 2)
+}
+
+// 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)
+	}
+}
+
+// A link change no newer than what the handle holds must not move its base
+// back; one stamped by another filer's clock says nothing about it and lands.
+func TestAdoptLinkChangeSkipsOlderVersion(t *testing.T) {
+	wfs, _, fh, _ := openLinkTestFile(t)
+	base := fh.baseEntry.Load()
+	fh.advanceEntryVersion(100, 5)
+	stored := proto.Clone(base).(*filer_pb.Entry)
+	wfs.mapPbIdFromLocalToFiler(stored)
+	stored.HardLinkId, stored.HardLinkCounter = []byte("link"), 2
+	fh.adoptLinkChange(stored, 100, 5)
+	if fh.baseEntry.Load() != base || len(fh.GetEntry().GetEntry().HardLinkId) != 0 {
+		t.Fatal("a link change older than the handle's version was adopted")
+	}
+	fh.adoptLinkChange(stored, 99, 7)
+	if got := fh.GetEntry().GetEntry(); string(got.HardLinkId) != "link" || got.HardLinkCounter != 2 {
+		t.Fatalf("a link change from another filer's clock was not adopted: %v", got)
+	}
+}