cache: Add ability to stop disk walkers

The disk cache spawns a set of walkers which regularly walk the cache to
check for cache members which are older than a given threshold. These
walkers currently cannot be stopped at all, leading to a Goroutine
leakage in our tests.

Implement a new `StopWalkers()` function which stops them and use it in
our tests.

3 files changed, 33 insertions, 2 deletions

diff --git a/internal/cache/diskcache.go b/internal/cache/diskcache.go
index 3fb5bf4bb..dffc38ca5 100644
--- a/internal/cache/diskcache.go
+++ b/internal/cache/diskcache.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus/ctxlogrus"
 	"github.com/prometheus/client_golang/prometheus"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/dontpanic"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/config"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/storage"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/safe"
@@ -79,6 +80,9 @@ type DiskCache struct {
 	af          activeFiles
 	cacheConfig cacheConfig
 
+	walkersDone chan struct{}
+	walkerLoops []*dontpanic.Forever
+
 	requestTotals              prometheus.Counter
 	missTotals                 prometheus.Counter
 	bytesStoredtotals          prometheus.Counter
@@ -107,6 +111,7 @@ func New(cfg config.Cfg, locator storage.Locator, opts ...Option) *DiskCache {
 			m:     map[string]int{},
 		},
 		cacheConfig: cacheConfig,
+		walkersDone: make(chan struct{}),
 
 		requestTotals: prometheus.NewCounter(
 			prometheus.CounterOpts{
diff --git a/internal/cache/walker.go b/internal/cache/walker.go
index 25403ad08..00d990c17 100644
--- a/internal/cache/walker.go
+++ b/internal/cache/walker.go
@@ -98,12 +98,27 @@ func (c *DiskCache) walkLoop(walkPath string) {
 	logger.Infof("Starting file walker for %s", walkPath)
 
 	walkTick := time.NewTicker(cleanWalkFrequency)
-	dontpanic.NewForever(time.Minute).Go(func() {
+
+	forever := dontpanic.NewForever(time.Minute)
+	forever.Go(func() {
+		select {
+		case <-c.walkersDone:
+			return
+		default:
+		}
+
 		if err := c.cleanWalk(walkPath); err != nil {
 			logger.Error(err)
 		}
-		<-walkTick.C
+
+		select {
+		case <-c.walkersDone:
+			return
+		case <-walkTick.C:
+		}
 	})
+
+	c.walkerLoops = append(c.walkerLoops, forever)
 }
 
 func (c *DiskCache) startCleanWalker(cacheDir, stateDir string) {
@@ -199,3 +214,12 @@ func (c *DiskCache) StartWalkers() error {
 
 	return nil
 }
+
+// StopWalkers stops all walkers started by StartWalkers.
+func (c *DiskCache) StopWalkers() {
+	close(c.walkersDone)
+	for _, walkerLoop := range c.walkerLoops {
+		walkerLoop.Cancel()
+	}
+	c.walkerLoops = nil
+}
diff --git a/internal/cache/walker_test.go b/internal/cache/walker_test.go
index 87e02d8fa..727e93d04 100644
--- a/internal/cache/walker_test.go
+++ b/internal/cache/walker_test.go
@@ -53,6 +53,7 @@ func TestDiskCacheObjectWalker(t *testing.T) {
 
 	cache := New(cfg, locator, withDisabledMoveAndClear())
 	require.NoError(t, cache.StartWalkers())
+	defer cache.StopWalkers()
 
 	pollCountersUntil(t, cache, 4)
 
@@ -78,6 +79,7 @@ func TestDiskCacheInitialClear(t *testing.T) {
 
 	cache := New(cfg, locator, withDisabledWalker())
 	require.NoError(t, cache.StartWalkers())
+	defer cache.StopWalkers()
 
 	require.NoFileExists(t, canary)
 }