shell: nfs.enable / nfs.disable

One command to make a filer path exportable over NFS: nfs.enable marks
the path with inode_index in filer.conf and backfills index rows for
the entries already under it, assigning inodes where missing, so
existing files resolve by filehandle right away. nfs.disable clears
the mark and deletes the rows, leaving no index data behind.

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

other/java/client/src/main/proto/filer.proto

@@ -527,6 +527,7 @@ message AssignVolumeResponse {
     string replication = 7;
     string error = 8;
     Location location = 9;
+    repeated Location replicas = 10;
 }
 
 message LookupVolumeRequest {
@@ -731,6 +732,7 @@ message FilerConf {
         bool worm = 14;
         uint64 worm_grace_period_seconds = 15;
         uint64 worm_retention_time_seconds = 16;
+        bool inode_index = 17;
     }
     repeated PathConf locations = 2;
 }

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

weed/pb/filer.proto

@@ -732,6 +732,7 @@ message FilerConf {
         bool worm = 14;
         uint64 worm_grace_period_seconds = 15;
         uint64 worm_retention_time_seconds = 16;
+        bool inode_index = 17;
     }
     repeated PathConf locations = 2;
 }

weed/pb/filer_pb/filer.pb.go

@@ -6645,6 +6645,7 @@ type FilerConf_PathConf struct {
 	Worm                     bool                   `protobuf:"varint,14,opt,name=worm,proto3" json:"worm,omitempty"`
 	WormGracePeriodSeconds   uint64                 `protobuf:"varint,15,opt,name=worm_grace_period_seconds,json=wormGracePeriodSeconds,proto3" json:"worm_grace_period_seconds,omitempty"`
 	WormRetentionTimeSeconds uint64                 `protobuf:"varint,16,opt,name=worm_retention_time_seconds,json=wormRetentionTimeSeconds,proto3" json:"worm_retention_time_seconds,omitempty"`
+	InodeIndex               bool                   `protobuf:"varint,17,opt,name=inode_index,json=inodeIndex,proto3" json:"inode_index,omitempty"`
 	unknownFields            protoimpl.UnknownFields
 	sizeCache                protoimpl.SizeCache
 }
@@ -6791,6 +6792,13 @@ func (x *FilerConf_PathConf) GetWormRetentionTimeSeconds() uint64 {
 	return 0
 }
 
+func (x *FilerConf_PathConf) GetInodeIndex() bool {
+	if x != nil {
+		return x.InodeIndex
+	}
+	return false
+}
+
 var File_filer_proto protoreflect.FileDescriptor
 
 const file_filer_proto_rawDesc = "" +
@@ -7253,10 +7261,10 @@ const file_filer_proto_rawDesc = "" +
 	"\x03key\x18\x01 \x01(\fR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\fR\x05value\"%\n" +
 	"\rKvPutResponse\x12\x14\n" +
-	"\x05error\x18\x01 \x01(\tR\x05error\"\xb2\x05\n" +
+	"\x05error\x18\x01 \x01(\tR\x05error\"\xd3\x05\n" +
 	"\tFilerConf\x12\x18\n" +
 	"\aversion\x18\x01 \x01(\x05R\aversion\x12:\n" +
-	"\tlocations\x18\x02 \x03(\v2\x1c.filer_pb.FilerConf.PathConfR\tlocations\x1a\xce\x04\n" +
+	"\tlocations\x18\x02 \x03(\v2\x1c.filer_pb.FilerConf.PathConfR\tlocations\x1a\xef\x04\n" +
 	"\bPathConf\x12'\n" +
 	"\x0flocation_prefix\x18\x01 \x01(\tR\x0elocationPrefix\x12\x1e\n" +
 	"\n" +
@@ -7277,7 +7285,9 @@ const file_filer_proto_rawDesc = "" +
 	"\x16disable_chunk_deletion\x18\r \x01(\bR\x14disableChunkDeletion\x12\x12\n" +
 	"\x04worm\x18\x0e \x01(\bR\x04worm\x129\n" +
 	"\x19worm_grace_period_seconds\x18\x0f \x01(\x04R\x16wormGracePeriodSeconds\x12=\n" +
-	"\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSeconds\"\xba\x01\n" +
+	"\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSeconds\x12\x1f\n" +
+	"\vinode_index\x18\x11 \x01(\bR\n" +
+	"inodeIndex\"\xba\x01\n" +
 	"&CacheRemoteObjectToLocalClusterRequest\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" +
 	"\x04name\x18\x02 \x01(\tR\x04name\x12+\n" +

weed/shell/command_nfs_disable.go

@@ -0,0 +1,177 @@
+package shell
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"sync/atomic"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/protobuf/proto"
+)
+
+func init() {
+	Commands = append(Commands, &commandNfsDisable{})
+}
+
+type commandNfsDisable struct {
+}
+
+func (c *commandNfsDisable) Name() string {
+	return "nfs.disable"
+}
+
+func (c *commandNfsDisable) Help() string {
+	return `disable NFS export support for a filer path
+
+	# preview the filer.conf change
+	nfs.disable -path=/exports
+
+	# persist it and delete the inode index rows under the path
+	nfs.disable -path=/exports -apply
+
+	Clears the inode_index mark that nfs.enable set on the path, then
+	walks the entries under it and removes their inode->path index rows,
+	so disabling leaves no index data behind. Pass -keepIndex to skip
+	the cleanup walk.
+`
+}
+
+func (c *commandNfsDisable) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandNfsDisable) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	nfsDisableCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	path := nfsDisableCommand.String("path", "", "filer path that nfs.enable was applied to")
+	apply := nfsDisableCommand.Bool("apply", false, "update filer.conf and delete the inode index rows")
+	keepIndex := nfsDisableCommand.Bool("keepIndex", false, "keep existing inode index rows")
+	if err = nfsDisableCommand.Parse(args); err != nil {
+		return nil
+	}
+	if !strings.HasPrefix(*path, "/") {
+		return fmt.Errorf("-path must be an absolute filer path")
+	}
+
+	fc, err := filer.ReadFilerConf(commandEnv.option.FilerAddress, commandEnv.option.GrpcDialOption, commandEnv.MasterClient)
+	if err != nil {
+		return err
+	}
+
+	if locConf, found := fc.GetLocationConf(*path); found && locConf.InodeIndex {
+		locConf = filer.ClonePathConf(locConf)
+		locConf.InodeIndex = false
+		if proto.Equal(locConf, &filer_pb.FilerConf_PathConf{LocationPrefix: *path}) {
+			fc.DeleteLocationConf(*path)
+		} else if err = fc.SetLocationConf(locConf); err != nil {
+			return err
+		}
+	} else {
+		fmt.Fprintf(writer, "inode_index is not set on %s in filer.conf\n", *path)
+	}
+
+	var buf bytes.Buffer
+	fc.ToText(&buf)
+	fmt.Fprint(writer, buf.String())
+	fmt.Fprintln(writer)
+
+	if !*apply {
+		infoAboutSimulationMode(writer, *apply, "-apply")
+		return nil
+	}
+
+	if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+	}); err != nil && err != filer_pb.ErrNotFound {
+		return err
+	}
+
+	if *keepIndex {
+		return nil
+	}
+	return cleanupInodeIndex(commandEnv, writer, util.FullPath(*path))
+}
+
+func cleanupInodeIndex(commandEnv *CommandEnv, writer io.Writer, exportPath util.FullPath) error {
+	var cleared atomic.Int64
+
+	if exportPath != "/" {
+		rootEntry, err := lookupPathEntry(commandEnv, exportPath)
+		if err == filer_pb.ErrNotFound {
+			fmt.Fprintf(writer, "%s does not exist; nothing to clean up\n", exportPath)
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if err := removeInodeIndexRow(commandEnv, exportPath, rootEntry.GetAttributes().GetInode(), &cleared); err != nil {
+			return err
+		}
+	}
+
+	err := filer_pb.TraverseBfs(context.Background(), commandEnv, exportPath, func(parentPath util.FullPath, entry *filer_pb.Entry) error {
+		fullPath := util.NewFullPath(string(parentPath), entry.Name)
+		if err := removeInodeIndexRow(commandEnv, fullPath, entry.GetAttributes().GetInode(), &cleared); err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("clean up inode index under %s: %v", exportPath, err)
+	}
+
+	fmt.Fprintf(writer, "cleared %d inode index rows\n", cleared.Load())
+	return nil
+}
+
+func removeInodeIndexRow(commandEnv *CommandEnv, fullPath util.FullPath, inode uint64, cleared *atomic.Int64) error {
+	if inode == 0 {
+		return nil
+	}
+	key := filer.InodeIndexKey(inode)
+	err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, kvErr := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: key})
+		if kvErr != nil {
+			return kvErr
+		}
+		if resp.GetError() != "" {
+			return errors.New(resp.GetError())
+		}
+		if len(resp.GetValue()) == 0 {
+			return nil
+		}
+		record, decodeErr := filer.DecodeInodeIndexRecord(resp.GetValue())
+		if decodeErr != nil || !record.RemovePath(fullPath) {
+			return nil
+		}
+		// A hard link can keep paths outside the export; only drop the row
+		// once no path remains. An empty KvPut value deletes the key.
+		var value []byte
+		if len(record.Paths) > 0 {
+			var encodeErr error
+			if value, encodeErr = record.Encode(); encodeErr != nil {
+				return encodeErr
+			}
+		}
+		putResp, putErr := client.KvPut(context.Background(), &filer_pb.KvPutRequest{Key: key, Value: value})
+		if putErr != nil {
+			return putErr
+		}
+		if putResp.GetError() != "" {
+			return errors.New(putResp.GetError())
+		}
+		cleared.Add(1)
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("clear inode %d for %s: %v", inode, fullPath, err)
+	}
+	return nil
+}

weed/shell/command_nfs_enable.go

@@ -0,0 +1,219 @@
+package shell
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+	"sync/atomic"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func init() {
+	Commands = append(Commands, &commandNfsEnable{})
+}
+
+type commandNfsEnable struct {
+}
+
+func (c *commandNfsEnable) Name() string {
+	return "nfs.enable"
+}
+
+func (c *commandNfsEnable) Help() string {
+	return `enable NFS export support for a filer path
+
+	# preview the filer.conf change
+	nfs.enable -path=/exports
+
+	# persist it and backfill the index for existing entries
+	nfs.enable -path=/exports -apply
+
+	The weed nfs gateway resolves NFS filehandles through an inode->path
+	index. This command marks the path with inode_index in filer.conf, so
+	every filer maintains index rows for entries under it (no restarts
+	needed), then walks the existing entries and writes their rows so
+	pre-existing files are immediately resolvable. Reverse with nfs.disable.
+`
+}
+
+func (c *commandNfsEnable) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandNfsEnable) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+
+	nfsEnableCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	path := nfsEnableCommand.String("path", "", "filer path to export over NFS")
+	apply := nfsEnableCommand.Bool("apply", false, "update filer.conf and backfill the inode index")
+	if err = nfsEnableCommand.Parse(args); err != nil {
+		return nil
+	}
+	if !strings.HasPrefix(*path, "/") {
+		return fmt.Errorf("-path must be an absolute filer path")
+	}
+
+	fc, err := filer.ReadFilerConf(commandEnv.option.FilerAddress, commandEnv.option.GrpcDialOption, commandEnv.MasterClient)
+	if err != nil {
+		return err
+	}
+	if err = fc.AddLocationConf(&filer_pb.FilerConf_PathConf{
+		LocationPrefix: *path,
+		InodeIndex:     true,
+	}); err != nil {
+		return err
+	}
+
+	var buf bytes.Buffer
+	fc.ToText(&buf)
+	fmt.Fprint(writer, buf.String())
+	fmt.Fprintln(writer)
+
+	if !*apply {
+		infoAboutSimulationMode(writer, *apply, "-apply")
+		return nil
+	}
+
+	if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes())
+	}); err != nil && err != filer_pb.ErrNotFound {
+		return err
+	}
+
+	return backfillInodeIndex(commandEnv, writer, util.FullPath(*path))
+}
+
+func backfillInodeIndex(commandEnv *CommandEnv, writer io.Writer, exportPath util.FullPath) error {
+	var indexed, assigned atomic.Int64
+
+	if exportPath != "/" {
+		rootEntry, err := lookupPathEntry(commandEnv, exportPath)
+		if err == filer_pb.ErrNotFound {
+			fmt.Fprintf(writer, "%s does not exist yet; nothing to backfill\n", exportPath)
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		changed, err := backfillEntryInodeIndex(commandEnv, exportPath, rootEntry, &assigned)
+		if err != nil {
+			return err
+		}
+		if changed {
+			indexed.Add(1)
+		}
+	}
+
+	err := filer_pb.TraverseBfs(context.Background(), commandEnv, exportPath, func(parentPath util.FullPath, entry *filer_pb.Entry) error {
+		fullPath := util.NewFullPath(string(parentPath), entry.Name)
+		changed, err := backfillEntryInodeIndex(commandEnv, fullPath, entry, &assigned)
+		if err != nil {
+			return err
+		}
+		if changed {
+			if count := indexed.Add(1); count%10000 == 0 {
+				fmt.Fprintf(writer, "indexed %d entries ...\n", count)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("backfill inode index under %s: %v", exportPath, err)
+	}
+
+	fmt.Fprintf(writer, "indexed %d entries, assigned %d missing inodes\n", indexed.Load(), assigned.Load())
+	return nil
+}
+
+func backfillEntryInodeIndex(commandEnv *CommandEnv, fullPath util.FullPath, entry *filer_pb.Entry, assigned *atomic.Int64) (changed bool, err error) {
+	inode := entry.GetAttributes().GetInode()
+	if inode == 0 {
+		// Entry predates filer-assigned inodes: re-save it so the filer
+		// assigns and persists one, then read the assignment back.
+		dir, name := fullPath.DirAndName()
+		if err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+			if _, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
+				Directory: dir,
+				Entry:     entry,
+			}); updateErr != nil {
+				return updateErr
+			}
+			resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+				Directory: dir,
+				Name:      name,
+			})
+			if lookupErr != nil {
+				return lookupErr
+			}
+			inode = resp.Entry.GetAttributes().GetInode()
+			return nil
+		}); err != nil {
+			return false, fmt.Errorf("assign inode for %s: %v", fullPath, err)
+		}
+		if inode == 0 {
+			return false, fmt.Errorf("filer did not assign an inode for %s", fullPath)
+		}
+		assigned.Add(1)
+	}
+	return upsertInodeIndexRow(commandEnv, fullPath, inode)
+}
+
+func lookupPathEntry(commandEnv *CommandEnv, fullPath util.FullPath) (entry *filer_pb.Entry, err error) {
+	dir, name := fullPath.DirAndName()
+	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if lookupErr != nil {
+			return lookupErr
+		}
+		entry = resp.Entry
+		return nil
+	})
+	return
+}
+
+func upsertInodeIndexRow(commandEnv *CommandEnv, fullPath util.FullPath, inode uint64) (changed bool, err error) {
+	key := filer.InodeIndexKey(inode)
+	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, kvErr := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: key})
+		if kvErr != nil {
+			return kvErr
+		}
+		if resp.GetError() != "" {
+			return errors.New(resp.GetError())
+		}
+		record, decodeErr := filer.DecodeInodeIndexRecord(resp.GetValue())
+		if decodeErr != nil {
+			// Unreadable row: rebuild it from this path.
+			record = &filer.InodeIndexRecord{}
+		}
+		if !record.AddPath(fullPath) {
+			return nil
+		}
+		value, encodeErr := record.Encode()
+		if encodeErr != nil {
+			return encodeErr
+		}
+		putResp, putErr := client.KvPut(context.Background(), &filer_pb.KvPutRequest{Key: key, Value: value})
+		if putErr != nil {
+			return putErr
+		}
+		if putResp.GetError() != "" {
+			return errors.New(putResp.GetError())
+		}
+		changed = true
+		return nil
+	})
+	if err != nil {
+		return false, fmt.Errorf("index inode %d for %s: %v", inode, fullPath, err)
+	}
+	return changed, nil
+}