From 4a75778c1cbdd932ad429d69948520ed812b64db Mon Sep 17 00:00:00 2001
From: Denis Arh <denis.arh@gmail.com>
Date: Tue, 5 Apr 2022 15:19:36 +0200
Subject: [PATCH] Address data race in pkg/scheduler

---
 pkg/scheduler/healthcheck.go  |  2 +-
 pkg/scheduler/service.go      | 66 +++++++++++++++++------------------
 pkg/scheduler/service_test.go |  6 +++-
 3 files changed, 39 insertions(+), 35 deletions(-)

diff --git a/pkg/scheduler/healthcheck.go b/pkg/scheduler/healthcheck.go
index 9b71c2682..96763536a 100644
--- a/pkg/scheduler/healthcheck.go
+++ b/pkg/scheduler/healthcheck.go
@@ -11,7 +11,7 @@ func Healthcheck(ctx context.Context) error {
 		return nil
 	}
 
-	if gScheduler.ticker == nil {
+	if gScheduler.ticker() == nil {
 		return fmt.Errorf("stopped")
 	}
 
diff --git a/pkg/scheduler/service.go b/pkg/scheduler/service.go
index 0add63665..9bf55d3cd 100644
--- a/pkg/scheduler/service.go
+++ b/pkg/scheduler/service.go
@@ -22,7 +22,7 @@ type (
 		l sync.RWMutex
 
 		// Simple chan to control if service is running or not
-		ticker *time.Ticker
+		t *time.Ticker
 	}
 
 	dispatcher interface {
@@ -83,29 +83,29 @@ func (svc *service) OnTick(events ...eventbus.Event) {
 }
 
 func (svc *service) Stop() {
+	svc.l.Lock()
+	defer svc.l.Unlock()
 
-	if svc.ticker == nil {
-		svc.log.Debug("already stopped")
-	} else {
-		svc.log.Debug("stopping")
-		svc.ticker.Stop()
-		svc.l.Lock()
-		svc.ticker = nil
-		defer svc.l.Unlock()
+	if svc.t == nil {
+		return
 	}
+
+	svc.log.Debug("stopping")
+	svc.t.Stop()
+	svc.t = nil
 }
 
 // Run starts event scheduler service
 func (svc *service) Start(ctx context.Context) {
-
-	if svc.ticker != nil {
+	if svc.ticker() != nil {
 		svc.log.Debug("already started")
 		return
 	}
 
 	svc.l.Lock()
-	svc.ticker = &time.Ticker{}
-	svc.l.Unlock()
+	defer svc.l.Unlock()
+	// setting un-configured ticker to mark scheduler service as stated
+	svc.t = &time.Ticker{}
 
 	go func() {
 		defer sentry.Recover()
@@ -121,9 +121,11 @@ func (svc *service) Start(ctx context.Context) {
 
 		// Wait until start of the next interval
 		time.Sleep(delay)
+
 		svc.l.Lock()
-		svc.ticker = time.NewTicker(svc.interval)
-		svc.l.Unlock()
+		defer svc.l.Unlock()
+		svc.t = time.NewTicker(svc.interval)
+
 		svc.log.Debug("started")
 
 		go svc.watch(ctx)
@@ -132,46 +134,44 @@ func (svc *service) Start(ctx context.Context) {
 
 func (svc *service) watch(ctx context.Context) {
 	defer sentry.Recover()
-	defer func() {
-		defer svc.log.Debug("stopped")
-		svc.ticker.Stop()
-		svc.l.Lock()
-		svc.ticker = nil
-		svc.l.Unlock()
-	}()
+	defer svc.Stop()
 
 	// start with first interval
 	svc.dispatch(ctx)
 
 	for {
-		select {
-		case <-svc.ticker.C:
-			svc.dispatch(ctx)
+		func() {
+			select {
+			case <-svc.ticker().C:
+				svc.dispatch(ctx)
 
-		case <-ctx.Done():
-			svc.log.Debug("done")
-			return
-		}
+			case <-ctx.Done():
+				svc.log.Debug("done")
+				return
+			}
+		}()
 	}
 }
 
-func (svc *service) Started() (started bool) {
+func (svc *service) ticker() *time.Ticker {
 	svc.l.RLock()
 	defer svc.l.RUnlock()
+	return svc.t
+}
 
-	return svc.ticker != nil
+func (svc *service) Started() (started bool) {
+	return svc.ticker() != nil
 }
 
 func (svc *service) dispatch(ctx context.Context) {
 	svc.l.RLock()
+	defer svc.l.RUnlock()
 
 	ee := make([]eventbus.Event, len(svc.events))
 	for e := range svc.events {
 		ee[e] = svc.events[e]
 	}
 
-	defer svc.l.RUnlock()
-
 	for _, ev := range ee {
 		go func(ev eventbus.Event) {
 			err := svc.dispatcher.WaitFor(ctx, ev)
diff --git a/pkg/scheduler/service_test.go b/pkg/scheduler/service_test.go
index 7d44b86cb..d3804e38b 100644
--- a/pkg/scheduler/service_test.go
+++ b/pkg/scheduler/service_test.go
@@ -43,7 +43,11 @@ func TestMainServiceFunctions(t *testing.T) {
 		actionWait   = loopInterval * 10
 	)
 
-	r.Nil(gScheduler)
+	if gScheduler != nil {
+		gScheduler.Stop()
+		gScheduler = nil
+	}
+
 	Setup(zap.NewNop(), eventbus.New(), loopInterval)
 	r.NotNil(gScheduler)
 	r.False(gScheduler.Started())