Fix leveldb test race (#10054)

Co-authored-by: Lauris BH <[email protected]>
Co-authored-by: techknowlogick <[email protected]>

Author: 3 people

modules/queue/queue_disk_test.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -24,6 +25,7 @@ func TestLevelQueue(t *testing.T) {
 		}
 	}
 
+	var lock sync.Mutex
 	queueShutdown := []func(){}
 	queueTerminate := []func(){}
 
@@ -46,9 +48,13 @@ func TestLevelQueue(t *testing.T) {
 	assert.NoError(t, err)
 
 	go queue.Run(func(_ context.Context, shutdown func()) {
+		lock.Lock()
 		queueShutdown = append(queueShutdown, shutdown)
+		lock.Unlock()
 	}, func(_ context.Context, terminate func()) {
+		lock.Lock()
 		queueTerminate = append(queueTerminate, terminate)
+		lock.Unlock()
 	})
 
 	test1 := testData{"A", 1}
@@ -72,9 +78,12 @@ func TestLevelQueue(t *testing.T) {
 	err = queue.Push(test1)
 	assert.Error(t, err)
 
+	lock.Lock()
 	for _, callback := range queueShutdown {
 		callback()
 	}
+	lock.Unlock()
+
 	time.Sleep(200 * time.Millisecond)
 	err = queue.Push(&test1)
 	assert.NoError(t, err)
@@ -85,9 +94,11 @@ func TestLevelQueue(t *testing.T) {
 		assert.Fail(t, "Handler processing should have stopped")
 	default:
 	}
+	lock.Lock()
 	for _, callback := range queueTerminate {
 		callback()
 	}
+	lock.Unlock()
 
 	// Reopen queue
 	queue, err = NewWrappedQueue(handle,
@@ -109,9 +120,13 @@ func TestLevelQueue(t *testing.T) {
 	assert.NoError(t, err)
 
 	go queue.Run(func(_ context.Context, shutdown func()) {
+		lock.Lock()
 		queueShutdown = append(queueShutdown, shutdown)
+		lock.Unlock()
 	}, func(_ context.Context, terminate func()) {
+		lock.Lock()
 		queueTerminate = append(queueTerminate, terminate)
+		lock.Unlock()
 	})
 
 	result3 := <-handleChan
@@ -121,10 +136,13 @@ func TestLevelQueue(t *testing.T) {
 	result4 := <-handleChan
 	assert.Equal(t, test2.TestString, result4.TestString)
 	assert.Equal(t, test2.TestInt, result4.TestInt)
+
+	lock.Lock()
 	for _, callback := range queueShutdown {
 		callback()
 	}
 	for _, callback := range queueTerminate {
 		callback()
 	}
+	lock.Unlock()
 }