mount: discard a path-cache insert that raced a purge (#10842)

* mount: discard a path-cache insert that raced a purge

The Windows adapter's walk resolves a component with a Lookup RPC and inserts
the result holding no lock, so a purge can land in between - and what the walk
just resolved is then the very name the purge removed. Anything opening the
old path concurrently with a rename repopulates the cache with the vacated
name, which the next stat is served from for up to a second. The release path
already guards its equivalent insert; the walk had nothing.

The cache counts purges now. A resolve snapshots the generation before its
lookups and insert discards the entry when any purge ran in between, parking
the reference in the graveyard so the in-flight caller keeps a valid inode
either way.

Seen once in CI as TestRenameOverExisting failing with 'source survived the
rename': every SeaweedFS layer is synchronous with the rename, but a
background open of the source - an antivirus scan of the just-written file
fits - can requalify the stale name through this window. The assertion also
reports what stat returned now, and whether it persisted, so a recurrence
indicts a specific layer instead of reading as a mystery.

* mount: cover the path-cache discard by key, and let a discard rest

Review follow-ups. The generation was global, so any purge between a walk's
snapshot and its insert discarded the entry whatever its name - and an open
retries resolve-then-steal only four times before failing with EIO, so
sustained unrelated churn could fail opens of untouched paths. Purges are
remembered by key now and only one that covers the inserted name discards it;
past the remembered window the insert is discarded without a check, which only
costs a retry.

A discard that itself tripped the sweep also handed its own reference straight
to forget while the walker was still using the inode. The graveyard holds two
generations now, so an appended reference always survives the sweep of the
call that appended it - which the displaced-entry and purge paths needed too.

Also restores the original path-cache test suite this branch had overwritten
instead of extended, and rewords the semantics-test failure so it no longer
claims the source survived when stat returned a transient error.

* mount: take an open's reference directly instead of stealing it back

resolveAndSteal cached the final component only to steal it back, so an open
depended on that insert surviving whatever purges raced it - four attempts and
then EIO. The keyed purge window narrowed how often an insert is discarded,
but past the window the discard is blind again, so the cliff had only moved.

A cached entry is still stolen; anything else is now looked up directly, with
the caller owning the reference from the start. No retry loop, and no way for
churn - covered, unrelated or overflowing the window - to fail an open.

Also covers the whole-cache purge: purge of the root with prefix set clears
every entry, but the covers check tested for a '/'-prefixed key that a
normalised key never has, so it covered no in-flight insert at all.

test/winfsp/semantics_test.go

@@ -175,8 +175,13 @@ func TestRenameOverExisting(t *testing.T) {
 	if string(got) != "new" {
 		t.Fatalf("target holds %q, want the source content", got)
 	}
-	if _, err := os.Stat(src); !os.IsNotExist(err) {
-		t.Fatal("source survived the rename")
+	if fi, err := os.Stat(src); !os.IsNotExist(err) {
+		// Both a stale positive and a transient non-ENOENT error land here, and
+		// they indict different layers; a second look says whether it persists.
+		time.Sleep(200 * time.Millisecond)
+		fi2, err2 := os.Stat(src)
+		t.Fatalf("stat of the renamed-away source did not return not-exist: stat=%v err=%v; 200ms later stat=%v err=%v",
+			describeFileInfo(fi), err, describeFileInfo(fi2), err2)
 	}
 }
 
@@ -515,3 +520,10 @@ func TestDeleteOnClose(t *testing.T) {
 		t.Fatalf("close after recreate: %v", err)
 	}
 }
+
+func describeFileInfo(fi os.FileInfo) string {
+	if fi == nil {
+		return "<nil>"
+	}
+	return fmt.Sprintf("{size=%d mode=%v mtime=%s}", fi.Size(), fi.Mode(), fi.ModTime().Format(time.RFC3339Nano))
+}

weed/mount/winfsp/fs_windows.go

@@ -32,11 +32,6 @@ const (
 	// WinFsp. Bounded so a directory with millions of children does not
 	// materialise in one slice.
 	readdirBatch = 4096
-
-	// stealAttempts bounds the resolve-then-steal loop in the open paths. A
-	// miss needs the entry to be evicted in the instant between the two calls,
-	// so a second round is already unlikely.
-	stealAttempts = 4
 )
 
 // never is a nil channel: receiving blocks forever, which is what the raw
@@ -192,6 +187,10 @@ func (w *WinFS) forget(inode uint64) {
 // to hold the inode longer steals the reference from the cache.
 func (w *WinFS) walk(parts []string) (uint64, fuse.Status) {
 	inode := uint64(rootInode)
+	// Taken before the lookups: a purge racing this walk - a rename or unlink
+	// landing between a Lookup and its insert - makes what was just resolved
+	// the very thing the purge removed.
+	gen := w.paths.snapshot()
 	for i, name := range parts {
 		key := strings.Join(parts[:i+1], "/")
 		if cached, _, ok := w.paths.lookup(key); ok {
@@ -202,7 +201,7 @@ func (w *WinFS) walk(parts []string) (uint64, fuse.Status) {
 		if status := w.wfs.Lookup(never, ptr(w.caller(inode)), name, &out); status != fuse.OK {
 			return 0, status
 		}
-		w.paths.insert(key, out.NodeId, out.Attr)
+		w.paths.insert(key, out.NodeId, out.Attr, gen)
 		inode = out.NodeId
 	}
 	return inode, fuse.OK
@@ -228,24 +227,29 @@ func (w *WinFS) resolveParent(path string) (uint64, string, fuse.Status) {
 	return parent, name, fuse.OK
 }
 
-// resolveAndSteal resolves path and takes over the cache's reference on the
-// final inode, for the open paths that hold it for the life of a handle. The
-// root is handed out without a reference; it does not need one.
+// resolveAndSteal resolves path to an inode reference the caller owns, for
+// the open paths that hold it for the life of a handle: a cached entry is
+// stolen, anything else is looked up directly so the reference never passes
+// through the cache - an open must not depend on an insert surviving the
+// purges racing it. The root is handed out without a reference; it does not
+// need one.
 func (w *WinFS) resolveAndSteal(path string) (uint64, fuse.Status) {
 	key := cacheKey(path)
-	for attempt := 0; attempt < stealAttempts; attempt++ {
-		inode, status := w.resolve(path)
-		if status != fuse.OK {
-			return 0, status
-		}
-		if key == "" {
-			return inode, fuse.OK
-		}
-		if stolen, ok := w.paths.steal(key); ok {
-			return stolen, fuse.OK
-		}
+	if key == "" {
+		return rootInode, fuse.OK
 	}
-	return 0, fuse.EIO
+	if stolen, ok := w.paths.steal(key); ok {
+		return stolen, fuse.OK
+	}
+	parent, name, status := w.resolveParent(path)
+	if status != fuse.OK {
+		return 0, status
+	}
+	var out fuse.EntryOut
+	if status := w.wfs.Lookup(never, ptr(w.caller(parent)), name, &out); status != fuse.OK {
+		return 0, status
+	}
+	return out.NodeId, fuse.OK
 }
 
 func (w *WinFS) attrToStat(attr *fuse.Attr, stat *cgofuse.Stat_t) {
@@ -392,7 +396,7 @@ func (w *WinFS) Mkdir(path string, mode uint32) int {
 	if status == fuse.OK {
 		w.purgeWithParent(path, false)
 		// The new directory is about to be filled; cache it, reference and all.
-		w.paths.insert(cacheKey(path), out.NodeId, out.Attr)
+		w.paths.insert(cacheKey(path), out.NodeId, out.Attr, w.paths.snapshot())
 	}
 	return toErrno(status)
 }
@@ -674,8 +678,8 @@ func (w *WinFS) Release(path string, fh uint64) int {
 		// path still names this inode: WinFsp reports the path the handle
 		// opened with, and after a delete-on-close or a rename caching it
 		// would resurrect an entry that is gone.
-		if key := cacheKey(path); attr != nil && w.stillNames(key, inode) {
-			w.paths.insert(key, inode, *attr)
+		if key, gen := cacheKey(path), w.paths.snapshot(); attr != nil && w.stillNames(key, inode) {
+			w.paths.insert(key, inode, *attr, gen)
 		} else {
 			w.forget(inode)
 		}

weed/mount/winfsp/pathcache.go

@@ -22,10 +22,60 @@ type pathCache struct {
 	ttl    time.Duration
 	forget func(inode uint64)
 
-	mu        sync.Mutex
-	entries   map[string]*pathCacheEntry
-	graveyard []uint64
-	lastSweep time.Time
+	mu      sync.Mutex
+	entries map[string]*pathCacheEntry
+	// Two graveyard generations: an appended reference always survives the
+	// sweep of the call that appended it, so a caller still using the inode -
+	// a walk mid-resolution, an operation that looked it up just before - is
+	// never holding a reference the same call just returned.
+	graveyard     []uint64
+	prevGraveyard []uint64
+	lastSweep     time.Time
+	// gen counts purges. A resolve holds no lock across its Lookup RPC, so a
+	// purge can run between the RPC and the insert; the insert then carries
+	// exactly the name the purge removed and has to be discarded. The recent
+	// purges are kept by key so only a purge that covers the inserted name
+	// discards it: unrelated churn must not starve resolveAndSteal into EIO.
+	gen          uint64
+	recentPurges []purgeRecord
+}
+
+type purgeRecord struct {
+	key    string
+	prefix bool
+}
+
+// maxRecentPurges bounds the purges remembered for the covers check. A resolve
+// older than the window is discarded without one, which only costs a retry.
+const maxRecentPurges = 128
+
+func (r purgeRecord) covers(key string) bool {
+	if r.key == key {
+		return true
+	}
+	if !r.prefix {
+		return false
+	}
+	// An empty key with prefix is the whole cache: purge clears every entry
+	// for it, so it has to cover every in-flight insert too.
+	return r.key == "" || strings.HasPrefix(key, r.key+"/")
+}
+
+// purgedSince reports whether any purge after gen covers key.
+func (c *pathCache) purgedSince(gen uint64, key string) bool {
+	dropped := c.gen - gen
+	if dropped == 0 {
+		return false
+	}
+	if dropped > uint64(len(c.recentPurges)) {
+		return true
+	}
+	for _, r := range c.recentPurges[uint64(len(c.recentPurges))-dropped:] {
+		if r.covers(key) {
+			return true
+		}
+	}
+	return false
 }
 
 type pathCacheEntry struct {
@@ -69,14 +119,32 @@ func (c *pathCache) lookup(key string) (inode uint64, attr fuse.Attr, ok bool) {
 	return entry.inode, entry.attr, true
 }
 
-// insert takes ownership of one lookup reference on inode.
-func (c *pathCache) insert(key string, inode uint64, attr fuse.Attr) {
+// snapshot returns the purge generation. A caller about to resolve outside
+// the lock passes it back to insert, which discards the entry if any purge ran
+// in between.
+func (c *pathCache) snapshot() uint64 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.gen
+}
+
+// insert takes ownership of one lookup reference on inode, in every outcome:
+// an entry a covering purge outdated goes to the graveyard instead of the map,
+// so the caller may keep using the inode for at least one sweep either way.
+func (c *pathCache) insert(key string, inode uint64, attr fuse.Attr, gen uint64) {
 	if key == "" {
 		c.forget(inode)
 		return
 	}
 	var pending []uint64
 	c.mu.Lock()
+	if c.purgedSince(gen, key) {
+		c.graveyard = append(c.graveyard, inode)
+		pending = c.sweepLocked()
+		c.mu.Unlock()
+		c.forgetAll(pending)
+		return
+	}
 	if existing, found := c.entries[key]; found {
 		c.graveyard = append(c.graveyard, existing.inode)
 	} else if len(c.entries) >= maxCachedPaths {
@@ -109,6 +177,13 @@ func (c *pathCache) steal(key string) (inode uint64, ok bool) {
 func (c *pathCache) purge(key string, prefix bool) {
 	var pending []uint64
 	c.mu.Lock()
+	// Recorded even when the map holds nothing: the point is as much the
+	// in-flight resolve about to insert this very name.
+	c.gen++
+	c.recentPurges = append(c.recentPurges, purgeRecord{key: key, prefix: prefix})
+	if len(c.recentPurges) > maxRecentPurges {
+		c.recentPurges = append(c.recentPurges[:0], c.recentPurges[len(c.recentPurges)-maxRecentPurges:]...)
+	}
 	if entry, found := c.entries[key]; found {
 		c.graveyard = append(c.graveyard, entry.inode)
 		delete(c.entries, key)
@@ -127,20 +202,24 @@ func (c *pathCache) purge(key string, prefix bool) {
 	c.forgetAll(pending)
 }
 
-// sweepLocked returns the previous graveyard for the caller to forget outside
-// the lock, and moves expired entries into the next one. Sweeps run at most
-// once per ttl, so a reference rests here for at least one full ttl.
+// sweepLocked returns the generation before last for the caller to forget
+// outside the lock, and retires the current one. Sweeps run at most once per
+// ttl and an appended reference sits out the sweep of its own call, so every
+// reference rests for at least one full ttl after its last possible use.
 func (c *pathCache) sweepLocked() []uint64 {
 	now := time.Now()
 	if now.Sub(c.lastSweep) < c.ttl {
 		return nil
 	}
 	c.lastSweep = now
-	pending := c.graveyard
+	pending := c.prevGraveyard
+	c.prevGraveyard = c.graveyard
 	c.graveyard = nil
 	for key, entry := range c.entries {
 		if now.After(entry.expires) {
-			c.graveyard = append(c.graveyard, entry.inode)
+			// Returned by the next sweep, a full ttl away: served from the map
+			// until a moment ago, so someone may still be holding it.
+			c.prevGraveyard = append(c.prevGraveyard, entry.inode)
 			delete(c.entries, key)
 		}
 	}

weed/mount/winfsp/pathcache_test.go

@@ -1,6 +1,7 @@
 package winfsp
 
 import (
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -29,7 +30,7 @@ func TestPathCacheLookupAndExpiry(t *testing.T) {
 	rec := &forgetRecorder{}
 	c := newPathCache(20*time.Millisecond, rec.forget)
 
-	c.insert("a/b", 7, fuse.Attr{Ino: 7, Size: 42})
+	c.insert("a/b", 7, fuse.Attr{Ino: 7, Size: 42}, c.snapshot())
 	inode, attr, ok := c.lookup("a/b")
 	if !ok || inode != 7 || attr.Size != 42 {
 		t.Fatalf("lookup = %d %v %v, want 7 size 42 true", inode, attr.Size, ok)
@@ -41,9 +42,9 @@ func TestPathCacheLookupAndExpiry(t *testing.T) {
 	}
 	// The reference survives at least one sweep past expiry before it is
 	// forgotten: one insert moves it to the graveyard, the next returns it.
-	c.insert("x", 8, fuse.Attr{})
+	c.insert("x", 8, fuse.Attr{}, c.snapshot())
 	time.Sleep(30 * time.Millisecond)
-	c.insert("y", 9, fuse.Attr{})
+	c.insert("y", 9, fuse.Attr{}, c.snapshot())
 	found := false
 	for _, inode := range rec.forgotten() {
 		if inode == 7 {
@@ -59,7 +60,7 @@ func TestPathCacheStealTransfersOwnership(t *testing.T) {
 	rec := &forgetRecorder{}
 	c := newPathCache(20*time.Millisecond, rec.forget)
 
-	c.insert("a", 5, fuse.Attr{})
+	c.insert("a", 5, fuse.Attr{}, c.snapshot())
 	inode, ok := c.steal("a")
 	if !ok || inode != 5 {
 		t.Fatalf("steal = %d %v, want 5 true", inode, ok)
@@ -70,7 +71,7 @@ func TestPathCacheStealTransfersOwnership(t *testing.T) {
 	// Drive several sweeps; the stolen reference must never be forgotten.
 	for i := 0; i < 4; i++ {
 		time.Sleep(25 * time.Millisecond)
-		c.insert("churn", uint64(100+i), fuse.Attr{})
+		c.insert("churn", uint64(100+i), fuse.Attr{}, c.snapshot())
 	}
 	for _, inode := range rec.forgotten() {
 		if inode == 5 {
@@ -83,14 +84,14 @@ func TestPathCacheReplaceReturnsOldReference(t *testing.T) {
 	rec := &forgetRecorder{}
 	c := newPathCache(20*time.Millisecond, rec.forget)
 
-	c.insert("a", 5, fuse.Attr{})
-	c.insert("a", 6, fuse.Attr{})
+	c.insert("a", 5, fuse.Attr{}, c.snapshot())
+	c.insert("a", 6, fuse.Attr{}, c.snapshot())
 	if inode, _, ok := c.lookup("a"); !ok || inode != 6 {
 		t.Fatalf("lookup after replace = %d %v, want 6 true", inode, ok)
 	}
 	for i := 0; i < 3; i++ {
 		time.Sleep(25 * time.Millisecond)
-		c.insert("churn", uint64(100+i), fuse.Attr{})
+		c.insert("churn", uint64(100+i), fuse.Attr{}, c.snapshot())
 	}
 	found := false
 	for _, inode := range rec.forgotten() {
@@ -110,10 +111,10 @@ func TestPathCachePurgePrefix(t *testing.T) {
 	rec := &forgetRecorder{}
 	c := newPathCache(time.Minute, rec.forget)
 
-	c.insert("dir", 2, fuse.Attr{})
-	c.insert("dir/a", 3, fuse.Attr{})
-	c.insert("dir/a/b", 4, fuse.Attr{})
-	c.insert("dirt", 5, fuse.Attr{})
+	c.insert("dir", 2, fuse.Attr{}, c.snapshot())
+	c.insert("dir/a", 3, fuse.Attr{}, c.snapshot())
+	c.insert("dir/a/b", 4, fuse.Attr{}, c.snapshot())
+	c.insert("dirt", 5, fuse.Attr{}, c.snapshot())
 
 	c.purge("dir", true)
 	if _, _, ok := c.lookup("dir"); ok {
@@ -132,7 +133,7 @@ func TestPathCacheRootNeverCached(t *testing.T) {
 	rec := &forgetRecorder{}
 	c := newPathCache(time.Minute, rec.forget)
 
-	c.insert("", 9, fuse.Attr{})
+	c.insert("", 9, fuse.Attr{}, c.snapshot())
 	if _, _, ok := c.lookup(""); ok {
 		t.Fatal("root was cached")
 	}
@@ -155,3 +156,114 @@ func TestCacheKey(t *testing.T) {
 		}
 	}
 }
+
+// A resolve runs its Lookup outside the cache lock, so a purge can land
+// between the RPC and the insert. An insert a purge covered must then be
+// discarded: it carries exactly the name the purge removed, and caching it
+// would serve a deleted entry for a full ttl. This is how a rename's source
+// briefly came back from the dead on the Windows mount.
+func TestPathCacheInsertAfterCoveringPurgeIsDiscarded(t *testing.T) {
+	c := newPathCache(time.Minute, func(uint64) {})
+
+	gen := c.snapshot()
+	c.purge("dir/src", false)
+	c.insert("dir/src", 42, fuse.Attr{}, gen)
+
+	if _, _, ok := c.lookup("dir/src"); ok {
+		t.Fatal("purged name served from an insert that started before the purge")
+	}
+}
+
+// A purge of a directory covers the resolves in flight under it.
+func TestPathCacheInsertUnderPurgedPrefixIsDiscarded(t *testing.T) {
+	c := newPathCache(time.Minute, func(uint64) {})
+
+	gen := c.snapshot()
+	c.purge("dir", true)
+	c.insert("dir/child", 7, fuse.Attr{}, gen)
+
+	if _, _, ok := c.lookup("dir/child"); ok {
+		t.Fatal("purged prefix served a child from a stale insert")
+	}
+}
+
+// A purge of something else entirely must not discard the insert: an open
+// retries resolve-then-steal only a few times before giving up with EIO, so
+// unrelated churn starving every insert would fail opens of untouched paths.
+func TestPathCacheUnrelatedPurgeDoesNotDiscard(t *testing.T) {
+	c := newPathCache(time.Minute, func(uint64) {})
+
+	gen := c.snapshot()
+	c.purge("elsewhere", true)
+	c.purge("dir/srcling", false)
+	c.insert("dir/src", 9, fuse.Attr{Size: 11}, gen)
+
+	inode, attr, ok := c.lookup("dir/src")
+	if !ok || inode != 9 || attr.Size != 11 {
+		t.Fatalf("lookup = %d,%d,%v, want the inserted entry", inode, attr.Size, ok)
+	}
+}
+
+// Purges beyond the remembered window cannot be checked by key, so the insert
+// is discarded rather than trusted.
+func TestPathCacheInsertOlderThanPurgeWindowIsDiscarded(t *testing.T) {
+	c := newPathCache(time.Minute, func(uint64) {})
+
+	gen := c.snapshot()
+	for i := 0; i <= maxRecentPurges; i++ {
+		c.purge(fmt.Sprintf("unrelated/%d", i), false)
+	}
+	c.insert("dir/src", 3, fuse.Attr{}, gen)
+
+	if _, _, ok := c.lookup("dir/src"); ok {
+		t.Fatal("insert older than the purge window was trusted")
+	}
+}
+
+// The discarded insert still owns a lookup reference, which has to come back
+// through the graveyard rather than leak - and not in the same call: the
+// walker is still using the inode.
+func TestPathCacheDiscardedInsertRestsBeforeForget(t *testing.T) {
+	rec := &forgetRecorder{}
+	c := newPathCache(10*time.Millisecond, rec.forget)
+
+	gen := c.snapshot()
+	c.purge("dir/src", false)
+	// Make a sweep due, so the discarding insert itself sweeps: the reference
+	// it just parked must not come back out of that same call.
+	time.Sleep(15 * time.Millisecond)
+	c.insert("dir/src", 42, fuse.Attr{}, gen)
+	for _, inode := range rec.forgotten() {
+		if inode == 42 {
+			t.Fatal("discarded insert forgotten in the same call; the walker still holds it")
+		}
+	}
+
+	for i := 0; i < 3; i++ {
+		time.Sleep(15 * time.Millisecond)
+		c.purge("churn", false)
+	}
+	found := false
+	for _, inode := range rec.forgotten() {
+		if inode == 42 {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("discarded insert leaked its lookup reference; forgot %v", rec.forgotten())
+	}
+}
+
+// A purge of the whole cache - the root, prefix set - covers every in-flight
+// insert, exactly as it removed every entry.
+func TestPathCacheRootPurgeCoversEveryInsert(t *testing.T) {
+	c := newPathCache(time.Minute, func(uint64) {})
+
+	gen := c.snapshot()
+	c.purge("", true)
+	c.insert("dir/src", 42, fuse.Attr{}, gen)
+
+	if _, _, ok := c.lookup("dir/src"); ok {
+		t.Fatal("root purge did not cover an in-flight insert")
+	}
+}