Merge pull request #1070 from Davoska/ocpbugs-32678-cvo-panics

OCPBUGS-32678: Fix a panic caused by a data race

Author: openshift-merge-bot[bot]

pkg/cvo/sync_worker.go

@@ -161,6 +161,9 @@ type SyncWorker struct {
 	// coordination between the sync loop and external callers
 	notify chan string
 	report chan SyncWorkerStatus
+	// startApply is used to start the initial attempt to apply a payload. It may be
+	// used consecutively to start additional attempts as well.
+	startApply chan string
 
 	// lock guards changes to these fields
 	lock     sync.Mutex
@@ -199,7 +202,8 @@ func NewSyncWorker(retriever PayloadRetriever, builder payload.ResourceBuilder,
 
 		minimumReconcileInterval: reconcileInterval,
 
-		notify: make(chan string, 1),
+		notify:     make(chan string, 1),
+		startApply: make(chan string, 1),
 		// report is a large buffered channel to improve local testing - most consumers should invoke
 		// Status() or use the result of calling Update() instead because the channel can be out of date
 		// if the reader is not fast enough.
@@ -547,7 +551,7 @@ func (w *SyncWorker) Update(ctx context.Context, generation int64, desired confi
 	}
 	msg := "new work is available"
 	select {
-	case w.notify <- msg:
+	case w.startApply <- msg:
 		klog.V(2).Info("Notify the sync worker that new work is available")
 	default:
 		klog.V(2).Info("The sync worker has already been notified that new work is available")
@@ -565,6 +569,23 @@ func (w *SyncWorker) Start(ctx context.Context, maxWorkers int) {
 	klog.V(2).Infof("Start: starting sync worker")
 
 	work := &SyncWork{}
+	initialStartApplyReceived := make(chan string, 1) // a local channel to not cause a potential deadlock
+
+	// Until Update() has finished at least once, we do nothing.
+	for loop := true; loop; {
+		select {
+		case <-ctx.Done():
+			klog.V(2).Infof("The sync worker was shut down while waiting for the initial signal")
+			return
+		case <-w.notify:
+			// Do not queue any retries until the worker has started
+			klog.V(2).Infof("The sync worker was notified; however, it is waiting for the initial signal")
+		case msg := <-w.startApply:
+			klog.V(2).Infof("The sync worker has received the initial signal")
+			initialStartApplyReceived <- msg
+			loop = false
+		}
+	}
 
 	wait.Until(func() {
 		consecutiveErrors := 0
@@ -582,6 +603,10 @@ func (w *SyncWorker) Start(ctx context.Context, maxWorkers int) {
 				klog.V(2).Infof("Wait finished")
 			case msg := <-w.notify:
 				klog.V(2).Info(msg)
+			case msg := <-w.startApply:
+				klog.V(2).Info(msg)
+			case msg := <-initialStartApplyReceived:
+				klog.V(2).Info(msg)
 			}
 
 			// determine whether we need to do work

pkg/cvo/sync_worker_test.go

@@ -476,3 +476,55 @@ func Test_equalDigest(t *testing.T) {
 		})
 	}
 }
+
+func Test_SyncWorkerShouldNotPanicDueToNotifySignalAtStartUp(t *testing.T) {
+	o, _, _, _, shutdownFn := setupCVOTest("testdata/panic")
+	defer shutdownFn()
+	syncChannel := make(chan bool)
+
+	// Start() should not cause a panic when the notify channel already contains elements at its startup
+	ctx, cancel := context.WithCancel(context.Background())
+	worker := o.configSync.(*SyncWorker)
+	worker.notify <- "Notify the sync worker: Cluster operator A changed versions"
+	go func() {
+		worker.Start(ctx, 1)
+		syncChannel <- true
+	}()
+
+	// A panic must not occur due to the notify signal; wait a reasonable time for a potential panic
+	<-time.After(3 * time.Second)
+
+	// Shut down the sync worker and wait for the confirmation
+	cancel()
+	select {
+	case <-syncChannel:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Sync worker did not shut down in time after its context was cancelled")
+	}
+}
+
+func Test_SyncWorkerShouldShutDownImmediatelyAtStartUpWhenContextCancelled(t *testing.T) {
+	o, _, _, _, shutdownFn := setupCVOTest("testdata/panic")
+	defer shutdownFn()
+	syncChannel := make(chan bool)
+
+	// Start() shuts down immediately while waiting for its initial signal
+	// Only the context cancellation signal is received
+	ctx, cancel := context.WithCancel(context.Background())
+	worker := o.configSync.(*SyncWorker)
+	go func() {
+		worker.Start(ctx, 1)
+		syncChannel <- true
+	}()
+
+	// A panic must not occur due to the notify signal; wait a reasonable time for a potential panic
+	<-time.After(3 * time.Second)
+
+	// Shut down the sync worker and wait for the confirmation
+	cancel()
+	select {
+	case <-syncChannel:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Sync worker did not shut down in time after its context was cancelled")
+	}
+}