mount: let a rename remove its source at the source's own version (#10973)

* mount: let a rename remove its source at the source's own version

A rename stamps the source and takes the name away at the same log position,
so the removal reaches the meta cache carrying exactly the version the source
already records. The version gate read that as a write already reflected and
dropped it, while the destination half of the same event still applied -- the
source stayed cached beside the destination, and readdir and stat went on
serving a name the filer no longer had:

    gate dropped removal of /winfsp-test-TestRenameOverExisting/src
      eventTs=1787761708173717200 record=1787761708173717200
      floor=1787761708173717200 tombstone=false

A removal asks a different question from a write. An entry still present at
exactly that version has the write reflected but not its removal, so only a
strictly newer record fences one out; a tombstone is the removal already
reflected and goes on fencing as before.

* mount: sweep a section's vanished name recorded at the snapshot

The refresh deletes the names its listing did not return, but asked the gate
whether a write at the snapshot was reflected. A name recorded at exactly that
version has the write reflected and not its removal, so it survived the sweep
and stayed cached until some later event happened to touch it.

Same reading as the rename source a commit earlier: the call site removes, so
it asks about a removal.

weed/mount/meta_cache/meta_cache.go

@@ -497,13 +497,19 @@ func (mc *MetaCache) entryVersionRecordLocked(ctx context.Context, fp util.FullP
 	return int64(util.BytesToUint64(value[:8])), tombstone, unversioned
 }
 
-// entryVersionBlocksLocked reports whether a write at tsNs is already
+// entryVersionBlocksLocked reports whether a change at tsNs is already
 // reflected at fp. The path's version is its own record or, lacking one, the
 // listing floors — which cover names a listing saw present and absent alike.
 // A tombstone fences with no entry present, and a later floor outranks it; a
 // plain record only counts while its entry exists: records linger after a
 // bulk folder wipe and must not fence a recreate.
-func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullPath, tsNs int64) bool {
+//
+// removal asks a different question from a write. An entry still present at
+// exactly tsNs has the write at that version reflected but not its removal --
+// a rename stamps the source and takes the name away at the same version --
+// so only a strictly newer record fences one out. A tombstone is the removal
+// already reflected, and fences at tsNs like any other write.
+func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullPath, tsNs int64, removal bool) bool {
 	recordTsNs, tombstone, unversioned := mc.entryVersionRecordLocked(ctx, fp)
 	if unversioned {
 		// Local content no log position describes: fence nothing, so any
@@ -513,7 +519,11 @@ func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullP
 	if !tombstone && !mc.entryExistsLocked(ctx, fp) {
 		recordTsNs = 0
 	}
-	return mc.entryVersionFloorLocked(fp, recordTsNs) >= tsNs
+	floorTsNs := mc.entryVersionFloorLocked(fp, recordTsNs)
+	if removal && !tombstone {
+		return floorTsNs > tsNs
+	}
+	return floorTsNs >= tsNs
 }
 
 // entryExistsLocked reports whether fp has a live entry, applying the same TTL
@@ -951,10 +961,10 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file
 	// already reflected in it, and applying it would roll the entry back while
 	// the version keeps the newer claim. Each half is gated independently.
 	if resp.TsNs != 0 {
-		if oldPath != "" && mc.entryVersionBlocksLocked(ctx, oldPath, resp.TsNs) {
+		if oldPath != "" && mc.entryVersionBlocksLocked(ctx, oldPath, resp.TsNs, true) {
 			oldPath = ""
 		}
-		if newEntry != nil && mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs) {
+		if newEntry != nil && mc.entryVersionBlocksLocked(ctx, newEntry.FullPath, resp.TsNs, false) {
 			newEntry = nil
 		}
 	}

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -868,7 +868,7 @@ func TestExpiredEntryIsJudgedByDirectoryFloor(t *testing.T) {
 	// is the accurate answer, and an event above the floor must apply.
 	mc.setEntryVersionLocked(context.Background(), util.FullPath("/dir/file"), 5000)
 	mc.dirVersionFloors[util.FullPath("/dir")] = 3000
-	blocks := mc.entryVersionBlocksLocked(context.Background(), util.FullPath("/dir/file"), 4000)
+	blocks := mc.entryVersionBlocksLocked(context.Background(), util.FullPath("/dir/file"), 4000, false)
 	mc.Unlock()
 
 	if blocks {

weed/mount/meta_cache/meta_cache_rename_gate_test.go

@@ -0,0 +1,95 @@
+package meta_cache
+
+import (
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// A rename stamps its source and takes the name away at the same log position,
+// so the removal arrives at exactly the version the source already records.
+// Read as a write that is already reflected, it was dropped and the source
+// stayed visible next to the destination.
+func TestRenameRemovesSourceAtItsOwnVersion(t *testing.T) {
+	const version = 1787761708173717200
+
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true})
+	defer mc.Shutdown()
+	ctx := context.Background()
+
+	write := func(name string, tsNs int64) *filer_pb.SubscribeMetadataResponse {
+		return &filer_pb.SubscribeMetadataResponse{
+			Directory: "/dir",
+			EventNotification: &filer_pb.EventNotification{
+				NewEntry: &filer_pb.Entry{
+					Name:       name,
+					Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3},
+				},
+			},
+			TsNs: tsNs,
+		}
+	}
+	for _, event := range []*filer_pb.SubscribeMetadataResponse{write("src", version), write("dst", version-1000)} {
+		if err := mc.ApplyMetadataResponseOwned(ctx, event, LocalMetadataResponseApplyOptions); err != nil {
+			t.Fatalf("seed %s: %v", event.EventNotification.NewEntry.Name, err)
+		}
+	}
+
+	rename := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "src"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "dst",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3},
+			},
+			NewParentPath: "/dir",
+		},
+		TsNs: version,
+	}
+	if err := mc.ApplyMetadataResponseOwned(ctx, rename, LocalMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("rename: %v", err)
+	}
+
+	if entry, _, err := mc.FindEntry(ctx, "/dir/src"); err == nil && entry != nil {
+		t.Error("the renamed-away source is still cached")
+	}
+	if entry, _, err := mc.FindEntry(ctx, "/dir/dst"); err != nil || entry == nil {
+		t.Errorf("the rename destination is missing: %v", err)
+	}
+}
+
+// A removal older than the entry's record is still stale and must not apply:
+// the record describes a write the removal never saw.
+func TestRemovalOlderThanTheRecordStillFences(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true})
+	defer mc.Shutdown()
+	ctx := context.Background()
+
+	if err := mc.ApplyMetadataResponseOwned(ctx, &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3},
+			},
+		},
+		TsNs: 2000,
+	}, LocalMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	if err := mc.ApplyMetadataResponseOwned(ctx, &filer_pb.SubscribeMetadataResponse{
+		Directory:         "/dir",
+		EventNotification: &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: "file"}},
+		TsNs:              1000,
+	}, LocalMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("stale removal: %v", err)
+	}
+
+	if entry, _, err := mc.FindEntry(ctx, "/dir/file"); err != nil || entry == nil {
+		t.Errorf("a removal older than the entry's record deleted it: %v", err)
+	}
+}

weed/mount/meta_cache/meta_cache_sections.go

@@ -362,7 +362,7 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu
 			mc.setEntryVersionLocked(ctx, entry.FullPath, 0)
 			continue
 		}
-		if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) {
+		if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs, false) {
 			continue
 		}
 		// An unversioned marker would bypass the section floor, so it cannot
@@ -404,7 +404,7 @@ func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.Fu
 			if mc.pinnedChildFn != nil && mc.pinnedChildFn(entry) {
 				continue
 			}
-			if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) {
+			if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs, true) {
 				continue
 			}
 			if err := mc.localStore.DeleteEntry(ctx, entry.FullPath); err != nil {

weed/mount/meta_cache/meta_cache_sections_test.go

@@ -927,3 +927,48 @@ func TestRenameAcrossSectionsCountsBoth(t *testing.T) {
 		t.Fatal("the landing side's section should be invalidated")
 	}
 }
+
+// A refresh sweeps the names its listing did not return. One recorded at
+// exactly the snapshot has its write reflected but not its removal, so reading
+// the gate as a write left it cached for good.
+func TestSectionRefreshSweepsANameRecordedAtTheSnapshot(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true, "/dir": true})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+
+	const snapshotTsNs = 5000
+	if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "b-vanish",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 3},
+			},
+		},
+		TsNs: snapshotTsNs,
+	}, SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("seed at the snapshot version: %v", err)
+	}
+
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh: &sectionRefresh{
+			hi: "m",
+			entries: []*filer.Entry{
+				{
+					FullPath: util.FullPath("/dir/b-keep"),
+					Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 7},
+				},
+			},
+			snapshotTsNs: snapshotTsNs,
+		},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-vanish")); err == nil && entry != nil {
+		t.Error("a name the refresh did not list is still cached")
+	}
+}