streamcache: Fix flaky TestCache_diskCleanup

Because implementation was used single sleep function for
both filestore cleanup and clean function invocation the
close of the channel to trigger "single" invocation of the
clean was broken. This is due to close will make channel
to return default results immediately. But for the test we
need to run it only once. We can't send two elements on
the single channel as we can't be sure both won't be consumed
by one of those functions. That is why we add separate sleep
function for each of those functions. By sending on each channel
we are sure each function will be triggered exactly once.

2 files changed, 22 insertions, 9 deletions

diff --git a/internal/streamcache/cache.go b/internal/streamcache/cache.go
index 0ea6ca3f0..5eb576e72 100644
--- a/internal/streamcache/cache.go
+++ b/internal/streamcache/cache.go
@@ -128,14 +128,20 @@ type cache struct {
 // New returns a new cache instance.
 func New(cfg config.StreamCacheConfig, logger logrus.FieldLogger) Cache {
 	if cfg.Enabled {
-		return newCacheWithSleep(cfg.Dir, cfg.MaxAge.Duration(), time.After, logger)
+		return newCacheWithSleep(cfg.Dir, cfg.MaxAge.Duration(), time.After, time.After, logger)
 	}
 
 	return NullCache{}
 }
 
-func newCacheWithSleep(dir string, maxAge time.Duration, sleep func(time.Duration) <-chan time.Time, logger logrus.FieldLogger) Cache {
-	fs := newFilestore(dir, maxAge, sleep, logger)
+func newCacheWithSleep(
+	dir string,
+	maxAge time.Duration,
+	filestoreSleep func(time.Duration) <-chan time.Time,
+	cleanSleep func(time.Duration) <-chan time.Time,
+	logger logrus.FieldLogger,
+) Cache {
+	fs := newFilestore(dir, maxAge, filestoreSleep, logger)
 
 	c := &cache{
 		maxAge:     maxAge,
@@ -148,7 +154,7 @@ func newCacheWithSleep(dir string, maxAge time.Duration, sleep func(time.Duratio
 	}
 
 	c.sleepLoop.Go(func() {
-		sleepLoop(c.stop, c.maxAge, sleep, c.clean)
+		sleepLoop(c.stop, c.maxAge, cleanSleep, c.clean)
 	})
 	go func() {
 		<-c.stop
diff --git a/internal/streamcache/cache_test.go b/internal/streamcache/cache_test.go
index f562dad42..013e13a7a 100644
--- a/internal/streamcache/cache_test.go
+++ b/internal/streamcache/cache_test.go
@@ -227,11 +227,17 @@ func TestCache_diskCleanup(t *testing.T) {
 		key = "test key"
 	)
 
-	timerCh := make(chan time.Time)
+	filestoreCleanTimerCh := make(chan time.Time)
+	filestoreClean := func(time.Duration) <-chan time.Time {
+		return filestoreCleanTimerCh
+	}
 
-	c := newCacheWithSleep(tmp, 0, func(time.Duration) <-chan time.Time {
-		return timerCh
-	}, log.Default())
+	cleanSleepTimerCh := make(chan time.Time)
+	cleanSleep := func(time.Duration) <-chan time.Time {
+		return cleanSleepTimerCh
+	}
+
+	c := newCacheWithSleep(tmp, 0, filestoreClean, cleanSleep, log.Default())
 	defer c.Stop()
 
 	content := func(i int) string { return fmt.Sprintf("content %d", i) }
@@ -251,7 +257,8 @@ func TestCache_diskCleanup(t *testing.T) {
 	requireCacheEntries(t, c, 1)
 
 	// Unblock cleanup goroutines so they run exactly once
-	close(timerCh)
+	cleanSleepTimerCh <- time.Time{}
+	filestoreCleanTimerCh <- time.Time{}
 	// Give them time to do their work
 	time.Sleep(10 * time.Millisecond)