add failed status in subscription engine

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -7,6 +7,7 @@
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\ClockBasedRetryStrategy;
+use Patchlevel\EventSourcing\Subscription\RetryStrategy\ConditionalRetryStrategy;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\RetryStrategy;
 use Patchlevel\EventSourcing\Subscription\RunMode;
 use Patchlevel\EventSourcing\Subscription\Status;
@@ -667,6 +668,7 @@ public function reactivate(SubscriptionEngineCriteria|null $criteria = null): Re
                 groups: $criteria->groups,
                 status: [
                     Status::Error,
+                    Status::Failed,
                     Status::Detached,
                     Status::Paused,
                     Status::Finished,
@@ -984,7 +986,12 @@ private function discoverNewSubscriptions(): void
 
     private function handleError(Subscription $subscription, Throwable $throwable): void
     {
-        $subscription->error($throwable);
+        if ($this->retryStrategy instanceof ConditionalRetryStrategy && !$this->retryStrategy->canRetry($subscription)) {
+            $subscription->failed($throwable);
+        } else {
+            $subscription->error($throwable);
+        }
+
         $this->subscriptionManager->update($subscription);
 
         if (!isset($this->batching[$subscription->id()])) {

src/Subscription/RetryStrategy/ClockBasedRetryStrategy.php

@@ -12,7 +12,7 @@
 use function round;
 use function sprintf;
 
-final class ClockBasedRetryStrategy implements RetryStrategy
+final class ClockBasedRetryStrategy implements ConditionalRetryStrategy
 {
     public const DEFAULT_BASE_DELAY = 5;
     public const DEFAULT_DELAY_FACTOR = 2;
@@ -30,9 +30,14 @@ public function __construct(
     ) {
     }
 
+    public function canRetry(Subscription $subscription): bool
+    {
+        return $subscription->retryAttempt() < $this->maxAttempts;
+    }
+
     public function shouldRetry(Subscription $subscription): bool
     {
-        if ($subscription->retryAttempt() >= $this->maxAttempts) {
+        if ($this->canRetry($subscription) === false) {
             return false;
         }
 

src/Subscription/RetryStrategy/NoRetryStrategy.php

@@ -6,10 +6,15 @@
 
 use Patchlevel\EventSourcing\Subscription\Subscription;
 
-final class NoRetryStrategy implements RetryStrategy
+final class NoRetryStrategy implements ConditionalRetryStrategy
 {
     public function shouldRetry(Subscription $subscription): bool
     {
         return false;
     }
+
+    public function canRetry(Subscription $subscription): bool
+    {
+        return false;
+    }
 }

src/Subscription/Status.php

@@ -13,4 +13,5 @@ enum Status: string
     case Finished = 'finished';
     case Detached = 'detached';
     case Error = 'error';
+    case Failed = 'failed';
 }

src/Subscription/Subscription.php

@@ -141,6 +141,25 @@ public function isError(): bool
         return $this->status === Status::Error;
     }
 
+    public function failed(Throwable|string $error): void
+    {
+        $previousStatus = $this->status;
+        $this->status = Status::Failed;
+
+        if ($error instanceof Throwable) {
+            $this->error = SubscriptionError::fromThrowable($previousStatus, $error);
+
+            return;
+        }
+
+        $this->error = new SubscriptionError($error, $previousStatus);
+    }
+
+    public function isFailed(): bool
+    {
+        return $this->status === Status::Failed;
+    }
+
     public function retryAttempt(): int
     {
         return $this->retryAttempt;

tests/Integration/Subscription/SubscriptionTest.php

@@ -252,6 +252,8 @@ public function testErrorHandling(): void
 
         $subscriber->subscribeError = true;
 
+        // first run, error
+
         $result = $engine->run();
 
         self::assertEquals(1, $result->processedMessages);
@@ -269,6 +271,8 @@ public function testErrorHandling(): void
         self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
         self::assertEquals(0, $subscription->retryAttempt());
 
+        // second run, time has not passed yet, no retry, no error
+
         $result = $engine->run();
 
         self::assertEquals(0, $result->processedMessages);
@@ -281,8 +285,9 @@ public function testErrorHandling(): void
         self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
         self::assertEquals(0, $subscription->retryAttempt());
 
-        $clock->sleep(5);
+        // third run, time has passed, 1. retry, error again
 
+        $clock->sleep(5);
         $result = $engine->run();
 
         self::assertEquals(1, $result->processedMessages);
@@ -300,8 +305,9 @@ public function testErrorHandling(): void
         self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
         self::assertEquals(1, $subscription->retryAttempt());
 
-        $clock->sleep(10);
+        // fourth run, time has passed, 2. retry, max retries reached, failed
 
+        $clock->sleep(10);
         $result = $engine->run();
 
         self::assertEquals(1, $result->processedMessages);
@@ -314,11 +320,28 @@ public function testErrorHandling(): void
 
         $subscription = self::findSubscription($engine->subscriptions(), 'error_producer');
 
-        self::assertEquals(Status::Error, $subscription->status());
+        self::assertEquals(Status::Failed, $subscription->status());
+        self::assertEquals('subscribe error', $subscription->subscriptionError()?->errorMessage);
+        self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
+        self::assertEquals(2, $subscription->retryAttempt());
+
+        // fifth run, time has passed, skip failed subscription
+
+        $clock->sleep(20);
+        $result = $engine->run();
+
+        self::assertEquals(0, $result->processedMessages);
+        self::assertEquals([], $result->errors);
+
+        $subscription = self::findSubscription($engine->subscriptions(), 'error_producer');
+
+        self::assertEquals(Status::Failed, $subscription->status());
         self::assertEquals('subscribe error', $subscription->subscriptionError()?->errorMessage);
         self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
         self::assertEquals(2, $subscription->retryAttempt());
 
+        // reactivated subscription
+
         $engine->reactivate(new SubscriptionEngineCriteria(
             ids: ['error_producer'],
         ));
@@ -329,6 +352,8 @@ public function testErrorHandling(): void
         self::assertEquals(null, $subscription->subscriptionError());
         self::assertEquals(0, $subscription->retryAttempt());
 
+        // sixth run, error again
+
         $result = $engine->run();
 
         self::assertEquals(1, $result->processedMessages);
@@ -346,6 +371,8 @@ public function testErrorHandling(): void
         self::assertEquals(Status::Active, $subscription->subscriptionError()?->previousStatus);
         self::assertEquals(0, $subscription->retryAttempt());
 
+        // seventh run, time has passed, error fixed, 1. retry, no error
+
         $clock->sleep(5);
         $subscriber->subscribeError = false;