diff --git a/Makefile b/Makefile
index 8bb475e..009c6e5 100644
--- a/Makefile
+++ b/Makefile
@@ -84,7 +84,7 @@ publisher:
 
 consumer:
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w /tests/simulation cli \
-		php worker.php --schema=high
+		php worker-with-catch.php --schema=high
 
 bench: ## bench
 	USER=$(USER) docker compose -f ./docker-compose.yml run --rm -u $(USER) -w / cli \
diff --git a/README.md b/README.md
index f11664a..18aeb13 100644
--- a/README.md
+++ b/README.md
@@ -51,6 +51,82 @@ composer require kuaukutsu/poc-queue-stream
 - https://redis.io/docs/latest/develop/data-types/streams/
 - https://habr.com/ru/articles/456270/
 
+### example
+
+`tests/simulation`
+
+#### common
+
+```php
+// необходим для внедрения зависимостей в объект задачи
+$container = new Container();
+$builder = (new Builder(new FactoryProxy($container)))
+    ->withConfig(
+        RedisConfig::fromUri('tcp://redis:6379')
+    );
+    
+// например из аргументов получаем схему - название очереди
+$schema = QueueSchemaStub::from((string)argument('schema', 'low'));
+```
+
+#### publish
+
+```php
+// где-то в своём проекте (app), через Interface и Decorator заводим сервис PublisherInterface
+$publisher = $builder->buildPublisher();
+
+// создание отложенной задачи
+$task = new QueueTask(
+    target: QueueHandlerStub::class,
+    arguments: [
+        'id' => 1,
+        'name' => 'test name',
+    ],
+);
+
+$publisher->push($schema, $task);
+```
+
+Или массовая публикация
+```php
+$batch = [];
+foreach (range(1, 100) as $item) {
+    $batch[] = new QueueTask(
+        target: QueueHandlerStub::class,
+        arguments: [
+            'id' => $item,
+            'name' => 'test batch',
+        ],
+    );
+}
+
+$publisher->pushBatch($schema, $batch);
+```
+
+#### consume
+
+consumer.php
+```php
+$consumer = $builder
+    // задаём обработчик ошибок, если необходимо и не устраивает глобальный TryCatchInterceptor
+    ->withCatch(
+        static function (?string $message, Throwable $exception): void {
+            echo sprintf("data: %s\nerror: %s", $message, $exception->getMessage());
+        }
+    )
+    ->withInterceptors(
+        new ArgumentsVerifyInterceptor(),
+        new ExactlyOnceInterceptor(createRedisClient('redis://redis:6379')),
+        new TryCatchInterceptor(),
+    )
+    ->buildConsumer();
+    
+$consumer->consume($schema);
+trapSignal([SIGTERM, SIGINT]);
+$consumer->disconnect();
+exit(0);
+```
+
 ### Benchmark
 
 ```
@@ -64,21 +140,21 @@ with PHP version 8.3.17, xdebug ✔, opcache ✔
 
 \kuaukutsu\poc\queue\stream\benchmarks\PublisherRedisBench
 
-    benchAsWhile............................I4 - Mo32.201ms (±5.32%)
-    benchAsBatch............................I4 - Mo29.561ms (±1.30%)
+    benchAsWhile............................I4 - Mo29.287ms (±15.55%)
+    benchAsBatch............................I4 - Mo17.785ms (±2.50%)
 
 \kuaukutsu\poc\queue\stream\benchmarks\PublisherValkeyBench
 
-    benchAsWhile............................I4 - Mo31.663ms (±2.65%)
-    benchAsBatch............................I4 - Mo30.349ms (±3.98%)
+    benchAsWhile............................I4 - Mo30.377ms (±8.72%)
+    benchAsBatch............................I4 - Mo17.267ms (±3.16%)
 
 Subjects: 4, Assertions: 0, Failures: 0, Errors: 0
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
-| benchmark            | subject      | set | revs | its | mem_peak | mode     | rstdev |
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
-| PublisherRedisBench  | benchAsWhile |     | 10   | 5   | 1.983mb  | 32.201ms | ±5.32% |
-| PublisherRedisBench  | benchAsBatch |     | 10   | 5   | 2.044mb  | 29.561ms | ±1.30% |
-| PublisherValkeyBench | benchAsWhile |     | 10   | 5   | 1.983mb  | 31.663ms | ±2.65% |
-| PublisherValkeyBench | benchAsBatch |     | 10   | 5   | 2.037mb  | 30.349ms | ±3.98% |
-+----------------------+--------------+-----+------+-----+----------+----------+--------+
++----------------------+--------------+-----+------+-----+----------+----------+---------+
+| benchmark            | subject      | set | revs | its | mem_peak | mode     | rstdev  |
++----------------------+--------------+-----+------+-----+----------+----------+---------+
+| PublisherRedisBench  | benchAsWhile |     | 10   | 5   | 2.053mb  | 29.287ms | ±15.55% |
+| PublisherRedisBench  | benchAsBatch |     | 10   | 5   | 4.589mb  | 17.785ms | ±2.50%  |
+| PublisherValkeyBench | benchAsWhile |     | 10   | 5   | 2.046mb  | 30.377ms | ±8.72%  |
+| PublisherValkeyBench | benchAsBatch |     | 10   | 5   | 4.589mb  | 17.267ms | ±3.16%  |
++----------------------+--------------+-----+------+-----+----------+----------+---------+
 ```
diff --git a/composer.json b/composer.json
index cdddc00..14d410b 100644
--- a/composer.json
+++ b/composer.json
@@ -18,7 +18,7 @@
     "php": "^8.3",
     "ext-redis": "*",
     "amphp/redis": "^2.0",
-    "kuaukutsu/queue-core": "^0.4"
+    "kuaukutsu/queue-core": "^0.5.2"
   },
   "require-dev": {
     "ext-pcntl": "*",
diff --git a/phpstan.neon b/phpstan.neon
index 8c6d17f..c7d77a7 100644
--- a/phpstan.neon
+++ b/phpstan.neon
@@ -6,4 +6,3 @@ parameters:
 	ignoreErrors:
 		- identifiers:
 		  - missingType.iterableValue
-		  - missingType.generics
diff --git a/src/Builder.php b/src/Builder.php
index 19e385c..f44264d 100644
--- a/src/Builder.php
+++ b/src/Builder.php
@@ -4,6 +4,7 @@
 
 namespace kuaukutsu\poc\queue\stream;
 
+use Closure;
 use Override;
 use Amp\Redis\RedisConfig;
 use Amp\Redis\RedisException;
@@ -12,10 +13,6 @@
 use kuaukutsu\queue\core\handler\Pipeline;
 use kuaukutsu\queue\core\interceptor\InterceptorInterface;
 use kuaukutsu\queue\core\BuilderInterface;
-use kuaukutsu\queue\core\SchemaInterface;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 
 use function Amp\Redis\createRedisClient;
 
@@ -30,6 +27,8 @@ final class Builder implements BuilderInterface
 
     private HandlerInterface $handler;
 
+    private ?Closure $catch = null;
+
     /**
      * @throws RedisException
      */
@@ -60,6 +59,14 @@ public function withStreamOptions(StreamOptions $options): self
         return $clone;
     }
 
+    #[Override]
+    public function withCatch(Closure $catch): BuilderInterface
+    {
+        $clone = clone $this;
+        $clone->catch = $catch;
+        return $clone;
+    }
+
     #[Override]
     public function withInterceptors(InterceptorInterface ...$interceptor): self
     {
@@ -75,15 +82,8 @@ public function buildPublisher(): Publisher
     }
 
     #[Override]
-    public function buildConsumer(SchemaInterface $schema): Consumer
+    public function buildConsumer(): Consumer
     {
-        $client = createRedisClient($this->config);
-
-        return new Consumer(
-            new RedisStream($client, $schema),
-            new RedisStreamGroup($client, $schema, $this->options),
-            new RedisString($client),
-            $this->handler,
-        );
+        return new Consumer(createRedisClient($this->config), $this->options, $this->handler, $this->catch);
     }
 }
diff --git a/src/Consumer.php b/src/Consumer.php
index eaa4455..5a7008d 100644
--- a/src/Consumer.php
+++ b/src/Consumer.php
@@ -4,66 +4,64 @@
 
 namespace kuaukutsu\poc\queue\stream;
 
+use Closure;
 use Override;
-use Throwable;
+use Amp\Redis\RedisClient;
 use Revolt\EventLoop;
 use kuaukutsu\queue\core\exception\QueueConsumeException;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\ConsumerInterface;
+use kuaukutsu\queue\core\SchemaInterface;
 use kuaukutsu\poc\queue\stream\internal\Context;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
-use kuaukutsu\poc\queue\stream\internal\workflow\TaskRunner;
+use kuaukutsu\poc\queue\stream\internal\workflow\TaskHandler;
+use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowCatch;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowClaim;
 use kuaukutsu\poc\queue\stream\internal\workflow\WorkflowMain;
 
 /**
  * @api
  */
-final readonly class Consumer implements ConsumerInterface
+final class Consumer implements ConsumerInterface
 {
-    private Context $ctx;
+    private ?Context $ctx = null;
 
-    private TaskRunner $runner;
+    private readonly TaskHandler $handler;
 
     public function __construct(
-        RedisStream $stream,
-        private RedisStreamGroup $streamGroup,
-        RedisString $string,
+        private readonly RedisClient $redis,
+        private readonly StreamOptions $options,
         HandlerInterface $handler,
+        ?Closure $catch = null,
     ) {
-        $this->ctx = new Context($string, $this->streamGroup);
-        $this->runner = new TaskRunner($string, $stream, $this->streamGroup, $handler);
+        $this->handler = new TaskHandler($handler, $catch);
     }
 
     /**
-     * @param ?callable(string, Throwable):void $catch
      * @throws QueueConsumeException
      */
     #[Override]
-    public function consume(?callable $catch = null): void
+    public function consume(SchemaInterface $schema): void
     {
-        $this->streamGroup->create();
+        $stream = new RedisStreamGroup($this->redis, $this->options, $schema);
+        $stream->create();
 
-        $worflow = new WorkflowMain(
-            $this->runner,
-            $this->streamGroup,
-        );
+        $this->ctx = new Context($schema, $stream, new RedisString($this->redis));
 
         EventLoop::queue(
-            $worflow(...),
-            $this->ctx->withCatch($catch),
-            new WorkflowClaim(
-                $this->runner,
-                $this->streamGroup,
-            ),
+            (new WorkflowMain($this->handler, $stream))(...),
+            $this->ctx,
+            new WorkflowClaim($this->handler, $stream),
+            new WorkflowCatch($stream),
         );
     }
 
+    #[Override]
     public function disconnect(): void
     {
-        $this->ctx->cancel();
-        $this->streamGroup->delConsumer();
+        if ($this->ctx instanceof Context) {
+            $this->ctx->cancel();
+        }
     }
 }
diff --git a/src/Publisher.php b/src/Publisher.php
index 78250c5..1afe852 100644
--- a/src/Publisher.php
+++ b/src/Publisher.php
@@ -6,28 +6,33 @@
 
 use Override;
 use Throwable;
-use WeakMap;
 use Amp\Redis\RedisClient;
-use kuaukutsu\poc\queue\stream\internal\Payload;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
 use kuaukutsu\queue\core\exception\QueuePublishException;
 use kuaukutsu\queue\core\PublisherInterface;
 use kuaukutsu\queue\core\QueueContext;
 use kuaukutsu\queue\core\QueueMessage;
 use kuaukutsu\queue\core\QueueTask;
 use kuaukutsu\queue\core\SchemaInterface;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
+use kuaukutsu\poc\queue\stream\internal\Payload;
+
+use function Amp\async;
+use function Amp\Future\await;
 
 /**
  * @api
  */
-final class Publisher implements PublisherInterface
+final readonly class Publisher implements PublisherInterface
 {
-    private WeakMap $map;
+    private RedisStream $stream;
 
-    public function __construct(private readonly RedisClient $client)
+    private RedisString $string;
+
+    public function __construct(RedisClient $redis)
     {
-        $this->map = new WeakMap();
+        $this->stream = new RedisStream($redis);
+        $this->string = new RedisString($redis);
     }
 
     /**
@@ -37,45 +42,32 @@ public function __construct(private readonly RedisClient $client)
     #[Override]
     public function push(SchemaInterface $schema, QueueTask $task, ?QueueContext $context = null): string
     {
-        $storage = new RedisString($this->client);
-        $storage->set(
+        $this->string->set(
             $task->getUuid(),
             QueueMessage::makeMessage($task, $context ?? QueueContext::make($schema)),
         );
 
         try {
-            $this->makeStream($schema)->add(
-                Payload::fromTask($task)->toArray()
-            );
+            $this->stream->add($schema, Payload::fromTask($task)->toArray());
         } catch (Throwable $exception) {
-            $storage->del($task->getUuid());
+            $this->string->del($task->getUuid());
             throw new QueuePublishException($schema, $exception);
         }
 
         return $task->getUuid();
     }
 
-    /**
-     * @param list<QueueTask> $taskBatch
-     * @return list<non-empty-string>
-     * @throws QueuePublishException
-     */
     #[Override]
     public function pushBatch(SchemaInterface $schema, array $taskBatch, ?QueueContext $context = null): array
     {
-        $list = [];
+        $groupAwait = [];
         foreach ($taskBatch as $task) {
-            $list[] = $this->push($schema, $task, $context);
+            $groupAwait[] = async($this->push(...), $schema, $task, $context);
         }
 
-        return $list;
-    }
-
-    private function makeStream(SchemaInterface $schema): RedisStream
-    {
         /**
-         * @var RedisStream
+         * @var list<non-empty-string>
          */
-        return $this->map[$schema] ??= new RedisStream($this->client, $schema);
+        return await($groupAwait);
     }
 }
diff --git a/src/StreamOptions.php b/src/StreamOptions.php
index 02b11c5..da9f394 100644
--- a/src/StreamOptions.php
+++ b/src/StreamOptions.php
@@ -12,6 +12,7 @@
      * @param positive-int $batchSize
      * @param positive-int $minIdleTime milliseconds
      * @param positive-int $timeoutBlocking milliseconds
+     * @param non-empty-string $templateDLQ
      */
     public function __construct(
         public string $consumerName,
@@ -19,6 +20,7 @@ public function __construct(
         public int $batchSize = 25,
         public int $minIdleTime = 60_000,
         public int $timeoutBlocking = 5_000,
+        public string $templateDLQ = '%s:dlq',
     ) {
     }
 }
diff --git a/src/interceptor/ExactlyOnceInterceptor.php b/src/interceptor/ExactlyOnceInterceptor.php
new file mode 100644
index 0000000..d347e11
--- /dev/null
+++ b/src/interceptor/ExactlyOnceInterceptor.php
@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\interceptor;
+
+use Override;
+use Amp\Redis\RedisClient;
+use Amp\Redis\Command\Option\SetOptions;
+use kuaukutsu\queue\core\handler\HandlerInterface;
+use kuaukutsu\queue\core\interceptor\InterceptorInterface;
+use kuaukutsu\queue\core\QueueMessage;
+
+/**
+ * @api
+ */
+final readonly class ExactlyOnceInterceptor implements InterceptorInterface
+{
+    /**
+     * @param non-negative-int $ttl Time in seconds. Default 3 min.
+     */
+    public function __construct(
+        private RedisClient $redis,
+        private int $ttl = 180,
+    ) {
+    }
+
+    #[Override]
+    public function intercept(QueueMessage $message, HandlerInterface $handler): void
+    {
+        $options = (new SetOptions())->withTtl($this->ttl)->withoutOverwrite();
+        $isSave = $this->redis->set('eac' . $message->task->getUuid(), '1', $options);
+        if ($isSave) {
+            $handler->handle($message);
+        }
+    }
+}
diff --git a/src/internal/Context.php b/src/internal/Context.php
index 1a31d53..4590212 100644
--- a/src/internal/Context.php
+++ b/src/internal/Context.php
@@ -5,10 +5,10 @@
 namespace kuaukutsu\poc\queue\stream\internal;
 
 use Closure;
-use Throwable;
 use Revolt\EventLoop;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
+use kuaukutsu\queue\core\SchemaInterface;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
@@ -30,33 +30,33 @@ final class Context
      */
     private array $payloadList = [];
 
+    /**
+     * @param non-negative-int $maxExceededAttempts
+     */
     public function __construct(
+        public readonly SchemaInterface $schema,
+        private readonly RedisStreamGroup $streamGroup,
         private readonly RedisString $string,
-        private readonly RedisStreamGroup $stream,
-        private readonly ?Closure $catch = null,
+        public readonly int $maxExceededAttempts = 3,
     ) {
     }
 
     /**
-     * @param ?callable(string, Throwable):void $catch
+     * @param non-empty-string $uuid
      */
-    public function withCatch(?callable $catch): self
+    public function getData(string $uuid): ?string
     {
-        if ($catch === null) {
-            return $this;
-        }
-
-        return new Context($this->string, $this->stream, $catch(...));
+        return $this->string->get($uuid);
     }
 
-    public function tryCatch(string $message, Throwable $throwable): bool
+    /**
+     * @param non-empty-string $source
+     * @param non-empty-string $destination
+     * @param positive-int $ttl
+     */
+    public function copyData(string $source, string $destination, int $ttl = 600): bool
     {
-        if (is_callable($this->catch)) {
-            call_user_func($this->catch, $message, $throwable);
-            return true;
-        }
-
-        return false;
+        return $this->string->copy($source, $destination, $ttl);
     }
 
     /**
@@ -72,7 +72,7 @@ public function setAck(string $identity, string $payloadUuid): void
     public function sendAck(): void
     {
         if ($this->identityList !== []) {
-            $this->stream->ack(array_shift($this->identityList), ...$this->identityList);
+            $this->streamGroup->ack(array_shift($this->identityList), ...$this->identityList);
             $this->identityList = [];
         }
 
@@ -90,16 +90,23 @@ public function defer(Closure $callback): void
         $callbackId = EventLoop::defer($callback);
         $this->callbackList[$callbackId] = true;
         // truncation tail
-        if (count($this->callbackList) > 3) {
-            $this->callbackList = array_slice($this->callbackList, -3, null, true);
+        if (count($this->callbackList) > 32) {
+            $this->callbackList = array_slice($this->callbackList, -32, null, true);
         }
     }
 
+    public function done(string $callbackId): void
+    {
+        unset($this->callbackList[$callbackId]);
+    }
+
     public function cancel(): void
     {
         $this->sendAck();
         foreach ($this->callbackList as $callbackId => $_) {
             EventLoop::cancel($callbackId);
         }
+
+        $this->streamGroup->delConsumer();
     }
 }
diff --git a/src/internal/stream/RedisStream.php b/src/internal/stream/RedisStream.php
index 931bcb5..3fc9ee8 100644
--- a/src/internal/stream/RedisStream.php
+++ b/src/internal/stream/RedisStream.php
@@ -17,76 +17,41 @@
 {
     use ForbidCloning;
     use ForbidSerialization;
+    use StreamUtils;
 
-    /**
-     * @var non-empty-string
-     */
-    private string $key;
-
-    public function __construct(
-        private RedisClient $client,
-        SchemaInterface $schema,
-    ) {
-        $this->key = StreamUtils::generateKey($schema);
+    public function __construct(private RedisClient $client)
+    {
     }
 
     /**
      * @param array<non-empty-string, float|int|string> $payload
      * @param positive-int|false $maxlen
+     * @return ?non-empty-string
      * @see https://redis.io/docs/latest/commands/xadd/
      */
-    public function add(array $payload, int|false $maxlen = 100_000): ?string
+    public function add(SchemaInterface $schema, array $payload, int|false $maxlen = 100_000): ?string
     {
-        $args = $this->preparePayload($payload);
-
         $identity = $maxlen > 0
             ? $this->client->execute(
                 'XADD',
-                $this->key,
+                $this->generateKey($schema),
                 'MAXLEN',
                 '~',
                 $maxlen,
                 '*',
-                ...$args
+                ...$this->preparePayload($payload)
             )
             : $this->client->execute(
                 'XADD',
-                $this->key,
+                $this->generateKey($schema),
                 '*',
-                ...$args
+                ...$this->preparePayload($payload)
             );
 
-        if (is_string($identity)) {
-            return $identity;
-        }
-
-        return null;
-    }
-
-    public function dlq(array $payload): ?string
-    {
-        $identity = $this->client->execute(
-            'XADD',
-            $this->key . ':dlq',
-            '*',
-            ...$this->preparePayload($payload)
-        );
-
-        if (is_string($identity)) {
+        if (is_string($identity) && $identity !== '') {
             return $identity;
         }
 
         return null;
     }
-
-    private function preparePayload(array $payload): array
-    {
-        $args = [];
-        foreach ($payload as $argKey => $argValue) {
-            $args[] = $argKey;
-            $args[] = $argValue;
-        }
-
-        return $args;
-    }
 }
diff --git a/src/internal/stream/RedisStreamGroup.php b/src/internal/stream/RedisStreamGroup.php
index aba4ae8..2e40b7c 100644
--- a/src/internal/stream/RedisStreamGroup.php
+++ b/src/internal/stream/RedisStreamGroup.php
@@ -20,6 +20,7 @@
 {
     use ForbidCloning;
     use ForbidSerialization;
+    use StreamUtils;
 
     /**
      * @var non-empty-string
@@ -28,10 +29,10 @@
 
     public function __construct(
         private RedisClient $client,
-        private SchemaInterface $schema,
         private StreamOptions $options,
+        private SchemaInterface $schema,
     ) {
-        $this->key = StreamUtils::generateKey($schema);
+        $this->key = $this->generateKey($this->schema);
     }
 
     /**
@@ -49,7 +50,7 @@ public function create(): bool
                 '$',
                 'MKSTREAM',
             );
-        } catch (QueryException $e) {
+        } /** @noinspection PhpRedundantCatchClauseInspection */ catch (QueryException $e) {
             if (str_contains($e->getMessage(), 'BUSYGROUP Consumer Group name already exists')) {
                 return false;
             }
@@ -61,10 +62,36 @@ public function create(): bool
     }
 
     /**
-     * @return ?array<array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}>
+     * @param array<non-empty-string, float|int|string> $payload
+     * @return ?non-empty-string
+     * @see https://redis.io/docs/latest/commands/xadd/
+     */
+    public function addDLQ(array $payload): ?string
+    {
+        $key = $this->key;
+        if (str_contains($this->options->templateDLQ, '%s')) {
+            $key = sprintf($this->options->templateDLQ, $key);
+        }
+
+        $identity = $this->client->execute(
+            'XADD',
+            $key,
+            '*',
+            ...$this->preparePayload($payload)
+        );
+
+        if (is_string($identity) && $identity !== '') {
+            return $identity;
+        }
+
+        return null;
+    }
+
+    /**
+     * @return array{}|array<array{0: non-empty-string, 1: ?list<array{0: non-empty-string, 1: string[]}>}>
      * @see https://redis.io/docs/latest/commands/xreadgroup/
      */
-    public function read(): ?array
+    public function read(): array
     {
         $result = $this->client->execute(
             'XREADGROUP',
@@ -82,19 +109,19 @@ public function read(): ?array
 
         if (is_array($result)) {
             /**
-             * @var array<array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}>
+             * @var array<array{0: non-empty-string, 1: ?list<array{0: non-empty-string, 1: string[]}>}>
              */
             return $result;
         }
 
-        return null;
+        return [];
     }
 
     /**
-     * @return null|array<empty>|array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}
+     * @return array{}|array{0: non-empty-string, 1: ?list<array{0: non-empty-string, 1: string[]}>}
      * @see https://redis.io/docs/latest/commands/xautoclaim/
      */
-    public function autoclaim(string $start = '0-0'): ?array
+    public function autoclaim(string $start = '0-0'): array
     {
         $result = $this->client->execute(
             'XAUTOCLAIM',
@@ -109,12 +136,12 @@ public function autoclaim(string $start = '0-0'): ?array
 
         if (is_array($result)) {
             /**
-             * @var array<empty>|array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}
+             * @var array{0: non-empty-string, 1: ?list<array{0: non-empty-string, 1: string[]}>}
              */
             return $result;
         }
 
-        return null;
+        return [];
     }
 
     /**
diff --git a/src/internal/stream/RedisString.php b/src/internal/stream/RedisString.php
index a53a18f..c490b03 100644
--- a/src/internal/stream/RedisString.php
+++ b/src/internal/stream/RedisString.php
@@ -13,9 +13,8 @@
  */
 final readonly class RedisString
 {
-    public function __construct(
-        private RedisClient $client,
-    ) {
+    public function __construct(private RedisClient $client)
+    {
     }
 
     /**
diff --git a/src/internal/stream/StreamUtils.php b/src/internal/stream/StreamUtils.php
index 1f04153..d63c1f6 100644
--- a/src/internal/stream/StreamUtils.php
+++ b/src/internal/stream/StreamUtils.php
@@ -9,17 +9,28 @@
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
  */
-final readonly class StreamUtils
+trait StreamUtils
 {
-    private function __construct()
+    /**
+     * @return non-empty-string
+     */
+    private function generateKey(SchemaInterface $schema): string
     {
+        return 'queue:' . $schema->getRoutingKey();
     }
 
     /**
-     * @return non-empty-string
+     * @param array<non-empty-string, float|int|string> $payload
+     * @return list<non-empty-string|float|int|string>
      */
-    public static function generateKey(SchemaInterface $schema): string
+    private function preparePayload(array $payload): array
     {
-        return 'queue:' . $schema->getRoutingKey();
+        $args = [];
+        foreach ($payload as $argKey => $argValue) {
+            $args[] = $argKey;
+            $args[] = $argValue;
+        }
+
+        return $args;
     }
 }
diff --git a/src/internal/workflow/TaskHandler.php b/src/internal/workflow/TaskHandler.php
new file mode 100644
index 0000000..ae95665
--- /dev/null
+++ b/src/internal/workflow/TaskHandler.php
@@ -0,0 +1,152 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\internal\workflow;
+
+use Closure;
+use Throwable;
+use Amp\CancelledException;
+use Amp\TimeoutCancellation;
+use kuaukutsu\queue\core\handler\HandlerInterface;
+use kuaukutsu\queue\core\QueueMessage;
+use kuaukutsu\poc\queue\stream\exception\WorkflowException;
+use kuaukutsu\poc\queue\stream\internal\Context;
+use kuaukutsu\poc\queue\stream\internal\Payload;
+
+use function Amp\async;
+
+/**
+ * @note если задан обработчик исключений (tryCatch), считаем что отвественность за ошибки лежит на клиенте.
+ * Иначе пробуем несколько раз выполнить, и перекидываем в DLQ.
+ * @psalm-internal kuaukutsu\poc\queue\stream
+ */
+final readonly class TaskHandler
+{
+    public function __construct(
+        private HandlerInterface $handler,
+        private ?Closure $catch,
+    ) {
+    }
+
+    /**
+     * @param Closure(Context, non-empty-string, Payload, Throwable):void $catchHandle
+     * @param non-empty-string $identity
+     * @return bool TRUE отправить ACK; FALSE не отправлять ACK
+     */
+    public function run(Closure $catchHandle, Context $context, string $identity, Payload $payload): bool
+    {
+        $message = $context->getData($payload->uuid);
+        if ($message === null || $message === '') {
+            $exception = new WorkflowException("[$payload->uuid] The payload message must not be empty.");
+            if ($this->tryCatch($message, $exception)) {
+                return true;
+            }
+
+            $this->handleCatch(
+                $catchHandle,
+                $context,
+                $identity,
+                $payload,
+                $exception,
+            );
+
+            return true;
+        }
+
+        try {
+            $queueMessage = QueueMessage::makeFromMessage($message);
+        } catch (Throwable $exception) {
+            if ($this->tryCatch($message, $exception)) {
+                return true;
+            }
+
+            $this->handleCatch(
+                $catchHandle,
+                $context,
+                $identity,
+                $payload,
+                new WorkflowException("[$payload->uuid] The payload message is corrupted.", $exception),
+            );
+
+            return true;
+        }
+
+        $cancellation = null;
+        if ($queueMessage->context->timeout > 0) {
+            $cancellation = new TimeoutCancellation($queueMessage->context->timeout);
+        }
+
+        try {
+            async(
+                $this->handler->handle(...),
+                $queueMessage
+            )->await($cancellation);
+        } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) {
+            if ($this->tryCatch($message, $exception)) {
+                return true;
+            }
+
+            $this->handleCatch(
+                $catchHandle,
+                $context,
+                $identity,
+                $payload,
+                $exception,
+            );
+
+            return true;
+        } catch (Throwable $exception) {
+            if ($this->tryCatch($message, $exception)) {
+                return true;
+            }
+
+            $this->handleCatch(
+                $catchHandle,
+                $context,
+                $identity,
+                $payload,
+                $exception,
+            );
+
+            return false;
+        }
+
+        return true;
+    }
+
+    private function tryCatch(?string $message, Throwable $throwable): bool
+    {
+        if (is_callable($this->catch)) {
+            call_user_func($this->catch, $message, $throwable);
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * @param Closure(Context, non-empty-string, Payload, Throwable):void $handle
+     * @param non-empty-string $identity
+     */
+    private function handleCatch(
+        Closure $handle,
+        Context $context,
+        string $identity,
+        Payload $payload,
+        Throwable $exception,
+    ): void {
+        $context->defer(
+            static function (string $callbackId) use (
+                $handle,
+                $context,
+                $identity,
+                $payload,
+                $exception,
+            ): void {
+                $handle($context, $identity, $payload, $exception);
+                $context->done($callbackId);
+            }
+        );
+    }
+}
diff --git a/src/internal/workflow/TaskRunner.php b/src/internal/workflow/TaskRunner.php
deleted file mode 100644
index 5c5f59a..0000000
--- a/src/internal/workflow/TaskRunner.php
+++ /dev/null
@@ -1,167 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace kuaukutsu\poc\queue\stream\internal\workflow;
-
-use Throwable;
-use Amp\CancelledException;
-use Amp\TimeoutCancellation;
-use kuaukutsu\queue\core\handler\HandlerInterface;
-use kuaukutsu\queue\core\QueueMessage;
-use kuaukutsu\poc\queue\stream\exception\WorkflowException;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStream;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
-use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
-use kuaukutsu\poc\queue\stream\internal\Context;
-use kuaukutsu\poc\queue\stream\internal\Payload;
-
-use function Amp\async;
-
-/**
- * @note если задан обработчик исключений (tryCatch), считаем что отвественность за ошибки лежит на клиенте.
- * Иначе пробуем несколько раз выполнить, и перекидываем в DLQ.
- * @psalm-internal kuaukutsu\poc\queue\stream
- */
-final readonly class TaskRunner
-{
-    public function __construct(
-        private RedisString $string,
-        private RedisStream $stream,
-        private RedisStreamGroup $streamGroup,
-        private HandlerInterface $handler,
-    ) {
-    }
-
-    /**
-     * @param non-empty-string $identity
-     * @param non-negative-int $maxExceededAttempts
-     * @return bool TRUE отправить ACK; FALSE не отправлять ACK
-     */
-    public function run(Context $ctx, string $identity, Payload $payload, int $maxExceededAttempts = 0): bool
-    {
-        $message = $this->string->get($payload->uuid);
-        if ($message === null || $message === '') {
-            $exception = new WorkflowException("[$payload->uuid] The payload message must not be empty.");
-            if ($ctx->tryCatch('<NULL>', $exception)) {
-                return true;
-            }
-
-            // The payload message is empty.
-            return $this->pushDLQ($identity, $payload, $exception->getMessage());
-        }
-
-        try {
-            $queueMessage = QueueMessage::makeFromMessage($message);
-        } catch (Throwable $exception) {
-            if ($ctx->tryCatch($message, $exception)) {
-                return true;
-            }
-
-            // The payload message is corrupted.
-            return $this->pushDLQ($identity, $payload, $exception->getMessage());
-        }
-
-        $cancellation = null;
-        if ($queueMessage->context->timeout > 0) {
-            $cancellation = new TimeoutCancellation($queueMessage->context->timeout);
-        }
-
-        try {
-            async(
-                $this->handler->handle(...),
-                $queueMessage
-            )->await($cancellation);
-        } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) {
-            if ($ctx->tryCatch($message, $exception)) {
-                return true;
-            }
-
-            // Handler Timeout
-            return $this->pushDLQ($identity, $payload, $exception->getMessage());
-        } catch (Throwable $exception) {
-            if ($ctx->tryCatch($message, $exception)) {
-                return true;
-            }
-
-            return $maxExceededAttempts > 0
-                && $this->isExceededAttempts(
-                    $maxExceededAttempts,
-                    $identity,
-                    $payload,
-                    $exception->getMessage(),
-                );
-        }
-
-        return true;
-    }
-
-    private function pushDLQ(string $identity, Payload $payload, string $reason): bool
-    {
-        $newUuid = $this->copyPayload($payload);
-        if ($newUuid === false) {
-            return false;
-        }
-
-        try {
-            $this->stream->dlq(
-                [
-                    'id' => $identity,
-                    'reason' => $reason,
-                    'uuid' => $newUuid,
-                    'target' => $payload->target,
-                ],
-            );
-        } catch (Throwable) {
-            $this->string->del($newUuid);
-            return false;
-        }
-
-        return true;
-    }
-
-    /**
-     * @return non-empty-string|false
-     */
-    private function copyPayload(Payload $payload): string|false
-    {
-        $destination = preg_replace('/^\w{8}/', '0000000d', $payload->uuid);
-        if (is_string($destination) === false || empty($destination)) {
-            $destination = '0000000d' . $payload->uuid;
-        }
-
-        if ($this->string->copy($payload->uuid, $destination, 900)) {
-            return $destination;
-        }
-
-        return false;
-    }
-
-    /**
-     * @param positive-int $maxExceededAttempts
-     * @param non-empty-string $identity
-     */
-    private function isExceededAttempts(
-        int $maxExceededAttempts,
-        string $identity,
-        Payload $payload,
-        string $previousReason,
-    ): bool {
-        try {
-            $pendingState = $this->streamGroup->pending($identity);
-        } catch (Throwable) {
-            return false;
-        }
-
-        return $pendingState['deliveryCount'] >= $maxExceededAttempts
-            && $this->pushDLQ(
-                $identity,
-                $payload,
-                sprintf(
-                    '[%d] The number of attempts has been exceeded. %s',
-                    $pendingState['deliveryCount'],
-                    $previousReason,
-                ),
-            );
-    }
-}
diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php
new file mode 100644
index 0000000..45b64a9
--- /dev/null
+++ b/src/internal/workflow/WorkflowCatch.php
@@ -0,0 +1,119 @@
+<?php
+
+declare(strict_types=1);
+
+namespace kuaukutsu\poc\queue\stream\internal\workflow;
+
+use Throwable;
+use Amp\CancelledException;
+use kuaukutsu\poc\queue\stream\exception\WorkflowException;
+use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup;
+use kuaukutsu\poc\queue\stream\internal\Context;
+use kuaukutsu\poc\queue\stream\internal\Payload;
+
+use function Amp\async;
+
+/**
+ * @psalm-internal kuaukutsu\poc\queue\stream
+ */
+final readonly class WorkflowCatch
+{
+    public function __construct(private RedisStreamGroup $stream)
+    {
+    }
+
+    /**
+     * @param non-empty-string $identity
+     */
+    public function __invoke(Context $ctx, string $identity, Payload $payload, Throwable $exception): void
+    {
+        if ($exception instanceof WorkflowException) {
+            $this->dlq($ctx, $identity, $payload, $exception->getMessage());
+            return;
+        }
+
+        if ($exception instanceof CancelledException) {
+            $this->dlq($ctx, $identity, $payload, $exception->getMessage());
+            return;
+        }
+
+        if ($ctx->maxExceededAttempts === 0) {
+            $this->dlq($ctx, $identity, $payload, $exception->getMessage());
+            return;
+        }
+
+        $pending = $this->pending($this->stream, $identity);
+        $attempts = $pending['deliveryCount'] ?? 0;
+        if ($attempts >= $ctx->maxExceededAttempts) {
+            $this->dlq(
+                $ctx,
+                $identity,
+                $payload,
+                sprintf(
+                    '[%d] The number of attempts has been exceeded. %s',
+                    $attempts,
+                    $exception->getMessage(),
+                ),
+            );
+
+            $ctx->setAck($identity, $payload->uuid);
+            $ctx->sendAck();
+        }
+    }
+
+    private function dlq(Context $ctx, string $identity, Payload $payload, string $reason): void
+    {
+        $newUuid = preg_replace('/^\w{8}/', '0000000d', $payload->uuid);
+        if (is_string($newUuid) === false || empty($newUuid)) {
+            $newUuid = '0000000d' . $payload->uuid;
+        }
+
+        if ($ctx->copyData($payload->uuid, $newUuid, 1800) === false) {
+            return;
+        }
+
+        try {
+            $this->stream->addDLQ(
+                [
+                    'id' => $identity,
+                    'reason' => $reason,
+                    'uuid' => $newUuid,
+                    'target' => $payload->target,
+                ],
+            );
+        } catch (Throwable) {
+            $ctx->copyData($newUuid, $payload->uuid);
+        }
+    }
+
+    /**
+     * @return array{}|array{
+     *       "consumer": string,
+     *       "elapsedMilliseconds": int,
+     *       "deliveryCount": int,
+     *   }
+     */
+    public function pending(RedisStreamGroup $command, string $identity): array
+    {
+        /**
+         * @param non-empty-string $identity
+         */
+        $fn = static function (RedisStreamGroup $command, string $identity): array {
+            /** @var non-empty-string $identity */
+            try {
+                return $command->pending($identity);
+            } catch (Throwable) {
+                return [];
+            }
+        };
+
+        /**
+         * @var array{}|array{
+         *        "consumer": string,
+         *        "elapsedMilliseconds": int,
+         *        "deliveryCount": int,
+         *    }
+         */
+        return async($fn(...), $command, $identity)->await();
+    }
+}
diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php
index 3a71d63..2f98fa5 100644
--- a/src/internal/workflow/WorkflowClaim.php
+++ b/src/internal/workflow/WorkflowClaim.php
@@ -19,16 +19,17 @@
 final readonly class WorkflowClaim
 {
     public function __construct(
-        private TaskRunner $action,
+        private TaskHandler $action,
         private RedisStreamGroup $stream,
     ) {
     }
 
-    public function __invoke(Context $ctx): void
+    public function __invoke(Context $ctx, WorkflowCatch $catch): void
     {
-        $fn = static function (TaskRunner $action, Context $ctx, string $identity, Payload $payload): void {
+        $action = $this->action->run(...);
+        $workflow = static function (string $identity, Payload $payload) use ($action, $catch, $ctx): void {
             /** @var non-empty-string $identity */
-            if ($action->run($ctx, $identity, $payload, 3)) {
+            if ($action($catch(...), $ctx, $identity, $payload)) {
                 $ctx->setAck($identity, $payload->uuid);
             }
         };
@@ -36,7 +37,7 @@ public function __invoke(Context $ctx): void
         while (true) {
             $list = [];
             foreach ($this->autoclaim($this->stream) as $identity => $payload) {
-                $list[] = async($fn(...), $this->action, $ctx, $identity, $payload);
+                $list[] = async($workflow(...), $identity, $payload);
             }
 
             if ($list === []) {
@@ -61,13 +62,10 @@ private function autoclaim(RedisStreamGroup $command): iterable
     {
         $fn = static function (RedisStreamGroup $command): iterable {
             $batch = $command->autoclaim();
-            if ($batch === null) {
+            if ($batch === []) {
                 return;
             }
 
-            /**
-             * @var array<array{0: non-empty-string, 1: string[]}> $src
-             */
             $src = $batch[1] ?? [];
             foreach ($src as [$identity, $payload]) {
                 $data = [];
diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php
index 13adabf..6a46dec 100644
--- a/src/internal/workflow/WorkflowMain.php
+++ b/src/internal/workflow/WorkflowMain.php
@@ -19,17 +19,18 @@
 final readonly class WorkflowMain
 {
     public function __construct(
-        private TaskRunner $action,
+        private TaskHandler $action,
         private RedisStreamGroup $stream,
     ) {
     }
 
-    public function __invoke(Context $ctx, WorkflowClaim $workflowClaim): void
+    public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch): void
     {
         $lastAction = time();
-        $fn = static function (TaskRunner $action, Context $ctx, string $identity, Payload $payload): void {
+        $action = $this->action->run(...);
+        $workflow = static function (string $identity, Payload $payload) use ($action, $catch, $ctx): void {
             /** @var non-empty-string $identity */
-            if ($action->run($ctx, $identity, $payload)) {
+            if ($action($catch(...), $ctx, $identity, $payload)) {
                 $ctx->setAck($identity, $payload->uuid);
             }
         };
@@ -38,7 +39,7 @@ public function __invoke(Context $ctx, WorkflowClaim $workflowClaim): void
         while (true) {
             $list = [];
             foreach ($this->read($this->stream) as $identity => $payload) {
-                $list[] = async($fn(...), $this->action, $ctx, $identity, $payload);
+                $list[] = async($workflow(...), $identity, $payload);
             }
 
             try {
@@ -51,8 +52,9 @@ public function __invoke(Context $ctx, WorkflowClaim $workflowClaim): void
 
             if ($list === [] && $lastAction < strtotime('-30 seconds')) {
                 $ctx->defer(
-                    static function () use ($workflowClaim, $ctx): void {
-                        $workflowClaim($ctx);
+                    static function (string $callbackId) use ($claim, $ctx, $catch): void {
+                        $claim($ctx, $catch);
+                        $ctx->done($callbackId);
                     }
                 );
 
@@ -68,13 +70,10 @@ private function read(RedisStreamGroup $command): iterable
     {
         $fn = static function (RedisStreamGroup $command): iterable {
             $batch = $command->read();
-            if ($batch === null || $batch === []) {
+            if ($batch === []) {
                 return;
             }
 
-            /**
-             * @var array<array{0: non-empty-string, 1: string[]}> $src
-             */
             $src = $batch[0][1] ?? [];
             foreach ($src as [$identity, $payload]) {
                 $data = [];
diff --git a/tests/simulation/publisher-batch.php b/tests/simulation/publisher-batch.php
new file mode 100644
index 0000000..e102502
--- /dev/null
+++ b/tests/simulation/publisher-batch.php
@@ -0,0 +1,39 @@
+<?php
+
+/**
+ * Publisher.
+ * @var Builder $builder bootstrap.php
+ */
+
+declare(strict_types=1);
+
+use kuaukutsu\poc\queue\stream\Builder;
+use kuaukutsu\poc\queue\stream\tests\stub\QueueHandlerStub;
+use kuaukutsu\poc\queue\stream\tests\stub\QueueSchemaStub;
+use kuaukutsu\queue\core\QueueContext;
+use kuaukutsu\queue\core\QueueTask;
+
+use function kuaukutsu\poc\queue\stream\tests\argument;
+
+require dirname(__DIR__) . '/bootstrap.php';
+
+$schema = QueueSchemaStub::from((string)argument('schema', 'low'));
+echo 'publisher run: ' . $schema->getRoutingKey() . PHP_EOL;
+
+$batch = [];
+foreach (range(1, 100) as $item) {
+    $batch[] = new QueueTask(
+        target: QueueHandlerStub::class,
+        arguments: [
+            'id' => $item,
+            'name' => 'test batch',
+        ],
+    );
+}
+
+$publisher = $builder->buildPublisher();
+$publisher->pushBatch(
+    $schema,
+    $batch,
+    QueueContext::make($schema)
+);
diff --git a/tests/simulation/publisher.php b/tests/simulation/publisher.php
index ab099bc..a4694e1 100644
--- a/tests/simulation/publisher.php
+++ b/tests/simulation/publisher.php
@@ -34,18 +34,17 @@
 
 // range
 foreach (range(1, 100) as $item) {
-    $publisher
-        ->push(
-            $schema,
-            new QueueTask(
-                target: QueueHandlerStub::class,
-                arguments: [
-                    'id' => $item,
-                    'name' => 'test range',
-                ],
-            ),
-            QueueContext::make($schema)
-                ->withExternal(['requestId' => $item])
-                ->withTimeout(300)
-        );
+    $publisher->push(
+        $schema,
+        new QueueTask(
+            target: QueueHandlerStub::class,
+            arguments: [
+                'id' => $item,
+                'name' => 'test range',
+            ],
+        ),
+        QueueContext::make($schema)
+            ->withExternal(['requestId' => $item])
+            ->withTimeout(300)
+    );
 }
diff --git a/tests/simulation/worker-redis.php b/tests/simulation/worker-redis.php
index 281db87..59bd39b 100644
--- a/tests/simulation/worker-redis.php
+++ b/tests/simulation/worker-redis.php
@@ -22,8 +22,8 @@
         RedisConfig::fromUri('tcp://redis:6379')
     );
 
-$consumer = $builder->buildConsumer($schema);
-$consumer->consume();
+$consumer = $builder->buildConsumer();
+$consumer->consume($schema);
 
 /** @noinspection PhpUnhandledExceptionInspection */
 trapSignal([SIGTERM, SIGINT]);
diff --git a/tests/simulation/worker-valkey.php b/tests/simulation/worker-valkey.php
index 859529e..7d4cc1e 100644
--- a/tests/simulation/worker-valkey.php
+++ b/tests/simulation/worker-valkey.php
@@ -22,8 +22,8 @@
         RedisConfig::fromUri('tcp://valkey:6379')
     );
 
-$consumer = $builder->buildConsumer($schema);
-$consumer->consume();
+$consumer = $builder->buildConsumer();
+$consumer->consume($schema);
 
 /** @noinspection PhpUnhandledExceptionInspection */
 trapSignal([SIGTERM, SIGINT]);
diff --git a/tests/simulation/worker-with-catch.php b/tests/simulation/worker-with-catch.php
index fddb1ad..b7ba4cd 100644
--- a/tests/simulation/worker-with-catch.php
+++ b/tests/simulation/worker-with-catch.php
@@ -7,9 +7,10 @@
 
 declare(strict_types=1);
 
-use kuaukutsu\poc\queue\stream\Builder;
+use kuaukutsu\queue\core\interceptor\ArgumentsVerifyInterceptor;
 use kuaukutsu\poc\queue\stream\tests\stub\QueueSchemaStub;
 use kuaukutsu\poc\queue\stream\tests\stub\TryCatchInterceptor;
+use kuaukutsu\poc\queue\stream\Builder;
 
 use function Amp\trapSignal;
 use function kuaukutsu\poc\queue\stream\tests\argument;
@@ -20,16 +21,18 @@
 echo 'consumer run: ' . $schema->getRoutingKey() . PHP_EOL;
 
 $consumer = $builder
+    ->withCatch(
+        static function (?string $message, Throwable $exception): void {
+            echo sprintf("data: %s\nerror: %s", $message, $exception->getMessage());
+        }
+    )
     ->withInterceptors(
+        new ArgumentsVerifyInterceptor(),
         new TryCatchInterceptor(),
     )
-    ->buildConsumer($schema);
+    ->buildConsumer();
 
-$consumer->consume(
-    static function (string $message, Throwable $exception): void {
-        echo sprintf("data: %s\nerror: %s", $message, $exception->getMessage());
-    }
-);
+$consumer->consume($schema);
 
 /** @noinspection PhpUnhandledExceptionInspection */
 trapSignal([SIGTERM, SIGINT]);
diff --git a/tests/simulation/worker.php b/tests/simulation/worker.php
index 97bf317..82eea02 100644
--- a/tests/simulation/worker.php
+++ b/tests/simulation/worker.php
@@ -7,9 +7,12 @@
 
 declare(strict_types=1);
 
-use kuaukutsu\poc\queue\stream\Builder;
+use kuaukutsu\queue\core\interceptor\ArgumentsVerifyInterceptor;
+use kuaukutsu\poc\queue\stream\interceptor\ExactlyOnceInterceptor;
 use kuaukutsu\poc\queue\stream\tests\stub\QueueSchemaStub;
+use kuaukutsu\poc\queue\stream\Builder;
 
+use function Amp\Redis\createRedisClient;
 use function Amp\trapSignal;
 use function kuaukutsu\poc\queue\stream\tests\argument;
 
@@ -18,9 +21,14 @@
 $schema = QueueSchemaStub::from((string)argument('schema', 'low'));
 echo 'consumer run: ' . $schema->getRoutingKey() . PHP_EOL;
 
-$consumer = $builder->buildConsumer($schema);
-$consumer->consume();
+$consumer = $builder
+    ->withInterceptors(
+        new ArgumentsVerifyInterceptor(),
+        new ExactlyOnceInterceptor(createRedisClient('tcp://redis:6379')),
+    )
+    ->buildConsumer();
 
+$consumer->consume($schema);
 /** @noinspection PhpUnhandledExceptionInspection */
 trapSignal([SIGTERM, SIGINT]);
 $consumer->disconnect();