fix(scheduler): fix data race (#2085)

* fix(scheduler): data race when pushing new tasks

the problem here is that scheduler can be closed in two ways:
- canceling the context given as argument to scheduler.RunScheduler()
- running scheduler.Shutdown()

because of this shutdown can trigger a data race between calling scheduler.inShutdown()
and actually pushing tasks into the pool workers

solved that by keeping a quit channel and listening on both quit channel and ctx.Done()
and closing the worker chan and scheduler afterwards.

Signed-off-by: Petu Eusebiu <peusebiu@cisco.com>

* refactor(scheduler): refactor into a single shutdown

before this we could stop scheduler either by closing the context
provided to RunScheduler(ctx) or by running Shutdown().

simplify things by getting rid of the external context in RunScheduler().
keep an internal context in the scheduler itself and pass it down to all tasks.

Signed-off-by: Petu Eusebiu <peusebiu@cisco.com>

---------

Signed-off-by: Petu Eusebiu <peusebiu@cisco.com>

pkg/extensions/extension_image_trust_test.go

@@ -257,7 +257,7 @@ func RunSignatureUploadAndVerificationTests(t *testing.T, cacheDriverParams map[
 			[]string{fmt.Sprintf("localhost:%s/%s@%s", port, repo, image.DigestStr())})
 		So(err, ShouldBeNil)
 
-		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 10*time.Second)
+		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 30*time.Second)
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
@@ -369,7 +369,7 @@ func RunSignatureUploadAndVerificationTests(t *testing.T, cacheDriverParams map[
 		err = signature.SignWithNotation(certName, imageURL, rootDir, true)
 		So(err, ShouldBeNil)
 
-		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 10*time.Second)
+		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 30*time.Second)
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
@@ -502,7 +502,7 @@ func RunSignatureUploadAndVerificationTests(t *testing.T, cacheDriverParams map[
 		err = signature.SignWithNotation(certName, imageURL, rootDir, false)
 		So(err, ShouldBeNil)
 
-		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 10*time.Second)
+		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 30*time.Second)
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
@@ -672,7 +672,7 @@ func RunSignatureUploadAndVerificationTests(t *testing.T, cacheDriverParams map[
 			[]string{fmt.Sprintf("localhost:%s/%s@%s", port, repo, image.DigestStr())})
 		So(err, ShouldBeNil)
 
-		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 10*time.Second)
+		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 30*time.Second)
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
@@ -883,7 +883,7 @@ func RunSignatureUploadAndVerificationTests(t *testing.T, cacheDriverParams map[
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
-		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 10*time.Second)
+		found, err = test.ReadLogFileAndSearchString(logFile.Name(), "update signatures validity", 30*time.Second)
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 

pkg/extensions/monitoring/monitoring_test.go

@@ -4,7 +4,6 @@
 package monitoring_test
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"math/rand"
@@ -463,8 +462,7 @@ func TestPopulateStorageMetrics(t *testing.T) {
 
 		metrics := monitoring.NewMetricsServer(true, ctlr.Log)
 		sch := scheduler.NewScheduler(conf, metrics, ctlr.Log)
-		ctx, cancel := context.WithCancel(context.Background())
-		sch.RunScheduler(ctx)
+		sch.RunScheduler()
 
 		generator := &common.StorageMetricsInitGenerator{
 			ImgStore: ctlr.StoreController.DefaultStore,
@@ -485,7 +483,7 @@ func TestPopulateStorageMetrics(t *testing.T) {
 		So(err, ShouldBeNil)
 		So(found, ShouldBeTrue)
 
-		cancel()
+		sch.Shutdown()
 		alpineSize, err := monitoring.GetDirSize(path.Join(rootDir, "alpine"))
 		So(err, ShouldBeNil)
 		busyboxSize, err := monitoring.GetDirSize(path.Join(rootDir, "busybox"))

pkg/extensions/scrub/scrub_test.go

@@ -70,13 +70,11 @@ func TestScrubExtension(t *testing.T) {
 
 		cm := test.NewControllerManager(ctlr)
 		cm.StartAndWait(port)
-		time.Sleep(6 * time.Second)
-
 		defer cm.StopServer()
 
-		data, err := os.ReadFile(logFile.Name())
+		found, err := test.ReadLogFileAndSearchString(logFile.Name(), "blobs/manifest ok", 60*time.Second)
+		So(found, ShouldBeTrue)
 		So(err, ShouldBeNil)
-		So(string(data), ShouldContainSubstring, "blobs/manifest ok")
 	})
 
 	Convey("Blobs integrity affected", t, func(c C) {
@@ -122,13 +120,11 @@ func TestScrubExtension(t *testing.T) {
 
 		cm := test.NewControllerManager(ctlr)
 		cm.StartAndWait(port)
-		time.Sleep(6 * time.Second)
-
 		defer cm.StopServer()
 
-		data, err := os.ReadFile(logFile.Name())
+		found, err := test.ReadLogFileAndSearchString(logFile.Name(), "blobs/manifest affected", 60*time.Second)
+		So(found, ShouldBeTrue)
 		So(err, ShouldBeNil)
-		So(string(data), ShouldContainSubstring, "blobs/manifest affected")
 	})
 
 	Convey("Generator error - not enough permissions to access root directory", t, func(c C) {
@@ -170,13 +166,11 @@ func TestScrubExtension(t *testing.T) {
 
 		cm := test.NewControllerManager(ctlr)
 		cm.StartAndWait(port)
-		time.Sleep(6 * time.Second)
-
 		defer cm.StopServer()
 
-		data, err := os.ReadFile(logFile.Name())
+		found, err := test.ReadLogFileAndSearchString(logFile.Name(), "failed to execute generator", 60*time.Second)
+		So(found, ShouldBeTrue)
 		So(err, ShouldBeNil)
-		So(string(data), ShouldContainSubstring, "failed to execute generator")
 
 		So(os.Chmod(path.Join(dir, repoName), 0o755), ShouldBeNil)
 	})

pkg/extensions/search/cve/scan_test.go

@@ -432,11 +432,9 @@ func TestScanGeneratorWithMockedData(t *testing.T) { //nolint: gocyclo
 
 		sch.SubmitGenerator(generator, 10*time.Second, scheduler.MediumPriority)
 
-		ctx, cancel := context.WithCancel(context.Background())
+		sch.RunScheduler()
 
-		sch.RunScheduler(ctx)
-
-		defer cancel()
+		defer sch.Shutdown()
 
 		// Make sure the scanner generator has completed despite errors
 		found, err := test.ReadLogFileAndSearchString(logPath,
@@ -529,11 +527,9 @@ func TestScanGeneratorWithRealData(t *testing.T) {
 		// Start the generator
 		sch.SubmitGenerator(generator, 120*time.Second, scheduler.MediumPriority)
 
-		ctx, cancel := context.WithCancel(context.Background())
+		sch.RunScheduler()
 
-		sch.RunScheduler(ctx)
-
-		defer cancel()
+		defer sch.Shutdown()
 
 		// Make sure the scanner generator has completed
 		found, err := test.ReadLogFileAndSearchString(logPath,

pkg/extensions/search/cve/update_test.go

@@ -63,11 +63,9 @@ func TestCVEDBGenerator(t *testing.T) {
 
 		sch.SubmitGenerator(generator, 12000*time.Millisecond, scheduler.HighPriority)
 
-		ctx, cancel := context.WithCancel(context.Background())
+		sch.RunScheduler()
 
-		sch.RunScheduler(ctx)
-
-		defer cancel()
+		defer sch.Shutdown()
 
 		// Wait for trivy db to download
 		found, err := test.ReadLogFileAndCountStringOccurence(logPath,

pkg/extensions/search/search_test.go

@@ -681,16 +681,14 @@ func TestRepoListWithNewestImage(t *testing.T) {
 		ctlr := api.NewController(conf)
 		ctlr.Log.Logger = ctlr.Log.Output(writers)
 
-		ctx := context.Background()
-
-		if err := ctlr.Init(ctx); err != nil {
+		if err := ctlr.Init(); err != nil {
 			panic(err)
 		}
 
 		ctlr.CveScanner = getMockCveScanner(ctlr.MetaDB)
 
 		go func() {
-			if err := ctlr.Run(ctx); !errors.Is(err, http.ErrServerClosed) {
+			if err := ctlr.Run(); !errors.Is(err, http.ErrServerClosed) {
 				panic(err)
 			}
 		}()
@@ -3373,16 +3371,14 @@ func TestGlobalSearch(t *testing.T) {
 		ctlr := api.NewController(conf)
 		ctlr.Log.Logger = ctlr.Log.Output(writers)
 
-		ctx := context.Background()
-
-		if err := ctlr.Init(ctx); err != nil {
+		if err := ctlr.Init(); err != nil {
 			panic(err)
 		}
 
 		ctlr.CveScanner = getMockCveScanner(ctlr.MetaDB)
 
 		go func() {
-			if err := ctlr.Run(ctx); !errors.Is(err, http.ErrServerClosed) {
+			if err := ctlr.Run(); !errors.Is(err, http.ErrServerClosed) {
 				panic(err)
 			}
 		}()
@@ -6205,16 +6201,14 @@ func TestImageSummary(t *testing.T) {
 		configDigest := godigest.FromBytes(configBlob)
 		So(errConfig, ShouldBeNil) // marshall success, config is valid JSON
 
-		ctx := context.Background()
-
-		if err := ctlr.Init(ctx); err != nil {
+		if err := ctlr.Init(); err != nil {
 			panic(err)
 		}
 
 		ctlr.CveScanner = getMockCveScanner(ctlr.MetaDB)
 
 		go func() {
-			if err := ctlr.Run(ctx); !errors.Is(err, http.ErrServerClosed) {
+			if err := ctlr.Run(); !errors.Is(err, http.ErrServerClosed) {
 				panic(err)
 			}
 		}()

pkg/extensions/sync/service.go

@@ -44,7 +44,7 @@ func New(
 	storeController storage.StoreController,
 	metadb mTypes.MetaDB,
 	log log.Logger,
-) (Service, error) {
+) (*BaseService, error) {
 	service := &BaseService{}
 
 	service.config = opts

pkg/extensions/sync/sync_internal_test.go

@@ -170,6 +170,29 @@ func TestService(t *testing.T) {
 	})
 }
 
+func TestSyncRepo(t *testing.T) {
+	Convey("trigger context error", t, func() {
+		conf := syncconf.RegistryConfig{
+			URLs: []string{"http://localhost"},
+		}
+
+		service, err := New(conf, "", os.TempDir(), storage.StoreController{}, mocks.MetaDBMock{}, log.Logger{})
+		So(err, ShouldBeNil)
+
+		service.remote = mocks.SyncRemote{
+			GetRepoTagsFn: func(repo string) ([]string, error) {
+				return []string{"repo1", "repo2"}, nil
+			},
+		}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		err = service.SyncRepo(ctx, "repo")
+		So(err, ShouldEqual, ctx.Err())
+	})
+}
+
 func TestDestinationRegistry(t *testing.T) {
 	Convey("make StoreController", t, func() {
 		dir := t.TempDir()

pkg/extensions/sync/sync_test.go

@@ -1901,15 +1901,15 @@ func TestConfigReloader(t *testing.T) {
 			hotReloader, err := cli.NewHotReloader(dctlr, cfgfile.Name())
 			So(err, ShouldBeNil)
 
-			reloadCtx := hotReloader.Start()
+			hotReloader.Start()
 
 			go func() {
 				// this blocks
-				if err := dctlr.Init(reloadCtx); err != nil {
+				if err := dctlr.Init(); err != nil {
 					return
 				}
 
-				if err := dctlr.Run(reloadCtx); err != nil {
+				if err := dctlr.Run(); err != nil {
 					return
 				}
 			}()
@@ -2051,15 +2051,15 @@ func TestConfigReloader(t *testing.T) {
 			hotReloader, err := cli.NewHotReloader(dctlr, cfgfile.Name())
 			So(err, ShouldBeNil)
 
-			reloadCtx := hotReloader.Start()
+			hotReloader.Start()
 
 			go func() {
 				// this blocks
-				if err := dctlr.Init(reloadCtx); err != nil {
+				if err := dctlr.Init(); err != nil {
 					return
 				}
 
-				if err := dctlr.Run(reloadCtx); err != nil {
+				if err := dctlr.Run(); err != nil {
 					return
 				}
 			}()
@@ -3938,6 +3938,12 @@ func TestPeriodicallySignaturesErr(t *testing.T) {
 					}
 				}
 
+				// start downstream server
+				updateDuration, err = time.ParseDuration("1s")
+				So(err, ShouldBeNil)
+
+				syncConfig.Registries[0].PollInterval = updateDuration
+
 				// start downstream server
 				dctlr, destBaseURL, _, _ := makeDownstreamServer(t, false, syncConfig)
 
@@ -4030,6 +4036,61 @@ func TestPeriodicallySignaturesErr(t *testing.T) {
 				So(err, ShouldBeNil)
 				So(len(index.Manifests), ShouldEqual, 0)
 			})
+
+			Convey("of type OCI image, error on downstream in canSkipReference()", func() { //nolint: dupl
+				// start downstream server
+				updateDuration, err = time.ParseDuration("1s")
+				So(err, ShouldBeNil)
+
+				syncConfig.Registries[0].PollInterval = updateDuration
+				dctlr, _, destDir, _ := makeDownstreamServer(t, false, syncConfig)
+
+				dcm := test.NewControllerManager(dctlr)
+				dcm.StartAndWait(dctlr.Config.HTTP.Port)
+				defer dcm.StopServer()
+
+				found, err := test.ReadLogFileAndSearchString(dctlr.Config.Log.Output,
+					"finished syncing all repos", 15*time.Second)
+				if err != nil {
+					panic(err)
+				}
+
+				if !found {
+					data, err := os.ReadFile(dctlr.Config.Log.Output)
+					So(err, ShouldBeNil)
+
+					t.Logf("downstream log: %s", string(data))
+				}
+
+				So(found, ShouldBeTrue)
+
+				time.Sleep(time.Second)
+
+				blob := referrers.Manifests[0]
+				blobsDir := path.Join(destDir, repoName, "blobs", string(blob.Digest.Algorithm()))
+				blobPath := path.Join(blobsDir, blob.Digest.Encoded())
+				err = os.MkdirAll(blobsDir, storageConstants.DefaultDirPerms)
+				So(err, ShouldBeNil)
+				err = os.WriteFile(blobPath, []byte("blob"), storageConstants.DefaultFilePerms)
+				So(err, ShouldBeNil)
+				err = os.Chmod(blobPath, 0o000)
+				So(err, ShouldBeNil)
+
+				found, err = test.ReadLogFileAndSearchString(dctlr.Config.Log.Output,
+					"couldn't check if the upstream oci references for image can be skipped", 30*time.Second)
+				if err != nil {
+					panic(err)
+				}
+
+				if !found {
+					data, err := os.ReadFile(dctlr.Config.Log.Output)
+					So(err, ShouldBeNil)
+
+					t.Logf("downstream log: %s", string(data))
+				}
+
+				So(found, ShouldBeTrue)
+			})
 		})
 	})
 }
@@ -5200,78 +5261,6 @@ func TestOnDemandPullsOnce(t *testing.T) {
 	})
 }
 
-func TestError(t *testing.T) {
-	Convey("Verify periodically sync pushSyncedLocalImage() error", t, func() {
-		updateDuration, _ := time.ParseDuration("30m")
-
-		sctlr, srcBaseURL, _, _, _ := makeUpstreamServer(t, false, false)
-
-		scm := test.NewControllerManager(sctlr)
-		scm.StartAndWait(sctlr.Config.HTTP.Port)
-		defer scm.StopServer()
-
-		regex := ".*"
-		semver := true
-		var tlsVerify bool
-
-		syncRegistryConfig := syncconf.RegistryConfig{
-			Content: []syncconf.Content{
-				{
-					Prefix: testImage,
-					Tags: &syncconf.Tags{
-						Regex:  &regex,
-						Semver: &semver,
-					},
-				},
-			},
-			URLs:         []string{srcBaseURL},
-			PollInterval: updateDuration,
-			TLSVerify:    &tlsVerify,
-			CertDir:      "",
-		}
-
-		defaultVal := true
-		syncConfig := &syncconf.Config{
-			Enable:     &defaultVal,
-			Registries: []syncconf.RegistryConfig{syncRegistryConfig},
-		}
-
-		dctlr, _, destDir, _ := makeDownstreamServer(t, false, syncConfig)
-
-		dcm := test.NewControllerManager(dctlr)
-		dcm.StartAndWait(dctlr.Config.HTTP.Port)
-		defer dcm.StopServer()
-
-		// give permission denied on pushSyncedLocalImage()
-		localRepoPath := path.Join(destDir, testImage, "blobs")
-		err := os.MkdirAll(localRepoPath, 0o755)
-		So(err, ShouldBeNil)
-
-		err = os.Chmod(localRepoPath, 0o000)
-		So(err, ShouldBeNil)
-
-		defer func() {
-			err = os.Chmod(localRepoPath, 0o755)
-			So(err, ShouldBeNil)
-		}()
-
-		found, err := test.ReadLogFileAndSearchString(dctlr.Config.Log.Output,
-			"couldn't commit image to local image store", 30*time.Second)
-		if err != nil {
-			panic(err)
-		}
-
-		if !found {
-			data, err := os.ReadFile(dctlr.Config.Log.Output)
-			So(err, ShouldBeNil)
-
-			t.Logf("downstream log: %s", string(data))
-		}
-
-		So(found, ShouldBeTrue)
-	})
-}
-
 func TestSignaturesOnDemand(t *testing.T) {
 	Convey("Verify sync signatures on demand feature", t, func() {
 		sctlr, srcBaseURL, srcDir, _, _ := makeUpstreamServer(t, false, false)