Merge branch '3.6.x' into 3.7.x

patchlevel · Dec 29, 2024 · 5cc8133 · 5cc8133
2 parents 8ab3314 + 94a73ae
commit 5cc8133
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 13 deletions.
diff --git a/src/Subscription/Engine/DefaultSubscriptionEngine.php b/src/Subscription/Engine/DefaultSubscriptionEngine.php
@@ -21,7 +21,6 @@
 use Throwable;
 
 use function count;
-use function in_array;
 use function sprintf;
 
 final class DefaultSubscriptionEngine implements SubscriptionEngine
@@ -60,7 +59,7 @@ public function setup(SubscriptionEngineCriteria|null $criteria = null, bool $sk
         );
 
         $this->discoverNewSubscriptions();
-        $this->retrySubscriptions($criteria);
+        $this->retrySubscriptions($criteria, Status::New);
 
         return $this->subscriptionManager->findForUpdate(
             new SubscriptionCriteria(
@@ -169,7 +168,7 @@ public function boot(
             );
 
             $this->discoverNewSubscriptions();
-            $this->retrySubscriptions($criteria);
+            $this->retrySubscriptions($criteria, Status::Booting);
 
             return $this->subscriptionManager->findForUpdate(
                 new SubscriptionCriteria(
@@ -340,7 +339,7 @@ public function run(
 
             $this->discoverNewSubscriptions();
             $this->markDetachedSubscriptions($criteria);
-            $this->retrySubscriptions($criteria);
+            $this->retrySubscriptions($criteria, Status::Active);
 
             return $this->subscriptionManager->findForUpdate(
                 new SubscriptionCriteria(
@@ -901,15 +900,15 @@ function (SubscriptionCollection $subscriptions): void {
         );
     }
 
-    private function retrySubscriptions(SubscriptionEngineCriteria $criteria): void
+    private function retrySubscriptions(SubscriptionEngineCriteria $criteria, Status $previousStatus): void
     {
         $this->subscriptionManager->findForUpdate(
             new SubscriptionCriteria(
                 ids: $criteria->ids,
                 groups: $criteria->groups,
                 status: [Status::Error],
             ),
-            function (SubscriptionCollection $subscriptions): void {
+            function (SubscriptionCollection $subscriptions) use ($previousStatus): void {
                 /** @var Subscription $subscription */
                 foreach ($subscriptions as $subscription) {
                     $error = $subscription->subscriptionError();
@@ -918,13 +917,7 @@ function (SubscriptionCollection $subscriptions): void {
                         continue;
                     }
 
-                    $retryable = in_array(
-                        $error->previousStatus,
-                        [Status::New, Status::Booting, Status::Active],
-                        true,
-                    );
-
-                    if (!$retryable) {
+                    if ($error->previousStatus !== $previousStatus) {
                         continue;
                     }
 

diff --git a/tests/Unit/Subscription/Engine/DefaultSubscriptionEngineTest.php b/tests/Unit/Subscription/Engine/DefaultSubscriptionEngineTest.php
@@ -3050,6 +3050,58 @@ public function subscribe(): void
         self::assertEquals(1, $update2->retryAttempt());
     }
 
+    #[DataProvider('statusProvider')]
+    public function testShouldNotRetryOtherStatus(string $method, string $status): void
+    {
+        $subscriptionId = 'test';
+        $subscriber = new #[Subscriber('test', RunMode::FromBeginning)]
+        class {
+        };
+
+        $streamableStore = $this->prophesize(Store::class);
+
+        $subscription = new Subscription(
+            $subscriptionId,
+            Subscription::DEFAULT_GROUP,
+            RunMode::FromBeginning,
+            Status::Error,
+            0,
+            new SubscriptionError('ERROR', Status::from($status)),
+        );
+
+        $subscriptionStore = new DummySubscriptionStore([$subscription]);
+
+        $retryStrategy = $this->prophesize(RetryStrategy::class);
+        $retryStrategy->shouldRetry($subscription)->shouldNotBeCalled();
+
+        $engine = new DefaultSubscriptionEngine(
+            $streamableStore->reveal(),
+            $subscriptionStore,
+            new MetadataSubscriberAccessorRepository([$subscriber]),
+            $retryStrategy->reveal(),
+            new NullLogger(),
+        );
+
+        $result = match ($method) {
+            'setup' => $engine->setup(),
+            'boot' => $engine->boot(),
+            'run' => $engine->run(),
+        };
+
+        self::assertCount(0, $result->errors);
+        self::assertEquals([], $subscriptionStore->updatedSubscriptions);
+    }
+
+    public static function statusProvider(): Generator
+    {
+        yield 'setup_booting' => ['setup', 'booting'];
+        yield 'setup_active' => ['setup', 'active'];
+        yield 'boot_new' => ['boot', 'new'];
+        yield 'boot_active' => ['boot', 'active'];
+        yield 'run_new' => ['run', 'new'];
+        yield 'run_booting' => ['run', 'booting'];
+    }
+
     public function testShouldNotRetry(): void
     {
         $subscriptionId = 'test';