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>

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.

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

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

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.

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)