rename file.

Bumps [actions/setup-go](https://github.com/actions/setup-go) from 3 to 4.
- [Release notes](https://github.com/actions/setup-go/releases)
- [Commits](https://github.com/actions/setup-go/compare/v3...v4)

---
updated-dependencies:
- dependency-name: actions/setup-go
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

pkg/cache/memory/memory.go

@@ -5,8 +5,6 @@ import (
 	"time"
 
 	"github.com/patrickmn/go-cache"
-
-	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 )
 
 const (
@@ -27,9 +25,7 @@ func New() *Cache {
 }
 
 // NewWithData constructs a new in-memory cache with existing data.
-func NewWithData(ctx context.Context, data []string) *Cache {
-	ctx.Logger().V(3).Info("Loading cache", "num-items", len(data))
-
+func NewWithData(data []string) *Cache {
 	items := make(map[string]cache.Item, len(data))
 	for _, d := range data {
 		items[d] = cache.Item{Object: d, Expiration: int64(defaultExpiration)}

pkg/cache/memory/memory_test.go

@@ -7,8 +7,6 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
-
-	logContext "github.com/trufflesecurity/trufflehog/v3/pkg/context"
 )
 
 func TestCache(t *testing.T) {
@@ -66,7 +64,7 @@ func TestCache(t *testing.T) {
 }
 
 func TestCache_NewWithData(t *testing.T) {
-	c := NewWithData(logContext.Background(), []string{"key1", "key2", "key3"})
+	c := NewWithData([]string{"key1", "key2", "key3"})
 
 	// Test the count.
 	if c.Count() != 3 {

pkg/sources/gcs/gcs.go

@@ -1,12 +1,12 @@
 package gcs
 
 import (
-	"encoding/json"
 	"fmt"
 	"io"
 	"os"
-	"sort"
+	"strings"
 	"sync"
+	"sync/atomic"
 
 	diskbufferreader "github.com/bill-rich/disk-buffer-reader"
 	"github.com/go-errors/errors"
@@ -14,6 +14,7 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/trufflesecurity/trufflehog/v3/pkg/cache"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/handlers"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/pb/source_metadatapb"
@@ -21,23 +22,6 @@ import (
 	"github.com/trufflesecurity/trufflehog/v3/pkg/sources"
 )
 
-// Source represents a GCS source.
-type Source struct {
-	name        string
-	jobId       int64
-	sourceId    int64
-	concurrency int
-	verify      bool
-
-	gcsManager objectManager
-	stats      *attributes
-	log        logr.Logger
-	chunksCh   chan *sources.Chunk
-
-	mu               sync.Mutex
-	sources.Progress // not safe for concurrent access.
-}
-
 // Ensure the Source satisfies the interface at compile time.
 var _ sources.Source = (*Source)(nil)
 
@@ -57,6 +41,25 @@ func (s *Source) JobID() int64 {
 	return s.jobId
 }
 
+// Source represents a GCS source.
+type Source struct {
+	name        string
+	jobId       int64
+	sourceId    int64
+	concurrency int
+	verify      bool
+
+	gcsManager objectManager
+	stats      *attributes
+	log        logr.Logger
+	chunksCh   chan *sources.Chunk
+
+	processedObjects int32
+	cache            cache.Cache
+
+	sources.Progress
+}
+
 // Init returns an initialized GCS source.
 func (s *Source) Init(aCtx context.Context, name string, id int64, sourceID int64, verify bool, connection *anypb.Any, concurrency int) error {
 	s.log = aCtx.Logger()
@@ -73,6 +76,30 @@ func (s *Source) Init(aCtx context.Context, name string, id int64, sourceID int6
 		return errors.WrapPrefix(err, "error unmarshalling connection", 0)
 	}
 
+	gcsManager, err := configureGCSManager(aCtx, &conn, concurrency)
+	if err != nil {
+		return err
+	}
+	s.gcsManager = gcsManager
+
+	s.log.V(2).Info("enumerating buckets and objects")
+	if err := s.enumerate(aCtx); err != nil {
+		return fmt.Errorf("error enumerating buckets and objects: %w", err)
+	}
+
+	// For runs that aren't associated with a job, we don't need progress or a cache.
+	if id != 0 {
+		s.cache = configureCache(aCtx, &s.Progress)
+	}
+
+	return nil
+}
+
+func configureGCSManager(aCtx context.Context, conn *sourcespb.GCS, concurrency int) (*gcsManager, error) {
+	if conn == nil {
+		return nil, fmt.Errorf("GCS connection is nil, cannot configure GCS manager")
+	}
+
 	var gcsManagerAuthOption gcsManagerOption
 
 	switch conn.Credential.(type) {
@@ -81,7 +108,7 @@ func (s *Source) Init(aCtx context.Context, name string, id int64, sourceID int6
 	case *sourcespb.GCS_ServiceAccountFile:
 		b, err := os.ReadFile(conn.GetServiceAccountFile())
 		if err != nil {
-			return fmt.Errorf("error reading GCS JSON Service Account file: %w", err)
+			return nil, fmt.Errorf("error reading GCS JSON Service Account file: %w", err)
 		}
 		gcsManagerAuthOption = withJSONServiceAccount(aCtx, b)
 	case *sourcespb.GCS_JsonServiceAccount:
@@ -91,39 +118,28 @@ func (s *Source) Init(aCtx context.Context, name string, id int64, sourceID int6
 	case *sourcespb.GCS_Unauthenticated:
 		gcsManagerAuthOption = withoutAuthentication()
 	default:
-		return fmt.Errorf("unknown GCS authentication type: %T", conn.Credential)
+		return nil, fmt.Errorf("unknown GCS authentication type: %T", conn.Credential)
 
 	}
 
-	resume, err := setResumeBucketOffset(s.Progress.EncodedResumeInfo)
-	if err != nil {
-		return fmt.Errorf("error setting resume info: %w", err)
-	}
-
 	gcsManagerOpts := []gcsManagerOption{
 		withConcurrency(concurrency),
-		withBucketOffsets(resume),
 		withMaxObjectSize(conn.MaxObjectSize),
 		gcsManagerAuthOption,
 	}
-	if setGCSManagerBucketOptions(&conn) != nil {
-		gcsManagerOpts = append(gcsManagerOpts, setGCSManagerBucketOptions(&conn))
+	if setGCSManagerBucketOptions(conn) != nil {
+		gcsManagerOpts = append(gcsManagerOpts, setGCSManagerBucketOptions(conn))
 	}
-	if setGCSManagerObjectOptions(&conn) != nil {
-		gcsManagerOpts = append(gcsManagerOpts, setGCSManagerObjectOptions(&conn))
+	if setGCSManagerObjectOptions(conn) != nil {
+		gcsManagerOpts = append(gcsManagerOpts, setGCSManagerObjectOptions(conn))
 	}
 
-	if s.gcsManager, err = newGCSManager(conn.ProjectId, gcsManagerOpts...); err != nil {
-		return fmt.Errorf("error creating GCS manager: %w", err)
+	gcsManager, err := newGCSManager(conn.ProjectId, gcsManagerOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("error creating GCS manager: %w", err)
 	}
 
-	if len(resume) > 0 {
-		s.log.V(2).Info("resuming from previous run")
-		return nil
-	}
-
-	s.log.V(2).Info("new run, enumerating buckets and objects")
-	return s.enumerate(aCtx)
+	return gcsManager, nil
 }
 
 func setGCSManagerBucketOptions(conn *sourcespb.GCS) gcsManagerOption {
@@ -147,140 +163,13 @@ func setGCSManagerOptions(include, exclude []string, includeFn, excludeFn func([
 	return nil
 }
 
-type progressInfo struct {
-	processedBucketCnt,
-	totalBucketsCnt,
-	processedObjectsCnt,
-	totalObjectsCnt int
-
-	// bucketObjects tracks processing and processed objects for each bucket.
-	bucketObjects map[string]*objectsProgress
-}
-
-// objectsProgress keeps track of the progress of processing objects in a bucket.
-// It is marshalled/unmarshalled to/from a string as part of the Source's Progress.
-type objectsProgress struct {
-	// IsBuckedProcessed is true if all objects in the bucket have been processed.
-	IsBucketProcessed bool
-	// ProcessedCnt is the number of objects that have been processed.
-	ProcessedCnt,
-	// TotalCnt is the total number of objects in the bucket.
-	TotalCnt int
-	// Processed is the last object that was fully processed.
-	Processed string
-	// Processing are the objects that were not fully processed.
-	Processing map[string]struct{}
-}
-
-func newObjectsProgress(cnt int) *objectsProgress {
-	return &objectsProgress{TotalCnt: cnt, Processing: map[string]struct{}{}}
-}
-
-// newProgressInfo constructs a progressInfo from the Source's Progress information.
-func newProgressInfo(ctx context.Context, s *sources.Progress) (*progressInfo, error) {
-	totalObjCnt := s.SectionsRemaining
-	processedObjCnt := s.SectionsCompleted
-	info := &progressInfo{
-		totalObjectsCnt:     int(totalObjCnt),
-		processedObjectsCnt: int(processedObjCnt),
+func configureCache(ctx context.Context, p *sources.Progress) cache.Cache {
+	if p.EncodedResumeInfo != "" {
+		ctx.Logger().V(3).Info("creating storage service with existing progress")
+		return newPersistableCache(p, withMemoryLoadedPersistableCache(strings.Split(p.EncodedResumeInfo, ",")))
 	}
 
-	encodeResume := s.EncodedResumeInfo
-	if encodeResume == "" {
-		info.bucketObjects = make(map[string]*objectsProgress)
-		return info, nil
-	}
-
-	var m map[string]*objectsProgress
-	if err := json.Unmarshal([]byte(encodeResume), &m); err != nil {
-		ctx.Logger().V(1).Info("error unmarshalling progress info", "error", err)
-		return info, nil
-	}
-	info.bucketObjects = m
-
-	return info, nil
-}
-
-type progressStateFn func(string, string, *progressInfo)
-
-func (p *progressInfo) setProcessStatus(obj object, fn progressStateFn) {
-	fn(obj.bucket, obj.name, p)
-}
-
-func setProcessingBucketObject(bucket, obj string, progress *progressInfo) {
-	progress.processing(bucket, obj)
-}
-
-func (p *progressInfo) processing(bkt, obj string) {
-	p.bucketObjects[bkt].Processing[obj] = struct{}{}
-}
-
-func setProcessedBucketObject(bucket, obj string, progress *progressInfo) {
-	progress.processed(bucket, obj)
-}
-
-func (p *progressInfo) processed(bkt, obj string) {
-	// Remove the object from the processing list and increment the processed count.
-	delete(p.bucketObjects[bkt].Processing, obj)
-
-	p.bucketObjects[bkt].ProcessedCnt++
-	// If all objects in the bucket have been processed, mark the bucket as processed.
-	if p.bucketObjects[bkt].ProcessedCnt == p.bucketObjects[bkt].TotalCnt {
-		p.bucketObjects[bkt].IsBucketProcessed = true
-		p.processedBucketCnt++
-	}
-	p.processedObjectsCnt++
-
-	// If the object is not greater (lexicographically) than the last processed object, we can skip it.
-	// This ensures we keep the greatest object name as the last processed object.
-	if obj < p.bucketObjects[bkt].Processed {
-		return
-	}
-	p.bucketObjects[bkt].Processed = obj
-}
-
-func setResumeBucketOffset(s string) (map[string]offsetInfo, error) {
-	resumeInfo := progressInfo{}
-	if s != "" {
-		if err := json.Unmarshal([]byte(s), &resumeInfo.bucketObjects); err != nil {
-			return nil, fmt.Errorf("error unmarshalling resume info: %w", err)
-		}
-	}
-
-	return calcBktOffset(resumeInfo.bucketObjects)
-}
-
-// In order to calculate the bucket offset, we need to know the last object
-// that was processed for each bucket, as well as any objects that were processing.
-// If the bucket was fully processed, we don't need to set an offset.
-// If there were no objects processing, we can just use the last object that was processed.
-// If there were objects processing, we need to find the first object (lexicographically)
-// that was processing, as that is the next object that needs to be processed.
-func calcBktOffset(resumeInfo map[string]*objectsProgress) (map[string]offsetInfo, error) {
-	bucketOffset := make(map[string]offsetInfo, len(resumeInfo))
-	for objName, progress := range resumeInfo {
-		info := offsetInfo{}
-		if progress.IsBucketProcessed {
-			info.isBucketProcessed = true
-			bucketOffset[objName] = info
-			continue
-		}
-		if len(progress.Processing) == 0 {
-			info.lastProcessedObject = progress.Processed
-			bucketOffset[objName] = info
-			continue
-		}
-
-		processing := make([]string, 0, len(progress.Processing))
-		for k := range progress.Processing {
-			processing = append(processing, k)
-		}
-		sort.Strings(processing)
-		info.lastProcessedObject = processing[0]
-		bucketOffset[objName] = info
-	}
-
-	return bucketOffset, nil
+	return newPersistableCache(p, withMemoryPersistableCache())
 }
 
 // enumerate all the objects and buckets in the source and use the results to
@@ -293,18 +182,7 @@ func (s *Source) enumerate(ctx context.Context) error {
 	}
 	s.stats = stats
 
-	info := progressInfo{
-		totalBucketsCnt: int(s.stats.numBuckets),
-		totalObjectsCnt: int(s.stats.numObjects),
-	}
-
-	progress := make(map[string]*objectsProgress, len(s.stats.bucketObjects))
-	for k, v := range s.stats.bucketObjects {
-		progress[k] = newObjectsProgress(int(v))
-	}
-	info.bucketObjects = progress
-
-	return s.setProgress(ctx, &info, fmt.Sprintf("enumerated %d buckets, %d objects", s.stats.numBuckets, s.stats.numObjects))
+	return nil
 }
 
 // Chunks emits chunks of bytes over a channel.
@@ -314,11 +192,7 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		return fmt.Errorf("error listing objects: %w", err)
 	}
 	s.chunksCh = chunksChan
-
-	progress, err := newProgressInfo(ctx, &s.Progress)
-	if err != nil {
-		return fmt.Errorf("error constructing progress info: %w", err)
-	}
+	s.Progress.Message = "starting to process objects..."
 
 	var wg sync.WaitGroup
 	for obj := range objectCh {
@@ -329,55 +203,46 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 			continue
 		}
 
-		if err := s.startProcessing(ctx, progress, o); err != nil {
-			ctx.Logger().V(1).Info("GCS source error starting to process objects", "error", err)
+		if s.cache != nil && s.cache.Exists(o.md5) {
+			ctx.Logger().V(5).Info("skipping object, object already processed", "name", o.name)
 			continue
 		}
+
 		wg.Add(1)
-		go func(obj object, resume *progressInfo) {
+		go func(obj object) {
 			defer wg.Done()
 
 			if err := s.processObject(ctx, o); err != nil {
 				ctx.Logger().V(1).Info("error setting start progress progress", "name", o.name, "error", err)
 				return
 			}
-			if err := s.endProcessing(ctx, resume, o); err != nil {
-				ctx.Logger().V(1).Info("error setting end progress progress", "name", o.name, "error", err)
+			if s.cache != nil {
+				s.setProgress(ctx, o.name, o.md5)
 			}
-		}(o, progress)
+		}(o)
 	}
 	wg.Wait()
 
-	ctx.Logger().Info("GCS source finished processing", "name", s.name)
+	s.completeProgress(ctx)
 	return nil
 }
 
-func (s *Source) startProcessing(ctx context.Context, progress *progressInfo, o object) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+func (s *Source) setProgress(ctx context.Context, objName, md5 string) {
+	atomic.AddInt32(&s.processedObjects, 1)
+	ctx.Logger().V(5).Info("setting progress for object", "object-name", objName, "md5", md5)
 
-	progress.setProcessStatus(o, setProcessingBucketObject)
-	return s.setProgress(ctx, progress, fmt.Sprintf("GCS source beginning to process object %s in bucket %s", o.name, o.bucket))
+	s.cache.Set(md5, md5)
+	processed := atomic.LoadInt32(&s.processedObjects)
+	s.Progress.SectionsCompleted = processed
+	s.Progress.SectionsRemaining = int32(s.stats.numObjects)
+	s.Progress.PercentComplete = int64(float64(processed) / float64(s.stats.numObjects) * 100)
 }
 
-func (s *Source) endProcessing(ctx context.Context, progress *progressInfo, o object) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	progress.setProcessStatus(o, setProcessedBucketObject)
-	return s.setProgress(ctx, progress, fmt.Sprintf("GCS source finished processing object %s in bucket %s", o.name, o.bucket))
-}
-
-func (s *Source) setProgress(ctx context.Context, progress *progressInfo, msg string) error {
-	ctx.Logger().V(5).Info(msg)
-
-	b, err := json.Marshal(progress.bucketObjects)
-	if err != nil {
-		return fmt.Errorf("error marshalling resume info: %w", err)
-	}
-	s.SetProgressComplete(progress.processedObjectsCnt, progress.totalObjectsCnt, msg, string(b))
-
-	return nil
+func (s *Source) completeProgress(ctx context.Context) {
+	msg := fmt.Sprintf("GCS source finished processing %d objects", s.stats.numObjects)
+	ctx.Logger().Info(msg)
+	s.Progress.Message = msg
+	s.cache.Clear()
 }
 
 func (s *Source) processObject(ctx context.Context, o object) error {

pkg/sources/gcs/gcs_manager.go

@@ -2,6 +2,7 @@ package gcs
 
 import (
 	aCtx "context"
+	"encoding/base64"
 	"fmt"
 	"io"
 	"runtime"
@@ -357,6 +358,7 @@ type object struct {
 	contentType string
 	owner       string
 	link        string
+	md5         string
 	// acl represents an ACLEntities.
 	// https://pkg.go.dev/cloud.google.com/go/storage#ACLEntity
 	acl       []string
@@ -634,6 +636,7 @@ func (g *gcsManager) constructObject(ctx context.Context, obj *storage.ObjectHan
 	o.contentType = attrs.ContentType
 	o.owner = attrs.Owner
 	o.link = attrs.MediaLink
+	o.md5 = base64.StdEncoding.EncodeToString(attrs.MD5[:])
 	o.createdAt = attrs.Created
 	o.updatedAt = attrs.Updated
 	o.acl = objectACLs(attrs.ACL)

pkg/sources/gcs/gcs_manager_test.go

@@ -473,6 +473,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/aws1.txt?generation=1677870994890594&alt=media",
 					acl:         []string{},
+					md5:         "ek30X4Oc1ZSsI2sZUkzskg==",
 				},
 				{
 					name:        "moar2.txt",
@@ -481,6 +482,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        12,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/moar2.txt?generation=1677871000378542&alt=media",
 					acl:         []string{},
+					md5:         "5Z/5eUEET4XfUpfhwwLSYA==",
 				},
 				{
 					name:        "aws3.txt",
@@ -489,6 +491,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th2/o/aws3.txt?generation=1677871022489611&alt=media",
 					acl:         []string{},
+					md5:         "JusDN4oMjpnuDOz5FVrOgQ==",
 				},
 				{
 					name:        "moar.txt",
@@ -497,6 +500,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        6,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th3/o/moar.txt?generation=1677871042896804&alt=media",
 					acl:         []string{},
+					md5:         "CffgLxKQviEdpweiZvFTsw==",
 				},
 				{
 					name:        "AMAZON_FASHION_5.json",
@@ -505,6 +509,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        1413469,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th4/o/AMAZON_FASHION_5.json?generation=1677871063457469&alt=media",
 					acl:         []string{},
+					md5:         "Wh2PSDqjRpUlY3kFDUKitw==",
 				},
 			},
 			wantNumBkt: 4,
@@ -522,6 +527,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/aws1.txt?generation=1677870994890594&alt=media",
 					acl:         []string{},
+					md5:         "ek30X4Oc1ZSsI2sZUkzskg==",
 				},
 				{
 					name:        "moar2.txt",
@@ -530,6 +536,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        12,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/moar2.txt?generation=1677871000378542&alt=media",
 					acl:         []string{},
+					md5:         "5Z/5eUEET4XfUpfhwwLSYA==",
 				},
 			},
 			wantNumBkt: 1,
@@ -547,6 +554,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        6,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th3/o/moar.txt?generation=1677871042896804&alt=media",
 					acl:         []string{},
+					md5:         "CffgLxKQviEdpweiZvFTsw==",
 				},
 				{
 					name:        "AMAZON_FASHION_5.json",
@@ -555,6 +563,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        1413469,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th4/o/AMAZON_FASHION_5.json?generation=1677871063457469&alt=media",
 					acl:         []string{},
+					md5:         "Wh2PSDqjRpUlY3kFDUKitw==",
 				},
 			},
 			wantNumBkt: 2,
@@ -576,6 +585,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th2/o/aws3.txt?generation=1677871022489611&alt=media",
 					acl:         []string{},
+					md5:         "JusDN4oMjpnuDOz5FVrOgQ==",
 				},
 				{
 					name:        "moar.txt",
@@ -584,6 +594,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        6,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th3/o/moar.txt?generation=1677871042896804&alt=media",
 					acl:         []string{},
+					md5:         "CffgLxKQviEdpweiZvFTsw==",
 				},
 				{
 					name:        "AMAZON_FASHION_5.json",
@@ -592,6 +603,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        1413469,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th4/o/AMAZON_FASHION_5.json?generation=1677871063457469&alt=media",
 					acl:         []string{},
+					md5:         "Wh2PSDqjRpUlY3kFDUKitw==",
 				},
 			},
 			wantNumBkt: 4, // We still list objects in all buckets.
@@ -613,6 +625,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/aws1.txt?generation=1677870994890594&alt=media",
 					acl:         []string{},
+					md5:         "ek30X4Oc1ZSsI2sZUkzskg==",
 				},
 			},
 			wantNumBkt: 1,
@@ -634,6 +647,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        150,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/aws1.txt?generation=1677870994890594&alt=media",
 					acl:         []string{},
+					md5:         "ek30X4Oc1ZSsI2sZUkzskg==",
 				},
 			},
 			wantNumBkt: 4,
@@ -656,6 +670,7 @@ func TestGCSManagerListObjects(t *testing.T) {
 					size:        12,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/moar2.txt?generation=1677871000378542&alt=media",
 					acl:         []string{},
+					md5:         "5Z/5eUEET4XfUpfhwwLSYA==",
 				},
 			},
 			wantNumBkt: 1,
@@ -743,6 +758,7 @@ func TestGCSManagerListObjects_Resuming(t *testing.T) {
 					size:        12,
 					link:        "https://storage.googleapis.com/download/storage/v1/b/test-bkt-th/o/moar2.txt?generation=1677871000378542&alt=media",
 					acl:         []string{},
+					md5:         "5Z/5eUEET4XfUpfhwwLSYA==",
 				},
 			},
 			wantNumBkt: 1,

pkg/sources/gcs/gcs_test.go

@@ -1,10 +1,10 @@
 package gcs
 
 import (
-	"encoding/json"
 	"fmt"
 	"io"
 	"sort"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -31,7 +31,7 @@ func TestSourceInit(t *testing.T) {
 	source, conn := createTestSource(&sourcespb.GCS{
 		ProjectId: testProjectID,
 		IncludeBuckets: []string{
-			"bucket1",
+			publicBucket,
 		},
 		ExcludeBuckets: []string{
 			perfTestBucketGlob,
@@ -39,22 +39,46 @@ func TestSourceInit(t *testing.T) {
 		ExcludeObjects: []string{
 			"object1",
 		},
-		Credential: &sourcespb.GCS_ApiKey{
-			ApiKey: testAPIKey,
-		},
+		Credential: &sourcespb.GCS_Unauthenticated{},
 	})
 
-	// Set the EncodeResumeInfo to prevent enumeration.
-	b, err := json.Marshal(map[string]objectsProgress{"other": {}})
-	assert.Nil(t, err)
-	source.Progress.EncodedResumeInfo = string(b)
+	testCases := []struct {
+		name      string
+		jobID     int64
+		conn      *anypb.Any
+		source    *Source
+		wantGCS   bool
+		wantCache bool
+	}{
+		{
+			name:      "Non-cacheable GCS source init",
+			jobID:     0,
+			conn:      conn,
+			source:    source,
+			wantGCS:   true,
+			wantCache: false,
+		},
+		{
+			name:      "Cacheable GCS source init",
+			jobID:     1,
+			conn:      conn,
+			source:    source,
+			wantGCS:   true,
+			wantCache: true,
+		},
+	}
 
-	err = source.Init(context.Background(), "test", 1, 1, true, conn, 8)
-	assert.Nil(t, err)
-	assert.NotNil(t, source.gcsManager)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.source.Init(context.Background(), "test", tc.jobID, 1, true, tc.conn, 8)
+			assert.Nil(t, err)
+			assert.Equal(t, tc.wantGCS, tc.source.gcsManager != nil)
+			assert.Equal(t, tc.wantCache, tc.source.cache != nil)
+		})
+	}
 }
 
-func TestSourceInit_Conn(t *testing.T) {
+func TestConfigureGCSManager(t *testing.T) {
 	testCases := []struct {
 		name    string
 		conn    *sourcespb.GCS
@@ -77,7 +101,6 @@ func TestSourceInit_Conn(t *testing.T) {
 			want: &gcsManager{
 				projectID:      testProjectID,
 				excludeBuckets: map[string]struct{}{perfTestBucketGlob: {}},
-				hasEnumerated:  true,
 			},
 		},
 		{
@@ -95,7 +118,6 @@ func TestSourceInit_Conn(t *testing.T) {
 			want: &gcsManager{
 				projectID:      testProjectID,
 				includeBuckets: map[string]struct{}{"bucket1": {}},
-				hasEnumerated:  true,
 			},
 		},
 		{
@@ -117,27 +139,20 @@ func TestSourceInit_Conn(t *testing.T) {
 				projectID:      testProjectID,
 				includeObjects: map[string]struct{}{"object1": {}},
 				excludeBuckets: map[string]struct{}{perfTestBucketGlob: {}},
-				hasEnumerated:  true,
 			},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			source, conn := createTestSource(tc.conn)
-			// Set the EncodeResumeInfo to prevent enumeration.
-			b, err := json.Marshal(map[string]objectsProgress{"other": {}})
-			assert.Nil(t, err)
-			source.Progress.EncodedResumeInfo = string(b)
-
-			err = source.Init(context.Background(), "test", 1, 1, true, conn, 8)
+			ctx := context.Background()
+			got, err := configureGCSManager(ctx, tc.conn, 8)
 			if err != nil && !tc.wantErr {
-				t.Errorf("source.Init() got: %v, want: %v", err, nil)
-				return
+				t.Errorf("source.configureGCSManager() error = %v", err)
 			}
 
 			if !tc.wantErr {
-				if diff := cmp.Diff(tc.want, source.gcsManager,
+				if diff := cmp.Diff(tc.want, got,
 					cmp.AllowUnexported(gcsManager{}),
 					cmpopts.IgnoreFields(gcsManager{}, "client", "workerPool", "concurrency", "buckets", "maxObjectSize", "attr"),
 				); diff != "" {
@@ -148,65 +163,6 @@ func TestSourceInit_Conn(t *testing.T) {
 	}
 }
 
-func TestSourceInit_Progress(t *testing.T) {
-	complete := map[string]objectsProgress{testBucket: {
-		Processed:         object2,
-		IsBucketProcessed: true,
-		ProcessedCnt:      1,
-		TotalCnt:          1,
-	}}
-	completeEncodeResumeInfo, err := json.Marshal(complete)
-	assert.Nil(t, err)
-
-	processing := map[string]objectsProgress{
-		testBucket: {
-			IsBucketProcessed: false,
-			ProcessedCnt:      1,
-			Processed:         object2,
-			Processing:        map[string]struct{}{object1: {}},
-			TotalCnt:          2,
-		},
-	}
-	processingEncodeResumeInfo, err := json.Marshal(processing)
-	assert.Nil(t, err)
-
-	testCases := []struct {
-		name              string
-		conn              *sourcespb.GCS
-		encodeResumeInfo  string
-		gcsManagerBuckets map[string]bucket
-	}{
-		{
-			name:              "no progress",
-			conn:              &sourcespb.GCS{ProjectId: testProjectID, Credential: &sourcespb.GCS_Adc{}, ExcludeBuckets: []string{perfTestBucketGlob}},
-			gcsManagerBuckets: map[string]bucket{},
-		},
-		{
-			name:              "progress set, no processing objects in any buckets",
-			conn:              &sourcespb.GCS{ProjectId: testProjectID, Credential: &sourcespb.GCS_Adc{}},
-			encodeResumeInfo:  string(completeEncodeResumeInfo),
-			gcsManagerBuckets: map[string]bucket{testBucket: {name: testBucket, startOffset: "", shouldInclude: false}},
-		},
-		{
-			name:              "progress set, processing objects in some buckets",
-			conn:              &sourcespb.GCS{ProjectId: testProjectID, Credential: &sourcespb.GCS_Adc{}},
-			encodeResumeInfo:  string(processingEncodeResumeInfo),
-			gcsManagerBuckets: map[string]bucket{testBucket: {name: testBucket, startOffset: object1, shouldInclude: true}},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			source, conn := createTestSource(tc.conn)
-			source.Progress.EncodedResumeInfo = tc.encodeResumeInfo
-
-			err := source.Init(context.Background(), "test", 1, 1, true, conn, 8)
-			assert.Nil(t, err)
-			assert.Equal(t, tc.gcsManagerBuckets, source.gcsManager.(*gcsManager).buckets)
-		})
-	}
-}
-
 type mockObjectManager struct {
 	wantErr bool
 }
@@ -264,6 +220,7 @@ func createTestObject(id int) object {
 		link:        fmt.Sprintf("https://storage.googleapis.com/%s/%s", testBucket, fmt.Sprintf("object%d", id)),
 		acl:         []string{"authenticatedUsers"},
 		size:        42,
+		md5:         fmt.Sprintf("md5hash%d", id),
 		Reader:      &mockReader{data: []byte(fmt.Sprintf("hello world %d", id))},
 	}
 }
@@ -297,7 +254,9 @@ func TestSourceChunks_ListObjects(t *testing.T) {
 	source := &Source{
 		gcsManager: &mockObjectManager{},
 		chunksCh:   chunksCh,
+		Progress:   sources.Progress{},
 	}
+	source.cache = newPersistableCache(&source.Progress, withMemoryPersistableCache())
 
 	err := source.enumerate(ctx)
 	assert.Nil(t, err)
@@ -349,11 +308,6 @@ func TestSourceInit_Enumerate(t *testing.T) {
 	assert.Equal(t, uint64(5), source.stats.numObjects)
 	assert.Equal(t, uint32(1), source.stats.numBuckets)
 	assert.Equal(t, uint64(5), source.stats.bucketObjects[testBucket])
-	// Ensure progress is initialized correctly.
-	p := map[string]*objectsProgress{testBucket: {TotalCnt: 5, Processing: map[string]struct{}{}}}
-	b, err := json.Marshal(p)
-	assert.Nil(t, err)
-	assert.Equal(t, string(b), source.Progress.EncodedResumeInfo)
 }
 
 func TestSourceChunks_ListObjects_Error(t *testing.T) {
@@ -370,11 +324,14 @@ func TestSourceChunks_ListObjects_Error(t *testing.T) {
 func TestSourceChunks_ProgressSet(t *testing.T) {
 	ctx := context.Background()
 	chunksCh := make(chan *sources.Chunk, 1)
-
 	source := &Source{
 		gcsManager: &mockObjectManager{},
 		chunksCh:   chunksCh,
+		Progress:   sources.Progress{},
 	}
+
+	source.cache = newPersistableCache(&source.Progress, withMemoryPersistableCache(), withCustomIncrement(5))
+
 	err := source.enumerate(ctx)
 	assert.Nil(t, err)
 
@@ -397,26 +354,22 @@ func TestSourceChunks_ProgressSet(t *testing.T) {
 	// Ensure we get 5 objects back.
 	assert.Equal(t, len(want), len(got))
 
-	processing := map[string]struct{}{"object0": {}, "object1": {}, "object2": {}, "object3": {}, "object4": {}}
+	processing := map[string]struct{}{"md5hash0": {}, "md5hash1": {}, "md5hash2": {}, "md5hash3": {}, "md5hash4": {}}
 	for i := 0; i < 5; i++ {
 		delete(processing, fmt.Sprintf("object%d", i))
 	}
 
 	// Test that the resume progress is set.
 	// The processed values should be the greatest object name lexicographically.
-	resume := map[string]objectsProgress{
-		testBucket: {
-			Processed:         "object4",
-			Processing:        processing,
-			IsBucketProcessed: true,
-			ProcessedCnt:      5,
-			TotalCnt:          5,
-		},
-	}
-	b, err := json.Marshal(resume)
-	assert.Nil(t, err)
+	progress := "md5hash0,md5hash1,md5hash2,md5hash3,md5hash4"
+	objs := strings.Split(progress, ",")
 
-	assert.Equal(t, string(b), source.Progress.EncodedResumeInfo)
+	encodeResume := strings.Split(source.Progress.EncodedResumeInfo, ",")
+	sort.Strings(encodeResume)
+
+	assert.Equal(t, objs, encodeResume)
 	assert.Equal(t, int32(5), source.Progress.SectionsCompleted)
 	assert.Equal(t, int64(100), source.Progress.PercentComplete)
+	assert.Equal(t, 0, source.cache.Count())
+	assert.Equal(t, fmt.Sprintf("GCS source finished processing %d objects", 5), source.Progress.Message)
 }

pkg/sources/gcs/persistable_cache.go

@@ -0,0 +1,70 @@
+package gcs
+
+import (
+	"github.com/trufflesecurity/trufflehog/v3/pkg/cache"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/cache/memory"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/sources"
+)
+
+const defaultCachePersistIncrement = 2500
+
+type persistibleCacheConfiguration func(c *persistableCache)
+
+// persistableCache handles all cache operations with some additional bookkeeping.
+// The threshold value is the percentage of objects that must be processed
+// before the cache is persisted.
+type persistableCache struct {
+	persistIncrement int
+	cache.Cache
+	*sources.Progress
+}
+
+func newPersistableCache(p *sources.Progress, cfgs ...persistibleCacheConfiguration) *persistableCache {
+	pc := &persistableCache{
+		persistIncrement: defaultCachePersistIncrement,
+		Progress:         p,
+	}
+
+	for _, cfg := range cfgs {
+		cfg(pc)
+	}
+	return pc
+}
+
+// withMemoryPersistableCache configures the persistableCache to use an in-memory cache.
+func withMemoryPersistableCache() persistibleCacheConfiguration {
+	return func(pc *persistableCache) {
+		pc.Cache = memory.New()
+	}
+}
+
+// withMemoryLoadedPersistableCache configures the persistableCache to use an in-memory cache with the given contents.
+func withMemoryLoadedPersistableCache(contents []string) persistibleCacheConfiguration {
+	return func(pc *persistableCache) {
+		pc.Cache = memory.NewWithData(contents)
+	}
+}
+
+// withCustomIncrement configures the persistableCache to use the given increment.
+func withCustomIncrement(increment int) persistibleCacheConfiguration {
+	return func(pc *persistableCache) {
+		pc.persistIncrement = increment
+	}
+}
+
+// Set overrides the default Set method in order to persist the cache contents
+// on each persistIncrement.
+func (c *persistableCache) Set(key, value string) {
+	c.Cache.Set(key, value)
+
+	if ok, val := c.shouldPersist(); ok {
+		c.EncodedResumeInfo = val
+	}
+}
+
+func (c *persistableCache) shouldPersist() (bool, string) {
+	if c.Count()%c.persistIncrement != 0 {
+		return false, ""
+	}
+	return true, c.Contents()
+}

pkg/sources/gcs/persistable_cache_test.go

@@ -0,0 +1,74 @@
+package gcs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/trufflesecurity/trufflehog/v3/pkg/cache/memory"
+	"github.com/trufflesecurity/trufflehog/v3/pkg/sources"
+)
+
+func TestPersistableCache(t *testing.T) {
+	// Create a new Progress object for testing.
+	progress := sources.Progress{}
+
+	tests := []struct {
+		name string
+		cfgs []persistibleCacheConfiguration
+		fn   func(cache *persistableCache)
+	}{
+		{
+			name: "withMemoryPersistableCache",
+			cfgs: []persistibleCacheConfiguration{withMemoryPersistableCache()},
+			fn: func(cache *persistableCache) {
+				assert.NotNil(t, cache.Cache)
+				assert.IsType(t, &memory.Cache{}, cache.Cache)
+			},
+		},
+		{
+			name: "withMemoryLoadedPersistableCache",
+			cfgs: []persistibleCacheConfiguration{withMemoryLoadedPersistableCache([]string{"foo", "bar"})},
+			fn: func(cache *persistableCache) {
+				assert.NotNil(t, cache.Cache)
+				assert.IsType(t, &memory.Cache{}, cache.Cache)
+				assert.Contains(t, cache.Cache.Contents(), "foo,bar")
+			},
+		},
+		{
+			name: "withCustomIncrement",
+			cfgs: []persistibleCacheConfiguration{withCustomIncrement(100)},
+			fn: func(cache *persistableCache) {
+				assert.Equal(t, 100, cache.persistIncrement)
+			},
+		},
+		{
+			name: "Set",
+			cfgs: []persistibleCacheConfiguration{withMemoryPersistableCache(), withCustomIncrement(3)},
+			fn: func(cache *persistableCache) {
+				cache.Set("foo", "bar")
+				assert.Equal(t, 1, cache.Count())
+
+				cache.Set("baz", "qux")
+				assert.Equal(t, 2, cache.Count())
+
+				// The cache should not persist until the increment (3) is reached.
+				assert.False(t, progress.EncodedResumeInfo != "")
+
+				cache.Set("quux", "corge")
+				assert.Equal(t, 3, cache.Count())
+
+				// The cache should now persist.
+				assert.True(t, progress.EncodedResumeInfo != "")
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cache := newPersistableCache(&progress, test.cfgs...)
+			assert.NotNil(t, cache)
+			test.fn(cache)
+		})
+	}
+}