From 6c05c4a00b91aa542267d8e32a8254774799d68d Mon Sep 17 00:00:00 2001 From: Cody Rose Date: Mon, 9 Mar 2026 16:02:57 -0400 Subject: [PATCH] Stop growing filesystem resume data (#4797) #4742 (4563dde124c011b7ab615dbe531b45f3a6193b96) introduced a change to the filesystem source resumption tracking that caused it to start growing linearly with subdirectory count - which causes the payload to get intractably big on large data sets. This commit is an attempt to resolve the issue. Note that the resumption code still has a bug that can cause data to get inadvertently skipped due to mishandling of the internal parallelization of the scan. This bug has been present for a long time and is present in other sources, so fixing it is out of scope here. This commit _also_ introduces a new bug related to the fact that lexicographic sorting is not completely appropriate for the resumption check. This needs to be cleaned up as a fast follow, but it's still less serious than the current bug that prevents all scans of large data sets. --- pkg/sources/filesystem/filesystem.go | 84 +++++-- pkg/sources/filesystem/filesystem_test.go | 284 +++++++++++++++++++++- pkg/sources/sources.go | 17 -- 3 files changed, 348 insertions(+), 37 deletions(-) diff --git a/pkg/sources/filesystem/filesystem.go b/pkg/sources/filesystem/filesystem.go index e93ac6154..e54ccc220 100644 --- a/pkg/sources/filesystem/filesystem.go +++ b/pkg/sources/filesystem/filesystem.go @@ -5,6 +5,7 @@ import ( "io" "os" "path/filepath" + "strings" "github.com/go-errors/errors" "github.com/go-logr/logr" @@ -136,7 +137,7 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ . initialDepth := 1 err = s.scanSymlink(ctx, cleanPath, chunksChan, workerPool, initialDepth, path) _ = workerPool.Wait() - s.ClearEncodedResumeContainingId(path + "#") + s.ClearEncodedResumeInfoFor(path) } else if fileInfo.IsDir() { ctx.Logger().V(5).Info("Root path is a dir", "path", cleanPath) workerPool := new(errgroup.Group) @@ -144,7 +145,7 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk, _ . initialDepth := 1 err = s.scanDir(ctx, cleanPath, chunksChan, workerPool, initialDepth, path) _ = workerPool.Wait() - s.ClearEncodedResumeContainingId(path + "#") + s.ClearEncodedResumeInfoFor(path) } else { if !fileInfo.Mode().IsRegular() { logger.Info("skipping non-regular file", "path", cleanPath) @@ -217,13 +218,11 @@ func (s *Source) scanSymlink( if s.filter != nil && !s.filter.Pass(resolvedPath) { return nil } - resumptionKey := rootPath + "#" + path - startState := s.GetEncodedResumeInfoFor(resumptionKey) - resuming := startState != "" - if resuming && startState == resolvedPath { - ctx.Logger().V(5).Info("skipping symlink, already scanned", "path", resolvedPath) - return nil - } + + // Use a single resumption key for the entire scan rooted at rootPath. + // Resume checks are handled by the calling scanDir function. + resumptionKey := rootPath + workerPool.Go(func() error { if !fileInfo.Mode().Type().IsRegular() { ctx.Logger().V(5).Info("skipping non-regular file", "path", resolvedPath) @@ -232,7 +231,7 @@ func (s *Source) scanSymlink( if err := s.scanFile(ctx, resolvedPath, chunksChan); err != nil { ctx.Logger().Error(err, "error scanning file", "path", resolvedPath) } - s.SetEncodedResumeInfoFor(resumptionKey, resolvedPath) + s.SetEncodedResumeInfoFor(resumptionKey, path) return nil }) @@ -249,12 +248,35 @@ func (s *Source) scanDir( ) error { // check if the full path is not matching any pattern in include // FilterRuleSet and matching any exclude FilterRuleSet. - resumptionKey := rootPath + "#" + path if s.filter != nil && s.filter.ShouldExclude(path) { return nil } - startState := s.GetEncodedResumeInfoFor(resumptionKey) - resuming := startState != "" + + // Use a single resumption key for the entire scan rooted at rootPath. + // The value stored is the full path of the last successfully scanned file. + // This avoids accumulating separate entries for each subdirectory visited. + resumptionKey := rootPath + resumeAfter := s.GetEncodedResumeInfoFor(resumptionKey) + + // Only consider resumption if the resume point is within this directory's subtree. + // Since os.ReadDir returns entries sorted by filename: + // - If we're scanning /root/ccc and the resume point is /root/bbb/file.txt, + // we've already passed it (bbb < ccc) and should process ccc normally. + // - If we're scanning /root/aaa and the resume point is /root/bbb/file.txt, + // we haven't reached it yet (aaa < bbb), so aaa was already fully scanned + // and should be skipped entirely. + if resumeAfter != "" && !strings.HasPrefix(resumeAfter, path+string(filepath.Separator)) && resumeAfter != path { + // Resume point is not in this subtree. Compare paths to determine if we + // should skip this directory (already scanned) or process it (already passed). + if path < resumeAfter { + // This directory comes before the resume point lexicographically, + // meaning it was already fully scanned. Skip it entirely. + return nil + } + // This directory comes after the resume point, so we've already passed + // the resume point. Process this directory normally. + resumeAfter = "" + } ctx.Logger().V(5).Info("Full path found is", "fullPath", path) @@ -271,11 +293,35 @@ func (s *Source) scanDir( } } - if resuming { - if entryPath == startState { - resuming = false + // Skip entries until we pass the resume point. + // We don't clear the resume info when we find the resume point - instead we + // keep it set until a new file is scanned. This ensures we don't lose progress + // if the scan is interrupted between finding the resume point and scanning + // the next file. + if resumeAfter != "" { + // If this entry is the resume point, stop skipping. + if entryPath == resumeAfter { + resumeAfter = "" + continue // Skip the resume point itself since it was already processed. } - } else if entry.Type()&os.ModeSymlink != 0 { + // If the resume point is within this entry (a descendant), we need to + // traverse into it to find where to resume. + if entry.IsDir() && strings.HasPrefix(resumeAfter, entryPath+string(filepath.Separator)) { + // Recurse into this directory to find the resume point. + if err := s.scanDir(ctx, entryPath, chunksChan, workerPool, depth, rootPath); err != nil { + ctx.Logger().Error(err, "error scanning directory", "path", entryPath) + } + // After recursing, clear local resumeAfter. The child scanDir will have + // handled resumption within its subtree, and subsequent entries in this + // directory should be processed normally. + resumeAfter = "" + continue + } + // Skip this entry - it comes before the resume point in traversal order. + continue + } + + if entry.Type()&os.ModeSymlink != 0 { ctx.Logger().V(5).Info("Entry found is a symlink", "path", entryPath) if !s.canFollowSymlinks() { // If the file or directory is a symlink but the followSymlinks is disable ignore the path @@ -401,7 +447,7 @@ func (s *Source) ChunkUnit(ctx context.Context, unit sources.SourceUnit, reporte initialDepth := 1 scanErr = s.scanSymlink(ctx, cleanPath, ch, workerPool, initialDepth, path) _ = workerPool.Wait() - s.ClearEncodedResumeContainingId(path + "#") + s.ClearEncodedResumeInfoFor(path) } else if fileInfo.IsDir() { ctx.Logger().V(5).Info("Root path is a dir", "path", cleanPath) @@ -411,7 +457,7 @@ func (s *Source) ChunkUnit(ctx context.Context, unit sources.SourceUnit, reporte // TODO: Finer grain error tracking of individual chunks. scanErr = s.scanDir(ctx, cleanPath, ch, workerPool, initialDepth, path) _ = workerPool.Wait() - s.ClearEncodedResumeContainingId(path + "#") + s.ClearEncodedResumeInfoFor(path) } else { ctx.Logger().V(5).Info("Root path is a file", "path", cleanPath) // TODO: Finer grain error tracking of individual diff --git a/pkg/sources/filesystem/filesystem_test.go b/pkg/sources/filesystem/filesystem_test.go index 07803ca72..8d7cb9129 100644 --- a/pkg/sources/filesystem/filesystem_test.go +++ b/pkg/sources/filesystem/filesystem_test.go @@ -1,9 +1,12 @@ package filesystem import ( + "encoding/json" + "fmt" "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -394,7 +397,7 @@ func TestScanSubDirFile(t *testing.T) { // Create an IncludePathsFile with the absolute path of the file includeFilePath := filepath.Join(testDir, "include.txt") err = os.WriteFile(includeFilePath, []byte(strings.ReplaceAll(filePath, `\`, `\\`)+"\n"), 0644) - require.NoError(t, err) + require.NoError(t, err) conn, err := anypb.New(&sourcespb.Filesystem{ IncludePathsFile: includeFilePath, @@ -464,6 +467,285 @@ func TestSkipBinaries(t *testing.T) { require.NotContains(t, processedFiles, binaryFile, "Binary file should be skipped") } +func TestResumptionInfoDoesNotGrowWithSubdirectories(t *testing.T) { + ctx := context.AddLogger(t.Context()) + + // Create a deeply nested directory structure with files at each level. + // Structure: root/dir0/dir1/dir2/.../dir9, each containing a file. + rootDir, err := os.MkdirTemp("", "trufflehog-resumption-test") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(rootDir) }) + + const numSubdirs = 10 + currentDir := rootDir + for i := 0; i < numSubdirs; i++ { + // Create a file in the current directory + filePath := filepath.Join(currentDir, fmt.Sprintf("file%d.txt", i)) + err := os.WriteFile(filePath, []byte(fmt.Sprintf("content %d", i)), 0644) + require.NoError(t, err) + + // Create the next subdirectory + subDir := filepath.Join(currentDir, fmt.Sprintf("subdir%d", i)) + err = os.Mkdir(subDir, 0755) + require.NoError(t, err) + currentDir = subDir + } + // Create a file in the deepest directory + err = os.WriteFile(filepath.Join(currentDir, "deepest.txt"), []byte("deepest"), 0644) + require.NoError(t, err) + + conn, err := anypb.New(&sourcespb.Filesystem{MaxSymlinkDepth: 0}) + require.NoError(t, err) + + // Initialize the source. + s := Source{} + err = s.Init(ctx, "test resumption growth", 0, 0, true, conn, 1) + require.NoError(t, err) + + // Track the maximum size of EncodedResumeInfo during the scan. + var maxResumeInfoSize int + var mu sync.Mutex + + // We need to periodically check the resume info size during scanning. + // Run ChunkUnit in a goroutine and poll the progress. + done := make(chan struct{}) + go func() { + defer close(done) + reporter := sourcestest.TestReporter{} + err := s.ChunkUnit(ctx, sources.CommonSourceUnit{ + ID: rootDir, + }, &reporter) + require.NoError(t, err) + }() + + // Poll the resume info size while scanning is in progress. + ticker := time.NewTicker(1 * time.Millisecond) + defer ticker.Stop() + +polling: + for { + select { + case <-done: + break polling + case <-ticker.C: + progress := s.GetProgress() + mu.Lock() + if len(progress.EncodedResumeInfo) > maxResumeInfoSize { + maxResumeInfoSize = len(progress.EncodedResumeInfo) + } + mu.Unlock() + } + } + + // After scan completes, check the final state. + finalProgress := s.GetProgress() + t.Logf("Final EncodedResumeInfo length: %d", len(finalProgress.EncodedResumeInfo)) + t.Logf("Max EncodedResumeInfo length during scan: %d", maxResumeInfoSize) + + // Parse the resume info to count entries if it's not empty. + if maxResumeInfoSize > 0 { + var resumeMap map[string]string + err := json.Unmarshal([]byte(finalProgress.EncodedResumeInfo), &resumeMap) + if err == nil { + t.Logf("Final resume info entries: %d", len(resumeMap)) + } + } + + // The key assertion: resumption info should NOT grow proportionally with + // the number of subdirectories. During the scan, it should only track the + // current position, not accumulate entries for every directory visited. + // + // With proper implementation, resume info should have at most a few entries + // (e.g., one per directory being actively scanned), not one entry per + // directory that has ever been visited. + // + // A reasonable upper bound for resume info size: each entry is roughly + // "rootPath#subPath": "filePath". With temp paths ~50 chars, one entry is + // ~150 bytes with JSON overhead. For 10 directories, accumulation would + // mean ~1500+ bytes. A non-accumulating implementation should stay well + // under that. + const maxAcceptableResumeInfoSize = 300 // bytes - allows for ~2 entries max + assert.LessOrEqual(t, maxResumeInfoSize, maxAcceptableResumeInfoSize, + "Resume info grew to %d bytes during scan, suggesting accumulation across %d subdirectories. "+ + "Resume info should not accumulate entries for each subdirectory visited.", + maxResumeInfoSize, numSubdirs) +} + +func TestResumptionSkipsAlreadyScannedFiles(t *testing.T) { + ctx := context.Background() + + // Create a directory with files that have predictable alphabetical order. + rootDir, err := os.MkdirTemp("", "trufflehog-resumption-test") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(rootDir) }) + + // Create files with predictable names for sorting. + files := []string{"aaa.txt", "bbb.txt", "ccc.txt", "ddd.txt"} + for _, name := range files { + filePath := filepath.Join(rootDir, name) + err := os.WriteFile(filePath, []byte("content of "+name), 0644) + require.NoError(t, err) + } + + conn, err := anypb.New(&sourcespb.Filesystem{}) + require.NoError(t, err) + + // Initialize the source. + s := Source{} + err = s.Init(ctx, "test resumption", 0, 0, true, conn, 1) + require.NoError(t, err) + + // Pre-set the resume point to simulate a previous interrupted scan. + // Setting it to bbb.txt means we should skip aaa.txt and bbb.txt, + // and only scan ccc.txt and ddd.txt. + resumePoint := filepath.Join(rootDir, "bbb.txt") + s.SetEncodedResumeInfoFor(rootDir, resumePoint) + + // Run the scan. + reporter := sourcestest.TestReporter{} + err = s.ChunkUnit(ctx, sources.CommonSourceUnit{ID: rootDir}, &reporter) + require.NoError(t, err) + + // Collect scanned file names. + scannedFiles := make(map[string]bool) + for _, chunk := range reporter.Chunks { + file := chunk.SourceMetadata.GetFilesystem().GetFile() + scannedFiles[filepath.Base(file)] = true + } + + // Assert only files after the resume point were scanned. + assert.False(t, scannedFiles["aaa.txt"], "aaa.txt should have been skipped (before resume point)") + assert.False(t, scannedFiles["bbb.txt"], "bbb.txt should have been skipped (the resume point itself)") + assert.True(t, scannedFiles["ccc.txt"], "ccc.txt should have been scanned (after resume point)") + assert.True(t, scannedFiles["ddd.txt"], "ddd.txt should have been scanned (after resume point)") + assert.Equal(t, 2, len(reporter.Chunks), "expected exactly 2 files to be scanned") +} + +func TestResumptionWithNestedDirectories(t *testing.T) { + ctx := context.Background() + + // Create a nested directory structure: + // root/ + // aaa/ + // file1.txt + // bbb/ + // file2.txt + // ccc/ + // file3.txt + rootDir, err := os.MkdirTemp("", "trufflehog-resumption-nested-test") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(rootDir) }) + + dirs := []string{"aaa", "bbb", "ccc"} + for i, dir := range dirs { + dirPath := filepath.Join(rootDir, dir) + err := os.Mkdir(dirPath, 0755) + require.NoError(t, err) + + filePath := filepath.Join(dirPath, fmt.Sprintf("file%d.txt", i+1)) + err = os.WriteFile(filePath, []byte(fmt.Sprintf("content of file%d", i+1)), 0644) + require.NoError(t, err) + } + + conn, err := anypb.New(&sourcespb.Filesystem{}) + require.NoError(t, err) + + // Initialize the source. + s := Source{} + err = s.Init(ctx, "test resumption nested", 0, 0, true, conn, 1) + require.NoError(t, err) + + // Pre-set the resume point to bbb/file2.txt. + // This should skip aaa/file1.txt and bbb/file2.txt, only scanning ccc/file3.txt. + resumePoint := filepath.Join(rootDir, "bbb", "file2.txt") + s.SetEncodedResumeInfoFor(rootDir, resumePoint) + + // Run the scan. + reporter := sourcestest.TestReporter{} + err = s.ChunkUnit(ctx, sources.CommonSourceUnit{ID: rootDir}, &reporter) + require.NoError(t, err) + + // Collect scanned file names. + scannedFiles := make(map[string]bool) + for _, chunk := range reporter.Chunks { + file := chunk.SourceMetadata.GetFilesystem().GetFile() + scannedFiles[filepath.Base(file)] = true + } + + // Assert only file3.txt was scanned. + assert.False(t, scannedFiles["file1.txt"], "file1.txt should have been skipped (in aaa/, before resume point)") + assert.False(t, scannedFiles["file2.txt"], "file2.txt should have been skipped (the resume point itself)") + assert.True(t, scannedFiles["file3.txt"], "file3.txt should have been scanned (in ccc/, after resume point)") + assert.Equal(t, 1, len(reporter.Chunks), "expected exactly 1 file to be scanned") +} + +func TestResumptionWithOutOfSubtreeResumePoint(t *testing.T) { + ctx := context.Background() + + // Create a directory structure: + // root/ + // aaa/ + // file1.txt + // bbb/ + // file2.txt + // ccc/ + // file3.txt + // + // This test verifies correct behavior when scanDir is called for a directory + // with a resume point OUTSIDE that directory's subtree. Since os.ReadDir + // returns entries sorted by filename, directories that lexicographically + // precede the resume point were already fully scanned and should be skipped. + rootDir, err := os.MkdirTemp("", "trufflehog-resumption-subtree-test") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(rootDir) }) + + dirs := []string{"aaa", "bbb", "ccc"} + for i, dir := range dirs { + dirPath := filepath.Join(rootDir, dir) + err := os.Mkdir(dirPath, 0755) + require.NoError(t, err) + + filePath := filepath.Join(dirPath, fmt.Sprintf("file%d.txt", i+1)) + err = os.WriteFile(filePath, []byte(fmt.Sprintf("content of file%d", i+1)), 0644) + require.NoError(t, err) + } + + conn, err := anypb.New(&sourcespb.Filesystem{}) + require.NoError(t, err) + + // Initialize the source. + s := Source{} + err = s.Init(ctx, "test resumption subtree", 0, 0, true, conn, 1) + require.NoError(t, err) + + // Pre-set the resume point to bbb/file2.txt using aaaDir as the key. + // This simulates an edge case where scanDir is called directly for a + // directory with a resume point outside its subtree. + aaaDir := filepath.Join(rootDir, "aaa") + resumePoint := filepath.Join(rootDir, "bbb", "file2.txt") + s.SetEncodedResumeInfoFor(aaaDir, resumePoint) + + // Scan the aaa directory with a resume point outside its subtree. + reporter := sourcestest.TestReporter{} + err = s.ChunkUnit(ctx, sources.CommonSourceUnit{ID: aaaDir}, &reporter) + require.NoError(t, err) + + // Collect scanned file names. + scannedFiles := make(map[string]bool) + for _, chunk := range reporter.Chunks { + file := chunk.SourceMetadata.GetFilesystem().GetFile() + scannedFiles[filepath.Base(file)] = true + } + + // file1.txt should NOT be scanned because aaa/ comes before bbb/ + // lexicographically, meaning aaa/ would have been fully processed + // before reaching the resume point. + assert.False(t, scannedFiles["file1.txt"], + "file1.txt should NOT be scanned because aaa/ comes before resume point bbb/file2.txt lexicographically") + assert.Equal(t, 0, len(reporter.Chunks), + "expected 0 files to be scanned since aaa/ was already fully processed before the resume point") +} + // createTempFile is a helper function to create a temporary file in the given // directory with the provided contents. If dir is "", the operating system's // temp directory is used. diff --git a/pkg/sources/sources.go b/pkg/sources/sources.go index 6a8c2b1d5..88b11b85a 100644 --- a/pkg/sources/sources.go +++ b/pkg/sources/sources.go @@ -4,7 +4,6 @@ import ( "encoding/json" "errors" "runtime" - "strings" "sync" "google.golang.org/protobuf/types/known/anypb" @@ -595,22 +594,6 @@ func (p *Progress) ClearEncodedResumeInfoFor(id string) { p.EncodedResumeInfo = marshalEncodedResumeInfo(p.encodedResumeInfoByID) } -// ClearEncodedResumeContainingId removes the encoded resume information -// entries that contain the id -func (p *Progress) ClearEncodedResumeContainingId(id string) { - p.mut.Lock() - defer p.mut.Unlock() - p.ensureEncodedResumeInfoByID() - - for key := range p.encodedResumeInfoByID { - if strings.Contains(key, id) { - delete(p.encodedResumeInfoByID, key) - } - } - - p.EncodedResumeInfo = marshalEncodedResumeInfo(p.encodedResumeInfoByID) -} - // ensureEncodedResumeInfoByID ensures the encodedResumeInfoByID attribute is a // non-nil map. The mutex must be held when calling this function. func (p *Progress) ensureEncodedResumeInfoByID() {