Incremental re-indexing over full re-indexes

client/go.mod

@@ -11,6 +11,7 @@ require (
 	friendnet.org/updater v0.0.0
 	friendnet.org/upnp v0.0.0
 	friendnet.org/webui v0.0.0
+	github.com/fsnotify/fsnotify v1.10.1
 	github.com/google/uuid v1.6.0
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c
 	github.com/quic-go/quic-go v0.61.0
@@ -22,7 +23,6 @@ require (
 
 require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
-	github.com/fsnotify/fsnotify v1.10.1 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/ncruces/go-strftime v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect

client/go.sum

@@ -23,38 +23,34 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmd
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw=
-github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
+github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0=
+github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk=
+github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA=
 github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=
 go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
-golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts=
-golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos=
+golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
 golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
 golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa h1:Zt3DZoOFFYkKhDT3v7Lm9FDMEV06GpzjG2jrqW+QTE0=
 golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA=
-golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
-golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
-golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60=
-golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM=
+golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
+golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
+golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
 golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
-golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
-golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
-golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
 golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
-golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
 golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
-golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
-golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
+golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
+golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

client/share/manager.go

@@ -5,10 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"os"
 	"strings"
 	"sync"
-	"syscall"
 	"time"
 
 	"friendnet.org/client/storage"
@@ -31,6 +29,9 @@ var ErrTooManyFiles = errors.New("too many files in share, indexing canceled")
 // ErrInvalidShareName is returned when trying to create a share with an invalid name.
 var ErrInvalidShareName = errors.New("invalid share name")
 
+// ErrWrongShare is returned when the wrong share is used in reference to an operation involving indices.
+var ErrWrongShare = errors.New("wrong share referenced in index operation")
+
 type shareData struct {
 	share       Share
 	record      storage.ShareRecord
@@ -75,20 +76,6 @@ func NewManager(
 		return nil, fmt.Errorf(`failed to get share records for server %q: %w`, serverUuid, err)
 	}
 
-	shareMap := make(map[string]*shareData, len(records))
-	for _, record := range records {
-		var share Share
-		share, err = NewDirShare(
-			record.Name,
-			record.Path.String(),
-			record.FollowLinks,
-		)
-		shareMap[record.Name] = &shareData{
-			share:  share,
-			record: record,
-		}
-	}
-
 	m := &Manager{
 		ctx:       ctx,
 		ctxCancel: ctxCancel,
@@ -98,14 +85,33 @@ func NewManager(
 		serverUuid: serverUuid,
 		storage:    storage,
 
-		shareMap: shareMap,
-
 		indexerInterval:         1 * time.Hour,
 		indexingShares:          make(map[string]struct{}),
 		indexerMaxFiles:         1_000_000,
 		orphanedIndexGcInterval: 10 * time.Minute,
 	}
 
+	m.shareMap = make(map[string]*shareData, len(records))
+	for _, record := range records {
+		var share Share
+		share, err = NewDirShare(
+			ctx,
+			record.Name,
+			record.Path.String(),
+			record.FollowLinks,
+		)
+
+		if record.EnableIndexing && share.SupportsWatching() {
+			share.OnNeedIndex(m.buildIndexCallback(ctx, record.Name))
+			share.OnDelete(m.buildDeleteCallback(ctx, record.Uuid))
+		}
+
+		m.shareMap[record.Name] = &shareData{
+			share:  share,
+			record: record,
+		}
+	}
+
 	go m.indexerDaemon()
 	go m.orphanedIndexGc()
 
@@ -204,22 +210,163 @@ func (m *Manager) orphanedIndexGc() {
 	}
 }
 
+// indexShareFromPath recursively indexes a share from a starting path.
+// If indexStartPath is true, it will also index startPath.
+// If indexStartPath is true and the start path points to a file, it just indexes that file and is done.
+// If replaceIndices is true, it will delete existing indices for files it finds and replace them with the new ones.
+// It will stop if it reaches maxFiles and return ErrTooManyFiles.
+func (m *Manager) indexShareFromPath(
+	ctx context.Context,
+	shareDat *shareData,
+	startPath common.ProtoPath,
+	indexStartPath bool,
+	replaceIndices bool,
+	maxFiles int,
+) (count int, err error) {
+	share := shareDat.share
+	shareUuid := shareDat.record.Uuid
+
+	if indexStartPath {
+		meta, err := share.GetFileMeta(startPath)
+		if err != nil {
+			return 0, fmt.Errorf(`failed to get metadata for start path %q in share %q: %w`, startPath.String(), shareUuid, err)
+		}
+
+		err = m.storage.InsertShareIndex(
+			ctx,
+			shareUuid,
+			shareDat.lastIndexId,
+			startPath.String(),
+			meta.IsDir,
+			int64(meta.Size),
+		)
+		if err != nil {
+			return 0, fmt.Errorf(`failed to insert share %q index for file %q: %w`, shareUuid, startPath.String(), err)
+		}
+
+		if replaceIndices {
+			err = m.storage.DeleteShareIndexByPath(ctx, shareUuid, startPath.String())
+			if err != nil {
+				return 1, fmt.Errorf(`failed to delete old share %q index for file %q: %w`, shareUuid, startPath.String(), err)
+			}
+		}
+
+		// If we know that this isn't a directory, just exit early.
+		if !meta.IsDir {
+			return 1, nil
+		}
+
+		count = 1
+	}
+
+	err = WalkShareDir(shareDat.share, startPath, func(path common.ProtoPath, meta *pb.MsgFileMeta) (bool, error) {
+		if count >= maxFiles {
+			return false, ErrTooManyFiles
+		}
+
+		count++
+
+		err = m.storage.InsertShareIndex(
+			ctx,
+			shareUuid,
+			shareDat.lastIndexId,
+			path.String(),
+			meta.IsDir,
+			int64(meta.Size),
+		)
+		if err != nil {
+			return false, fmt.Errorf(`failed to insert share %q index for file %q: %w`, shareUuid, path, err)
+		}
+
+		if replaceIndices {
+			err = m.storage.DeleteShareIndexByPath(ctx, shareUuid, path.String())
+			if err != nil {
+				return false, fmt.Errorf(`failed to delete old share %q index for file %q: %w`, shareUuid, startPath.String(), err)
+			}
+		}
+
+		return true, nil
+	})
+	if err != nil {
+		return count, err
+	}
+
+	return count, nil
+}
+
+// indexShareFile will index a single file in the share with the specified name.
+// This function is not optimized for bulk indexing. In that case, use indexShare.
+// This function should only be called from file watcher callbacks (see share.go).
+// This function expects the provided path to point to a file.
+// Refuses to index the file if it resides outside of the given share, returning ErrWrongShare.
+// Refuses to index the file if the share has indexing disabled, returning ErrIndexingDisabled.
+func (m *Manager) indexShareFile(ctx context.Context, shareName string, path common.ProtoPath) error {
+	if path.IsZero() || path.IsRoot() {
+		return nil
+	}
+
+	m.mu.RLock()
+	shareDat, has := m.shareMap[shareName]
+	m.mu.RUnlock()
+
+	if !has {
+		return nil
+	}
+
+	count, err := m.indexShareFromPath(ctx, shareDat, path, true, true, m.indexerMaxFiles)
+	if err != nil {
+		return fmt.Errorf(`failed to index share %q new directory %q: %w`, shareDat.record.Uuid, path.String(), err)
+	}
+
+	if count > 2 {
+		// It indexed more than just a couple of items, optimize the index.
+		err = m.storage.OptimizeShareIndex(ctx)
+		if err != nil {
+			optErr := m.storage.OptimizeShareIndex(ctx)
+			if optErr != nil {
+				m.logger.Warn("failed to optimize share index",
+					"service", "share.Manager",
+					"share_uuid", shareDat.record.Uuid,
+					"err", optErr,
+				)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (m *Manager) buildIndexCallback(ctx context.Context, name string) func(path common.ProtoPath) {
+	return func(path common.ProtoPath) {
+		_ = m.indexShareFile(ctx, name, path)
+	}
+}
+
+func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string) func(path common.ProtoPath) {
+	return func(path common.ProtoPath) {
+		if path.IsZero() || path.IsRoot() {
+			return
+		}
+
+		_ = m.storage.DeleteShareIndexByPath(ctx, uuid, path.String())
+	}
+}
+
 // indexShare indexes all files in the share with the specified name.
 // It returns the number of files indexed, whether the share existed, and any error that occurred.
 // Refuses to index the share if it has indexing disabled, returning ErrIndexingDisabled.
 func (m *Manager) indexShare(ctx context.Context, name string) (count int, hasShare bool, err error) {
 	m.mu.Lock()
-	val, has := m.shareMap[name]
+	shareDat, has := m.shareMap[name]
 	if !has {
 		m.mu.Unlock()
 		return 0, false, nil
 	}
 	curIndexId := time.Now().UnixMilli()
-	val.lastIndexId = curIndexId
+	shareDat.lastIndexId = curIndexId
 	m.mu.Unlock()
 
-	share := val.share
-	rec := val.record
+	rec := shareDat.record
 
 	if !rec.EnableIndexing {
 		return 0, true, ErrIndexingDisabled
@@ -254,52 +401,13 @@ func (m *Manager) indexShare(ctx context.Context, name string) (count int, hasSh
 		}
 	}()
 
-	dirs := []string{"/"}
-
-	for len(dirs) > 0 {
-		dir := dirs[0]
-		dirs = dirs[1:]
-
-		var files []*pb.MsgFileMeta
-		files, err = share.DirFiles(common.UncheckedCreateProtoPath(dir))
-		if err != nil {
-			// Skip files that were removed or we do not have permission to access.
-			if os.IsNotExist(err) || os.IsPermission(err) || errors.Is(err, syscall.ESRCH) {
-				continue
-			}
-
-			return count, true, fmt.Errorf("failed to read directory %q: %w", dir, err)
+	count, err = m.indexShareFromPath(ctx, shareDat, common.RootProtoPath, false, false, m.indexerMaxFiles)
+	if err != nil {
+		if errors.Is(err, ErrTooManyFiles) {
+			shouldClearOld = true
 		}
-		for _, file := range files {
-			if count >= m.indexerMaxFiles {
-				shouldClearOld = true
-				return count, true, ErrTooManyFiles
-			}
-
-			count++
-
-			var path string
-			if dir == "/" {
-				path = "/" + file.Name
-			} else {
-				path = dir + "/" + file.Name
-			}
-
-			if file.IsDir {
-				dirs = append(dirs, path)
-			}
 
-			err = m.storage.InsertShareIndex(ctx,
-				rec.Uuid,
-				curIndexId,
-				path,
-				file.IsDir,
-				int64(file.Size),
-			)
-			if err != nil {
-				return count, true, fmt.Errorf(`failed to insert share %q index for file %q: %w`, rec.Uuid, path, err)
-			}
-		}
+		return count, true, err
 	}
 
 	shouldClearOld = true
@@ -478,6 +586,7 @@ func (m *Manager) Add(
 
 	// Create instance.
 	share, err := NewDirShare(
+		ctx,
 		name,
 		path,
 		followLinks,
@@ -494,6 +603,12 @@ func (m *Manager) Add(
 	m.mu.Unlock()
 
 	if rec.EnableIndexing {
+		// Add requisite callbacks for file watcher
+		if share.SupportsWatching() {
+			share.OnNeedIndex(m.buildIndexCallback(ctx, name))
+			share.OnDelete(m.buildDeleteCallback(ctx, rec.Uuid))
+		}
+
 		go func() {
 			m.indexShareWithLockAndLogging(rec)
 		}()