mount: tell windows about changes made elsewhere (#10553)

* mount: tell windows about changes made elsewhere Nothing invalidates a Windows client's cache from this side, so a file created or removed by another mount, the S3 gateway or the filer API stayed invisible in Explorer until the user refreshed by hand. The mount already receives those events; they just had nowhere to go. WFS gains a listener for every applied metadata event, and on Windows that turns into the WinFsp notification for the path. A rename reports both ends, since the destination's own event may never arrive when it falls outside this mount. * mount: report a removed directory as a directory Entry is nil once a path is vacated, so asking it whether the thing that went away was a directory always answered no and every removal was reported as a file. Windows watches the two through different filters, so a folder removed elsewhere never refreshed. The invalidation now carries what used to be there, which the event already knew and simply was not passing on. * mount: report a rename destination once The event stream already carries a second invalidation describing the new path, so reporting RenamedTo here sent the destination twice — and always as a create, so a moved directory arrived as a create followed by a mkdir.
2026-08-16 04:06:44 +00:00 · 2026-08-03 22:17:09 -07:00
parent a0e278f86f
commit b8cba2982c
7 changed files with 171 additions and 4 deletions
@@ -108,6 +108,10 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		ExtraOptions: option.extraOptions,
 	})

+	// Windows caches entries on its own side and the mount cannot invalidate
+	// that cache directly, so changes made elsewhere have to be pushed out.
+	host.Notify(seaweedFileSystem)
+
 	grace.OnInterrupt(func() {
 		// The signal handler exits the process as soon as the hooks return, so
 		// anything still queued has to be flushed here rather than after Serve.
@@ -795,6 +795,10 @@ type EntryInvalidation struct {
 	// Signatures from the event. The filer that logged it appends its own, so
 	// this identifies the clock domain TsNs belongs to.
 	Signatures []int32
+	// WasDirectory records what used to be at a vacated path. Entry is nil
+	// once the path is empty, so this is the only thing left saying whether a
+	// directory or a file went away.
+	WasDirectory bool
 }

 type metadataResponseSideEffects struct {
@@ -1250,7 +1254,7 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []Entry
 		}
 		if message.OldEntry.Name != message.NewEntry.Name || resp.Directory != newDir {
 			newKey := util.NewFullPath(newDir, message.NewEntry.Name)
-			invalidations = append(invalidations, EntryInvalidation{Path: oldKey, TsNs: resp.TsNs, Signatures: signatures, RenamedTo: newKey})
+			invalidations = append(invalidations, EntryInvalidation{Path: oldKey, TsNs: resp.TsNs, Signatures: signatures, RenamedTo: newKey, WasDirectory: message.OldEntry.IsDirectory})
 			invalidations = append(invalidations, EntryInvalidation{Path: newKey, Entry: message.NewEntry, TsNs: resp.TsNs, Signatures: signatures})
 		} else {
 			invalidations = append(invalidations, EntryInvalidation{Path: oldKey, Entry: message.NewEntry, TsNs: resp.TsNs, Signatures: signatures})
@@ -1269,7 +1273,7 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []Entry

 	if filer_pb.IsDelete(resp) && message.OldEntry != nil {
 		oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
-		invalidations = append(invalidations, EntryInvalidation{Path: oldKey, TsNs: resp.TsNs, Deleted: true, Signatures: signatures})
+		invalidations = append(invalidations, EntryInvalidation{Path: oldKey, TsNs: resp.TsNs, Deleted: true, Signatures: signatures, WasDirectory: message.OldEntry.IsDirectory})
 	}

 	return invalidations
@@ -185,7 +185,12 @@ type WFS struct {

 	// asyncFlushWg tracks pending background flush work items for writebackCache mode.
 	// Must be waited on before unmount cleanup to prevent data loss.
-	asyncFlushWg    sync.WaitGroup
+	asyncFlushWg sync.WaitGroup
+
+	// entryChanged is notified of every applied metadata event, for a front
+	// end that has to push invalidations to its own client.
+	entryChangeMu   sync.RWMutex
+	entryChanged    func(meta_cache.EntryInvalidation)
 	asyncFlushClose sync.Once

 	// asyncFlushCh is a bounded work queue for background flush operations.
@@ -305,7 +310,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		}, func(path util.FullPath) bool {
 			return wfs.inodeToPath.IsChildrenCached(path)
-		}, wfs.invalidateOpenFileHandle, func(dirPath util.FullPath) {
+		}, wfs.onEntryInvalidation, func(dirPath util.FullPath) {
 			if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) {
 				wfs.markDirectoryReadThrough(dirPath)
 			}
@@ -777,6 +782,35 @@ func sameEntryContent(a, b *filer_pb.Entry) bool {
 // invalidateOpenFileHandle refreshes an open file handle from a metadata
 // subscription event. No filer lookup here: it can fail transiently, and with
 // the subscription cursor already past the event, nothing would retry.
+
+// SetEntryChangeListener registers a callback for every metadata event this
+// mount applies. A front end whose client caches entries on its own side, and
+// which the mount cannot invalidate directly, uses it to push the change out.
+func (wfs *WFS) SetEntryChangeListener(fn func(meta_cache.EntryInvalidation)) {
+	wfs.entryChangeMu.Lock()
+	wfs.entryChanged = fn
+	wfs.entryChangeMu.Unlock()
+}
+
+// onEntryInvalidation runs for every applied event, whether or not the path is
+// open here, so a listener sees changes made anywhere in the cluster.
+func (wfs *WFS) onEntryInvalidation(invalidation meta_cache.EntryInvalidation) {
+	wfs.entryChangeMu.RLock()
+	listener := wfs.entryChanged
+	wfs.entryChangeMu.RUnlock()
+	if listener != nil {
+		listener(invalidation)
+	}
+	wfs.invalidateOpenFileHandle(invalidation)
+}
+
+// MountRoot is the filer path this mount is rooted at. Event paths are absolute
+// on the filer; a front end that addresses files relative to the mount needs it
+// to translate them.
+func (wfs *WFS) MountRoot() util.FullPath {
+	return util.FullPath(wfs.option.FilerMountRootPath)
+}
+
 func (wfs *WFS) invalidateOpenFileHandle(invalidation meta_cache.EntryInvalidation) {
 	filePath, eventEntry, eventTsNs := invalidation.Path, invalidation.Entry, invalidation.TsNs
 	inode, inodeFound := wfs.inodeToPath.GetInode(filePath)
@@ -51,6 +51,13 @@ func New(wfs *mount.WFS, options Options) *Host {
 	return &Host{host: host, options: options}
 }

+// Notify wires the mount's metadata events to Windows. Called once before
+// Serve; the host has to exist first, which is why it is not done in New.
+func (h *Host) Notify(wfs *mount.WFS) {
+	n := &notifier{host: h.host, mountRoot: wfs.MountRoot()}
+	wfs.SetEntryChangeListener(n.notify)
+}
+
 // Serve attaches the filesystem at mountPoint, which is a drive letter ("S:"),
 // a directory that does not yet exist, or a UNC path. It blocks until the
 // filesystem is unmounted.
@@ -0,0 +1,25 @@
+package winfsp
+
+import (
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// relativeToMount rebases a filer path onto the mount root. Metadata events
+// cover the whole subscription, so a path outside this mount is not ours to
+// report, and a sibling whose name merely starts with the root is not inside
+// it either.
+func relativeToMount(root, path util.FullPath) (string, bool) {
+	full, prefix := string(path), string(root)
+	if prefix == "" || prefix == "/" {
+		return full, full != ""
+	}
+	if full == prefix {
+		return "/", true
+	}
+	if !strings.HasPrefix(full, prefix+"/") {
+		return "", false
+	}
+	return full[len(prefix):], true
+}
@@ -0,0 +1,34 @@
+package winfsp
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// Event paths are absolute on the filer and cover the whole subscription, so
+// rebasing them decides both what Windows is told and what is none of its
+// business.
+func TestRelativeToMount(t *testing.T) {
+	cases := []struct {
+		root string
+		path string
+		want string
+		ok   bool
+	}{
+		{"/", "/a/b.txt", "/a/b.txt", true},
+		{"", "/a/b.txt", "/a/b.txt", true},
+		{"/buckets/data", "/buckets/data/a/b.txt", "/a/b.txt", true},
+		{"/buckets/data", "/buckets/data", "/", true},
+		{"/buckets/data", "/buckets/other/b.txt", "", false},
+		// A sibling whose name merely starts with the root must not match.
+		{"/buckets/data", "/buckets/data2/b.txt", "", false},
+		{"/buckets/data", "/elsewhere", "", false},
+	}
+	for _, c := range cases {
+		got, ok := relativeToMount(util.FullPath(c.root), util.FullPath(c.path))
+		if ok != c.ok || (ok && got != c.want) {
+			t.Errorf("root %q path %q = (%q, %v), want (%q, %v)", c.root, c.path, got, ok, c.want, c.ok)
+		}
+	}
+}
@@ -0,0 +1,59 @@
+package winfsp
+
+import (
+	cgofuse "github.com/winfsp/cgofuse/fuse"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// notifier turns the mount's metadata events into the notifications Windows
+// listens for. Nothing invalidates a Windows client's cache from this side, so
+// without it a change made by another mount, the S3 gateway or the filer API
+// stays invisible until the user refreshes by hand.
+type notifier struct {
+	host      *cgofuse.FileSystemHost
+	mountRoot util.FullPath
+}
+
+// notify reports one applied event. Windows wants the path relative to the
+// mount, and an action describing what happened to it.
+func (n *notifier) notify(invalidation meta_cache.EntryInvalidation) {
+	if n.host == nil {
+		return
+	}
+	// A rename arrives as two invalidations, one vacating the old path and one
+	// describing the new, so reporting RenamedTo here as well would send the
+	// destination twice — and as a create even when a directory moved.
+	if path, ok := relativeToMount(n.mountRoot, invalidation.Path); ok {
+		n.send(path, n.action(invalidation))
+	}
+}
+
+func (n *notifier) send(path string, action uint32) {
+	if !n.host.Notify(path, action) {
+		// A rejected notification only costs a stale view until the client
+		// looks again, so it is not worth failing an operation over.
+		glog.V(4).Infof("winfsp notify %s action %d rejected", path, action)
+	}
+}
+
+// action picks what to tell Windows happened. It cannot distinguish a created
+// entry from a modified one, and Windows treats an unexpected create on an
+// existing name as a refresh, so create is the safe report.
+func (n *notifier) action(invalidation meta_cache.EntryInvalidation) uint32 {
+	if invalidation.Deleted || invalidation.Entry == nil {
+		// The entry is gone, so only WasDirectory still says what it was, and
+		// Windows watches directory and file removals through different
+		// filters.
+		if invalidation.WasDirectory {
+			return cgofuse.NOTIFY_RMDIR
+		}
+		return cgofuse.NOTIFY_UNLINK
+	}
+	if invalidation.Entry.GetIsDirectory() {
+		return cgofuse.NOTIFY_MKDIR
+	}
+	return cgofuse.NOTIFY_CREATE
+}