Message failure handling middleware pipeline (yiisoft#64)

* Initial implementation

* Apply fixes from StyleCI

* General implementation

* Make internal method private

* Fix tests

* remove redundant use

* The first test

* Start simple strategy tests with data provider

* Create SendAgainStrategy

* Upgrade PayloadFactory

* Upgrade existing strategy test

* Rename BehaviorRemovingStrategy

* Apply fixes from StyleCI

* Create queue sending test draft

* More strategy tests

* Divide different tests into different classes

* Remove redundant tests

* Apply fixes from StyleCI

* Remove unused uses

* Fix dispatcher creating

* Create FailedJobsHandler instead of Queue::jobRetry()

* Add typehint to property

* Make DispatcherFactory final

* Add PhpDoc with type to a property

* Disallow empty default pipeline in DispatcherFactory

* Apply fixes from StyleCI

* Allow zero initial delay

* Add integrational failure strategy test

* Divide namespaces

* Add dispatcher tests

* Improve strategy tests

* Apply fixes from StyleCI

* Add BehaviorAddingStrategy

* Apply fixes from StyleCI

* Actualizing

* Apply fixes from StyleCI

* Actualizing

* Default configuration

* Pipeline fix

* Apply fixes from StyleCI

* FailureStrategyFactory fix

* FailureStrategyFactory fix

* Bugfixes and tests

* Fix more tests

* Apply fixes from StyleCI

* Fix types in a test

* Cosmetics

* Cosmetics

* Remove redundant test

* One more test

* Apply fixes from StyleCI

* Cosmetics

* Test coverage increasing and bugfixes

* Docs and bugfixes

* Apply fixes from StyleCI

* Exception message dots

* PhpDoc

* Apply fixes from StyleCI

* Fix PhpDoc

* Apply fixes from StyleCI

* Convert Fail Strategies into middlewares

* Convert into middlewares, iteration 1

* Apply fixes from StyleCI

* Fix tests and naming

* Apply fixes from StyleCI

* Bugfix

* Apply fixes from StyleCI

* Style fix

* More tests and bugfixes

Co-authored-by: StyleCI Bot <[email protected]>

* Docs, configs and bugfixes

* Apply fixes from StyleCI

* Improve exception messages

* Bugfix

* Apply docs suggestions from code review

Co-authored-by: Alexander Makarov <[email protected]>

* Fix PhpDocs

Co-authored-by: Alexander Makarov <[email protected]>
Co-authored-by: StyleCI Bot <[email protected]>
Co-authored-by: Alexander Makarov <[email protected]>

Author: 3 people

README.md

@@ -215,8 +215,8 @@ twice for a queue message. That means you can add extra functionality on message
 of the two classes: `PushMiddlewareDispatcher` and `ConsumeMiddlewareDispatcher` respectively.
 
 You can use any of these formats to define a middleware:
-- A ready-to-use middleware object: `new FooMiddleware()`. It must implement either `MiddlewarePushInterface`,
-    or `MiddlewareConsumeInterface` depending on the place you use it.
+- A ready-to-use middleware object: `new FooMiddleware()`. It must implement `MiddlewarePushInterface`,
+ `MiddlewareConsumeInterface` or `MiddlewareFailureInterface` depending on the place you use it.
 - An array in the format of [yiisoft/definitions](https://github.com/yiisoft/definitions).
     **Only if you use yiisoft/definitions and yiisoft/di**.
 - A `callable`: `fn() => // do stuff`, `$object->foo(...)`, etc. It will be executed through the
@@ -256,7 +256,9 @@ in the `PushRequest` object. You will get a `AdapterNotConfiguredException`, if
 You have three places to define push middlewares:
 
 1. `PushMiddlewareDispatcher`. You can pass it either to the constructor, or to the `withMiddlewares()` method, which  
-creates a completely new dispatcher object with only those middlewares, which are passed as arguments.
+creates a completely new dispatcher object with only those middlewares, which are passed as arguments. 
+If you use [yiisoft/config](yiisoft/config), you can add middleware to the `middlewares-push` key of the 
+`yiisoft/yii-queue` array in the `params`.
 2. Pass middlewares to either `Queue::withMiddlewares()` or `Queue::withMiddlewaresAdded()` methods. The difference is 
 that the former will completely replace an existing middleware stack, while the latter will add passed middlewares to 
 the end of the existing stack. These middlewares will be executed after the common ones, passed directly to the 
@@ -269,7 +271,17 @@ along with them.
 
 ### Consume pipeline
 
-You can set a middleware pipeline for a message when it will be consumed from a queue server. This is useful to collect metrics, modify message data, etc. In pair with a Push middleware you can deduplicate messages in the queue, calculate time from push to consume, handle errors (push to a queue again, redirect failed message to another queue, send a notification, etc.). Unless Push pipeline, you have only one place to define the middleware stack: in the `ConsumeMiddlewareDispatcher`, either in the constructor, or in the `withMiddlewares()` method.
+You can set a middleware pipeline for a message when it will be consumed from a queue server. This is useful to collect metrics, modify message data, etc. In pair with a Push middleware you can deduplicate messages in the queue, calculate time from push to consume, handle errors (push to a queue again, redirect failed message to another queue, send a notification, etc.). Unless Push pipeline, you have only one place to define the middleware stack: in the `ConsumeMiddlewareDispatcher`, either in the constructor, or in the `withMiddlewares()` method. If you use [yiisoft/config](yiisoft/config), you can add middleware to the `middlewares-consume` key of the `yiisoft/yii-queue` array in the `params`.
+
+### Error handling pipeline
+
+Often when some job is failing, we want to retry its execution a couple more times or redirect it to another queue channel. This can be done in `yiisoft/yii-queue` with Failure middleware pipeline. They are triggered each time message processing via the Consume middleware pipeline is interrupted with any `Throwable`. The key differences from the previous two pipelines:
+- You should set up the middleware pipeline separately for each queue channel. That means, the format should be `['channel-name' => [FooMiddleware::class]]` instead of `[FooMiddleware::class]`, like for the other two pipelines. There is also a default key, which will be used for those channels without their own one: `FailureMiddlewareDispatcher::DEFAULT_PIPELINE`.
+- The last middleware will throw the exception, which will come with the `FailureHandlingRequest` object. If you don't want the exception to be thrown, your middlewares should `return` a request without calling `$handler->handleFailure()`.
+
+You can declare error handling middleware pipeline in the `FailureMiddlewareDispatcher`, either in the constructor, or in the `withMiddlewares()` method. If you use [yiisoft/config](yiisoft/config), you can add middleware to the `middlewares-fail` key of the `yiisoft/yii-queue` array in the `params`.
+
+See [error handling docs](docs/guide/error-handling.md) for details.
 
 ## Extra
 

config/common.php

@@ -6,10 +6,15 @@
 use Yiisoft\Yii\Queue\Cli\LoopInterface;
 use Yiisoft\Yii\Queue\Cli\SignalLoop;
 use Yiisoft\Yii\Queue\Cli\SimpleLoop;
+use Yiisoft\Yii\Queue\Middleware\Consume\ConsumeMiddlewareDispatcher;
 use Yiisoft\Yii\Queue\Middleware\Consume\MiddlewareFactoryConsume;
 use Yiisoft\Yii\Queue\Middleware\Consume\MiddlewareFactoryConsumeInterface;
+use Yiisoft\Yii\Queue\Middleware\FailureHandling\FailureMiddlewareDispatcher;
+use Yiisoft\Yii\Queue\Middleware\FailureHandling\MiddlewareFactoryFailure;
+use Yiisoft\Yii\Queue\Middleware\FailureHandling\MiddlewareFactoryFailureInterface;
 use Yiisoft\Yii\Queue\Middleware\Push\MiddlewareFactoryPush;
 use Yiisoft\Yii\Queue\Middleware\Push\MiddlewareFactoryPushInterface;
+use Yiisoft\Yii\Queue\Middleware\Push\PushMiddlewareDispatcher;
 use Yiisoft\Yii\Queue\Queue;
 use Yiisoft\Yii\Queue\QueueFactory;
 use Yiisoft\Yii\Queue\QueueFactoryInterface;
@@ -37,4 +42,14 @@
     QueueInterface::class => Queue::class,
     MiddlewareFactoryPushInterface::class => MiddlewareFactoryPush::class,
     MiddlewareFactoryConsumeInterface::class => MiddlewareFactoryConsume::class,
+    MiddlewareFactoryFailureInterface::class => MiddlewareFactoryFailure::class,
+    PushMiddlewareDispatcher::class => [
+        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/yii-queue']['middlewares-push']],
+    ],
+    ConsumeMiddlewareDispatcher::class => [
+        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/yii-queue']['middlewares-consume']],
+    ],
+    FailureMiddlewareDispatcher::class => [
+        '__construct()' => ['middlewareDefinitions' => $params['yiisoft/yii-queue']['middlewares-fail']],
+    ],
 ];

config/events-console.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+use Yiisoft\Yii\Queue\Event\JobFailure;
+use Yiisoft\Yii\Queue\FailureStrategy\FailedJobsHandler;
+
+return [
+    JobFailure::class => [
+        [FailedJobsHandler::class, 'handle'],
+    ],
+];

config/params.php

@@ -15,5 +15,8 @@
     'yiisoft/yii-queue' => [
         'handlers' => [],
         'channel-definitions' => [],
+        'middlewares-push' => [],
+        'middlewares-consume' => [],
+        'middlewares-fail' => [],
     ],
 ];

docs/guide/README.md

@@ -6,6 +6,6 @@ An extension for running tasks asynchronously via queues.
 
 * [Usage basics](usage.md)
 * [Migrating from `yii2-queue`](migrating-from-yii2-queue.md)
-* [Errors and retryable jobs](retryable.md)
+* [Errors and retryable jobs](error-handling.md)
 * [Workers](worker.md)
 * [Adapter list](adapter-list.md)

docs/guide/error-handling.md

@@ -0,0 +1,90 @@
+# Error handling on message processing
+
+Often when some message handling is failing, we want to retry its execution a couple more times or redirect it to another queue channel. This can be done in `yiisoft/yii-queue` with _Failure Handling Middleware Pipeline_. It is triggered each time message processing via Consume Middleware Pipeline is interrupted with any `Throwable`. 
+
+## Configuration
+
+Here below is configuration via `yiisoft/config`. If you don't use it - you should add middleware definition list (in the `middlewares-fail` key here) to the `FailureMiddlewareDispatcher` by your own.
+
+Configuration should be passed to the `yiisoft/yii-queue.fail-strategy-pipelines` key of the `params` config to work with the `yiisoft/config`. You can define different failure handling pipelines for each queue channel. Let's see and describe an example:
+
+```php
+'yiisoft/yii-queue' => [
+    'middlewares-fail' => [
+        FailureMiddlewareDispatcher::DEFAULT_PIPELINE => [
+            [
+                'class' => SendAgainMiddleware::class,
+                '__construct()' => ['id' => 'default-first-resend', 'queue' => null], 
+            ],
+            static fn (QueueFactoryInterface $factory) => new SendAgainMiddleware(
+                id: 'default-second-resend', 
+                queue: $factory->get('failed-messages'),
+            ),
+        ],
+        
+        'failed-messages' => [
+            [
+                'class' => ExponentialDelayMiddleware::class,
+                '__construct()' => [
+                    'id' => 'failed-messages',
+                    'maxAttempts' => 30,
+                    'delayInitial' => 5,
+                    'delayMaximum' => 60,
+                    'exponent' => 1.5,
+                    'queue' => null,
+                ], 
+            ],
+        ],
+    ],
+]
+```
+
+Keys here except `FailureMiddlewareDispatcher::DEFAULT_PIPELINE` are queue channel names, and values are lists of `FailureMiddlewareInterface` definitions. `FailureMiddlewareDispatcher::DEFAULT_PIPELINE` defines a default pipeline to apply to channels without explicitly defined failure strategy pipeline. Each middleware definition must be one of:
+- A ready-to-use `MiddlewareFailureInterface` object like `new FooMiddleware()`.
+- A valid definition for the [yiisoft/definitions](https://github.com/yiisoft/definitions). It must describe an object, implementing the `MiddlewareFailureInterface`.
+- A callable: `fn() => // do stuff`, `$object->foo(...)`, etc. It will be executed through the `yiisoft/injector`, so all the dependencies of your callable will be resolved. You can also define a "callable-looking" array, where object will be instantiated with a DI container: `[FooMiddleware::class, 'handle']`.
+- A string for your DI container to resolve the middleware, e.g. `FooMiddleware::class`.
+
+In the example above failures will be handled this way (look the concrete middleware description below):
+
+1. For the first time message will be resent to the same queue channel immediately.
+2. If it fails again, it will be resent to the queue channel named `failed-messages`.
+3. From now on it will be resent to the same queue channel (`failed-messages`) up to 30 times with a delay from 5 to 60 seconds, increased 1.5 times each time the message fails again.
+4. If the message handler throw an exception one more (33rd) time, the exception will not be caught.
+
+Failures of messages, which are initially sent to the `failed-messages` channel, will only be handled by the 3rd and the 4th points of this list.
+
+## Default failure handling strategies
+
+Let's see the built-in defaults.
+
+### SendAgainMiddleware
+
+This strategy simply resends the given message to a queue. Let's see the constructor parameters through which it's configured:
+
+- `id` - A unique string. Allows to use this strategy more than once for the same message, just like in example above.
+- `maxAttempts` - Maximum attempts count for this strategy with the given $id before it will give up.
+- `queue` - The strategy will send the message to the given queue when it's not `null`. That means you can use this strategy to push a message not to the same queue channel it came from. When the `queue` parameter is set to `null`, a message will be sent to the same channel it came from.
+
+### ExponentialDelayMiddleware
+
+This strategy does the same thing as the `SendAgainMiddleware` with a single difference: it resends a message with an exponentially increasing delay. The delay **must** be implemented by the used `AdapterInterface` implementation.
+
+It's configured via constructor parameters, too. Here they are:
+
+- `id` - A unique string allows to use this strategy more than once for the same message, just like in example above.
+- `maxAttempts` - Maximum attempts count for this strategy with the given $id before it will give up.
+- `delayInitial` - The initial delay that will be applied to a message for the first time. It must be a positive float. 
+- `delayMaximum` - The maximum delay which can be applied to a single message. Must be above the `delayInitial`.
+- `exponent` - Message handling delay will be muliplied by exponent each time it fails.
+- `queue` - The strategy will send the message to the given queue when it's not `null`. That means you can use this strategy to push a message not to the same queue channel it came from. When the `queue` parameter is set to `null`, a message will be sent to the same channel it came from.
+
+## How to create a custom Failure Middleware?
+
+All you need is to implement the `MiddlewareFailureInterface` and add your implementation definition to the [configuration](#configuration).
+This interface has the only method `handle`. And the method has these parameters:
+- `ConsumeRequest $request` - a request for a message handling. It consists of a message and a queue the message came from.
+- `Throwable $exception` - an exception thrown on the `request` handling
+- `MessageFailureHandlerInterface $handler` - failure strategy pipeline continuation. Your Middleware should call `$pipeline->handle()` when it shouldn't interrupt failure pipeline execution.
+
+> Note: your strategy have to check by its own if it should be applied. Look into [`SendAgainMiddleware::suites()`](../../src/Middleware/Implementation/FailureMiddleware/Middleware/SendAgainMiddleware.php#L52) for an example.

docs/guide/retryable.md

@@ -6,24 +6,25 @@
 
 final class Message implements MessageInterface
 {
-    private ?string $id = null;
-
+    /**
+     * @param string $handlerName
+     * @param mixed $data Message data, encodable by a queue adapter
+     * @param array $metadata Message metadata, encodable by a queue adapter
+     * @param string|null $id Message id
+     */
     public function __construct(
         private string $handlerName,
         private mixed $data,
+        private array $metadata = [],
+        private ?string $id = null
     ) {
-        $this->handlerName = $handlerName;
-        $this->data = $data;
     }
 
     public function getHandlerName(): string
     {
         return $this->handlerName;
     }
 
-    /**
-     * @return mixed
-     */
     public function getData(): mixed
     {
         return $this->data;
@@ -38,4 +39,9 @@ public function getId(): ?string
     {
         return $this->id;
     }
+
+    public function getMetadata(): array
+    {
+        return $this->metadata;
+    }
 }

src/Message/Message.php

@@ -27,5 +27,12 @@ public function getHandlerName(): string;
      *
      * @return mixed
      */
-    public function getData();
+    public function getData(): mixed;
+
+    /**
+     * Returns message metadata: timings, attempts count, metrics, etc.
+     *
+     * @return array
+     */
+    public function getMetadata(): array;
 }

src/Message/MessageInterface.php

@@ -9,7 +9,7 @@
 /**
  * @internal
  */
-final class ConsumeHandler implements MessageHandlerConsumeInterface
+final class ConsumeFinalHandler implements MessageHandlerConsumeInterface
 {
     public function __construct(private Closure $handler)
     {