test(cooler): Pull concurrency, expiry and prefix isolation

- TestPullConcurrent: 64 concurrent Pulls of one key, one winner, 20 rounds,
  on memory and miniredis
- TestPullExpired: an expired value is never returned and is gone afterwards
- TestPullPrefixIsolation: Pull touches only its own <prefix>cache:<key>
- TestPullDocker: the same on redis:7-alpine (GETDEL)

modules/cooler/docker_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -131,3 +133,74 @@ func TestRedisCacheDocker(t *testing.T) {
 		}
 	}
 }
+
+// TestPullDocker runs cooler.Cache.Pull against redis:7-alpine (GETDEL
+// needs Redis 6.2 or newer): a value is found once, a second Pull finds
+// nothing, an expired key is never returned, and 64 concurrent Pulls of one
+// key have exactly one winner, 10 rounds (D-08).
+func TestPullDocker(t *testing.T) {
+	client := realRedis(t)
+	ctx := t.Context()
+	c := cooler.NewRedisCache(client, "acme:")
+
+	if err := c.Set(ctx, "code", []byte("v"), time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	if v, ok, err := c.Pull(ctx, "code"); err != nil || !ok || string(v) != "v" {
+		t.Fatalf("first Pull = %q %v %v", v, ok, err)
+	}
+	if v, ok, err := c.Pull(ctx, "code"); err != nil || ok || v != nil {
+		t.Fatalf("second Pull = %q %v %v, want nil false nil", v, ok, err)
+	}
+	if n, err := client.Exists(ctx, "acme:cache:code").Result(); err != nil || n != 0 {
+		t.Fatalf("the key survived Pull (%d, %v)", n, err)
+	}
+
+	if err := c.Set(ctx, "short", []byte("x"), 100*time.Millisecond); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(200 * time.Millisecond)
+	if v, ok, err := c.Pull(ctx, "short"); err != nil || ok || v != nil {
+		t.Fatalf("Pull of an expired key = %q %v %v, want nil false nil", v, ok, err)
+	}
+
+	const pullers, rounds = 64, 10
+	for round := range rounds {
+		key := "race-" + strconv.Itoa(round)
+		want := "one-time-" + strconv.Itoa(round)
+		if err := c.Set(ctx, key, []byte(want), time.Minute); err != nil {
+			t.Fatal(err)
+		}
+		var (
+			wg      sync.WaitGroup
+			winners atomic.Int32
+		)
+		start := make(chan struct{})
+		errs := make(chan error, pullers)
+		for range pullers {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				<-start
+				v, ok, err := c.Pull(ctx, key)
+				switch {
+				case err != nil:
+					errs <- err
+				case ok && string(v) != want:
+					errs <- fmt.Errorf("winner got %q, want %q", v, want)
+				case ok:
+					winners.Add(1)
+				}
+			}()
+		}
+		close(start)
+		wg.Wait()
+		close(errs)
+		for err := range errs {
+			t.Fatal(err)
+		}
+		if n := winners.Load(); n != 1 {
+			t.Fatalf("round %d: %d of %d Pulls found the value, want exactly 1", round, n, pullers)
+		}
+	}
+}