streamcache: Unlock waiters after cache keys have been prunedpks-streamcache-fix-flaky-test

When creating a new cache entry the writer is running in a Goroutine to
populate the cache file. After it has ran we then unlock any waiters
which have been waiting for the writer to finish. If the writer failed,
we also remove its respective cache key from the cache so that it's not
used by anything else anymore.

The current order is that we first unblock waiters and then delete the
key. This is creating a race in our tests when checking whether we try
to reuse the failed cache entry because the cache key still exists for a
brief period after the original failing writer has exited.

Fix this flake by only unblocking waiters after the cache key has been
removed. This also feels like the right thing to do outside of this
flaky test: it's inconsistent to treat the write as finished but failed,
but to still have the cache entry around at that point in time.

Changelog: fixed

2 files changed, 5 insertions, 3 deletions

diff --git a/internal/streamcache/cache.go b/internal/streamcache/cache.go
index a6e39233b..333be63c7 100644
--- a/internal/streamcache/cache.go
+++ b/internal/streamcache/cache.go
@@ -320,7 +320,11 @@ func (c *cache) newEntry(key string, create func(io.Writer) error) (_ *Stream, _
 
 	go func() {
 		err := runCreate(e.pipe, create)
-		e.waiter.SetError(err)
+
+		// We defer this until after we have removed the cache entry so that the waiter is
+		// only unblocked when the cache key has already been pruned from the cache.
+		defer e.waiter.SetError(err)
+
 		if err != nil {
 			c.logger.WithError(err).Error("create cache entry")
 			c.m.Lock()
diff --git a/internal/streamcache/cache_test.go b/internal/streamcache/cache_test.go
index a90c4be3b..fa7f1da46 100644
--- a/internal/streamcache/cache_test.go
+++ b/internal/streamcache/cache_test.go
@@ -339,8 +339,6 @@ func TestCache_failedWrite(t *testing.T) {
 			require.NoError(t, r1.Close(), "errors on the write end are not propagated via Close()")
 			require.Error(t, r1.Wait(ctx), "error propagation happens via Wait()")
 
-			time.Sleep(10 * time.Millisecond)
-
 			const happy = "all is good"
 			r2, created, err := c.FindOrCreate(tc.desc, writeString(happy))
 			require.NoError(t, err)