From 65d4bd748658126f59c5218f50082cfbe8e4697e Mon Sep 17 00:00:00 2001 From: ten Date: Fri, 31 Jul 2026 02:45:30 -0400 Subject: [PATCH 1/4] get started on indexing single files on fs events. still need to handle gc and symlinks --- client/go.mod | 2 +- client/go.sum | 28 ++++----- client/share/manager.go | 126 +++++++++++++++++++++++++++++++++++----- client/share/share.go | 110 +++++++++++++++++++++++++++++++++-- common/path.go | 27 +++++++++ common/path_test.go | 85 +++++++++++++++++++++++++++ 6 files changed, 341 insertions(+), 37 deletions(-) diff --git a/client/go.mod b/client/go.mod index e515a8b3..0aaba1eb 100644 --- a/client/go.mod +++ b/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 diff --git a/client/go.sum b/client/go.sum index c6be9276..1407aa56 100644 --- a/client/go.sum +++ b/client/go.sum @@ -23,8 +23,9 @@ 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= @@ -32,29 +33,24 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu 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= diff --git a/client/share/manager.go b/client/share/manager.go index 1ace28c8..998fc96c 100644 --- a/client/share/manager.go +++ b/client/share/manager.go @@ -31,6 +31,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 +78,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 +87,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)) + } + + m.shareMap[record.Name] = &shareData{ + share: share, + record: record, + } + } + go m.indexerDaemon() go m.orphanedIndexGc() @@ -204,6 +212,85 @@ func (m *Manager) orphanedIndexGc() { } } +// 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, name string, path common.ProtoPath) error { + if path.IsZero() || path.IsRoot() { + return nil + } + + m.mu.Lock() + val, has := m.shareMap[name] + if !has { + m.mu.Unlock() + return nil + } + + pathContains, pathIdx := common.PathContains(val.record.Path, path) + if !pathContains { + m.mu.Unlock() + return ErrWrongShare + } + + relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) + if err != nil { + m.mu.Unlock() + return err + } + + curIndexId := time.Now().UnixMilli() + val.lastIndexId = curIndexId + m.mu.Unlock() + + share := val.share + rec := val.record + + if !rec.EnableIndexing { + return ErrIndexingDisabled + } + + meta, err := share.GetFileMeta(relPath) + if err != nil { + return err + } + + // TODO Consider doing garbage collection after this + // TODO Figure out symlinks + + err = m.storage.InsertShareIndex( + ctx, + rec.Uuid, + curIndexId, + relPath.String(), + false, + int64(meta.GetSize()), + ) + if err != nil { + return err + } + + m.logger.Debug("indexed share index", "service", "share.Manager", "path", relPath.String()) + + 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) + } +} + +// TODO What do we actually do on a delete? +func (m *Manager) buildDeleteCallback(ctx context.Context) func(path common.ProtoPath) { + return func(path common.ProtoPath) { + m.storage.OptimizeShareIndex(ctx) + } +} + // 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. @@ -478,6 +565,7 @@ func (m *Manager) Add( // Create instance. share, err := NewDirShare( + ctx, name, path, followLinks, @@ -494,6 +582,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)) + } + go func() { m.indexShareWithLockAndLogging(rec) }() diff --git a/client/share/share.go b/client/share/share.go index 5b837d7d..378b0407 100644 --- a/client/share/share.go +++ b/client/share/share.go @@ -1,20 +1,25 @@ package share import ( + "context" "errors" "io" "io/fs" "os" "path/filepath" "strings" + "sync" "friendnet.org/common" pb "friendnet.org/protocol/pb/v1" + "github.com/fsnotify/fsnotify" ) // ErrShareClosed is returned by Share methods when the share is closed. var ErrShareClosed = errors.New("share closed") +type ShareCallback func(path common.ProtoPath) + // Share is a shared filesystem. // A share only has the concepts of files and directories. // It has no way of representing symlinks or pipes. @@ -60,26 +65,47 @@ type Share interface { // // May return ErrShareClosed if the share is closed, depending on the implementation. GetFile(path common.ProtoPath, offset uint64, limit uint64) (*pb.MsgFileMeta, io.ReadCloser, error) + + // SupportsWatching will return true if the Share implementation supports filesystem event watching + SupportsWatching() bool + + // OnNeedIndex subscribes a callback to a filesystem event listener. + // The callbacks will fire, in order of subscription, when a new file in a watched directory is created or if an existing file has been modified. + OnNeedIndex(callback ShareCallback) + + // OnDelete subscribes a callback to a filesystem event listener. + // The callbacks will fire, in order of subscription, when a file in a watched directory is deleted. + OnDelete(callback ShareCallback) } // DirShare is an implementation of Share backed by a directory. type DirShare struct { + ctx context.Context + name string dir string followLinks bool fsys fs.FS + + // Watching related members + mu sync.RWMutex + + watcher *fsnotify.Watcher + onIndexHdlrs []ShareCallback + onDeleteHdlrs []ShareCallback } var _ Share = (*DirShare)(nil) -// Close is no-op because DirShare is stateless. func (s *DirShare) Close() error { - return nil + return s.watcher.Close() } // NewDirShare creates a new DirShare backed by the specified directory. +// It will also initialize a filesystem watcher. // If followLinks is false, symlinks will be treated as if they do not exist. func NewDirShare( + ctx context.Context, name string, dir string, followLinks bool, @@ -89,12 +115,88 @@ func NewDirShare( return nil, err } - return &DirShare{ + // Setup watcher + watcher, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + share := &DirShare{ + ctx: ctx, name: name, dir: abs, followLinks: followLinks, fsys: os.DirFS(abs), - }, nil + watcher: watcher, + } + + err = watcher.Add(abs) + if err != nil { + return nil, err + } + + // Init watcher + // On errors, just kill the watcher + go func() { + for { + select { + case <-ctx.Done(): + return + case event, ok := <-watcher.Events: + if !ok { + break + } + + share.mu.RLock() + + if len(event.Name) == 0 { + continue + } + + // TODO Is this the best way to reject a directory + if event.Name[len(event.Name)-1] == '/' { + continue + } + + path, err := common.NormalizePath(event.Name) + if err != nil { + break + } + + if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Chmod) { + for _, cb := range share.onIndexHdlrs { + cb(path) + } + } else if event.Has(fsnotify.Remove) { + for _, cb := range share.onDeleteHdlrs { + cb(path) + } + } + + share.mu.RUnlock() + } + } + }() + + return share, nil +} + +func (s *DirShare) SupportsWatching() bool { + return true +} + +func (s *DirShare) OnNeedIndex(callback ShareCallback) { + s.mu.Lock() + defer s.mu.Unlock() + + s.onIndexHdlrs = append(s.onIndexHdlrs, callback) +} + +func (s *DirShare) OnDelete(callback ShareCallback) { + s.mu.Lock() + defer s.mu.Unlock() + + s.onDeleteHdlrs = append(s.onDeleteHdlrs, callback) } func (s *DirShare) isInfoOk(info fs.FileInfo) bool { diff --git a/common/path.go b/common/path.go index 932e0503..a21fce17 100644 --- a/common/path.go +++ b/common/path.go @@ -262,3 +262,30 @@ func JoinPaths(paths ...ProtoPath) ProtoPath { } return UncheckedCreateProtoPath(sb.String()) } + +// PathContains returns true when path is contained wholly in parent and the array index in the path array corresponding to the full parent path. +func PathContains(parent ProtoPath, path ProtoPath) (bool, int) { + if parent.IsZero() || path.IsZero() { + return false, 0 + } + + if parent.IsRoot() { + return true, 0 + } + + parentDiv := parent.ToSegments() + pathDiv := path.ToSegments() + + if len(parentDiv) > len(pathDiv) { + return false, 0 + } + + i := 0 + for ; i < len(parentDiv); i++ { + if parentDiv[i] != pathDiv[i] { + return false, i + } + } + + return true, i +} diff --git a/common/path_test.go b/common/path_test.go index b9d7a458..3fc00b2e 100644 --- a/common/path_test.go +++ b/common/path_test.go @@ -2,6 +2,91 @@ package common import "testing" +//goland:noinspection ALL +func TestPathContains(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + parent string + path string + wantBool bool + wantIdx int + }{ + // --- valid --- + { + name: "valid_root_whatever", + parent: "/", + path: "/this/is/anything", + wantBool: true, + wantIdx: 0, + }, + { + name: "valid_first_degree", + parent: "/foo/", + path: "/foo/bar", + wantBool: true, + wantIdx: 1, + }, + { + name: "valid_second_degree", + parent: "/foo/", + path: "/foo/bar/qux", + wantBool: true, + wantIdx: 1, + }, + // --- invalid --- + { + name: "invalid_longer_parent", + parent: "/foo/bar/qux", + path: "/foo/", + wantBool: false, + wantIdx: 0, + }, + { + name: "invalid_wrong_first_degree", + parent: "/qux/", + path: "/foo/bar/qux", + wantBool: false, + wantIdx: 0, + }, + { + name: "invalid_wrong_second_degree", + parent: "/qux/foo", + path: "/foo/bar/qux", + wantBool: false, + wantIdx: 0, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + parent, err := NormalizePath(tt.parent) + if err != nil { + t.Fatalf("PathContains parent normalize fail: %v", err) + } + + path, err := NormalizePath(tt.path) + if err != nil { + t.Fatalf("PathContains path normalize fail: %v", err) + } + + res, idx := PathContains(parent, path) + + if res != tt.wantBool { + t.Fatalf("expected result %t, got %t", tt.wantBool, res) + } + + if idx != tt.wantIdx { + t.Fatalf("expected parent path cutoff index %d, got %d", tt.wantIdx, idx) + } + }) + } +} + //goland:noinspection ALL func TestValidatePath(t *testing.T) { t.Parallel() From 178b08e65bde30241a0fb8bb7dd42b785008c0d0 Mon Sep 17 00:00:00 2001 From: ten Date: Fri, 31 Jul 2026 03:44:43 -0400 Subject: [PATCH 2/4] delete callback deletes --- client/share/manager.go | 26 +++++++++++++++++++++----- client/storage/storage.go | 8 ++++++++ 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/client/share/manager.go b/client/share/manager.go index 998fc96c..4e359863 100644 --- a/client/share/manager.go +++ b/client/share/manager.go @@ -105,7 +105,7 @@ func NewManager( if record.EnableIndexing && share.SupportsWatching() { share.OnNeedIndex(m.buildIndexCallback(ctx, record.Name)) - share.OnDelete(m.buildDeleteCallback(ctx)) + share.OnDelete(m.buildDeleteCallback(ctx, record.Uuid, record.Path)) } m.shareMap[record.Name] = &shareData{ @@ -284,10 +284,26 @@ func (m *Manager) buildIndexCallback(ctx context.Context, name string) func(path } } -// TODO What do we actually do on a delete? -func (m *Manager) buildDeleteCallback(ctx context.Context) func(path common.ProtoPath) { +func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string, parent common.ProtoPath) func(path common.ProtoPath) { return func(path common.ProtoPath) { - m.storage.OptimizeShareIndex(ctx) + if path.IsZero() || path.IsRoot() { + return + } + + pathContains, pathIdx := common.PathContains(parent, path) + if !pathContains { + return + } + + relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) + if err != nil { + return + } + if relPath.IsZero() || relPath.IsRoot() { + return + } + + _ = m.storage.DeleteShareIndexByPath(ctx, uuid, relPath.String()) } } @@ -585,7 +601,7 @@ func (m *Manager) Add( // Add requisite callbacks for file watcher if share.SupportsWatching() { share.OnNeedIndex(m.buildIndexCallback(ctx, name)) - share.OnDelete(m.buildDeleteCallback(ctx)) + share.OnDelete(m.buildDeleteCallback(ctx, rec.Uuid, rec.Path)) } go func() { diff --git a/client/storage/storage.go b/client/storage/storage.go index e24de2f1..12b9a741 100644 --- a/client/storage/storage.go +++ b/client/storage/storage.go @@ -298,6 +298,14 @@ func (s *Storage) DeleteShareByUuid( return err } +func (s *Storage) DeleteShareIndexByPath(ctx context.Context, uuid string, path string) error { + _, err := s.Exec(ctx, `delete from share_index_fts where share = ? and path = ?`, uuid, path) + if err != nil { + return fmt.Errorf("failed to delete index for %q in share %q: %w", path, uuid, err) + } + return nil +} + // ClearShareIndex clears the search index for the share with the specified UUID. // It excludes all indexes that have an index ID lower than curIndexId. func (s *Storage) ClearShareIndex(ctx context.Context, uuid string, curIndexId int64) error { From dfff802fc2ccacc15703585690e9e37fbd51e14f Mon Sep 17 00:00:00 2001 From: ten Date: Sat, 1 Aug 2026 02:56:55 -0400 Subject: [PATCH 3/4] fundamentally misunderstood protopath, refactored share reindex functions --- client/share/manager.go | 223 ++++++++++++++++++++-------------------- client/share/share.go | 87 ++++++++++++---- client/share/walker.go | 58 +++++++++++ common/path.go | 27 ----- common/path_test.go | 85 --------------- 5 files changed, 240 insertions(+), 240 deletions(-) create mode 100644 client/share/walker.go diff --git a/client/share/manager.go b/client/share/manager.go index 4e359863..7cb3d330 100644 --- a/client/share/manager.go +++ b/client/share/manager.go @@ -5,10 +5,8 @@ import ( "errors" "fmt" "log/slog" - "os" "strings" "sync" - "syscall" "time" "friendnet.org/client/storage" @@ -212,69 +210,129 @@ 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, name string, path common.ProtoPath) error { +func (m *Manager) indexShareFile(ctx context.Context, shareName string, path common.ProtoPath) error { if path.IsZero() || path.IsRoot() { return nil } - m.mu.Lock() - val, has := m.shareMap[name] + m.mu.RLock() + shareDat, has := m.shareMap[shareName] + m.mu.RUnlock() + if !has { - m.mu.Unlock() return nil } - pathContains, pathIdx := common.PathContains(val.record.Path, path) - if !pathContains { - m.mu.Unlock() - return ErrWrongShare - } - - relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) + count, err := m.indexShareFromPath(ctx, shareDat, path, true, true, m.indexerMaxFiles) if err != nil { - m.mu.Unlock() - return err + return fmt.Errorf(`failed to index share %q new directory %q: %w`, shareDat.record.Uuid, path.String(), err) } - curIndexId := time.Now().UnixMilli() - val.lastIndexId = curIndexId - m.mu.Unlock() - - share := val.share - rec := val.record - - if !rec.EnableIndexing { - return ErrIndexingDisabled - } - - meta, err := share.GetFileMeta(relPath) - if err != nil { - return err - } - - // TODO Consider doing garbage collection after this - // TODO Figure out symlinks - - err = m.storage.InsertShareIndex( - ctx, - rec.Uuid, - curIndexId, - relPath.String(), - false, - int64(meta.GetSize()), - ) - if err != nil { - return 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, + ) + } + } } - m.logger.Debug("indexed share index", "service", "share.Manager", "path", relPath.String()) - return nil } @@ -284,26 +342,13 @@ func (m *Manager) buildIndexCallback(ctx context.Context, name string) func(path } } -func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string, parent common.ProtoPath) func(path common.ProtoPath) { +func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string) func(path common.ProtoPath) { return func(path common.ProtoPath) { if path.IsZero() || path.IsRoot() { return } - pathContains, pathIdx := common.PathContains(parent, path) - if !pathContains { - return - } - - relPath, err := common.SegmentsToPath(path.ToSegments()[pathIdx:]) - if err != nil { - return - } - if relPath.IsZero() || relPath.IsRoot() { - return - } - - _ = m.storage.DeleteShareIndexByPath(ctx, uuid, relPath.String()) + _ = m.storage.DeleteShareIndexByPath(ctx, uuid, path.String()) } } @@ -312,17 +357,16 @@ func (m *Manager) buildDeleteCallback(ctx context.Context, uuid string, parent c // 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 @@ -357,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 diff --git a/client/share/share.go b/client/share/share.go index 378b0407..e67b406d 100644 --- a/client/share/share.go +++ b/client/share/share.go @@ -3,12 +3,15 @@ package share import ( "context" "errors" + "fmt" "io" "io/fs" + "math" "os" "path/filepath" "strings" "sync" + "time" "friendnet.org/common" pb "friendnet.org/protocol/pb/v1" @@ -66,7 +69,7 @@ type Share interface { // May return ErrShareClosed if the share is closed, depending on the implementation. GetFile(path common.ProtoPath, offset uint64, limit uint64) (*pb.MsgFileMeta, io.ReadCloser, error) - // SupportsWatching will return true if the Share implementation supports filesystem event watching + // SupportsWatching will return true if the Share implementation supports filesystem event watching. SupportsWatching() bool // OnNeedIndex subscribes a callback to a filesystem event listener. @@ -138,6 +141,21 @@ func NewDirShare( // Init watcher // On errors, just kill the watcher go func() { + var ( + dedupDelay = 100 * time.Millisecond + dedupTimerMu sync.Mutex + dedupTimers = make(map[string]*time.Timer) + ) + + // Crawl for subdirectories to add to watcher + filepath.WalkDir(abs, func(path string, d fs.DirEntry, err error) error { + if d.IsDir() { + _ = watcher.Add(path) + } + + return nil + }) + for { select { case <-ctx.Done(): @@ -147,33 +165,64 @@ func NewDirShare( break } - share.mu.RLock() - - if len(event.Name) == 0 { + if !event.Has(fsnotify.Create) && !event.Has(fsnotify.Write) && !event.Has(fsnotify.Chmod) { continue } - // TODO Is this the best way to reject a directory - if event.Name[len(event.Name)-1] == '/' { + share.mu.RLock() + + if len(event.Name) == 0 { continue } - path, err := common.NormalizePath(event.Name) - if err != nil { - break - } + dedupTimerMu.Lock() + t, ok := dedupTimers[event.Name] + dedupTimerMu.Unlock() - if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Chmod) { - for _, cb := range share.onIndexHdlrs { - cb(path) - } - } else if event.Has(fsnotify.Remove) { - for _, cb := range share.onDeleteHdlrs { - cb(path) - } + // If timer for item doesn't exist, create + if !ok { + t = time.AfterFunc(math.MaxInt64, func() { + evtPath := event.Name + + // If this is a directory, add it to the watches. + stat, err := os.Stat(evtPath) + if err == nil && stat.IsDir() { + _ = watcher.Add(evtPath) + } + + relPath, err := filepath.Rel(abs, evtPath) + if err != nil { + return + } + + fmt.Printf("event: %s %s\n", event.Op.String(), relPath) + + path, err := common.NormalizePath(relPath) + if err != nil { + return + } + + if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Chmod) { + for _, cb := range share.onIndexHdlrs { + cb(path) + } + } else if event.Has(fsnotify.Remove) { + for _, cb := range share.onDeleteHdlrs { + cb(path) + } + } + + share.mu.RUnlock() + }) + + t.Stop() + + dedupTimerMu.Lock() + dedupTimers[event.Name] = t + dedupTimerMu.Unlock() } - share.mu.RUnlock() + t.Reset(dedupDelay) } } }() diff --git a/client/share/walker.go b/client/share/walker.go new file mode 100644 index 00000000..67730285 --- /dev/null +++ b/client/share/walker.go @@ -0,0 +1,58 @@ +package share + +import ( + "errors" + "fmt" + "os" + "syscall" + + "friendnet.org/common" + pb "friendnet.org/protocol/pb/v1" +) + +// WalkShareDir walks files in a share's directory. +// The path can be root. +// fn will NOT be called for the initial path. +// If fn returns an error, it will be returned immediately and walking will stop. +// If fn returns false, walking will stop without an error. +// If fn returns true, walking will continue for another file. +func WalkShareDir(share Share, path common.ProtoPath, fn func(path common.ProtoPath, meta *pb.MsgFileMeta) (bool, error)) error { + dirs := []string{path.String()} + + for len(dirs) > 0 { + dir := dirs[0] + dirs = dirs[1:] + + 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 fmt.Errorf("failed to read share %q directory %q before walking: %w", share.Name(), dir, err) + } + for _, file := range files { + var path string + if dir == "/" { + path = "/" + file.Name + } else { + path = dir + "/" + file.Name + } + + if file.IsDir { + dirs = append(dirs, path) + } + + ok, err := fn(common.UncheckedCreateProtoPath(path), file) + if err != nil { + return err + } + if !ok { + return nil + } + } + } + + return nil +} diff --git a/common/path.go b/common/path.go index a21fce17..932e0503 100644 --- a/common/path.go +++ b/common/path.go @@ -262,30 +262,3 @@ func JoinPaths(paths ...ProtoPath) ProtoPath { } return UncheckedCreateProtoPath(sb.String()) } - -// PathContains returns true when path is contained wholly in parent and the array index in the path array corresponding to the full parent path. -func PathContains(parent ProtoPath, path ProtoPath) (bool, int) { - if parent.IsZero() || path.IsZero() { - return false, 0 - } - - if parent.IsRoot() { - return true, 0 - } - - parentDiv := parent.ToSegments() - pathDiv := path.ToSegments() - - if len(parentDiv) > len(pathDiv) { - return false, 0 - } - - i := 0 - for ; i < len(parentDiv); i++ { - if parentDiv[i] != pathDiv[i] { - return false, i - } - } - - return true, i -} diff --git a/common/path_test.go b/common/path_test.go index 3fc00b2e..b9d7a458 100644 --- a/common/path_test.go +++ b/common/path_test.go @@ -2,91 +2,6 @@ package common import "testing" -//goland:noinspection ALL -func TestPathContains(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - parent string - path string - wantBool bool - wantIdx int - }{ - // --- valid --- - { - name: "valid_root_whatever", - parent: "/", - path: "/this/is/anything", - wantBool: true, - wantIdx: 0, - }, - { - name: "valid_first_degree", - parent: "/foo/", - path: "/foo/bar", - wantBool: true, - wantIdx: 1, - }, - { - name: "valid_second_degree", - parent: "/foo/", - path: "/foo/bar/qux", - wantBool: true, - wantIdx: 1, - }, - // --- invalid --- - { - name: "invalid_longer_parent", - parent: "/foo/bar/qux", - path: "/foo/", - wantBool: false, - wantIdx: 0, - }, - { - name: "invalid_wrong_first_degree", - parent: "/qux/", - path: "/foo/bar/qux", - wantBool: false, - wantIdx: 0, - }, - { - name: "invalid_wrong_second_degree", - parent: "/qux/foo", - path: "/foo/bar/qux", - wantBool: false, - wantIdx: 0, - }, - } - - for _, tt := range tests { - tt := tt - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - parent, err := NormalizePath(tt.parent) - if err != nil { - t.Fatalf("PathContains parent normalize fail: %v", err) - } - - path, err := NormalizePath(tt.path) - if err != nil { - t.Fatalf("PathContains path normalize fail: %v", err) - } - - res, idx := PathContains(parent, path) - - if res != tt.wantBool { - t.Fatalf("expected result %t, got %t", tt.wantBool, res) - } - - if idx != tt.wantIdx { - t.Fatalf("expected parent path cutoff index %d, got %d", tt.wantIdx, idx) - } - }) - } -} - //goland:noinspection ALL func TestValidatePath(t *testing.T) { t.Parallel() From fd499c8a4f6f5df88b2f520c35cad32de3dc6e25 Mon Sep 17 00:00:00 2001 From: ten Date: Sat, 1 Aug 2026 02:57:35 -0400 Subject: [PATCH 4/4] typo --- client/share/manager.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client/share/manager.go b/client/share/manager.go index 7cb3d330..be958395 100644 --- a/client/share/manager.go +++ b/client/share/manager.go @@ -103,7 +103,7 @@ func NewManager( if record.EnableIndexing && share.SupportsWatching() { share.OnNeedIndex(m.buildIndexCallback(ctx, record.Name)) - share.OnDelete(m.buildDeleteCallback(ctx, record.Uuid, record.Path)) + share.OnDelete(m.buildDeleteCallback(ctx, record.Uuid)) } m.shareMap[record.Name] = &shareData{ @@ -606,7 +606,7 @@ func (m *Manager) Add( // Add requisite callbacks for file watcher if share.SupportsWatching() { share.OnNeedIndex(m.buildIndexCallback(ctx, name)) - share.OnDelete(m.buildDeleteCallback(ctx, rec.Uuid, rec.Path)) + share.OnDelete(m.buildDeleteCallback(ctx, rec.Uuid)) } go func() {