Remove expired members from subscription topic stored in Redis set map

src/Subscriptions/Storage/RedisStorageManager.php

@@ -65,6 +65,9 @@ public function subscribersByTopic(string $topic): Collection
             return new Collection();
         }
 
+        // Store all keys as missing keys to remove the ones which are expired later.
+        $missingKeys = $subscriberIds;
+
         // Since we store the individual subscribers with a prefix,
         // but not in the set, we have to add the prefix here.
         $subscriberIds = array_map([$this, 'channelKey'], $subscriberIds);
@@ -73,22 +76,35 @@ public function subscribersByTopic(string $topic): Collection
         // This is like using multiple get calls (getSubscriber uses the get command).
         $subscribers = $this->connection->command('mget', [$subscriberIds]);
 
-        return (new Collection($subscribers))
+        $subscribersCollection = (new Collection($subscribers))
             ->filter()
-            ->map(static function (?string $subscriber): ?Subscriber {
-                // Some entries may be expired
+            ->map(static function (?string $subscriber) use (&$missingKeys): ?Subscriber {
+                // Some entries may be expired.
                 if ($subscriber === null) {
                     return null;
                 }
 
                 // Other entries may contain invalid values
                 try {
-                    return unserialize($subscriber);
+                    $subscriber = unserialize($subscriber);
+
+                    // This key exists so remove it from the list of missing keys.
+                    $missingKeys = array_diff($missingKeys, [$subscriber->channel]);
+
+                    return $subscriber;
                 } catch (\ErrorException) {
                     return null;
                 }
             })
             ->filter();
+
+        // Remove expired subscribers from the set of subscribers of this topic.
+        $this->connection->command('srem', [
+            $this->topicKey($topic),
+            ...$missingKeys,
+        ]);
+
+        return $subscribersCollection;
     }
 
     public function storeSubscriber(Subscriber $subscriber, string $topic): void

tests/Unit/Subscriptions/Storage/RedisStorageManagerTest.php

@@ -195,18 +195,20 @@ public function testSubscribersByTopic(): void
             $subscriber2,
         ];
 
-        $redisConnection->expects($this->exactly(2))
+        $redisConnection->expects($this->exactly(3))
             ->method('command')
             ->with(...$this->withConsecutive(
                 ['smembers', ["graphql.topic.{$topic}"]],
                 ['mget', [[
                     'graphql.subscriber.foo1',
                     'graphql.subscriber.foo2',
                     'graphql.subscriber.foo3',
+                    'graphql.subscriber.foo4',
                 ]]],
+                ['srem', ["graphql.topic.{$topic}", 'foo3', 'foo4']],
             ))
             ->willReturnOnConsecutiveCalls(
-                ['foo1', 'foo2', 'foo3'],
+                ['foo1', 'foo2', 'foo3', 'foo4'],
                 [
                     serialize($subscriber1),
                     serialize($subscriber2),
@@ -217,6 +219,7 @@ public function testSubscribersByTopic(): void
                     // mget non-existing-entry
                     false,
                 ],
+                null,
             );
 
         $this->assertEquals(