From ef8828a07258a8b6c06ef248d2f4f3d0025284f2 Mon Sep 17 00:00:00 2001 From: ashmrtn <3891298+ashmrtn@users.noreply.github.com> Date: Sat, 4 Jun 2022 06:36:25 -0700 Subject: [PATCH] refactor(snapshots): Remove remaining internal uses of Readdir (#1986) * Remove remaining internal uses of Readdir * Remove old helpers and interface functions. * Update tests for updated fs.Directory interface * Fix index out of range error in snapshot walker Record one error if an error occurred and it's not limiting errors * Use helper functions more; exit loops early Follow up on reviewer comments and reduce code duplication, use more targetted functions like Directory.Child, and exit directory iteration early if possible. * Remove fs.Entries type and unused functions Leave some functions dealing with sorting and finding entries in fs package. This retains tests for those functions while still allowing mockfs to access them. * Simplify function return --- .golangci.yml | 5 +- cli/command_ls.go | 13 +- cli/command_snapshot_create.go | 2 +- fs/cachefs/cache.go | 41 ++- fs/cachefs/cache_test.go | 10 +- fs/cachefs/cachefs.go | 22 +- fs/entry.go | 130 +++------- fs/entry_test.go | 236 +----------------- fs/ignorefs/ignorefs.go | 4 - fs/ignorefs/ignorefs_test.go | 12 +- fs/localfs/local_fs.go | 4 - fs/localfs/local_fs_test.go | 69 +++-- fs/localfs/localfs_benchmark_test.go | 5 +- fs/localfs/shallow_fs.go | 4 - fs/loggingfs/loggingfs.go | 18 +- fs/virtualfs/virtualfs.go | 19 +- fs/virtualfs/virtualfs_test.go | 4 +- internal/diff/diff.go | 8 +- internal/fshasher/fshasher.go | 17 +- internal/fusemount/fusefs.go | 18 +- internal/mockfs/mockfs.go | 25 +- internal/webdavmount/webdavmount.go | 31 ++- snapshot/restore/restore.go | 2 +- snapshot/snapshotfs/all_sources.go | 21 +- snapshot/snapshotfs/estimate.go | 26 +- snapshot/snapshotfs/objref.go | 3 +- snapshot/snapshotfs/repofs.go | 8 +- snapshot/snapshotfs/snapshot_tree_walker.go | 29 ++- .../snapshotfs/snapshot_tree_walker_test.go | 49 +++- snapshot/snapshotfs/source_directories.go | 20 +- .../snapshotfs/source_directories_test.go | 70 +++--- snapshot/snapshotfs/source_snapshots.go | 20 +- snapshot/snapshotfs/upload_test.go | 2 +- tests/tools/kopiaclient/kopiaclient.go | 2 +- 34 files changed, 381 insertions(+), 568 deletions(-) diff --git a/.golangci.yml b/.golangci.yml index 1bbcb3375..13de4545b 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -134,9 +134,6 @@ issues: - text: "Errors unhandled" linters: - gosec - - text: "unwrapped: sig: func github.com/kopia/kopia/fs.ReaddirToIterate" - linters: - - wrapcheck - - text: "unwrapped: sig: func github.com/kopia/kopia/fs.IterateEntriesToReaddir" + - text: "unwrapped: sig: func github.com/kopia/kopia/fs.GetAllEntries" linters: - wrapcheck diff --git a/cli/command_ls.go b/cli/command_ls.go index 34d926893..c650015e6 100644 --- a/cli/command_ls.go +++ b/cli/command_ls.go @@ -54,15 +54,10 @@ func (c *commandList) run(ctx context.Context, rep repo.Repository) error { } func (c *commandList) listDirectory(ctx context.Context, d fs.Directory, prefix, indent string) error { - entries, err := d.Readdir(ctx) - if err != nil { - return errors.Wrap(err, "error reading directory") - } - - for _, e := range entries { - if err := c.printDirectoryEntry(ctx, e, prefix, indent); err != nil { - return errors.Wrap(err, "unable to print directory entry") - } + if err := d.IterateEntries(ctx, func(innerCtx context.Context, e fs.Entry) error { + return c.printDirectoryEntry(innerCtx, e, prefix, indent) + }); err != nil { + return err // nolint:wrapcheck } if dws, ok := d.(fs.DirectoryWithSummary); ok && c.errorSummary { diff --git a/cli/command_snapshot_create.go b/cli/command_snapshot_create.go index ba2bf31a1..8b7e5b62c 100644 --- a/cli/command_snapshot_create.go +++ b/cli/command_snapshot_create.go @@ -273,7 +273,7 @@ func (c *commandSnapshotCreate) snapshotSingleSource(ctx context.Context, rep re if c.snapshotCreateStdinFileName != "" { // stdin source will be snapshotted using a virtual static root directory with a single streaming file entry // Create a new static directory with the given name and add a streaming file entry with os.Stdin reader - fsEntry = virtualfs.NewStaticDirectory(sourceInfo.Path, fs.Entries{ + fsEntry = virtualfs.NewStaticDirectory(sourceInfo.Path, []fs.Entry{ virtualfs.StreamingFileFromReader(c.snapshotCreateStdinFileName, os.Stdin), }) setManual = true diff --git a/fs/cachefs/cache.go b/fs/cachefs/cache.go index 58d9af0b0..c46b7c400 100644 --- a/fs/cachefs/cache.go +++ b/fs/cachefs/cache.go @@ -21,7 +21,7 @@ type cacheEntry struct { next *cacheEntry expireAfter time.Time - entries fs.Entries + entries []fs.Entry } // Cache maintains in-memory cache of recently-read data to speed up filesystem operations. @@ -78,25 +78,44 @@ func (c *Cache) remove(e *cacheEntry) { } // Loader provides data to be stored in the cache. -type Loader func(ctx context.Context) (fs.Entries, error) +type Loader func(ctx context.Context) ([]fs.Entry, error) // EntryWrapper allows an fs.Entry to be modified before inserting into the cache. type EntryWrapper func(entry fs.Entry) fs.Entry -// Readdir reads the contents of a provided directory using ObjectID of a directory (if any) to cache -// the results. -func (c *Cache) Readdir(ctx context.Context, d fs.Directory, w EntryWrapper) (fs.Entries, error) { +// IterateEntries reads the contents of a provided directory using ObjectID of a directory (if any) to cache +// the results. The given callback is invoked on each item in the directory. +func (c *Cache) IterateEntries(ctx context.Context, d fs.Directory, w EntryWrapper, callback func(context.Context, fs.Entry) error) error { if h, ok := d.(object.HasObjectID); ok { cacheID := h.ObjectID().String() - return c.getEntries(ctx, cacheID, dirCacheExpiration, d.Readdir, w) + entries, err := c.getEntries( + ctx, + cacheID, + dirCacheExpiration, + func(innerCtx context.Context) ([]fs.Entry, error) { + return fs.GetAllEntries(innerCtx, d) + }, + w, + ) + if err != nil { + return err + } + + for _, e := range entries { + err = callback(ctx, e) + if err != nil { + return err + } + } + + return nil } - // nolint:wrapcheck - return d.Readdir(ctx) + return d.IterateEntries(ctx, callback) // nolint:wrapcheck } -func (c *Cache) getEntriesFromCacheLocked(ctx context.Context, id string) fs.Entries { +func (c *Cache) getEntriesFromCacheLocked(ctx context.Context, id string) []fs.Entry { if v, ok := c.data[id]; id != "" && ok { if clock.Now().Before(v.expireAfter) { c.moveToHead(v) @@ -121,7 +140,7 @@ func (c *Cache) getEntriesFromCacheLocked(ctx context.Context, id string) fs.Ent // getEntries consults the cache and either retrieves the contents of directory listing from the cache // or invokes the provides callback and adds the results to cache. -func (c *Cache) getEntries(ctx context.Context, id string, expirationTime time.Duration, cb Loader, w EntryWrapper) (fs.Entries, error) { +func (c *Cache) getEntries(ctx context.Context, id string, expirationTime time.Duration, cb Loader, w EntryWrapper) ([]fs.Entry, error) { if c == nil { return cb(ctx) } @@ -142,7 +161,7 @@ func (c *Cache) getEntries(ctx context.Context, id string, expirationTime time.D return nil, err } - wrapped := make(fs.Entries, len(raw)) + wrapped := make([]fs.Entry, len(raw)) for i, entry := range raw { wrapped[i] = w(entry) } diff --git a/fs/cachefs/cache_test.go b/fs/cachefs/cache_test.go index dd06c87ca..1442fad8c 100644 --- a/fs/cachefs/cache_test.go +++ b/fs/cachefs/cache_test.go @@ -20,7 +20,7 @@ import ( const expirationTime = 10 * time.Hour type cacheSource struct { - data map[string]fs.Entries + data map[string][]fs.Entry callCounter map[string]int } @@ -28,8 +28,8 @@ func identityWrapper(e fs.Entry) fs.Entry { return e } -func (cs *cacheSource) get(id string) func(ctx context.Context) (fs.Entries, error) { - return func(context.Context) (fs.Entries, error) { +func (cs *cacheSource) get(id string) func(ctx context.Context) ([]fs.Entry, error) { + return func(context.Context) ([]fs.Entry, error) { cs.callCounter[id]++ d, ok := cs.data[id] @@ -42,7 +42,7 @@ func (cs *cacheSource) get(id string) func(ctx context.Context) (fs.Entries, err } func (cs *cacheSource) setEntryCount(id string, cnt int) { - var fakeEntries fs.Entries + var fakeEntries []fs.Entry var fakeEntry fs.Entry @@ -56,7 +56,7 @@ func (cs *cacheSource) setEntryCount(id string, cnt int) { func newCacheSource() *cacheSource { return &cacheSource{ - data: make(map[string]fs.Entries), + data: make(map[string][]fs.Entry), callCounter: make(map[string]int), } } diff --git a/fs/cachefs/cachefs.go b/fs/cachefs/cachefs.go index c9a75f4cf..b6045ce7b 100644 --- a/fs/cachefs/cachefs.go +++ b/fs/cachefs/cachefs.go @@ -9,7 +9,7 @@ import ( // DirectoryCacher reads and potentially caches directory entries for a given directory. type DirectoryCacher interface { - Readdir(ctx context.Context, d fs.Directory, w EntryWrapper) (fs.Entries, error) + IterateEntries(ctx context.Context, d fs.Directory, w EntryWrapper, callback func(context.Context, fs.Entry) error) error } type cacheContext struct { @@ -31,17 +31,17 @@ func (d *directory) Child(ctx context.Context, name string) (fs.Entry, error) { return wrapWithContext(e, d.ctx), nil } -func (d *directory) Readdir(ctx context.Context) (fs.Entries, error) { - entries, err := d.ctx.cacher.Readdir(ctx, d.Directory, func(e fs.Entry) fs.Entry { - return wrapWithContext(e, d.ctx) - }) - if err != nil { - // nolint:wrapcheck - return entries, err - } +func (d *directory) IterateEntries(ctx context.Context, callback func(context.Context, fs.Entry) error) error { + err := d.ctx.cacher.IterateEntries( + ctx, + d.Directory, + func(e fs.Entry) fs.Entry { + return wrapWithContext(e, d.ctx) + }, + callback, + ) - // nolint:wrapcheck - return entries, err + return err // nolint:wrapcheck } type file struct { diff --git a/fs/entry.go b/fs/entry.go index f48a1a9e2..857ee89af 100644 --- a/fs/entry.go +++ b/fs/entry.go @@ -33,9 +33,6 @@ type DeviceInfo struct { Rdev uint64 `json:"rdev"` } -// Entries is a list of entries sorted by name. -type Entries []Entry - // Reader allows reading from a file and retrieving its up-to-date file info. type Reader interface { io.ReadCloser @@ -60,7 +57,6 @@ type StreamingFile interface { type Directory interface { Entry Child(ctx context.Context, name string) (Entry, error) - Readdir(ctx context.Context) (Entries, error) IterateEntries(ctx context.Context, cb func(context.Context, Entry) error) error } @@ -76,53 +72,48 @@ type ErrorEntry interface { ErrorInfo() error } -// ErrEntryNotFound is returned when an entry is not found. -var ErrEntryNotFound = errors.New("entry not found") - -// ReaddirToIterate is an adapter for a naive Readdir -> IterateEntries implementation. -func ReaddirToIterate(ctx context.Context, d Directory, cb func(context.Context, Entry) error) error { - entries, err := d.Readdir(ctx) - if err != nil { - return errors.Wrap(err, "error reading directory") - } - - for _, e := range entries { - if err := cb(ctx, e); err != nil { - return err - } - } - - return nil -} - -// IterateEntriesToReaddir is an adapter for a naive IterateEntries -> Readdir implementation. -func IterateEntriesToReaddir(ctx context.Context, d Directory) (Entries, error) { - var entries Entries +// GetAllEntries uses IterateEntries to return all entries in a Directory. +func GetAllEntries(ctx context.Context, d Directory) ([]Entry, error) { + entries := []Entry{} err := d.IterateEntries(ctx, func(ctx context.Context, e Entry) error { entries = append(entries, e) return nil }) - entries.Sort() - return entries, err // nolint:wrapcheck } -// ReadDirAndFindChild reads all entries from a directory and returns one by name. +// ErrEntryNotFound is returned when an entry is not found. +var ErrEntryNotFound = errors.New("entry not found") + +// IterateEntriesAndFindChild iterates through entries from a directory and returns one by name. // This is a convenience function that may be helpful in implementations of Directory.Child(). -func ReadDirAndFindChild(ctx context.Context, d Directory, name string) (Entry, error) { - children, err := d.Readdir(ctx) - if err != nil { +func IterateEntriesAndFindChild(ctx context.Context, d Directory, name string) (Entry, error) { + type errStop struct { + error + } + + var result Entry + + err := d.IterateEntries(ctx, func(c context.Context, e Entry) error { + if result == nil && e.Name() == name { + result = e + return errStop{errors.New("")} + } + return nil + }) + + var stopped errStop + if err != nil && !errors.As(err, &stopped) { return nil, errors.Wrap(err, "error reading directory") } - e := children.FindByName(name) - if e == nil { + if result == nil { return nil, ErrEntryNotFound } - return e, nil + return result, nil } // MaxFailedEntriesPerDirectorySummary is the maximum number of failed entries per directory summary. @@ -166,74 +157,25 @@ type Symlink interface { Readlink(ctx context.Context) (string, error) } -// FindByName returns an entry with a given name, or nil if not found. -func (e Entries) FindByName(n string) Entry { +// FindByName returns an entry with a given name, or nil if not found. Assumes +// the given slice of fs.Entry is sorted. +func FindByName(entries []Entry, n string) Entry { i := sort.Search( - len(e), + len(entries), func(i int) bool { - return e[i].Name() >= n + return entries[i].Name() >= n }, ) - if i < len(e) && e[i].Name() == n { - return e[i] + if i < len(entries) && entries[i].Name() == n { + return entries[i] } return nil } -// Update returns a copy of Entries with the provided entry included, by either replacing -// existing entry with the same name or inserted in the appropriate place to maintain sorted order. -func (e Entries) Update(newEntry Entry) Entries { - name := newEntry.Name() - pos := sort.Search(len(e), func(i int) bool { - return e[i].Name() >= name - }) - - // append at the end - if pos >= len(e) { - return append(append(Entries(nil), e...), newEntry) - } - - if e[pos].Name() == name { - if pos > 0 { - return append(append(append(Entries(nil), e[0:pos]...), newEntry), e[pos+1:]...) - } - - return append(append(Entries(nil), newEntry), e[pos+1:]...) - } - - if pos > 0 { - return append(append(append(Entries(nil), e[0:pos]...), newEntry), e[pos:]...) - } - - return append(append(Entries(nil), newEntry), e[pos:]...) -} - -// Remove returns a copy of Entries with the provided entry removed, while maintaining sorted order. -func (e Entries) Remove(name string) Entries { - pos := sort.Search(len(e), func(i int) bool { - return e[i].Name() >= name - }) - - // not found - if pos >= len(e) { - return e - } - - if e[pos].Name() != name { - return e - } - - if pos > 0 { - return append(append(Entries(nil), e[0:pos]...), e[pos+1:]...) - } - - return append(Entries(nil), e[pos+1:]...) -} - // Sort sorts the entries by name. -func (e Entries) Sort() { - sort.Slice(e, func(i, j int) bool { - return e[i].Name() < e[j].Name() +func Sort(entries []Entry) { + sort.Slice(entries, func(i, j int) bool { + return entries[i].Name() < entries[j].Name() }) } diff --git a/fs/entry_test.go b/fs/entry_test.go index 290ac52e4..b25e061ac 100644 --- a/fs/entry_test.go +++ b/fs/entry_test.go @@ -18,40 +18,40 @@ func (e testEntry) Name() string { } func TestEntriesFindByName(t *testing.T) { - entries := Entries{ + entries := []Entry{ testEntry{n: "aa", value: 3}, testEntry{n: "cc", value: 5}, testEntry{n: "ee", value: 6}, } want := testEntry{n: "aa", value: 3} - if got := entries.FindByName("aa"); got != want { + if got := FindByName(entries, "aa"); got != want { t.Errorf("a") } - if got := entries.FindByName("aa0"); got != nil { + if got := FindByName(entries, "aa0"); got != nil { t.Errorf("unexpected result when looking for non-existent: %v", got) } - if got := entries.FindByName("dd"); got != nil { + if got := FindByName(entries, "dd"); got != nil { t.Errorf("unexpected result when looking for non-existent: %v", got) } - if got := entries.FindByName("ff"); got != nil { + if got := FindByName(entries, "ff"); got != nil { t.Errorf("unexpected result when looking for non-existent: %v", got) } } func TestEntriesSort(t *testing.T) { - entries := Entries{ + entries := []Entry{ testEntry{n: "cc"}, testEntry{n: "bb"}, testEntry{n: "aa"}, } - entries.Sort() + Sort(entries) - want := Entries{ + want := []Entry{ testEntry{n: "aa"}, testEntry{n: "bb"}, testEntry{n: "cc"}, @@ -61,223 +61,3 @@ func TestEntriesSort(t *testing.T) { t.Errorf("unexpected output diff (-got, +want): %v", diff) } } - -func TestEntriesUpdate(t *testing.T) { - cases := []struct { - desc string - base Entries - updatedEntry Entry - want Entries - }{ - { - desc: "update existing - first", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - updatedEntry: testEntry{n: "aa", value: 3}, - want: Entries{ - testEntry{n: "aa", value: 3}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "update existing in the middle", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - updatedEntry: testEntry{n: "bb", value: 3}, - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb", value: 3}, - testEntry{n: "cc"}, - }, - }, - { - desc: "update existing - last", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - updatedEntry: testEntry{n: "cc", value: 3}, - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc", value: 3}, - }, - }, - { - desc: "insert before first", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - updatedEntry: testEntry{n: "00", value: 3}, - want: Entries{ - testEntry{n: "00", value: 3}, - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "insert between 2 existing", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "cc"}, - testEntry{n: "dd"}, - }, - updatedEntry: testEntry{n: "bb", value: 3}, - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb", value: 3}, - testEntry{n: "cc"}, - testEntry{n: "dd"}, - }, - }, - { - desc: "insert after last first", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - updatedEntry: testEntry{n: "dd", value: 3}, - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - testEntry{n: "dd", value: 3}, - }, - }, - { - desc: "append to empty", - base: Entries{}, - updatedEntry: testEntry{n: "dd", value: 3}, - want: Entries{ - testEntry{n: "dd", value: 3}, - }, - }, - } - - for _, tc := range cases { - t.Logf("starting %q", tc.desc) - updated := tc.base.Update(tc.updatedEntry) - - if diff := pretty.Compare(updated, tc.want); diff != "" { - t.Errorf("unexpected output for %q diff (-got, +want): %v", tc.desc, diff) - } - } -} - -func TestEntriesRemove(t *testing.T) { - cases := []struct { - desc string - base Entries - removedEntry string - want Entries - }{ - { - desc: "remove existing - first", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - removedEntry: "aa", - want: Entries{ - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "remove existing in the middle", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - removedEntry: "bb", - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "remove existing - last", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - removedEntry: "cc", - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - }, - }, - { - desc: "non-existent before first", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - removedEntry: "00", - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "non-existent in the middle", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "dd"}, - }, - removedEntry: "cc", - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "dd"}, - }, - }, - { - desc: "non-existent after last", - base: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - removedEntry: "dd", - want: Entries{ - testEntry{n: "aa"}, - testEntry{n: "bb"}, - testEntry{n: "cc"}, - }, - }, - { - desc: "empty", - base: Entries{}, - removedEntry: "zz", - want: Entries{}, - }, - } - - for _, tc := range cases { - t.Logf("starting %q", tc.desc) - updated := tc.base.Remove(tc.removedEntry) - - if diff := pretty.Compare(updated, tc.want); diff != "" { - t.Errorf("unexpected output for %q diff (-got, +want): %v", tc.desc, diff) - } - } -} diff --git a/fs/ignorefs/ignorefs.go b/fs/ignorefs/ignorefs.go index 79e513452..25c236763 100644 --- a/fs/ignorefs/ignorefs.go +++ b/fs/ignorefs/ignorefs.go @@ -210,10 +210,6 @@ func (d *ignoreDirectory) Child(ctx context.Context, name string) (fs.Entry, err return nil, fs.ErrEntryNotFound } -func (d *ignoreDirectory) Readdir(ctx context.Context) (fs.Entries, error) { - return fs.IterateEntriesToReaddir(ctx, d) -} - func (d *ignoreDirectory) buildContext(ctx context.Context) (*ignoreContext, error) { effectiveDotIgnoreFiles := d.parentContext.dotIgnoreFiles diff --git a/fs/ignorefs/ignorefs_test.go b/fs/ignorefs/ignorefs_test.go index 3669fa93d..7b1d18083 100644 --- a/fs/ignorefs/ignorefs_test.go +++ b/fs/ignorefs/ignorefs_test.go @@ -2,6 +2,7 @@ package ignorefs_test import ( "bytes" + "context" "sort" "testing" @@ -548,12 +549,7 @@ func walkTree(t *testing.T, dir fs.Directory) []string { walk = func(path string, d fs.Directory) error { output = append(output, path+"/") - entries, err := d.Readdir(testlogging.Context(t)) - if err != nil { - return err - } - - for _, e := range entries { + return d.IterateEntries(testlogging.Context(t), func(innerCtx context.Context, e fs.Entry) error { relPath := path + "/" + e.Name() if subdir, ok := e.(fs.Directory); ok { @@ -563,9 +559,9 @@ func walkTree(t *testing.T, dir fs.Directory) []string { } else { output = append(output, relPath) } - } - return nil + return nil + }) } if err := walk(".", dir); err != nil { diff --git a/fs/localfs/local_fs.go b/fs/localfs/local_fs.go index fe25d123c..d18742db7 100644 --- a/fs/localfs/local_fs.go +++ b/fs/localfs/local_fs.go @@ -140,10 +140,6 @@ func toDirEntryOrNil(dirEntry os.DirEntry, prefix string) (fs.Entry, error) { return entryFromDirEntry(fi, prefix), nil } -func (fsd *filesystemDirectory) Readdir(ctx context.Context) (fs.Entries, error) { - return fs.IterateEntriesToReaddir(ctx, fsd) -} - func (fsd *filesystemDirectory) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { fullPath := fsd.fullPath() diff --git a/fs/localfs/local_fs_test.go b/fs/localfs/local_fs_test.go index c201c77d1..5a8381586 100644 --- a/fs/localfs/local_fs_test.go +++ b/fs/localfs/local_fs_test.go @@ -17,6 +17,11 @@ import ( "github.com/kopia/kopia/internal/testutil" ) +type fileEnt struct { + size int64 + isFile bool +} + //nolint:gocyclo func TestFiles(t *testing.T) { ctx := testlogging.Context(t) @@ -39,7 +44,7 @@ func TestFiles(t *testing.T) { t.Errorf("error when dir empty directory: %v", err) } - entries, err := dir.Readdir(ctx) + entries, err := fs.GetAllEntries(ctx, dir) if err != nil { t.Errorf("error gettind dir Entries: %v", err) } @@ -56,35 +61,61 @@ func TestFiles(t *testing.T) { assertNoError(t, os.Mkdir(filepath.Join(tmp, "z"), 0o777)) assertNoError(t, os.Mkdir(filepath.Join(tmp, "y"), 0o777)) + expected := map[string]fileEnt{ + "f1": { + size: 5, + isFile: true, + }, + "f2": { + size: 4, + isFile: true, + }, + "f3": { + size: 3, + isFile: true, + }, + "y": { + size: 0, + isFile: false, + }, + "z": { + size: 0, + isFile: false, + }, + } + dir, err = Directory(tmp) if err != nil { t.Errorf("error when dir directory with files: %v", err) } - entries, err = dir.Readdir(ctx) + entries, err = fs.GetAllEntries(ctx, dir) if err != nil { t.Errorf("error gettind dir Entries: %v", err) } goodCount := 0 - if entries[0].Name() == "f1" && entries[0].Size() == 5 && entries[0].Mode().IsRegular() { - goodCount++ - } + for _, found := range entries { + wanted, ok := expected[found.Name()] + if !ok { + continue + } - if entries[1].Name() == "f2" && entries[1].Size() == 4 && entries[1].Mode().IsRegular() { - goodCount++ - } + if found.Size() != wanted.size { + continue + } - if entries[2].Name() == "f3" && entries[2].Size() == 3 && entries[2].Mode().IsRegular() { - goodCount++ - } + if wanted.isFile { + if !found.Mode().IsRegular() { + continue + } + } else { + if !found.Mode().IsDir() { + continue + } + } - if entries[3].Name() == "y" && entries[3].Size() == 0 && entries[3].Mode().IsDir() { - goodCount++ - } - - if entries[4].Name() == "z" && entries[4].Size() == 0 && entries[4].Mode().IsDir() { goodCount++ } @@ -193,12 +224,12 @@ func verifyChild(t *testing.T, dir fs.Directory) { t.Errorf("unexpected child size: %v, want %v", got, want) } - if _, err = fs.ReadDirAndFindChild(ctx, dir, "f4"); !errors.Is(err, fs.ErrEntryNotFound) { + if _, err = fs.IterateEntriesAndFindChild(ctx, dir, "f4"); !errors.Is(err, fs.ErrEntryNotFound) { t.Errorf("unexpected child error: %v", err) } - // read child again, this time using ReadAndFindChild - child2, err := fs.ReadDirAndFindChild(ctx, dir, "f3") + // read child again, this time using IterateEntriesAndFindChild + child2, err := fs.IterateEntriesAndFindChild(ctx, dir, "f3") if err != nil { t.Errorf("child2 error: %v", err) } diff --git a/fs/localfs/localfs_benchmark_test.go b/fs/localfs/localfs_benchmark_test.go index a38e0f28f..6d5d2ad11 100644 --- a/fs/localfs/localfs_benchmark_test.go +++ b/fs/localfs/localfs_benchmark_test.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" + "github.com/kopia/kopia/fs" "github.com/kopia/kopia/fs/localfs" ) @@ -56,6 +57,8 @@ func benchmarkReadDirWithCount(b *testing.B, fileCount int) { for i := 0; i < b.N; i++ { dir, _ := localfs.Directory(td) - dir.Readdir(ctx) + dir.IterateEntries(ctx, func(context.Context, fs.Entry) error { + return nil + }) } } diff --git a/fs/localfs/shallow_fs.go b/fs/localfs/shallow_fs.go index a40985ec8..058da4fee 100644 --- a/fs/localfs/shallow_fs.go +++ b/fs/localfs/shallow_fs.go @@ -119,10 +119,6 @@ func (fsd *shallowFilesystemDirectory) Child(ctx context.Context, name string) ( return nil, errors.New("shallowFilesystemDirectory.Child not supported") } -func (fsd *shallowFilesystemDirectory) Readdir(ctx context.Context) (fs.Entries, error) { - return nil, errors.New("shallowFilesystemDirectory.Readdir not supported") -} - func (fsd *shallowFilesystemDirectory) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { return errors.New("shallowFilesystemDirectory.IterateEntries not supported") } diff --git a/fs/loggingfs/loggingfs.go b/fs/loggingfs/loggingfs.go index c732bd405..869183a58 100644 --- a/fs/loggingfs/loggingfs.go +++ b/fs/loggingfs/loggingfs.go @@ -33,19 +33,23 @@ func (ld *loggingDirectory) Child(ctx context.Context, name string) (fs.Entry, e return wrapWithOptions(entry, ld.options, ld.relativePath+"/"+entry.Name()), nil } -func (ld *loggingDirectory) Readdir(ctx context.Context) (fs.Entries, error) { +func (ld *loggingDirectory) IterateEntries(ctx context.Context, callback func(context.Context, fs.Entry) error) error { timer := timetrack.StartTimer() - entries, err := ld.Directory.Readdir(ctx) + entries, err := fs.GetAllEntries(ctx, ld.Directory) dt := timer.Elapsed() ld.options.printf(ld.options.prefix+"Readdir(%v) took %v and returned %v items", ld.relativePath, dt, len(entries)) - loggingEntries := make(fs.Entries, len(entries)) - for i, entry := range entries { - loggingEntries[i] = wrapWithOptions(entry, ld.options, ld.relativePath+"/"+entry.Name()) + if err != nil { + return err } - // nolint:wrapcheck - return loggingEntries, err + for _, e := range entries { + if err2 := callback(ctx, wrapWithOptions(e, ld.options, ld.relativePath+"/"+e.Name())); err2 != nil { + return err2 + } + } + + return nil } type loggingFile struct { diff --git a/fs/virtualfs/virtualfs.go b/fs/virtualfs/virtualfs.go index 8569ebbb3..b2ce67062 100644 --- a/fs/virtualfs/virtualfs.go +++ b/fs/virtualfs/virtualfs.go @@ -64,26 +64,27 @@ func (e *virtualEntry) LocalFilesystemPath() string { // staticDirectory is an in-memory implementation of fs.Directory. type staticDirectory struct { virtualEntry - entries fs.Entries + entries []fs.Entry } // Child gets the named child of a directory. func (sd *staticDirectory) Child(ctx context.Context, name string) (fs.Entry, error) { // nolint:wrapcheck - return fs.ReadDirAndFindChild(ctx, sd, name) -} - -// Readdir gets the contents of a directory. -func (sd *staticDirectory) Readdir(ctx context.Context) (fs.Entries, error) { - return append(fs.Entries(nil), sd.entries...), nil + return fs.IterateEntriesAndFindChild(ctx, sd, name) } func (sd *staticDirectory) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { - return fs.ReaddirToIterate(ctx, sd, cb) + for _, e := range append([]fs.Entry{}, sd.entries...) { + if err := cb(ctx, e); err != nil { + return err + } + } + + return nil } // NewStaticDirectory returns a virtual static directory. -func NewStaticDirectory(name string, entries fs.Entries) fs.Directory { +func NewStaticDirectory(name string, entries []fs.Entry) fs.Directory { return &staticDirectory{ virtualEntry: virtualEntry{ name: name, diff --git a/fs/virtualfs/virtualfs_test.go b/fs/virtualfs/virtualfs_test.go index 27868d86b..a88e5115e 100644 --- a/fs/virtualfs/virtualfs_test.go +++ b/fs/virtualfs/virtualfs_test.go @@ -28,7 +28,7 @@ func TestStreamingFile(t *testing.T) { filename := "stream-file" f := StreamingFileFromReader(filename, r) - rootDir := NewStaticDirectory("root", fs.Entries{f}) + rootDir := NewStaticDirectory("root", []fs.Entry{f}) e, err := rootDir.Child(context.TODO(), filename) if err != nil { @@ -39,7 +39,7 @@ func TestStreamingFile(t *testing.T) { t.Fatalf("did not get expected filename: (actual) %v != %v (expected)", e.Name(), filename) } - entries, err := rootDir.Readdir(context.TODO()) + entries, err := fs.GetAllEntries(context.TODO(), rootDir) if err != nil { t.Fatalf("error getting dir entries %v", err) } diff --git a/internal/diff/diff.go b/internal/diff/diff.go index 414c60dcd..148293271 100644 --- a/internal/diff/diff.go +++ b/internal/diff/diff.go @@ -52,19 +52,19 @@ func maybeOID(e fs.Entry) string { func (c *Comparer) compareDirectories(ctx context.Context, dir1, dir2 fs.Directory, parent string) error { log(ctx).Debugf("comparing directories %v (%v and %v)", parent, maybeOID(dir1), maybeOID(dir2)) - var entries1, entries2 fs.Entries + var entries1, entries2 []fs.Entry var err error if dir1 != nil { - entries1, err = dir1.Readdir(ctx) + entries1, err = fs.GetAllEntries(ctx, dir1) if err != nil { return errors.Wrapf(err, "unable to read first directory %v", parent) } } if dir2 != nil { - entries2, err = dir2.Readdir(ctx) + entries2, err = fs.GetAllEntries(ctx, dir2) if err != nil { return errors.Wrapf(err, "unable to read second directory %v", parent) } @@ -206,7 +206,7 @@ func compareEntry(e1, e2 fs.Entry, fullpath string, out io.Writer) bool { return equal } -func (c *Comparer) compareDirectoryEntries(ctx context.Context, entries1, entries2 fs.Entries, dirPath string) error { +func (c *Comparer) compareDirectoryEntries(ctx context.Context, entries1, entries2 []fs.Entry, dirPath string) error { e1byname := map[string]fs.Entry{} for _, e1 := range entries1 { e1byname[e1.Name()] = e1 diff --git a/internal/fshasher/fshasher.go b/internal/fshasher/fshasher.go index 9334d2288..14dd84a4a 100644 --- a/internal/fshasher/fshasher.go +++ b/internal/fshasher/fshasher.go @@ -110,18 +110,15 @@ func header(ctx context.Context, fullpath string, e os.FileInfo) (*tar.Header, e } func writeDirectory(ctx context.Context, tw *tar.Writer, fullpath string, d fs.Directory) error { - entries, err := d.Readdir(ctx) - if err != nil { - return err - } - - for _, e := range entries { - if err := write(ctx, tw, path.Join(fullpath, e.Name()), e); err != nil { - return err + err := d.IterateEntries(ctx, func(c context.Context, e fs.Entry) error { + if err2 := write(c, tw, path.Join(fullpath, e.Name()), e); err2 != nil { + return err2 } - } - return nil + return nil + }) + + return err } func writeFile(ctx context.Context, w io.Writer, f fs.File) error { diff --git a/internal/fusemount/fusefs.go b/internal/fusemount/fusefs.go index 7c08c7820..feed3adbb 100644 --- a/internal/fusemount/fusefs.go +++ b/internal/fusemount/fusefs.go @@ -114,7 +114,7 @@ func (dir *fuseDirectoryNode) directory() fs.Directory { } func (dir *fuseDirectoryNode) Lookup(ctx context.Context, fileName string, out *fuse.EntryOut) (*gofusefs.Inode, syscall.Errno) { - entries, err := dir.directory().Readdir(ctx) + e, err := dir.directory().Child(ctx, fileName) if err != nil { if os.IsNotExist(err) { return nil, syscall.ENOENT @@ -125,7 +125,6 @@ func (dir *fuseDirectoryNode) Lookup(ctx context.Context, fileName string, out * return nil, syscall.EIO } - e := entries.FindByName(fileName) if e == nil { return nil, syscall.ENOENT } @@ -147,18 +146,19 @@ func (dir *fuseDirectoryNode) Lookup(ctx context.Context, fileName string, out * } func (dir *fuseDirectoryNode) Readdir(ctx context.Context) (gofusefs.DirStream, syscall.Errno) { - entries, err := dir.directory().Readdir(ctx) - if err != nil { - log(ctx).Errorf("error reading directory %v: %v", dir.entry.Name(), err) - return nil, syscall.EIO - } - + // TODO: Slice not required as DirStream is also an iterator. result := []fuse.DirEntry{} - for _, e := range entries { + + err := dir.directory().IterateEntries(ctx, func(innerCtx context.Context, e fs.Entry) error { result = append(result, fuse.DirEntry{ Name: e.Name(), Mode: entryToFuseMode(e), }) + return nil + }) + if err != nil { + log(ctx).Errorf("error reading directory %v: %v", dir.entry.Name(), err) + return nil, syscall.EIO } return gofusefs.NewListDirStream(result), gofusefs.OK diff --git a/internal/mockfs/mockfs.go b/internal/mockfs/mockfs.go index 6aaee8cae..da5a8c70d 100644 --- a/internal/mockfs/mockfs.go +++ b/internal/mockfs/mockfs.go @@ -81,7 +81,7 @@ func (e *entry) LocalFilesystemPath() string { type Directory struct { entry - children fs.Entries + children []fs.Entry readdirError error onReaddir func() } @@ -209,7 +209,7 @@ func (imd *Directory) addChild(e fs.Entry) { } imd.children = append(imd.children, e) - imd.children.Sort() + fs.Sort(imd.children) } func (imd *Directory) resolveSubdir(name string) (parent *Directory, leaf string) { @@ -226,7 +226,7 @@ func (imd *Directory) Subdir(name ...string) *Directory { i := imd for _, n := range name { - i2 := i.children.FindByName(n) + i2 := fs.FindByName(i.children, n) if i2 == nil { panic(fmt.Sprintf("'%s' not found in '%s'", n, i.Name())) } @@ -255,7 +255,7 @@ func (imd *Directory) Remove(name string) { imd.children = newChildren } -// FailReaddir causes the subsequent Readdir() calls to fail with the specified error. +// FailReaddir causes the subsequent IterateEntries() calls to fail with the specified error. func (imd *Directory) FailReaddir(err error) { imd.readdirError = err } @@ -267,7 +267,7 @@ func (imd *Directory) OnReaddir(cb func()) { // Child gets the named child of a directory. func (imd *Directory) Child(ctx context.Context, name string) (fs.Entry, error) { - e := imd.children.FindByName(name) + e := fs.FindByName(imd.children, name) if e != nil { return e, nil } @@ -277,20 +277,21 @@ func (imd *Directory) Child(ctx context.Context, name string) (fs.Entry, error) // IterateEntries calls the given callback on each entry in the directory. func (imd *Directory) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { - return fs.ReaddirToIterate(ctx, imd, cb) -} - -// Readdir gets the contents of a directory. -func (imd *Directory) Readdir(ctx context.Context) (fs.Entries, error) { if imd.readdirError != nil { - return nil, imd.readdirError + return imd.readdirError } if imd.onReaddir != nil { imd.onReaddir() } - return append(fs.Entries(nil), imd.children...), nil + for _, e := range append([]fs.Entry{}, imd.children...) { + if err := cb(ctx, e); err != nil { + return err + } + } + + return nil } // File is an in-memory fs.File capable of simulating failures. diff --git a/internal/webdavmount/webdavmount.go b/internal/webdavmount/webdavmount.go index 024c74a7f..1d5591e26 100644 --- a/internal/webdavmount/webdavmount.go +++ b/internal/webdavmount/webdavmount.go @@ -109,28 +109,32 @@ type webdavDir struct { // nolint:gochecknoglobals var symlinksAreUnsupportedLogged = new(int32) +// TODO: (bug) This incorrectly truncates the entries in the directory and does not allow pagination. func (d *webdavDir) Readdir(n int) ([]os.FileInfo, error) { - entries, err := d.entry.Readdir(d.ctx) - if err != nil { - return nil, errors.Wrap(err, "error reading directory") - } - - if n > 0 && n < len(entries) { - entries = entries[0:n] - } - var fis []os.FileInfo - for _, e := range entries { + foundEntries := 0 + + err := d.entry.IterateEntries(d.ctx, func(innerCtx context.Context, e fs.Entry) error { + if n > 0 && n <= foundEntries { + return nil + } + + foundEntries++ + if _, isSymlink := e.(fs.Symlink); isSymlink { if atomic.AddInt32(symlinksAreUnsupportedLogged, 1) == 1 { log(d.ctx).Errorf("Mounting directories containing symbolic links using WebDAV is not supported. The link entries will be skipped.") } - continue + return nil } fis = append(fis, &webdavFileInfo{e}) + return nil + }) + if err != nil { + return nil, errors.Wrap(err, "error reading directory") } return fis, nil @@ -213,12 +217,13 @@ func (w *webdavFS) findEntry(ctx context.Context, path string) (fs.Entry, error) return nil, errors.Errorf("%q not found in %q (not a directory)", p, strings.Join(parts[0:i], "/")) } - entries, err := d.Readdir(ctx) + var err error + + e, err = d.Child(ctx, p) if err != nil { return nil, errors.Wrap(err, "error reading directory") } - e = entries.FindByName(p) if e == nil { return nil, errors.Errorf("%q not found in %q (not found)", p, strings.Join(parts[0:i], "/")) } diff --git a/snapshot/restore/restore.go b/snapshot/restore/restore.go index 664b4cda6..b306375d7 100644 --- a/snapshot/restore/restore.go +++ b/snapshot/restore/restore.go @@ -257,7 +257,7 @@ func (c *copier) copyDirectory(ctx context.Context, d fs.Directory, targetPath s } func (c *copier) copyDirectoryContent(ctx context.Context, d fs.Directory, targetPath string, currentdepth, maxdepth int32, onCompletion parallelwork.CallbackFunc) error { - entries, err := d.Readdir(ctx) + entries, err := fs.GetAllEntries(ctx, d) if err != nil { return errors.Wrap(err, "error reading directory") } diff --git a/snapshot/snapshotfs/all_sources.go b/snapshot/snapshotfs/all_sources.go index 572253aff..44c848a87 100644 --- a/snapshot/snapshotfs/all_sources.go +++ b/snapshot/snapshotfs/all_sources.go @@ -55,17 +55,13 @@ func (s *repositoryAllSources) LocalFilesystemPath() string { func (s *repositoryAllSources) Child(ctx context.Context, name string) (fs.Entry, error) { // nolint:wrapcheck - return fs.ReadDirAndFindChild(ctx, s, name) + return fs.IterateEntriesAndFindChild(ctx, s, name) } func (s *repositoryAllSources) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { - return fs.ReaddirToIterate(ctx, s, cb) -} - -func (s *repositoryAllSources) Readdir(ctx context.Context) (fs.Entries, error) { srcs, err := snapshot.ListSources(ctx, s.rep) if err != nil { - return nil, errors.Wrap(err, "error listing sources") + return errors.Wrap(err, "error listing sources") } users := map[string]bool{} @@ -82,18 +78,19 @@ func (s *repositoryAllSources) Readdir(ctx context.Context) (fs.Entries, error) name2safe = disambiguateSafeNames(name2safe) - var result fs.Entries for u := range users { - result = append(result, &sourceDirectories{ + e := &sourceDirectories{ rep: s.rep, userHost: u, name: name2safe[u], - }) + } + + if err2 := cb(ctx, e); err2 != nil { + return err2 + } } - result.Sort() - - return result, nil + return nil } // AllSourcesEntry returns fs.Directory that contains the list of all snapshot sources found in the repository. diff --git a/snapshot/snapshotfs/estimate.go b/snapshot/snapshotfs/estimate.go index 4fbe92492..d1fc20220 100644 --- a/snapshot/snapshotfs/estimate.go +++ b/snapshot/snapshotfs/estimate.go @@ -6,6 +6,8 @@ import ( "path/filepath" "sync/atomic" + "github.com/pkg/errors" + "github.com/kopia/kopia/fs" "github.com/kopia/kopia/fs/ignorefs" "github.com/kopia/kopia/internal/units" @@ -106,6 +108,10 @@ func Estimate(ctx context.Context, rep repo.Repository, entry fs.Directory, poli } func estimate(ctx context.Context, relativePath string, entry fs.Entry, policyTree *policy.Tree, stats *snapshot.Stats, ib, eb SampleBuckets, ed *[]string, progress EstimateProgress, maxExamplesPerBucket int) error { + type processEntryError struct { + error + } + // see if the context got canceled select { case <-ctx.Done(): @@ -121,8 +127,20 @@ func estimate(ctx context.Context, relativePath string, entry fs.Entry, policyTr progress.Processing(ctx, relativePath) - children, err := entry.Readdir(ctx) + err := entry.IterateEntries(ctx, func(c context.Context, child fs.Entry) error { + if err2 := estimate(ctx, filepath.Join(relativePath, child.Name()), child, policyTree.Child(child.Name()), stats, ib, eb, ed, progress, maxExamplesPerBucket); err2 != nil { + return processEntryError{err2} + } + + return nil + }) + + var funcErr processEntryError if err != nil { + if errors.As(err, &funcErr) { + return funcErr.error + } + isIgnored := policyTree.EffectivePolicy().ErrorHandlingPolicy.IgnoreDirectoryErrors.OrDefault(false) if isIgnored { @@ -132,12 +150,6 @@ func estimate(ctx context.Context, relativePath string, entry fs.Entry, policyTr } progress.Error(ctx, relativePath, err, isIgnored) - } else { - for _, child := range children { - if err := estimate(ctx, filepath.Join(relativePath, child.Name()), child, policyTree.Child(child.Name()), stats, ib, eb, ed, progress, maxExamplesPerBucket); err != nil { - return err - } - } } progress.Stats(ctx, stats, ib, eb, *ed, false) diff --git a/snapshot/snapshotfs/objref.go b/snapshot/snapshotfs/objref.go index 69a901deb..0203f4c35 100644 --- a/snapshot/snapshotfs/objref.go +++ b/snapshot/snapshotfs/objref.go @@ -45,12 +45,11 @@ func GetNestedEntry(ctx context.Context, startingDir fs.Entry, pathElements []st return nil, errors.Errorf("entry not found %q: parent is not a directory", part) } - entries, err := dir.Readdir(ctx) + e, err := dir.Child(ctx, part) if err != nil { return nil, errors.Wrap(err, "error reading directory") } - e := entries.FindByName(part) if e == nil { return nil, errors.Errorf("entry not found: %q", part) } diff --git a/snapshot/snapshotfs/repofs.go b/snapshot/snapshotfs/repofs.go index 4baba4fec..992a50e9a 100644 --- a/snapshot/snapshotfs/repofs.go +++ b/snapshot/snapshotfs/repofs.go @@ -140,10 +140,6 @@ func (rd *repositoryDirectory) IterateEntries(ctx context.Context, cb func(conte return nil } -func (rd *repositoryDirectory) Readdir(ctx context.Context) (fs.Entries, error) { - return fs.IterateEntriesToReaddir(ctx, rd) -} - func (rd *repositoryDirectory) ensureDirEntriesLoaded(ctx context.Context) error { rd.mu.Lock() defer rd.mu.Unlock() @@ -288,7 +284,9 @@ func SnapshotRoot(rep repo.Repository, man *snapshot.Manifest) (fs.Entry, error) func AutoDetectEntryFromObjectID(ctx context.Context, rep repo.Repository, oid object.ID, maybeName string) fs.Entry { if IsDirectoryID(oid) { dirEntry := DirectoryEntry(rep, oid, nil) - if _, err := dirEntry.Readdir(ctx); err == nil { + if err := dirEntry.IterateEntries(ctx, func(context.Context, fs.Entry) error { + return nil + }); err == nil { repoFSLog(ctx).Debugf("%v auto-detected as directory", oid) return dirEntry } diff --git a/snapshot/snapshotfs/snapshot_tree_walker.go b/snapshot/snapshotfs/snapshot_tree_walker.go index 79a7e90ea..0adfe2e7b 100644 --- a/snapshot/snapshotfs/snapshot_tree_walker.go +++ b/snapshot/snapshotfs/snapshot_tree_walker.go @@ -48,7 +48,9 @@ func (w *TreeWalker) ReportError(ctx context.Context, entryPath string, err erro repoFSLog(ctx).Errorf("error processing %v: %v", entryPath, err) - if len(w.errors) < w.options.MaxErrors { + // Record one error if we can't get too many errors so that at least that one + // can be returned if it's the only one. + if len(w.errors) < w.options.MaxErrors || (w.options.MaxErrors <= 0 && len(w.errors) == 0) { w.errors = append(w.errors, err) } @@ -103,24 +105,20 @@ func (w *TreeWalker) processEntry(ctx context.Context, e fs.Entry, entryPath str } func (w *TreeWalker) processDirEntry(ctx context.Context, dir fs.Directory, entryPath string) { + type errStop struct { + error + } + var ag workshare.AsyncGroup defer ag.Wait() - entries, err := dir.Readdir(ctx) - if err != nil { - w.ReportError(ctx, entryPath, errors.Wrap(err, "error reading directory")) - return - } - - for _, ent := range entries { - ent := ent - + err := dir.IterateEntries(ctx, func(c context.Context, ent fs.Entry) error { if w.TooManyErrors() { - break + return errStop{errors.New("")} } if w.alreadyProcessed(ent) { - continue + return nil } childPath := path.Join(entryPath, ent.Name()) @@ -132,6 +130,13 @@ func (w *TreeWalker) processDirEntry(ctx context.Context, dir fs.Directory, entr } else { w.processEntry(ctx, ent, childPath) } + + return nil + }) + + var stopped errStop + if err != nil && !errors.As(err, &stopped) { + w.ReportError(ctx, entryPath, errors.Wrap(err, "error reading directory")) } } diff --git a/snapshot/snapshotfs/snapshot_tree_walker_test.go b/snapshot/snapshotfs/snapshot_tree_walker_test.go index 6c84e5baa..3d40b6892 100644 --- a/snapshot/snapshotfs/snapshot_tree_walker_test.go +++ b/snapshot/snapshotfs/snapshot_tree_walker_test.go @@ -146,7 +146,7 @@ func TestSnapshotTreeWalker_MultipleErrors(t *testing.T) { dir1.AddFile("file11", []byte{1, 2, 3}, 0o644) dir2.AddFile("file21", []byte{1, 2, 3, 4}, 0o644) - dir2.AddFile("file22", []byte{1, 2, 3}, 0o644) // same content as dir11/file11 + dir2.AddFile("file22", []byte{1, 2, 3, 4, 5}, 0o644) u := snapshotfs.NewUploader(env.RepositoryWriter) man, err := u.Upload(ctx, sourceRoot, nil, snapshot.SourceInfo{}) @@ -161,3 +161,50 @@ func TestSnapshotTreeWalker_MultipleErrors(t *testing.T) { require.Error(t, err) require.Equal(t, "encountered 2 errors", err.Error()) } + +func TestSnapshotTreeWalker_MultipleErrorsSameOID(t *testing.T) { + someErr1 := errors.Errorf("some error") + + w := snapshotfs.NewTreeWalker( + snapshotfs.TreeWalkerOptions{ + Parallelism: 1, + MaxErrors: -1, + EntryCallback: func(ctx context.Context, entry fs.Entry, oid object.ID, entryPath string) error { + if entryPath == "root-dir/dir1/file11" { + return someErr1 + } + + if entryPath == "root-dir/dir2/file22" { + return someErr1 + } + + return nil + }, + }) + defer w.Close() + + ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant) + + sourceRoot := mockfs.NewDirectory() + require.Error(t, w.Process(ctx, sourceRoot, "root-dir")) + + dir1 := sourceRoot.AddDir("dir1", 0o755) + dir2 := sourceRoot.AddDir("dir2", 0o755) + + dir1.AddFile("file11", []byte{1, 2, 3}, 0o644) + dir2.AddFile("file21", []byte{1, 2, 3, 4}, 0o644) + dir2.AddFile("file22", []byte{1, 2, 3}, 0o644) // same content as dir11/file11 + + u := snapshotfs.NewUploader(env.RepositoryWriter) + man, err := u.Upload(ctx, sourceRoot, nil, snapshot.SourceInfo{}) + require.NoError(t, err) + + uploadedRoot, err := snapshotfs.SnapshotRoot(env.Repository, man) + require.NoError(t, err) + + require.NoError(t, env.RepositoryWriter.Flush(ctx)) + + err = w.Process(ctx, uploadedRoot, "root-dir") + require.Error(t, err) + require.True(t, errors.Is(err, someErr1)) +} diff --git a/snapshot/snapshotfs/source_directories.go b/snapshot/snapshotfs/source_directories.go index 692f51d93..3e07d52da 100644 --- a/snapshot/snapshotfs/source_directories.go +++ b/snapshot/snapshotfs/source_directories.go @@ -59,17 +59,13 @@ func (s *sourceDirectories) LocalFilesystemPath() string { func (s *sourceDirectories) Child(ctx context.Context, name string) (fs.Entry, error) { // nolint:wrapcheck - return fs.ReadDirAndFindChild(ctx, s, name) + return fs.IterateEntriesAndFindChild(ctx, s, name) } func (s *sourceDirectories) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { - return fs.ReaddirToIterate(ctx, s, cb) -} - -func (s *sourceDirectories) Readdir(ctx context.Context) (fs.Entries, error) { sources0, err := snapshot.ListSources(ctx, s.rep) if err != nil { - return nil, errors.Wrap(err, "unable to list sources") + return errors.Wrap(err, "unable to list sources") } // step 1 - filter sources. @@ -92,15 +88,15 @@ func (s *sourceDirectories) Readdir(ctx context.Context) (fs.Entries, error) { name2safe = disambiguateSafeNames(name2safe) - var result fs.Entries - for _, src := range sources { - result = append(result, &sourceSnapshots{s.rep, src, name2safe[src.Path]}) + e := &sourceSnapshots{s.rep, src, name2safe[src.Path]} + + if err2 := cb(ctx, e); err2 != nil { + return err2 + } } - result.Sort() - - return result, nil + return nil } func disambiguateSafeNames(m map[string]string) map[string]string { diff --git a/snapshot/snapshotfs/source_directories_test.go b/snapshot/snapshotfs/source_directories_test.go index 7686dce71..31c82bd2e 100644 --- a/snapshot/snapshotfs/source_directories_test.go +++ b/snapshot/snapshotfs/source_directories_test.go @@ -50,50 +50,54 @@ func TestAllSources(t *testing.T) { as := AllSourcesEntry(env.RepositoryWriter) gotNames := iterateAllNames(ctx, t, as, "") - wantNames := []string{ - "another-user@some-host/", - "another-user@some-host/__root/", - "another-user@some-host/__root/20200101-120103/", - "another-user@some-host/tmp/", - "another-user@some-host/tmp/20200101-120103/", - "another-user@some-host/var/", - "another-user@some-host/var/20200101-120103/", - "some-user@some-host/", - "some-user@some-host/c_some_Path/", - "some-user@some-host/c_some_Path/20200101-120103/", - "some-user@some-host/c_some_path (2)/", - "some-user@some-host/c_some_path (2)/20200101-120103/", - "some-user@some-host/c_some_path (2)/20200101-120104/", - "some-user@some-host/c_some_path (2)/20200101-120105/", - "some_user@some-host/", - "some_user@some-host/c_some_path/", - "some_user@some-host/c_some_path/20200101-130103/", - "some_user@some-host/c_some_path/20200101-130104/", - "some_user@some-host (2)/", - "some_user@some-host (2)/c_some_path/", - "some_user@some-host (2)/c_some_path/20200101-140103/", - "some_user@some-host (2)/c_some_path/20200101-140104/", + wantNames := map[string]struct{}{ + "another-user@some-host/": {}, + "another-user@some-host/__root/": {}, + "another-user@some-host/__root/20200101-120103/": {}, + "another-user@some-host/tmp/": {}, + "another-user@some-host/tmp/20200101-120103/": {}, + "another-user@some-host/var/": {}, + "another-user@some-host/var/20200101-120103/": {}, + "some-user@some-host/": {}, + "some-user@some-host/c_some_Path/": {}, + "some-user@some-host/c_some_Path/20200101-120103/": {}, + "some-user@some-host/c_some_path (2)/": {}, + "some-user@some-host/c_some_path (2)/20200101-120103/": {}, + "some-user@some-host/c_some_path (2)/20200101-120104/": {}, + "some-user@some-host/c_some_path (2)/20200101-120105/": {}, + "some_user@some-host/": {}, + "some_user@some-host/c_some_path/": {}, + "some_user@some-host/c_some_path/20200101-130103/": {}, + "some_user@some-host/c_some_path/20200101-130104/": {}, + "some_user@some-host (2)/": {}, + "some_user@some-host (2)/c_some_path/": {}, + "some_user@some-host (2)/c_some_path/20200101-140103/": {}, + "some_user@some-host (2)/c_some_path/20200101-140104/": {}, } require.Equal(t, wantNames, gotNames) } -func iterateAllNames(ctx context.Context, t *testing.T, dir fs.Directory, prefix string) []string { +func iterateAllNames(ctx context.Context, t *testing.T, dir fs.Directory, prefix string) map[string]struct{} { t.Helper() - entries, err := dir.Readdir(ctx) - require.NoError(t, err) + result := map[string]struct{}{} - result := []string{} - - for _, ent := range entries { + err := dir.IterateEntries(ctx, func(innerCtx context.Context, ent fs.Entry) error { if ent.IsDir() { - result = append(result, prefix+ent.Name()+"/") - result = append(result, iterateAllNames(ctx, t, ent.(fs.Directory), prefix+ent.Name()+"/")...) + result[prefix+ent.Name()+"/"] = struct{}{} + childEntries := iterateAllNames(ctx, t, ent.(fs.Directory), prefix+ent.Name()+"/") + + for k, v := range childEntries { + result[k] = v + } } else { - result = append(result, prefix+ent.Name()) + result[prefix+ent.Name()] = struct{}{} } - } + + return nil + }) + require.NoError(t, err) return result } diff --git a/snapshot/snapshotfs/source_snapshots.go b/snapshot/snapshotfs/source_snapshots.go index 004a126a8..19d662564 100644 --- a/snapshot/snapshotfs/source_snapshots.go +++ b/snapshot/snapshotfs/source_snapshots.go @@ -57,21 +57,15 @@ func (s *sourceSnapshots) LocalFilesystemPath() string { func (s *sourceSnapshots) Child(ctx context.Context, name string) (fs.Entry, error) { // nolint:wrapcheck - return fs.ReadDirAndFindChild(ctx, s, name) + return fs.IterateEntriesAndFindChild(ctx, s, name) } func (s *sourceSnapshots) IterateEntries(ctx context.Context, cb func(context.Context, fs.Entry) error) error { - return fs.ReaddirToIterate(ctx, s, cb) -} - -func (s *sourceSnapshots) Readdir(ctx context.Context) (fs.Entries, error) { manifests, err := snapshot.ListSnapshots(ctx, s.rep, s.src) if err != nil { - return nil, errors.Wrap(err, "unable to list snapshots") + return errors.Wrap(err, "unable to list snapshots") } - var result fs.Entries - for _, m := range manifests { name := m.StartTime.Format("20060102-150405") if m.IncompleteReason != "" { @@ -90,12 +84,14 @@ func (s *sourceSnapshots) Readdir(ctx context.Context) (fs.Entries, error) { de.DirSummary = m.RootEntry.DirSummary } - result = append(result, EntryFromDirEntry(s.rep, de)) + e := EntryFromDirEntry(s.rep, de) + + if err2 := cb(ctx, e); err2 != nil { + return err2 + } } - result.Sort() - - return result, nil + return nil } var _ fs.Directory = (*sourceSnapshots)(nil) diff --git a/snapshot/snapshotfs/upload_test.go b/snapshot/snapshotfs/upload_test.go index 7f5832405..026019e05 100644 --- a/snapshot/snapshotfs/upload_test.go +++ b/snapshot/snapshotfs/upload_test.go @@ -663,7 +663,7 @@ func TestUpload_VirtualDirectoryWithStreamingFile(t *testing.T) { w.Close() - staticRoot := virtualfs.NewStaticDirectory("rootdir", fs.Entries{ + staticRoot := virtualfs.NewStaticDirectory("rootdir", []fs.Entry{ virtualfs.StreamingFileFromReader("stream-file", r), }) diff --git a/tests/tools/kopiaclient/kopiaclient.go b/tests/tools/kopiaclient/kopiaclient.go index 25ae25b4a..54fc7e4aa 100644 --- a/tests/tools/kopiaclient/kopiaclient.go +++ b/tests/tools/kopiaclient/kopiaclient.go @@ -209,7 +209,7 @@ func (kc *KopiaClient) getStorage(ctx context.Context, repoDir, bucketName strin // getSourceForKeyVal creates a virtual directory for `key` that contains a single virtual file that // reads its contents from `val`. func (kc *KopiaClient) getSourceForKeyVal(key string, val []byte) fs.Entry { - return virtualfs.NewStaticDirectory(key, fs.Entries{ + return virtualfs.NewStaticDirectory(key, []fs.Entry{ virtualfs.StreamingFileFromReader(dataFileName, bytes.NewReader(val)), }) }