filer: maintain an inode->path index for filer.conf inode_index locations

The NFS gateway resolves filehandles by reversing an inode to its path,
which the deterministic hash inode cannot do. Bring back the reverse
index, but scoped: rows are written only for entries under a filer.conf
location with inode_index set, so with no such rule the filer does no
extra work. All filers pick the rule up through the usual filer.conf
reload, with no per-process flags to keep in sync.

Claude-Session: https://claude.ai/code/session_01AQDAwEYZsGsmGZKo3myZxB

weed/filer/filer.go

@@ -165,7 +165,10 @@ func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*m
 }
 
 func (f *Filer) SetStore(store FilerStore) (isFresh bool) {
-	f.Store = NewFilerStoreWrapper(store)
+	fsw := NewFilerStoreWrapper(store)
+	// f.FilerConf is swapped on filer.conf reloads, so resolve it per call.
+	fsw.filerConfFn = func() *FilerConf { return f.FilerConf }
+	f.Store = fsw
 
 	return f.setOrLoadFilerStoreSignature(store)
 }

weed/filer/filer_conf.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
@@ -230,9 +231,28 @@ func ClonePathConf(src *filer_pb.FilerConf_PathConf) *filer_pb.FilerConf_PathCon
 		Worm:                     src.Worm,
 		WormGracePeriodSeconds:   src.WormGracePeriodSeconds,
 		WormRetentionTimeSeconds: src.WormRetentionTimeSeconds,
+		InodeIndex:               src.InodeIndex,
 	}
 }
 
+// InodeIndexActiveUnder reports whether entries inside dirPath may carry inode
+// index rows: either dirPath itself is covered by an inode_index rule, or such
+// a rule applies to a subtree inside dirPath.
+func (fc *FilerConf) InodeIndexActiveUnder(dirPath string) bool {
+	if fc.MatchStorageRule(dirPath).InodeIndex {
+		return true
+	}
+	active := false
+	fc.rules.Walk(func(key []byte, value *filer_pb.FilerConf_PathConf) bool {
+		if value.InodeIndex && strings.HasPrefix(string(key), dirPath) {
+			active = true
+			return false
+		}
+		return true
+	})
+	return active
+}
+
 // ApplyBucketQuotaReadOnly sets read-only when usedSize exceeds quota and clears it
 // once back under, reporting whether the flag changed. A non-positive quota is left
 // untouched so a manually locked bucket is never reopened.
@@ -292,6 +312,7 @@ func mergePathConf(a, b *filer_pb.FilerConf_PathConf) {
 	a.DataNode = util.Nvl(b.DataNode, a.DataNode)
 	a.DisableChunkDeletion = b.DisableChunkDeletion || a.DisableChunkDeletion
 	a.Worm = b.Worm || a.Worm
+	a.InodeIndex = b.InodeIndex || a.InodeIndex
 	if b.WormRetentionTimeSeconds > 0 {
 		a.WormRetentionTimeSeconds = b.WormRetentionTimeSeconds
 	}

weed/filer/filer_conf_test.go

@@ -71,6 +71,7 @@ func TestClonePathConf(t *testing.T) {
 		Worm:                     true,
 		WormGracePeriodSeconds:   3600,
 		WormRetentionTimeSeconds: 86400,
+		InodeIndex:               true,
 	}
 
 	clone := ClonePathConf(src)

weed/filer/filer_inode_index.go

@@ -0,0 +1,335 @@
+package filer
+
+import (
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"sort"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+const inodeIndexKeyPrefix = "filer.inode.path."
+const InodeIndexInitialGeneration uint64 = 1
+
+type inodeIndexEntry struct {
+	path  util.FullPath
+	inode uint64
+}
+
+// InodeIndexRecord is the inode→path reverse-lookup row consumed by the NFS
+// gateway to resolve filehandles. Rows are maintained only for paths covered
+// by an inode_index rule in filer.conf (see shell nfs.enable); everywhere else
+// the filer writes nothing.
+type InodeIndexRecord struct {
+	Generation uint64   `json:"generation,omitempty"`
+	Paths      []string `json:"paths,omitempty"`
+}
+
+func InodeIndexKey(inode uint64) []byte {
+	key := make([]byte, len(inodeIndexKeyPrefix)+8)
+	copy(key, inodeIndexKeyPrefix)
+	binary.BigEndian.PutUint64(key[len(inodeIndexKeyPrefix):], inode)
+	return key
+}
+
+func DecodeInodeIndexRecord(value []byte) (*InodeIndexRecord, error) {
+	if len(value) == 0 {
+		return &InodeIndexRecord{}, nil
+	}
+
+	// The first foundation slice stored the current path as raw bytes. Keep that
+	// format readable so existing records are transparently upgraded on write.
+	if value[0] != '{' {
+		record := &InodeIndexRecord{Generation: InodeIndexInitialGeneration}
+		record.AddPath(util.FullPath(value))
+		return record, nil
+	}
+
+	record := &InodeIndexRecord{}
+	if err := json.Unmarshal(value, record); err != nil {
+		return nil, err
+	}
+	record.normalize()
+	return record, nil
+}
+
+func (record *InodeIndexRecord) Encode() ([]byte, error) {
+	record.normalize()
+	return json.Marshal(record)
+}
+
+func (record *InodeIndexRecord) normalize() {
+	if len(record.Paths) == 0 {
+		return
+	}
+	if record.Generation == 0 {
+		record.Generation = InodeIndexInitialGeneration
+	}
+
+	sanitized := make([]string, 0, len(record.Paths))
+	for _, path := range record.Paths {
+		if path == "" {
+			continue
+		}
+		sanitized = append(sanitized, path)
+	}
+	if len(sanitized) == 0 {
+		record.Paths = nil
+		return
+	}
+
+	sort.Strings(sanitized)
+	deduped := sanitized[:1]
+	for _, path := range sanitized[1:] {
+		if path == deduped[len(deduped)-1] {
+			continue
+		}
+		deduped = append(deduped, path)
+	}
+	record.Paths = deduped
+}
+
+func (record *InodeIndexRecord) AddPath(path util.FullPath) bool {
+	if path == "" {
+		return false
+	}
+	record.normalize()
+	target := string(path)
+	index := sort.SearchStrings(record.Paths, target)
+	if index < len(record.Paths) && record.Paths[index] == target {
+		return false
+	}
+	record.Paths = append(record.Paths, "")
+	copy(record.Paths[index+1:], record.Paths[index:])
+	record.Paths[index] = target
+	return true
+}
+
+func (record *InodeIndexRecord) RemovePath(path util.FullPath) bool {
+	if len(record.Paths) == 0 || path == "" {
+		return false
+	}
+	record.normalize()
+	target := string(path)
+	index := sort.SearchStrings(record.Paths, target)
+	if index >= len(record.Paths) || record.Paths[index] != target {
+		return false
+	}
+	record.Paths = append(record.Paths[:index], record.Paths[index+1:]...)
+	if len(record.Paths) == 0 {
+		record.Paths = nil
+	}
+	return true
+}
+
+func (record *InodeIndexRecord) CanonicalPath() util.FullPath {
+	record.normalize()
+	if len(record.Paths) == 0 {
+		return ""
+	}
+	return util.FullPath(record.Paths[0])
+}
+
+func (record *InodeIndexRecord) FullPaths() []util.FullPath {
+	record.normalize()
+	if len(record.Paths) == 0 {
+		return nil
+	}
+	paths := make([]util.FullPath, 0, len(record.Paths))
+	for _, path := range record.Paths {
+		paths = append(paths, util.FullPath(path))
+	}
+	return paths
+}
+
+// inodeIndexInScope reports whether the entry at path gets an inode index row.
+// The index is opt-in per filer.conf inode_index location rules; with no
+// matching rule (the default) nothing is ever written.
+func (fsw *FilerStoreWrapper) inodeIndexInScope(path util.FullPath) bool {
+	if fsw.filerConfFn == nil {
+		return false
+	}
+	fc := fsw.filerConfFn()
+	if fc == nil {
+		return false
+	}
+	return fc.MatchStorageRule(string(path)).InodeIndex
+}
+
+// inodeIndexUnder reports whether any entry inside dirPath may carry an inode
+// index row, so DeleteFolderChildren can skip the housekeeping walk entirely
+// when the subtree does not intersect any inode_index rule.
+func (fsw *FilerStoreWrapper) inodeIndexUnder(dirPath util.FullPath) bool {
+	if fsw.filerConfFn == nil {
+		return false
+	}
+	fc := fsw.filerConfFn()
+	if fc == nil {
+		return false
+	}
+	return fc.InodeIndexActiveUnder(string(dirPath))
+}
+
+func (fsw *FilerStoreWrapper) lookupInodeIndex(ctx context.Context, inode uint64) (*InodeIndexRecord, error) {
+	if inode == 0 {
+		return nil, ErrKvNotFound
+	}
+
+	value, err := fsw.KvGet(ctx, InodeIndexKey(inode))
+	if err != nil {
+		return nil, err
+	}
+
+	return DecodeInodeIndexRecord(value)
+}
+
+func (fsw *FilerStoreWrapper) storeInodeIndex(ctx context.Context, path util.FullPath, inode uint64) error {
+	if inode == 0 || path == "" {
+		return nil
+	}
+
+	record, err := fsw.lookupInodeIndex(ctx, inode)
+	if err != nil {
+		if err != ErrKvNotFound {
+			return err
+		}
+		record = &InodeIndexRecord{Generation: InodeIndexInitialGeneration}
+	}
+	record.AddPath(path)
+
+	value, err := record.Encode()
+	if err != nil {
+		return err
+	}
+	return fsw.KvPut(ctx, InodeIndexKey(inode), value)
+}
+
+func (fsw *FilerStoreWrapper) lookupInodePath(ctx context.Context, inode uint64) (util.FullPath, error) {
+	record, err := fsw.lookupInodeIndex(ctx, inode)
+	if err != nil {
+		return "", err
+	}
+
+	path := record.CanonicalPath()
+	if path == "" {
+		return "", ErrKvNotFound
+	}
+	return path, nil
+}
+
+func (fsw *FilerStoreWrapper) lookupInodePaths(ctx context.Context, inode uint64) ([]util.FullPath, error) {
+	record, err := fsw.lookupInodeIndex(ctx, inode)
+	if err != nil {
+		return nil, err
+	}
+
+	paths := record.FullPaths()
+	if len(paths) == 0 {
+		return nil, ErrKvNotFound
+	}
+	return paths, nil
+}
+
+func (fsw *FilerStoreWrapper) removePathFromInodeIndex(ctx context.Context, path util.FullPath, inode uint64) error {
+	if inode == 0 || path == "" {
+		return nil
+	}
+
+	record, err := fsw.lookupInodeIndex(ctx, inode)
+	if err != nil {
+		if err == ErrKvNotFound {
+			return nil
+		}
+		return err
+	}
+
+	if !record.RemovePath(path) {
+		return nil
+	}
+	if len(record.Paths) == 0 {
+		return fsw.KvDelete(ctx, InodeIndexKey(inode))
+	}
+
+	value, err := record.Encode()
+	if err != nil {
+		return err
+	}
+	return fsw.KvPut(ctx, InodeIndexKey(inode), value)
+}
+
+func (fsw *FilerStoreWrapper) collectInodeIndexEntries(ctx context.Context, dirPath util.FullPath) ([]inodeIndexEntry, error) {
+	if !fsw.inodeIndexUnder(dirPath) {
+		return nil, nil
+	}
+	// Honor caller cancellation during the walk: a DeleteFolderChildren on a
+	// pathological directory could otherwise loop indefinitely gathering
+	// entries even after the client has given up, turning into a DoS vector.
+	// If the walk is aborted, the caller treats the index cleanup as
+	// best-effort and drops the partial result.
+	var collected []inodeIndexEntry
+	if err := fsw.collectInodeIndexEntriesRecursive(ctx, dirPath, &collected); err != nil {
+		return nil, err
+	}
+	return collected, nil
+}
+
+func (fsw *FilerStoreWrapper) collectInodeIndexEntriesRecursive(ctx context.Context, dirPath util.FullPath, collected *[]inodeIndexEntry) error {
+	actualStore := fsw.getActualStore(dirPath + "/")
+
+	lastFileName := ""
+	includeStartFile := false
+	for {
+		page := make([]*Entry, 0, PaginationSize)
+		nextLastFileName, err := actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, includeStartFile, PaginationSize, func(entry *Entry) (bool, error) {
+			page = append(page, entry)
+			return true, nil
+		})
+		if err != nil {
+			return err
+		}
+
+		for _, entry := range page {
+			if entry.Attr.Inode != 0 {
+				*collected = append(*collected, inodeIndexEntry{path: entry.FullPath, inode: entry.Attr.Inode})
+			}
+			if entry.IsDirectory() {
+				if err := fsw.collectInodeIndexEntriesRecursive(ctx, entry.FullPath, collected); err != nil {
+					return err
+				}
+			}
+		}
+
+		if len(page) < PaginationSize {
+			return nil
+		}
+		lastFileName = nextLastFileName
+		includeStartFile = false
+	}
+}
+
+// recordInodeIndexWrite updates the inode→path secondary index after the
+// primary store mutation has already succeeded. The index is best-effort: a
+// failure here must not surface as an operation error, because the caller
+// would then observe a failed create/update even though the entry was
+// persisted, and a retry cannot heal the index (DeleteEntry exits early once
+// the entry is gone). We log and let later writes rebuild the record.
+func (fsw *FilerStoreWrapper) recordInodeIndexWrite(ctx context.Context, op string, path util.FullPath, inode uint64) {
+	if inode == 0 || path == "" || !fsw.inodeIndexInScope(path) {
+		return
+	}
+	if err := fsw.storeInodeIndex(ctx, path, inode); err != nil {
+		glog.WarningfCtx(ctx, "%s: update inode index for %s (inode %d): %v", op, path, inode, err)
+	}
+}
+
+// recordInodeIndexRemoval mirrors recordInodeIndexWrite for removals.
+func (fsw *FilerStoreWrapper) recordInodeIndexRemoval(ctx context.Context, op string, path util.FullPath, inode uint64) {
+	if inode == 0 || path == "" || !fsw.inodeIndexInScope(path) {
+		return
+	}
+	if err := fsw.removePathFromInodeIndex(ctx, path, inode); err != nil {
+		glog.WarningfCtx(ctx, "%s: clear inode index for %s (inode %d): %v", op, path, inode, err)
+	}
+}

weed/filer/filer_inode_index_test.go

@@ -0,0 +1,301 @@
+package filer
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func inodeIndexConf(prefixes ...string) *FilerConf {
+	fc := NewFilerConf()
+	for _, prefix := range prefixes {
+		_ = fc.SetLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: prefix, InodeIndex: true})
+	}
+	return fc
+}
+
+func newInodeIndexWrapper(prefixes ...string) (*FilerStoreWrapper, *stubFilerStore) {
+	store := newStubFilerStore()
+	wrapper := NewFilerStoreWrapper(store)
+	fc := inodeIndexConf(prefixes...)
+	wrapper.filerConfFn = func() *FilerConf { return fc }
+	return wrapper, store
+}
+
+func countInodeIndexRows(store *stubFilerStore) int {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	count := 0
+	for key := range store.kv {
+		if strings.HasPrefix(key, inodeIndexKeyPrefix) {
+			count++
+		}
+	}
+	return count
+}
+
+func TestFilerStoreWrapperMaintainsInodeIndexLifecycle(t *testing.T) {
+	wrapper, _ := newInodeIndexWrapper("/")
+	ctx := context.Background()
+
+	created := &Entry{
+		FullPath: util.FullPath("/docs/report.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: 42,
+		},
+	}
+
+	require.NoError(t, wrapper.InsertEntry(ctx, created))
+	path, err := wrapper.lookupInodePath(ctx, created.Attr.Inode)
+	require.NoError(t, err)
+	assert.Equal(t, created.FullPath, path)
+	paths, err := wrapper.lookupInodePaths(ctx, created.Attr.Inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{created.FullPath}, paths)
+	record, err := wrapper.lookupInodeIndex(ctx, created.Attr.Inode)
+	require.NoError(t, err)
+	assert.Equal(t, InodeIndexInitialGeneration, record.Generation)
+
+	updated := &Entry{
+		FullPath: util.FullPath("/docs/report.txt"),
+		Attr: Attr{
+			Mode:  0o600,
+			Inode: 42,
+		},
+	}
+	require.NoError(t, wrapper.UpdateEntry(ctx, updated))
+	path, err = wrapper.lookupInodePath(ctx, updated.Attr.Inode)
+	require.NoError(t, err)
+	assert.Equal(t, updated.FullPath, path)
+
+	require.NoError(t, wrapper.DeleteEntry(ctx, created.FullPath))
+	_, err = wrapper.lookupInodePath(ctx, created.Attr.Inode)
+	require.ErrorIs(t, err, ErrKvNotFound)
+}
+
+func TestFilerStoreWrapperMaintainsMultiplePathsPerInode(t *testing.T) {
+	wrapper, _ := newInodeIndexWrapper("/")
+	ctx := context.Background()
+	inode := uint64(88)
+	hardLinkId := NewHardLinkId()
+
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/links/b.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: inode,
+		},
+		HardLinkId:      hardLinkId,
+		HardLinkCounter: 2,
+	}))
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/links/a.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: inode,
+		},
+		HardLinkId:      hardLinkId,
+		HardLinkCounter: 2,
+	}))
+
+	paths, err := wrapper.lookupInodePaths(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{"/links/a.txt", "/links/b.txt"}, paths)
+	record, err := wrapper.lookupInodeIndex(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, InodeIndexInitialGeneration, record.Generation)
+
+	path, err := wrapper.lookupInodePath(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, util.FullPath("/links/a.txt"), path)
+
+	require.NoError(t, wrapper.DeleteEntry(ctx, util.FullPath("/links/a.txt")))
+
+	paths, err = wrapper.lookupInodePaths(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{"/links/b.txt"}, paths)
+
+	path, err = wrapper.lookupInodePath(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, util.FullPath("/links/b.txt"), path)
+}
+
+func TestFilerStoreWrapperUpgradesLegacySinglePathInodeIndexRecords(t *testing.T) {
+	wrapper := NewFilerStoreWrapper(newStubFilerStore())
+	ctx := context.Background()
+	inode := uint64(91)
+
+	require.NoError(t, wrapper.KvPut(ctx, InodeIndexKey(inode), []byte("/legacy/path.txt")))
+
+	path, err := wrapper.lookupInodePath(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, util.FullPath("/legacy/path.txt"), path)
+
+	paths, err := wrapper.lookupInodePaths(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{"/legacy/path.txt"}, paths)
+
+	require.NoError(t, wrapper.storeInodeIndex(ctx, util.FullPath("/legacy/second.txt"), inode))
+
+	paths, err = wrapper.lookupInodePaths(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{"/legacy/path.txt", "/legacy/second.txt"}, paths)
+
+	value, err := wrapper.KvGet(ctx, InodeIndexKey(inode))
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"generation":1,"paths":["/legacy/path.txt","/legacy/second.txt"]}`, string(value))
+}
+
+func TestFilerStoreWrapperKeepsInodeIndexWhenDeleteArrivesAfterRenameInsert(t *testing.T) {
+	wrapper, _ := newInodeIndexWrapper("/")
+	ctx := context.Background()
+	inode := uint64(77)
+
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/old/name.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: inode,
+		},
+	}))
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/new/name.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: inode,
+		},
+	}))
+	require.NoError(t, wrapper.DeleteEntry(ctx, util.FullPath("/old/name.txt")))
+
+	path, err := wrapper.lookupInodePath(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, util.FullPath("/new/name.txt"), path)
+
+	paths, err := wrapper.lookupInodePaths(ctx, inode)
+	require.NoError(t, err)
+	assert.Equal(t, []util.FullPath{"/new/name.txt"}, paths)
+}
+
+func TestInodeIndexDisabledWritesNothing(t *testing.T) {
+	// No filer.conf wired at all, and a conf with no inode_index rules: both
+	// must leave the KV store free of index rows.
+	t.Run("no conf", func(t *testing.T) {
+		store := newStubFilerStore()
+		runInodeIndexDisabledScenario(t, NewFilerStoreWrapper(store), store)
+	})
+	t.Run("conf without rules", func(t *testing.T) {
+		wrapper, store := newInodeIndexWrapper()
+		runInodeIndexDisabledScenario(t, wrapper, store)
+	})
+}
+
+func runInodeIndexDisabledScenario(t *testing.T, wrapper *FilerStoreWrapper, store *stubFilerStore) {
+	ctx := context.Background()
+	entry := &Entry{
+		FullPath: util.FullPath("/docs/report.txt"),
+		Attr: Attr{
+			Mode:  0o644,
+			Inode: 42,
+		},
+	}
+	require.NoError(t, wrapper.InsertEntry(ctx, entry))
+	require.NoError(t, wrapper.UpdateEntry(ctx, entry))
+	assert.Zero(t, countInodeIndexRows(store))
+	require.NoError(t, wrapper.DeleteEntry(ctx, entry.FullPath))
+	assert.Zero(t, countInodeIndexRows(store))
+}
+
+func TestInodeIndexScopedToConfiguredPrefixes(t *testing.T) {
+	wrapper, store := newInodeIndexWrapper("/exports")
+	ctx := context.Background()
+
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/exports/docs/in.txt"),
+		Attr:     Attr{Mode: 0o644, Inode: 1001},
+	}))
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/other/out.txt"),
+		Attr:     Attr{Mode: 0o644, Inode: 1002},
+	}))
+
+	assert.Equal(t, 1, countInodeIndexRows(store))
+	path, err := wrapper.lookupInodePath(ctx, 1001)
+	require.NoError(t, err)
+	assert.Equal(t, util.FullPath("/exports/docs/in.txt"), path)
+	_, err = wrapper.lookupInodePath(ctx, 1002)
+	require.ErrorIs(t, err, ErrKvNotFound)
+
+	require.NoError(t, wrapper.InsertEntry(ctx, &Entry{
+		FullPath: util.FullPath("/exports/docs"),
+		Attr:     Attr{Mode: os.ModeDir | 0o755, Inode: 1000},
+	}))
+	require.NoError(t, wrapper.DeleteFolderChildren(ctx, util.FullPath("/exports")))
+	_, err = wrapper.lookupInodePath(ctx, 1001)
+	require.ErrorIs(t, err, ErrKvNotFound)
+}
+
+func TestRecursiveDeleteRemovesDescendantInodeIndexes(t *testing.T) {
+	f, _ := newTestFilerWithStubStore()
+	fc := inodeIndexConf("/tree")
+	f.Store.(*FilerStoreWrapper).filerConfFn = func() *FilerConf { return fc }
+	ctx := context.Background()
+
+	entries := []*Entry{
+		{
+			FullPath: util.FullPath("/tree"),
+			Attr: Attr{
+				Mode:  os.ModeDir | 0o755,
+				Inode: 100,
+			},
+		},
+		{
+			FullPath: util.FullPath("/tree/file.txt"),
+			Attr: Attr{
+				Mode:  0o644,
+				Inode: 101,
+			},
+		},
+		{
+			FullPath: util.FullPath("/tree/subdir"),
+			Attr: Attr{
+				Mode:  os.ModeDir | 0o755,
+				Inode: 102,
+			},
+		},
+		{
+			FullPath: util.FullPath("/tree/subdir/nested.txt"),
+			Attr: Attr{
+				Mode:  0o644,
+				Inode: 103,
+			},
+		},
+	}
+
+	for _, entry := range entries {
+		require.NoError(t, f.Store.InsertEntry(ctx, entry))
+	}
+
+	require.NoError(t, f.DeleteEntryMetaAndData(ctx, util.FullPath("/tree"), true, false, false, false, nil, 0))
+
+	for _, inode := range []uint64{100, 101, 102, 103} {
+		_, err := f.Store.(*FilerStoreWrapper).lookupInodePath(ctx, inode)
+		require.ErrorIs(t, err, ErrKvNotFound)
+	}
+}
+
+func TestInodeIndexActiveUnder(t *testing.T) {
+	fc := inodeIndexConf("/exports")
+	assert.True(t, fc.InodeIndexActiveUnder("/"))
+	assert.True(t, fc.InodeIndexActiveUnder("/exports"))
+	assert.True(t, fc.InodeIndexActiveUnder("/exports/sub"))
+	assert.False(t, fc.InodeIndexActiveUnder("/other"))
+
+	assert.False(t, NewFilerConf().InodeIndexActiveUnder("/"))
+}

weed/filer/filerstore_wrapper.go

@@ -37,7 +37,8 @@ type FilerStoreWrapper struct {
 	defaultStore         FilerStore
 	pathToStore          ptrie.Trie[string]
 	storeIdToStore       map[string]FilerStore
-	hasPathSpecificStore bool // fast check to skip MatchPrefix when no path-specific stores
+	hasPathSpecificStore bool              // fast check to skip MatchPrefix when no path-specific stores
+	filerConfFn          func() *FilerConf // set by Filer.SetStore; nil disables the inode index
 }
 
 func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
@@ -151,7 +152,11 @@ func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) err
 		return err
 	}
 
-	return actualStore.InsertEntry(ctx, entry)
+	if err := actualStore.InsertEntry(ctx, entry); err != nil {
+		return err
+	}
+	fsw.recordInodeIndexWrite(ctx, "InsertEntry", entry.FullPath, entry.Attr.Inode)
+	return nil
 }
 
 // InsertEntryKnownAbsent skips the pre-insert FindEntry path when the caller has
@@ -179,7 +184,11 @@ func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry
 		}
 	}
 
-	return actualStore.InsertEntry(ctx, entry)
+	if err := actualStore.InsertEntry(ctx, entry); err != nil {
+		return err
+	}
+	fsw.recordInodeIndexWrite(ctx, "InsertEntryKnownAbsent", entry.FullPath, entry.Attr.Inode)
+	return nil
 }
 
 func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
@@ -206,7 +215,11 @@ func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) err
 		return err
 	}
 
-	return actualStore.UpdateEntry(ctx, entry)
+	if err := actualStore.UpdateEntry(ctx, entry); err != nil {
+		return err
+	}
+	fsw.recordInodeIndexWrite(ctx, "UpdateEntry", entry.FullPath, entry.Attr.Inode)
+	return nil
 }
 
 func normalizeEntryMimeForStore(entry *Entry) {
@@ -258,6 +271,8 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath)
 	if findErr == filer_pb.ErrNotFound || existingEntry == nil {
 		return nil
 	}
+	inode := existingEntry.Attr.Inode
+	fullPath := existingEntry.FullPath
 	if len(existingEntry.HardLinkId) != 0 {
 		// remove hard link
 		op := ctx.Value("OP")
@@ -272,7 +287,11 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath)
 		}
 	}
 
-	return actualStore.DeleteEntry(ctx, fp)
+	if err := actualStore.DeleteEntry(ctx, fp); err != nil {
+		return err
+	}
+	fsw.recordInodeIndexRemoval(ctx, "DeleteEntry", fullPath, inode)
+	return nil
 }
 
 func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
@@ -280,6 +299,8 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry
 		return err
 	}
 	ctx = context.WithoutCancel(ctx)
+	fullPath := existingEntry.FullPath
+	inode := existingEntry.Attr.Inode
 	actualStore := fsw.getActualStore(existingEntry.FullPath)
 	stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
 	start := time.Now()
@@ -302,7 +323,11 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry
 		}
 	}
 
-	return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
+	if err := actualStore.DeleteEntry(ctx, existingEntry.FullPath); err != nil {
+		return err
+	}
+	fsw.recordInodeIndexRemoval(ctx, "DeleteOneEntry", fullPath, inode)
+	return nil
 }
 
 func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
@@ -317,7 +342,20 @@ func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.
 		stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
 	}()
 
-	return actualStore.DeleteFolderChildren(ctx, fp)
+	collected, collectErr := fsw.collectInodeIndexEntries(ctx, fp)
+	if collectErr != nil {
+		// Index collection is best-effort: a failure here only prevents inode
+		// index housekeeping, not the directory removal itself.
+		glog.WarningfCtx(ctx, "collectInodeIndexEntries %s: %v; deleting folder children without index cleanup", fp, collectErr)
+		collected = nil
+	}
+	if err := actualStore.DeleteFolderChildren(ctx, fp); err != nil {
+		return err
+	}
+	for _, entry := range collected {
+		fsw.recordInodeIndexRemoval(ctx, "DeleteFolderChildren", entry.path, entry.inode)
+	}
+	return nil
 }
 
 func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {