Atomic Redis writes, batched tag lookups and Redis in CI

.github/workflows/tests.yml

@@ -19,6 +19,17 @@ jobs:
 
     name: ${{ matrix.php }} - ${{ matrix.statamic }} - ${{ matrix.laravel }}
 
+    services:
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 3s
+          --health-retries 10
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v1
@@ -27,7 +38,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}
-          extensions: dom, curl, libxml, mbstring, zip, pcntl, pdo, sqlite, pdo_sqlite, bcmath, soap, intl, gd, exif, iconv, imagick
+          extensions: dom, curl, libxml, mbstring, zip, pcntl, pdo, sqlite, pdo_sqlite, bcmath, soap, intl, gd, exif, iconv, imagick, redis
 
       - name: Install dependencies
         run: |
@@ -36,3 +47,8 @@ jobs:
 
       - name: Run Pest
         run: vendor/bin/pest
+
+      - name: Run Pest against Redis
+        run: vendor/bin/pest
+        env:
+          CACHE_TRACKER_TEST_REDIS: 1

src/Tracker/Stores/CacheStore.php

@@ -116,8 +116,12 @@ public function matching(array $tags): array
         }
 
         $buckets = [];
-        foreach ($exact as $tag) {
-            $buckets[$tag] = array_keys($this->cache->get($this->tagKey($tag)) ?? []);
+        foreach (array_chunk($exact, self::CHUNK) as $chunk) {
+            $values = $this->cache->many(array_map(fn ($tag) => $this->tagKey($tag), $chunk));
+
+            foreach ($chunk as $tag) {
+                $buckets[$tag] = array_keys($values[$this->tagKey($tag)] ?? []);
+            }
         }
 
         $records = $this->many(array_unique(array_merge(...array_values($buckets ?: [[]]))));

src/Tracker/Stores/RedisStore.php

@@ -33,7 +33,7 @@ public function put(string $key, string $url, array $tags): void
         $old = $this->get($key);
         $removed = $old ? array_diff($old['tags'], $tags) : [];
 
-        $this->batch(function ($redis) use ($key, $url, $tags, $removed) {
+        $this->atomically(function ($redis) use ($key, $url, $tags, $removed) {
             $redis->hset($this->urlsKey(), $key, $this->encode($url, $tags));
 
             foreach ($tags as $tag) {
@@ -95,7 +95,7 @@ public function forget(array $keys): void
                 }
             }
 
-            $this->batch(function ($redis) use ($chunk, $byTag) {
+            $this->atomically(function ($redis) use ($chunk, $byTag) {
                 $redis->hdel($this->urlsKey(), ...$chunk);
 
                 foreach ($byTag as $tag => $keys) {
@@ -134,8 +134,8 @@ public function matching(array $tags): array
         }
 
         $keys = [];
-        foreach (array_unique($exact) as $tag) {
-            foreach ($this->redis->smembers($this->tagKey($tag)) ?: [] as $key) {
+        foreach ($this->membersOf(array_unique($exact)) as $members) {
+            foreach ($members as $key) {
                 $keys[$key] = true;
             }
         }
@@ -161,6 +161,39 @@ public function flush(): void
         $this->redis->del($this->tagsKey());
     }
 
+    /**
+     * Get the URL keys in each tag's set, pipelining the lookups so a long
+     * list of tags doesn't cost one round trip per tag.
+     *
+     * @return array<int, array> one list of URL keys per tag
+     */
+    private function membersOf(array $tags): array
+    {
+        $results = [];
+
+        foreach (array_chunk(array_values($tags), self::CHUNK) as $chunk) {
+            if ($this->isCluster()) {
+                foreach ($chunk as $tag) {
+                    $results[] = $this->redis->smembers($this->tagKey($tag)) ?: [];
+                }
+
+                continue;
+            }
+
+            $replies = $this->redis->pipeline(function ($redis) use ($chunk) {
+                foreach ($chunk as $tag) {
+                    $redis->smembers($this->tagKey($tag));
+                }
+            });
+
+            foreach ($replies as $members) {
+                $results[] = is_array($members) ? $members : [];
+            }
+        }
+
+        return $results;
+    }
+
     private function many(array $keys): array
     {
         $records = [];
@@ -184,7 +217,7 @@ private function many(array $keys): array
      */
     private function batch(callable $callback): void
     {
-        if ($this->redis instanceof PhpRedisClusterConnection || $this->redis instanceof PredisClusterConnection) {
+        if ($this->isCluster()) {
             $callback($this->redis);
 
             return;
@@ -193,6 +226,28 @@ private function batch(callable $callback): void
         $this->redis->pipeline($callback);
     }
 
+    /**
+     * Apply a record change and its tag set changes as one MULTI/EXEC, so two
+     * writes to the same URL can't interleave and leave the record listing a
+     * tag whose set no longer contains the URL. Cluster connections can't run
+     * a transaction across slots, so there the commands are sent one by one.
+     */
+    private function atomically(callable $callback): void
+    {
+        if ($this->isCluster()) {
+            $callback($this->redis);
+
+            return;
+        }
+
+        $this->redis->transaction($callback);
+    }
+
+    private function isCluster(): bool
+    {
+        return $this->redis instanceof PhpRedisClusterConnection || $this->redis instanceof PredisClusterConnection;
+    }
+
     private function encode(string $url, array $tags): string
     {
         return json_encode(['url' => $url, 'tags' => array_values($tags)], JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE);

tests/Unit/StoreTest.php

@@ -113,6 +113,23 @@ public function it_matches_wildcards_on_any_prefix(string $driver)
         $this->assertEqualsCanonicalizing(['b', 'c'], array_keys($store->matching(['product-*', 'articles:1'])));
     }
 
+    #[Test]
+    #[DataProvider('drivers')]
+    public function it_matches_a_long_list_of_tags(string $driver)
+    {
+        $store = $this->useDriver($driver);
+
+        for ($i = 0; $i < 50; $i++) {
+            $store->put("url-$i", "/url-$i", ["variant:$i", 'shared']);
+        }
+
+        // more tags than one chunk, most of them unknown
+        $tags = array_map(fn ($i) => "variant:$i", range(0, 2499));
+
+        $this->assertCount(50, $store->matching($tags));
+        $this->assertSame(['url-3', 'url-7'], array_keys($store->matching(['variant:3', 'missing', 'variant:7'])));
+    }
+
     #[Test]
     #[DataProvider('drivers')]
     public function it_imports_without_overwriting_existing_records(string $driver)