import assert.

go.mod

@@ -18,11 +18,11 @@ require (
 	github.com/bitfinexcom/bitfinex-api-go v0.0.0-20210608095005-9e0b26f200fb
 	github.com/bradleyfalzon/ghinstallation/v2 v2.1.0
 	github.com/crewjam/rfc5424 v0.1.0
-	github.com/denisenkom/go-mssqldb v0.12.3
-	github.com/envoyproxy/protoc-gen-validate v0.6.13
+	github.com/denisenkom/go-mssqldb v0.12.2
+	github.com/envoyproxy/protoc-gen-validate v0.6.8
 	github.com/fatih/color v1.13.0
 	github.com/felixge/fgprof v0.9.3
-	github.com/getsentry/sentry-go v0.14.0
+	github.com/getsentry/sentry-go v0.13.0
 	github.com/go-errors/errors v1.4.2
 	github.com/go-git/go-git/v5 v5.4.2
 	github.com/go-logr/logr v1.2.3
@@ -50,9 +50,9 @@ require (
 	github.com/stretchr/testify v1.8.0
 	github.com/tailscale/depaware v0.0.0-20210622194025-720c4b409502
 	github.com/xanzy/go-gitlab v0.73.1
-	go.mongodb.org/mongo-driver v1.10.3
+	go.mongodb.org/mongo-driver v1.10.2
 	go.uber.org/zap v1.23.0
-	golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be
+	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d
 	golang.org/x/exp v0.0.0-20221018205818-5c77f4b2bbd7
 	golang.org/x/oauth2 v0.0.0-20220909003341-f21342109be1
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
@@ -105,7 +105,7 @@ require (
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jpillora/s3 v1.1.4 // indirect
 	github.com/kevinburke/ssh_config v0.0.0-20201106050909-4977a11b4351 // indirect
-	github.com/klauspost/compress v1.15.11 // indirect
+	github.com/klauspost/compress v1.15.5 // indirect
 	github.com/klauspost/pgzip v1.2.5 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
@@ -126,10 +126,10 @@ require (
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
-	golang.org/x/net v0.0.0-20221002022538-bcab6841153b // indirect
-	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec // indirect
+	golang.org/x/net v0.0.0-20220909164309-bea034e7d591 // indirect
+	golang.org/x/sys v0.0.0-20220908150016-7ac13a9a928d // indirect
 	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
+	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 // indirect
 	golang.org/x/tools v0.1.12 // indirect
 	google.golang.org/api v0.96.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

go.sum

@@ -142,8 +142,8 @@ github.com/crewjam/rfc5424 v0.1.0/go.mod h1:RCi9M3xHVOeerf6ULZzqv2xOGRO/zYaVUeRy
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/denisenkom/go-mssqldb v0.12.3 h1:pBSGx9Tq67pBOTLmxNuirNTeB8Vjmf886Kx+8Y+8shw=
-github.com/denisenkom/go-mssqldb v0.12.3/go.mod h1:k0mtMFOnU+AihqFxPMiF05rtiDrorD1Vrm1KEz5hxDo=
+github.com/denisenkom/go-mssqldb v0.12.2 h1:1OcPn5GBIobjWNd+8yjfHNIaFX14B1pWI3F9HZy5KXw=
+github.com/denisenkom/go-mssqldb v0.12.2/go.mod h1:lnIw1mZukFRZDJYQ0Pb833QS2IaC3l5HkEfra2LJ+sk=
 github.com/dimchansky/utfbom v1.1.1 h1:vV6w1AhK4VMnhBno/TPVCoK9U/LP0PkLCS9tbxHdi/U=
 github.com/dimchansky/utfbom v1.1.1/go.mod h1:SxdoEBH5qIqFocHMyGOXVAybYJdr71b1Q/j0mACtrfE=
 github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ=
@@ -162,15 +162,15 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.m
 github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
 github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/envoyproxy/protoc-gen-validate v0.6.13 h1:TvDcILLkjuZV3ER58VkBmncKsLUBqBDxra/XctCzuMM=
-github.com/envoyproxy/protoc-gen-validate v0.6.13/go.mod h1:qEySVqXrEugbHKvmhI8ZqtQi75/RHSSRNpffvB4I6Bw=
+github.com/envoyproxy/protoc-gen-validate v0.6.8 h1:B2cR/FAaiMtYDHv5BQpaqtkjGuWQIgr2KQZtHQ7f6i8=
+github.com/envoyproxy/protoc-gen-validate v0.6.8/go.mod h1:0ZMblUx0cxNoWRswEEXoj9kHBmqX8pxGweMiyIAfR6A=
 github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/felixge/fgprof v0.9.3 h1:VvyZxILNuCiUCSXtPtYmmtGvb65nqXh2QFWc0Wpf2/g=
 github.com/felixge/fgprof v0.9.3/go.mod h1:RdbpDgzqYVh/T9fPELJyV7EYJuHB55UTEULNun8eiPw=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
-github.com/getsentry/sentry-go v0.14.0 h1:rlOBkuFZRKKdUnKO+0U3JclRDQKlRu5vVQtkWSQvC70=
-github.com/getsentry/sentry-go v0.14.0/go.mod h1:RZPJKSw+adu8PBNygiri/A98FqVr2HtRckJk9XVxJ9I=
+github.com/getsentry/sentry-go v0.13.0 h1:20dgTiUSfxRB/EhMPtxcL9ZEbM1ZdR+W/7f7NWD+xWo=
+github.com/getsentry/sentry-go v0.13.0/go.mod h1:EOsfu5ZdvKPfeHYV6pTVQnsjfp30+XA7//UooKNumH0=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gliderlabs/ssh v0.2.2 h1:6zsha5zo/TWhRhwqCD3+EarCAgZ2yN28ipRnGPnwkI0=
 github.com/gliderlabs/ssh v0.2.2/go.mod h1:U7qILu1NlMHj9FlMhZLlkCdDnU1DBEAqr0aevW3Awn0=
@@ -350,8 +350,8 @@ github.com/kevinburke/ssh_config v0.0.0-20201106050909-4977a11b4351/go.mod h1:CT
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.4.1/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
-github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
-github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
+github.com/klauspost/compress v1.15.5 h1:qyCLMz2JCrKADihKOh9FxnW3houKeNsp2h5OEz0QSEA=
+github.com/klauspost/compress v1.15.5/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/pgzip v1.2.5 h1:qnWYvvKqedOF2ulHpMG72XQol4ILEJ8k2wwRl/Km8oE=
 github.com/klauspost/pgzip v1.2.5/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
@@ -474,8 +474,8 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/yusufpapurcu/wmi v1.2.2 h1:KBNDSne4vP5mbSWnJbO+51IMOXJB67QiYCSBrubbPRg=
 github.com/yusufpapurcu/wmi v1.2.2/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
-go.mongodb.org/mongo-driver v1.10.3 h1:XDQEvmh6z1EUsXuIkXE9TaVeqHw6SwS1uf93jFs0HBA=
-go.mongodb.org/mongo-driver v1.10.3/go.mod h1:z4XpeoU6w+9Vht+jAFyLgVrD+jGSQQe0+CBWFHNiHt8=
+go.mongodb.org/mongo-driver v1.10.2 h1:4Wk3cnqOrQCn0P92L3/mmurMxzdvWWs5J9jinAVKD+k=
+go.mongodb.org/mongo-driver v1.10.2/go.mod h1:z4XpeoU6w+9Vht+jAFyLgVrD+jGSQQe0+CBWFHNiHt8=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
@@ -508,9 +508,8 @@ golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d h1:sK3txAijHtOK88l68nt020reeT1ZdKLIYetKl95FzVY=
 golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be h1:fmw3UbQh+nxngCAHrDCCztao/kbYFnWjoqop8dHx05A=
-golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -598,9 +597,8 @@ golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220907135653-1e95f45603a7/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
+golang.org/x/net v0.0.0-20220909164309-bea034e7d591 h1:D0B/7al0LLrVC8aWF4+oxpv/m8bc7ViFfVS8/gXGdqI=
 golang.org/x/net v0.0.0-20220909164309-bea034e7d591/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
-golang.org/x/net v0.0.0-20221002022538-bcab6841153b h1:6e93nYa3hNqAvLr0pD4PN1fFS+gKzp2zAXqrnTCstqU=
-golang.org/x/net v0.0.0-20221002022538-bcab6841153b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -709,9 +707,8 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220908150016-7ac13a9a928d h1:RoyzQTK76Rktm3p4xyZslc8T8I1tBz4UEjZCzeh57mM=
 golang.org/x/sys v0.0.0-20220908150016-7ac13a9a928d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI=
-golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -728,8 +725,8 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
-golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 h1:ftMN5LMiBFjbzleLqtoBZk7KdJwhuybIU+FckUHgoyQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=

hack/snifftest/main.go

@@ -156,7 +156,9 @@ func main() {
 		}
 
 		for _, repo := range strings.Split(*scanCmdRepo, ",") {
-			sem.Acquire(ctx, 1)
+			if err := sem.Acquire(ctx, 1); err != nil {
+				log.Fatal(err)
+			}
 			wgChunkers.Add(1)
 			go func(r string) {
 				defer sem.Release(1)

pkg/sources/filesystem/filesystem.go

@@ -79,8 +79,8 @@ func (s *Source) Init(aCtx context.Context, name string, jobId, sourceId int64,
 
 // Chunks emits chunks of bytes over a channel.
 func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) error {
-	for i, path := range s.paths {
-		s.SetProgressComplete(i, len(s.paths), fmt.Sprintf("Path: %s", path), "")
+	for _, path := range s.paths {
+		s.Update(len(s.paths), fmt.Sprintf("Path: %s", path), "")
 
 		cleanPath := filepath.Clean(path)
 		done := false

pkg/sources/git/git.go

@@ -132,8 +132,8 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		user := cred.BasicAuth.Username
 		token := cred.BasicAuth.Password
 
-		for i, repoURI := range s.conn.Repositories {
-			s.SetProgressComplete(i, len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
+		for _, repoURI := range s.conn.Repositories {
+			s.Update(len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
 			if len(repoURI) == 0 {
 				continue
 			}
@@ -150,8 +150,8 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 			}
 		}
 	case *sourcespb.Git_Unauthenticated:
-		for i, repoURI := range s.conn.Repositories {
-			s.SetProgressComplete(i, len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
+		for _, repoURI := range s.conn.Repositories {
+			s.Update(len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
 			if len(repoURI) == 0 {
 				continue
 			}
@@ -168,8 +168,8 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 			}
 		}
 	case *sourcespb.Git_SshAuth:
-		for i, repoURI := range s.conn.Repositories {
-			s.SetProgressComplete(i, len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
+		for _, repoURI := range s.conn.Repositories {
+			s.Update(len(s.conn.Repositories), fmt.Sprintf("Repo: %s", repoURI), "")
 			if len(repoURI) == 0 {
 				continue
 			}
@@ -189,8 +189,12 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		return errors.New("invalid connection type for git source")
 	}
 
-	for i, u := range s.conn.Directories {
-		s.SetProgressComplete(i, len(s.conn.Repositories), fmt.Sprintf("Repo: %s", u), "")
+	for _, u := range s.conn.Directories {
+		s.Counter.IncTotal()
+		go func() {
+			defer s.Counter.IncSuccess()
+		}()
+		s.Update(len(s.conn.Repositories), fmt.Sprintf("Repo: %s", u), "")
 
 		if len(u) == 0 {
 			continue
@@ -215,7 +219,7 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		}
 
 	}
-	s.SetProgressComplete(len(s.conn.Repositories), len(s.conn.Repositories), fmt.Sprintf("Completed scanning source %s", s.name), "")
+	s.Complete(fmt.Sprintf("Completed scanning source %s", s.name))
 	return nil
 }
 
@@ -322,7 +326,7 @@ func (s *Git) ScanCommits(ctx context.Context, repo *git.Repository, path string
 	var reachedBase = false
 	log.WithField("repo", urlMetadata).Debugf("Scanning repo")
 	for commit := range commitChan {
-		log.Tracef("Scanning commit %s", commit.Hash)
+		log.Debugf("Scanning commit %s", commit.Hash)
 		if scanOptions.MaxDepth > 0 && depth >= scanOptions.MaxDepth {
 			log.Debugf("reached max depth")
 			break

pkg/sources/github/github.go

@@ -496,15 +496,17 @@ func (s *Source) scan(ctx context.Context, installationClient *github.Client, ch
 	s.repos = reposToScan
 
 	var scanErrs []error
-	for i, repoURL := range s.repos {
-		i, repoURL := i, repoURL
+	repoCnt := len(s.repos)
+	for _, repoURL := range s.repos {
+		repoURL := repoURL
 		s.jobPool.Go(func() error {
 			if common.IsDone(ctx) {
 				return nil
 			}
+			cnt := s.Counter.IncTotal()
 
-			// TODO: set progress complete is being called concurrently with i
-			s.setProgressCompleteWithRepo(i, progressIndexOffset, repoURL)
+			defer s.Counter.IncSuccess()
+			s.setProgressCompleteWithRepo(progressIndexOffset, repoURL)
 			// Ensure the repo is removed from the resume info after being scanned.
 			defer func(s *Source, repoURL string) {
 				s.resumeInfoMutex.Lock()
@@ -512,15 +514,15 @@ func (s *Source) scan(ctx context.Context, installationClient *github.Client, ch
 				s.resumeInfoSlice = sources.RemoveRepoFromResumeInfo(s.resumeInfoSlice, repoURL)
 			}(s, repoURL)
 
+			var err error
 			if !strings.HasSuffix(repoURL, ".git") {
 				scanErrs = append(scanErrs, fmt.Errorf("repo %s does not end in .git", repoURL))
 				return nil
 			}
 
-			s.log.WithField("repo", repoURL).Debugf("attempting to clone repo %d/%d", i+1, len(s.repos))
+			s.log.WithField("repo", repoURL).Debugf("attempting to clone repo #%d of %d", cnt, repoCnt)
 			var path string
 			var repo *gogit.Repository
-			var err error
 
 			path, repo, err = s.cloneRepo(ctx, repoURL, installationClient)
 			if err != nil {
@@ -537,21 +539,20 @@ func (s *Source) scan(ctx context.Context, installationClient *github.Client, ch
 				git.ScanOptionHeadCommit(s.conn.Head),
 			)
 
+			log.Debugf("Starting to scan repo #%d of %d: %s", cnt, repoCnt, repo)
 			if err = s.git.ScanRepo(ctx, repo, path, scanOptions, chunksChan); err != nil {
 				log.WithError(err).Errorf("unable to scan repo, continuing")
 				return nil
 			}
 			atomic.AddUint64(&scanned, 1)
-			log.Debugf("scanned %d/%d repos", scanned, len(s.repos))
+			log.Debugf("scanned %d/%d repos", scanned, repoCnt)
 
 			return nil
 		})
 	}
 
 	_ = s.jobPool.Wait()
-	if len(scanErrs) == 0 {
-		s.SetProgressComplete(len(s.repos), len(s.repos), "Completed Github scan", "")
-	}
+	s.Complete(fmt.Sprintf("Completed scanning source %s", s.name))
 
 	return scanErrs
 }
@@ -1006,8 +1007,8 @@ func (s *Source) normalizeRepos(ctx context.Context) {
 	}
 }
 
-// setProgressCompleteWithRepo calls the s.SetProgressComplete after safely setting up the encoded resume info string.
-func (s *Source) setProgressCompleteWithRepo(index int, offset int, repoURL string) {
+// setProgressCompleteWithRepo calls the s.Update after safely setting up the encoded resume info string.
+func (s *Source) setProgressCompleteWithRepo(offset int, repoURL string) {
 	s.resumeInfoMutex.Lock()
 	defer s.resumeInfoMutex.Unlock()
 
@@ -1018,5 +1019,5 @@ func (s *Source) setProgressCompleteWithRepo(index int, offset int, repoURL stri
 	// Make the resume info string from the slice.
 	encodedResumeInfo := sources.EncodeResumeInfo(s.resumeInfoSlice)
 
-	s.SetProgressComplete(index+offset, len(s.repos)+offset, fmt.Sprintf("Repo: %s", repoURL), encodedResumeInfo)
+	s.Update(len(s.repos)+offset, fmt.Sprintf("Repo: %s", repoURL), encodedResumeInfo)
 }

pkg/sources/github/github_test.go

@@ -308,29 +308,6 @@ func TestEnumerateWithToken(t *testing.T) {
 	assert.True(t, gock.IsDone())
 }
 
-func TestEnumerateWithToken_IncludeRepos(t *testing.T) {
-	defer gock.Off()
-
-	gock.New("https://api.github.com").
-		Get("/user").
-		Reply(200).
-		JSON(map[string]string{"login": "super-secret-user"})
-
-	gock.New("https://api.github.com").
-		Get("/users/super-secret-user/gists").
-		Reply(200).
-		JSON([]map[string]string{{"clone_url": ""}})
-
-	s := initTestSource(nil)
-	s.repos = []string{"some-special-repo"}
-
-	err := s.enumerateWithToken(context.TODO(), "https://api.github.com", "token")
-	assert.Nil(t, err)
-	assert.Equal(t, 2, len(s.repos))
-	assert.Equal(t, []string{"some-special-repo", ""}, s.repos)
-	assert.True(t, gock.IsDone())
-}
-
 func TestEnumerateWithApp(t *testing.T) {
 	defer gock.Off()
 
@@ -406,7 +383,7 @@ func Test_setProgressCompleteWithRepo_resumeInfo(t *testing.T) {
 
 	for _, tt := range tests {
 		s.resumeInfoSlice = tt.startingResumeInfoSlice
-		s.setProgressCompleteWithRepo(0, 0, tt.repoURL)
+		s.setProgressCompleteWithRepo(0, tt.repoURL)
 		if !reflect.DeepEqual(s.resumeInfoSlice, tt.wantResumeInfoSlice) {
 			t.Errorf("s.setProgressCompleteWithRepo() got: %v, want: %v", s.resumeInfoSlice, tt.wantResumeInfoSlice)
 		}
@@ -417,7 +394,6 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	repos := []string{"a", "b", "c", "d", "e"}
 	tests := map[string]struct {
 		repos                 []string
-		index                 int
 		offset                int
 		wantPercentComplete   int64
 		wantSectionsCompleted int32
@@ -425,7 +401,6 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	}{
 		"starting from the beginning, no offset": {
 			repos:                 repos,
-			index:                 0,
 			offset:                0,
 			wantPercentComplete:   0,
 			wantSectionsCompleted: 0,
@@ -433,18 +408,16 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 		},
 		"resume from the third, offset 2": {
 			repos:                 repos[2:],
-			index:                 0,
 			offset:                2,
 			wantPercentComplete:   40,
 			wantSectionsCompleted: 2,
 			wantSectionsRemaining: 5,
 		},
-		"resume from the third, on last repo, offset 2": {
-			repos:                 repos[2:],
-			index:                 2,
-			offset:                2,
-			wantPercentComplete:   80,
-			wantSectionsCompleted: 4,
+		"resume from the third, on last repo, offset 3": {
+			repos:                 repos[3:],
+			offset:                3,
+			wantPercentComplete:   60,
+			wantSectionsCompleted: 3,
 			wantSectionsRemaining: 5,
 		},
 	}
@@ -453,22 +426,29 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	logger.Out = io.Discard
 
 	for _, tt := range tests {
-		s := &Source{
-			repos: tt.repos,
-			log:   logger.WithField("no", "output"),
-		}
+		t.Run("", func(t *testing.T) {
+			s := &Source{
+				repos: tt.repos,
+				log:   logger.WithField("no", "output"),
+			}
 
-		s.setProgressCompleteWithRepo(tt.index, tt.offset, "")
-		gotProgress := s.GetProgress()
-		if gotProgress.PercentComplete != tt.wantPercentComplete {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.PercentComplete, tt.wantPercentComplete)
-		}
-		if gotProgress.SectionsCompleted != tt.wantSectionsCompleted {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.SectionsCompleted, tt.wantSectionsCompleted)
-		}
-		if gotProgress.SectionsRemaining != tt.wantSectionsRemaining {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.SectionsRemaining, tt.wantSectionsRemaining)
-		}
+			// Increment the progress counter by number of repos already processed.
+			for i := 1; i <= tt.offset; i++ {
+				s.Counter.IncSuccess()
+			}
+
+			s.setProgressCompleteWithRepo(tt.offset, "")
+			gotProgress := s.GetProgress()
+			if gotProgress.PercentComplete != tt.wantPercentComplete {
+				t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.PercentComplete, tt.wantPercentComplete)
+			}
+			if gotProgress.SectionsCompleted != tt.wantSectionsCompleted {
+				t.Errorf("s.setProgressCompleteWithRepo() SectionsCompleted got: %v want: %v", gotProgress.SectionsCompleted, tt.wantSectionsCompleted)
+			}
+			if gotProgress.SectionsRemaining != tt.wantSectionsRemaining {
+				t.Errorf("s.setProgressCompleteWithRepo() SectionsRemaining got: %v want: %v", gotProgress.SectionsRemaining, tt.wantSectionsRemaining)
+			}
+		})
 	}
 }
 
@@ -477,16 +457,15 @@ func Test_scan_SetProgressComplete(t *testing.T) {
 		name         string
 		repos        []string
 		wantComplete bool
-		wantErr      bool
 	}{
 		{
 			name:         "no repos",
 			wantComplete: true,
 		},
 		{
-			name:    "one valid repo",
-			repos:   []string{"a"},
-			wantErr: true,
+			name:         "one valid repo",
+			repos:        []string{"a"},
+			wantComplete: true,
 		},
 	}
 
@@ -498,9 +477,6 @@ func Test_scan_SetProgressComplete(t *testing.T) {
 			src.jobPool = &errgroup.Group{}
 
 			_ = src.scan(context.Background(), nil, nil)
-			if !tc.wantErr {
-				assert.Equal(t, "", src.GetProgress().EncodedResumeInfo)
-			}
 
 			gotComplete := src.GetProgress().PercentComplete == 100
 			if gotComplete != tc.wantComplete {

pkg/sources/gitlab/gitlab.go

@@ -8,6 +8,9 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
+
+	"golang.org/x/sync/errgroup"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
 	"github.com/trufflesecurity/trufflehog/v3/pkg/context"
@@ -46,6 +49,7 @@ type Source struct {
 	resumeInfoMutex sync.Mutex
 	jobSem          *semaphore.Weighted
 	sources.Progress
+	jobPool *errgroup.Group
 }
 
 // Ensure the Source satisfies the interface at compile time.
@@ -73,7 +77,8 @@ func (s *Source) Init(aCtx context.Context, name string, jobId, sourceId int64,
 	s.sourceId = sourceId
 	s.jobId = jobId
 	s.verify = verify
-	s.jobSem = semaphore.NewWeighted(int64(concurrency))
+	s.jobPool = &errgroup.Group{}
+	s.jobPool.SetLimit(concurrency)
 
 	var conn sourcespb.GitLab
 	err := anypb.UnmarshalTo(connection, &conn, proto.UnmarshalOptions{})
@@ -269,34 +274,32 @@ func (s *Source) getRepos() ([]string, []error) {
 }
 
 func (s *Source) scanRepos(ctx context.Context, chunksChan chan *sources.Chunk) []error {
-	wg := sync.WaitGroup{}
-	var errs []error
-	var errsMut sync.Mutex
+	var scanned uint64
 
 	// If there is resume information available, limit this scan to only the repos that still need scanning.
 	reposToScan, progressIndexOffset := sources.FilterReposToResume(s.repos, s.GetProgress().EncodedResumeInfo)
 	s.repos = reposToScan
+	log.Debugf("Found %v total repos to scan", len(s.repos))
 
-	for i, repo := range s.repos {
-		if common.IsDone(ctx) {
-			// We are returning nil instead of the scanErrors slice here because
-			// we don't want to mark this scan as errored if we cancelled it.
-			return nil
-		}
-		if err := s.jobSem.Acquire(ctx, 1); err != nil {
-			log.WithError(err).Debug("could not acquire semaphore")
-			continue
-		}
-		wg.Add(1)
-		go func(ctx context.Context, repoURL string, i int) {
-			defer s.jobSem.Release(1)
-			defer wg.Done()
+	var errMut sync.Mutex
+	var scanErrs []error
+	repoCnt := len(s.repos)
+	for _, repoURL := range s.repos {
+		repoURL := repoURL
+		s.jobPool.Go(func() error {
+			if common.IsDone(ctx) {
+				// We are returning nil instead of the scanErrors slice here because
+				// we don't want to mark this scan as errored if we cancelled it.
+				return nil
+			}
 			if len(repoURL) == 0 {
-				return
+				return nil
 			}
 
-			s.setProgressCompleteWithRepo(i, progressIndexOffset, repoURL)
-			// Ensure the repo is removed from the resume info after being scanned.
+			cnt := s.Counter.IncTotal()
+			defer s.Counter.IncSuccess()
+			s.setProgressCompleteWithRepo(progressIndexOffset, repoURL)
+			// Ensure the repoURL is removed from the resume info after being scanned.
 			defer func(s *Source) {
 				s.resumeInfoMutex.Lock()
 				defer s.resumeInfoMutex.Unlock()
@@ -309,7 +312,7 @@ func (s *Source) scanRepos(ctx context.Context, chunksChan chan *sources.Chunk)
 			if s.authMethod == "UNAUTHENTICATED" {
 				path, repo, err = git.CloneRepoUsingUnauthenticated(repoURL)
 			} else {
-				// If a username is not provided we need to use a default one in order to clone a private repo.
+				// If a username is not provided we need to use a default one in order to clone a private repoURL.
 				// Not setting "placeholder" as s.user on purpose in case any downstream services rely on a "" value for s.user.
 				user := s.user
 				if user == "" {
@@ -319,28 +322,32 @@ func (s *Source) scanRepos(ctx context.Context, chunksChan chan *sources.Chunk)
 			}
 			defer os.RemoveAll(path)
 			if err != nil {
-				errsMut.Lock()
-				errs = append(errs, err)
-				errsMut.Unlock()
-				return
+				errMut.Lock()
+				scanErrs = append(scanErrs, err)
+				errMut.Unlock()
+				return nil
 			}
-			log.Debugf("Starting to scan repo %d/%d: %s", i+1, len(s.repos), repoURL)
+
+			log.Debugf("Starting to scan repoURL #%d of %d: %s", cnt, repoCnt, repo)
 			err = s.git.ScanRepo(ctx, repo, path, git.NewScanOptions(), chunksChan)
 			if err != nil {
-				errsMut.Lock()
-				errs = append(errs, err)
-				errsMut.Unlock()
-				return
+				errMut.Lock()
+				scanErrs = append(scanErrs, err)
+				errMut.Unlock()
+				return nil
 			}
-			log.Debugf("Completed scanning repo %d/%d: %s", i+1, len(s.repos), repoURL)
-		}(ctx, repo, i)
-	}
-	wg.Wait()
-	if len(errs) == 0 {
-		s.SetProgressComplete(len(s.repos), len(s.repos), fmt.Sprintf("Completed scanning source %s", s.name), "")
+			atomic.AddUint64(&scanned, 1)
+			log.Debugf("scanned %d/%d repos", scanned, repoCnt)
+			log.Debugf("Completed scanning repoURL # %d of %d: %s", cnt, repoCnt, repo)
+
+			return nil
+		})
 	}
 
-	return errs
+	_ = s.jobPool.Wait()
+	s.Complete(fmt.Sprintf("Completed scanning source %s", s.name))
+
+	return scanErrs
 }
 
 // Chunks emits chunks of bytes over a channel.
@@ -417,8 +424,8 @@ func (s *Source) basicAuthSuccessful(apiClient *gitlab.Client) bool {
 	return false
 }
 
-// setProgressCompleteWithRepo calls the s.SetProgressComplete after safely setting up the encoded resume info string.
-func (s *Source) setProgressCompleteWithRepo(index int, offset int, repoURL string) {
+// setProgressCompleteWithRepo calls the s.Update after safely setting up the encoded resume info string.
+func (s *Source) setProgressCompleteWithRepo(offset int, repoURL string) {
 	s.resumeInfoMutex.Lock()
 	defer s.resumeInfoMutex.Unlock()
 
@@ -430,5 +437,5 @@ func (s *Source) setProgressCompleteWithRepo(index int, offset int, repoURL stri
 	encodedResumeInfo := sources.EncodeResumeInfo(s.resumeInfoSlice)
 
 	// Add the offset to both the index and the repos to give the proper place and proper repo count.
-	s.SetProgressComplete(index+offset, len(s.repos)+offset, fmt.Sprintf("Repo: %s", repoURL), encodedResumeInfo)
+	s.Update(len(s.repos)+offset, fmt.Sprintf("Repo: %s", repoURL), encodedResumeInfo)
 }

pkg/sources/gitlab/gitlab_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/kylelemons/godebug/pretty"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-	"golang.org/x/sync/semaphore"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/trufflesecurity/trufflehog/v3/pkg/common"
@@ -187,7 +187,7 @@ func Test_setProgressCompleteWithRepo_resumeInfo(t *testing.T) {
 
 	for _, tt := range tests {
 		s.resumeInfoSlice = tt.startingResumeInfoSlice
-		s.setProgressCompleteWithRepo(0, 0, tt.repoURL)
+		s.setProgressCompleteWithRepo(0, tt.repoURL)
 		if !reflect.DeepEqual(s.resumeInfoSlice, tt.wantResumeInfoSlice) {
 			t.Errorf("s.setProgressCompleteWithRepo() got: %v, want: %v", s.resumeInfoSlice, tt.wantResumeInfoSlice)
 		}
@@ -198,7 +198,6 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	repos := []string{"a", "b", "c", "d", "e"}
 	tests := map[string]struct {
 		repos                 []string
-		index                 int
 		offset                int
 		wantPercentComplete   int64
 		wantSectionsCompleted int32
@@ -206,7 +205,6 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	}{
 		"starting from the beginning, no offset": {
 			repos:                 repos,
-			index:                 0,
 			offset:                0,
 			wantPercentComplete:   0,
 			wantSectionsCompleted: 0,
@@ -214,18 +212,16 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 		},
 		"resume from the third, offset 2": {
 			repos:                 repos[2:],
-			index:                 0,
 			offset:                2,
 			wantPercentComplete:   40,
 			wantSectionsCompleted: 2,
 			wantSectionsRemaining: 5,
 		},
 		"resume from the third, on last repo, offset 2": {
-			repos:                 repos[2:],
-			index:                 2,
-			offset:                2,
-			wantPercentComplete:   80,
-			wantSectionsCompleted: 4,
+			repos:                 repos[3:],
+			offset:                3,
+			wantPercentComplete:   60,
+			wantSectionsCompleted: 3,
 			wantSectionsRemaining: 5,
 		},
 	}
@@ -233,21 +229,28 @@ func Test_setProgressCompleteWithRepo_Progress(t *testing.T) {
 	log.SetOutput(io.Discard)
 
 	for _, tt := range tests {
-		s := &Source{
-			repos: tt.repos,
-		}
+		t.Run("", func(t *testing.T) {
+			s := &Source{
+				repos: tt.repos,
+			}
 
-		s.setProgressCompleteWithRepo(tt.index, tt.offset, "")
-		gotProgress := s.GetProgress()
-		if gotProgress.PercentComplete != tt.wantPercentComplete {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.PercentComplete, tt.wantPercentComplete)
-		}
-		if gotProgress.SectionsCompleted != tt.wantSectionsCompleted {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.SectionsCompleted, tt.wantSectionsCompleted)
-		}
-		if gotProgress.SectionsRemaining != tt.wantSectionsRemaining {
-			t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.SectionsRemaining, tt.wantSectionsRemaining)
-		}
+			// Increment the progress counter by number of repos already processed.
+			for i := 1; i <= tt.offset; i++ {
+				s.Counter.IncSuccess()
+			}
+
+			s.setProgressCompleteWithRepo(tt.offset, "")
+			gotProgress := s.GetProgress()
+			if gotProgress.PercentComplete != tt.wantPercentComplete {
+				t.Errorf("s.setProgressCompleteWithRepo() PercentComplete got: %v want: %v", gotProgress.PercentComplete, tt.wantPercentComplete)
+			}
+			if gotProgress.SectionsCompleted != tt.wantSectionsCompleted {
+				t.Errorf("s.setProgressCompleteWithRepo() SectionsCompleted got: %v want: %v", gotProgress.SectionsCompleted, tt.wantSectionsCompleted)
+			}
+			if gotProgress.SectionsRemaining != tt.wantSectionsRemaining {
+				t.Errorf("s.setProgressCompleteWithRepo() SectionsRemaining got: %v want: %v", gotProgress.SectionsRemaining, tt.wantSectionsRemaining)
+			}
+		})
 	}
 }
 
@@ -256,16 +259,15 @@ func Test_scanRepos_SetProgressComplete(t *testing.T) {
 		name         string
 		repos        []string
 		wantComplete bool
-		wantErr      bool
 	}{
 		{
 			name:         "no repos",
 			wantComplete: true,
 		},
 		{
-			name:    "one valid repo",
-			repos:   []string{"a"},
-			wantErr: true,
+			name:         "one valid repo",
+			repos:        []string{"a"},
+			wantComplete: true,
 		},
 	}
 
@@ -274,12 +276,9 @@ func Test_scanRepos_SetProgressComplete(t *testing.T) {
 			src := &Source{
 				repos: tc.repos,
 			}
-			src.jobSem = semaphore.NewWeighted(1)
+			src.jobPool = &errgroup.Group{}
 
 			_ = src.scanRepos(context.Background(), nil)
-			if !tc.wantErr {
-				assert.Equal(t, "", src.GetProgress().EncodedResumeInfo)
-			}
 
 			gotComplete := src.GetProgress().PercentComplete == 100
 			if gotComplete != tc.wantComplete {

pkg/sources/s3/s3.go

@@ -135,12 +135,15 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		return errors.Errorf("invalid configuration given for %s source", s.name)
 	}
 
-	for i, bucket := range bucketsToScan {
+	numBuckets := len(bucketsToScan)
+	log.Debugf("Found %v total buckets to scan", numBuckets)
+	for _, bucket := range bucketsToScan {
 		if common.IsDone(ctx) {
 			return nil
 		}
 
-		s.SetProgressComplete(i, len(bucketsToScan), fmt.Sprintf("Bucket: %s", bucket), "")
+		cnt := s.Counter.IncSuccess()
+		s.Update(len(bucketsToScan), fmt.Sprintf("Bucket: %s", bucket), "")
 
 		s.log.Debugf("Scanning bucket: %s", bucket)
 		region, err := s3manager.GetBucketRegionWithClient(context.Background(), client, bucket)
@@ -161,6 +164,7 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 		// pf := "public"
 		errorCount := sync.Map{}
 
+		log.Debugf("Starting to scan bucket #%d of %d: %s", cnt, numBuckets, bucket)
 		err = regionalClient.ListObjectsV2PagesWithContext(
 			ctx, &s3.ListObjectsV2Input{Bucket: &bucket},
 			func(page *s3.ListObjectsV2Output, last bool) bool {
@@ -172,8 +176,9 @@ func (s *Source) Chunks(ctx context.Context, chunksChan chan *sources.Chunk) err
 			s.log.WithError(err).Errorf("could not list objects in s3 bucket: %s", bucket)
 			return errors.WrapPrefix(err, fmt.Sprintf("could not list objects in s3 bucket: %s", bucket), 0)
 		}
+		log.Debugf("scanned %d/%d buckets", cnt, numBuckets)
 	}
-	s.SetProgressComplete(len(bucketsToScan), len(bucketsToScan), fmt.Sprintf("Completed scanning source %s", s.name), "")
+	s.Complete(fmt.Sprintf("Completed scanning source %s", s.name))
 
 	return nil
 }

pkg/sources/sources.go

@@ -2,6 +2,7 @@ package sources
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"google.golang.org/protobuf/types/known/anypb"
 
@@ -104,39 +105,71 @@ func NewConfig(opts ...func(*Config)) Config {
 	return *c
 }
 
+// Counter is a simple monotonically-increasing counter for tracking progress.
+// It includes a successCnt that records the number of successful scans.
+// It also includes a totalCnt that records the total number of scans.
+// This is necessary in order to display the number of concurrent scans
+// regardless of whether they are successful or not.
+type Counter struct {
+	successCnt, totalCnt uint32
+}
+
+// IncTotal increments the total field of counter and returns the new value.
+func (c *Counter) IncTotal() int {
+	return int(atomic.AddUint32(&c.totalCnt, 1))
+}
+
+// IncSuccess increments the success field of counter and returns the new value.
+func (c *Counter) IncSuccess() int {
+	return int(atomic.AddUint32(&c.successCnt, 1))
+}
+
+// Get returns the current successCnt.
+func (c *Counter) Get() uint32 {
+	return atomic.LoadUint32(&c.successCnt)
+}
+
 // Progress is used to update job completion progress across sources.
 type Progress struct {
-	mut               sync.Mutex
+	SectionsCompleted int32
+	SectionsRemaining int32
 	PercentComplete   int64
 	Message           string
 	EncodedResumeInfo string
-	SectionsCompleted int32
-	SectionsRemaining int32
+	mut               sync.Mutex
+	Counter           Counter
 }
 
-// SetProgressComplete sets job progress information for a running job based on the highest level objects in the source.
-// i is the current iteration in the loop of target scope
+// Update sets job progress information for a running job based on the highest level objects in the source.
 // scope should be the len(scopedItems)
 // message is the public facing user information about the current progress
 // encodedResumeInfo is an optional string representing any information necessary to resume the job if interrupted
-func (p *Progress) SetProgressComplete(i, scope int, message, encodedResumeInfo string) {
+func (p *Progress) Update(scope int, message, encodedResumeInfo string) {
 	p.mut.Lock()
 	defer p.mut.Unlock()
 
+	i := p.Counter.Get()
+
 	p.Message = message
 	p.EncodedResumeInfo = encodedResumeInfo
 	p.SectionsCompleted = int32(i)
 	p.SectionsRemaining = int32(scope)
 
-	// If the iteration and scope are both 0, completion is 100%.
-	if i == 0 && scope == 0 {
-		p.PercentComplete = 100
-		return
-	}
-
 	p.PercentComplete = int64((float64(i) / float64(scope)) * 100)
 }
 
+// Complete sets job progress information for a completed job.
+func (p *Progress) Complete(msg string) {
+	p.mut.Lock()
+	defer p.mut.Unlock()
+
+	p.SectionsRemaining = 0
+	p.SectionsCompleted = int32(p.Counter.Get())
+	p.Message = msg
+	p.EncodedResumeInfo = ""
+	p.PercentComplete = 100
+}
+
 // GetProgress gets job completion percentage for metrics reporting.
 func (p *Progress) GetProgress() *Progress {
 	p.mut.Lock()

pkg/sources/sources_test.go

@@ -0,0 +1,192 @@
+package sources
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"google.golang.org/protobuf/testing/protocmp"
+)
+
+func TestCounter_IncTotal(t *testing.T) {
+	tests := []struct {
+		name string
+		want int
+	}{
+		{
+			name: "increment total 50 times",
+			want: 50,
+		},
+		{
+			name: "increment total 0 times",
+		},
+		{
+			name: "increment total 1 times",
+			want: 1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &Counter{}
+
+			var wg sync.WaitGroup
+			wg.Add(tt.want)
+			for i := 0; i < tt.want; i++ {
+				go func(c *Counter) {
+					defer wg.Done()
+					c.IncTotal()
+				}(c)
+			}
+
+			wg.Wait()
+			if got := c.totalCnt; int(got) != tt.want {
+				t.Errorf("Counter.IncTotal() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestCounter_IncSuccess(t *testing.T) {
+	tests := []struct {
+		name string
+		want int
+	}{
+		{
+			name: "increment total 50 times",
+			want: 50,
+		},
+		{
+			name: "increment total 0 times",
+		},
+		{
+			name: "increment total 1 times",
+			want: 1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &Counter{}
+
+			var wg sync.WaitGroup
+			wg.Add(tt.want)
+			for i := 0; i < tt.want; i++ {
+				go func(c *Counter) {
+					defer wg.Done()
+					c.IncSuccess()
+				}(c)
+			}
+
+			wg.Wait()
+			if got := c.Get(); int(got) != tt.want {
+				t.Errorf("Counter.IncTotal() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestProgress_Update(t *testing.T) {
+	tests := []struct {
+		name, msg, encodeResumeInfo string
+		counter                     uint32
+		scope, routines, want       int
+	}{
+		{
+			name:     "update progress, no counter",
+			scope:    100,
+			routines: 10,
+			want:     10,
+		},
+		{
+			name:     "update progress, counter 5",
+			scope:    100,
+			routines: 10,
+			counter:  5,
+			want:     15,
+		},
+		{
+			name:     "update progress, counter 50",
+			scope:    100,
+			routines: 10,
+			counter:  50,
+			want:     60,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := &Progress{}
+			p.Counter.successCnt = tt.counter
+
+			var wg sync.WaitGroup
+			wg.Add(tt.routines)
+			for i := 0; i < tt.routines; i++ {
+				p.Counter.IncSuccess()
+				go func(p *Progress) {
+					defer wg.Done()
+					p.Update(tt.scope, tt.msg, tt.encodeResumeInfo)
+				}(p)
+			}
+
+			wg.Wait()
+			if got := p.PercentComplete; int(got) != tt.want {
+				t.Errorf("Progress.Update(%v, %v, %v): got %v, want %v", tt.scope, tt.msg, tt.encodeResumeInfo, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestProgress_Complete(t *testing.T) {
+	tests := []struct {
+		name,
+		msg string
+		curProgress, wantProgress *Progress
+	}{
+		{
+			name: "complete progress",
+			msg:  "completed scanning thinger",
+			curProgress: &Progress{
+				PercentComplete:   42,
+				Message:           "scanning thinger",
+				EncodedResumeInfo: "resume info",
+				SectionsRemaining: 12,
+				SectionsCompleted: 2,
+			},
+			wantProgress: &Progress{
+				PercentComplete:   100,
+				Message:           "completed scanning thinger",
+				EncodedResumeInfo: "",
+				SectionsRemaining: 0,
+				SectionsCompleted: 14,
+			},
+		},
+		{
+			name: "complete progress with no message",
+			curProgress: &Progress{
+				PercentComplete:   80,
+				Message:           "scanning thinger2",
+				EncodedResumeInfo: "resume info2",
+				SectionsRemaining: 2,
+				SectionsCompleted: 42,
+			},
+			wantProgress: &Progress{
+				PercentComplete:   100,
+				EncodedResumeInfo: "",
+				SectionsRemaining: 0,
+				SectionsCompleted: 44,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.curProgress.Counter.successCnt = uint32(tt.wantProgress.SectionsCompleted)
+			tt.curProgress.Complete(tt.msg)
+
+			ignoreOpts := cmpopts.IgnoreFields(Progress{}, "mut", "Counter")
+			if !cmp.Equal(tt.curProgress, tt.wantProgress, protocmp.Transform(), ignoreOpts) {
+				t.Errorf("Progress.Complete(%v): got %v, want %v", tt.msg, tt.curProgress, tt.wantProgress)
+			}
+		})
+	}
+}