kuaukutsu · kuaukutsu · Mar 17, 2026 · Mar 17, 2026
diff --git a/src/Consumer.php b/src/Consumer.php
@@ -8,6 +8,7 @@
 use Override;
 use Throwable;
 use Amp\Redis\RedisClient;
+use Amp\TimeoutCancellation;
 use Revolt\EventLoop;
 use kuaukutsu\queue\core\handler\HandlerInterface;
 use kuaukutsu\queue\core\ConsumerInterface;
@@ -55,6 +56,7 @@ public function consume(SchemaInterface $schema): void
             $stream,
             $string,
             $this->eventDispatcher,
+            new TimeoutCancellation(1800),
         );
 
         EventLoop::queue(

diff --git a/src/internal/Context.php b/src/internal/Context.php
@@ -5,16 +5,22 @@
 namespace kuaukutsu\poc\queue\stream\internal;
 
 use Closure;
+use Amp\Future;
+use Amp\Cancellation;
+use Amp\CancelledException;
 use Revolt\EventLoop;
 use kuaukutsu\queue\core\SchemaInterface;
 use kuaukutsu\poc\queue\stream\event\Event;
 use kuaukutsu\poc\queue\stream\event\EventInterface;
 use kuaukutsu\poc\queue\stream\event\EventDispatcher;
 use kuaukutsu\poc\queue\stream\event\MessageAckEvent;
 use kuaukutsu\poc\queue\stream\event\CallbackEvent;
+use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisString;
 
+use function Amp\Future\await;
+
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
  */
@@ -40,9 +46,20 @@ public function __construct(
         private readonly RedisConsume $streamGroup,
         private readonly RedisString $string,
         private readonly EventDispatcher $eventDispatcher,
+        private readonly ?Cancellation $cancellation = null,
     ) {
     }
 
+    public function trigger(Event $name, EventInterface $event): void
+    {
+        $fn = $this->eventDispatcher->trigger(...);
+        EventLoop::defer(
+            static function () use ($fn, $name, $event): void {
+                $fn($name, $event);
+            }
+        );
+    }
+
     /**
      * @param non-empty-string $uuid
      */
@@ -71,16 +88,6 @@ public function copyData(string $source, string $destination, int $ttl = 600): b
         return $this->string->copy($source, $destination, $ttl);
     }
 
-    public function trigger(Event $name, EventInterface $event): void
-    {
-        $fn = $this->eventDispatcher->trigger(...);
-        EventLoop::defer(
-            static function () use ($fn, $name, $event): void {
-                $fn($name, $event);
-            }
-        );
-    }
-
     /**
      * @param non-empty-string $identity
      * @param non-empty-string $payloadUuid
@@ -105,6 +112,43 @@ public function sendAck(): void
         }
     }
 
+    /**
+     * @template T
+     * @param Future<T> $future
+     * @return ?T
+     * @noinspection PhpRedundantCatchClauseInspection
+     */
+    public function awaitFuture(Future $future): mixed
+    {
+        try {
+            return $future->await($this->cancellation);
+        } catch (CancelledException $exception) {
+            $this->trigger(
+                Event::TimeoutCancellation,
+                new SystemExceptionEvent($exception),
+            );
+        }
+
+        return null;
+    }
+
+    /**
+     * @template T
+     * @param list<Future<T>> $futures
+     * @noinspection PhpRedundantCatchClauseInspection
+     */
+    public function awaitFutures(array $futures): void
+    {
+        try {
+            await($futures, $this->cancellation);
+        } catch (CancelledException $exception) {
+            $this->trigger(
+                Event::TimeoutCancellation,
+                new SystemExceptionEvent($exception),
+            );
+        }
+    }
+
     /**
      * @param Closure(string):void $callback
      */

diff --git a/src/internal/workflow/TaskHandler.php b/src/internal/workflow/TaskHandler.php
@@ -33,11 +33,11 @@ public function __construct(
     }
 
     /**
-     * @param Closure(Context, non-empty-string, Payload, Throwable):void $catchHandle
      * @param non-empty-string $identity
+     * @param Closure(Context, non-empty-string, Payload, Throwable):void $catchHandle
      * @return bool TRUE отправить ACK; FALSE не отправлять ACK
      */
-    public function run(Closure $catchHandle, Context $context, string $identity, Payload $payload): bool
+    public function run(Context $context, string $identity, Payload $payload, Closure $catchHandle): bool
     {
         $message = $context->getData($payload->uuid);
         if ($message === null || $message === '') {

diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php
@@ -1,5 +1,7 @@
 <?php
 
+/** @noinspection PhpRedundantCatchClauseInspection */
+
 declare(strict_types=1);
 
 namespace kuaukutsu\poc\queue\stream\internal\workflow;
@@ -66,7 +68,7 @@ public function __invoke(Context $ctx, string $identity, Payload $payload, Throw
             return;
         }
 
-        $attempts = $this->attempts($this->stream, $ctx, $identity);
+        $attempts = $this->attempts($ctx, $this->stream, $identity);
         $ctx->trigger(
             Event::MessageHandleError,
             new MessageErrorEvent($payload, $exception, $attempts),
@@ -155,29 +157,29 @@ private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt
      * @param non-empty-string $identity
      * @return positive-int
      */
-    private function attempts(RedisConsume $command, Context $ctx, string $identity): int
+    private function attempts(Context $ctx, RedisConsume $command, string $identity): int
     {
         /**
-         * @psalm-var Closure(RedisConsume, Context, non-empty-string): positive-int $fn
+         * @psalm-var Closure(Context, RedisConsume, non-empty-string): positive-int $fn
          */
-        static $fn = static function (RedisConsume $command, Context $ctx, string $identity): int {
+        static $fn = static function (Context $ctx, RedisConsume $command, string $identity): int {
             /** @var non-empty-string $identity */
             try {
                 $pending = $command->pending($identity);
+                return max(1, $pending['deliveryCount']);
             } catch (Throwable $exception) {
                 $ctx->trigger(
                     Event::RuntimeException,
                     new SystemExceptionEvent($exception),
                 );
-                return 1;
             }
 
-            return max(1, $pending['deliveryCount']);
+            return 1;
         };
 
         /**
          * @phpstan-var positive-int
          */
-        return async($fn(...), $command, $ctx, $identity)->await();
+        return $ctx->awaitFuture(async($fn(...), $ctx, $command, $identity)) ?? 1;
     }
 }
diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php
@@ -1,20 +1,17 @@
 <?php
 
+/** @noinspection PhpRedundantCatchClauseInspection */
+
 declare(strict_types=1);
 
 namespace kuaukutsu\poc\queue\stream\internal\workflow;
 
 use Closure;
-use Amp\CancelledException;
-use Amp\TimeoutCancellation;
-use kuaukutsu\poc\queue\stream\event\Event;
-use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\Context;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
 use function Amp\async;
-use function Amp\Future\await;
 
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
@@ -29,7 +26,7 @@ public function __construct(
 
     public function __invoke(Context $ctx, WorkflowCatch $catch): void
     {
-        $action = $this->action->run(...);
+        static $action = $this->action->run(...);
 
         /**
          * @psalm-var Closure(Context, string, Payload, callable, callable): void $workflow
@@ -42,15 +39,15 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void
             callable $catch,
         ): void {
             /** @var non-empty-string $identity */
-            if ($action($catch(...), $ctx, $identity, $payload)) {
+            if ($action($ctx, $identity, $payload, $catch(...))) {
                 $ctx->setAck($identity, $payload->uuid);
             }
         };
 
         $lastIdentity = '0-0';
         while (true) {
             $list = [];
-            foreach ($this->autoclaim($this->stream, $lastIdentity) as $identity => $payload) {
+            foreach ($this->autoclaim($ctx, $this->stream, $lastIdentity) as $identity => $payload) {
                 $list[] = async($workflow(...), $ctx, $identity, $payload, $action, $catch);
                 $lastIdentity = $identity;
             }
@@ -59,23 +56,15 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void
                 break;
             }
 
-            try {
-                await($list, new TimeoutCancellation(1800));
-            } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) {
-                $ctx->trigger(
-                    Event::TimeoutCancellation,
-                    new SystemExceptionEvent($exception),
-                );
-            }
-
+            $ctx->awaitFutures($list);
             $ctx->sendAck();
         }
     }
 
     /**
      * @return iterable<non-empty-string, Payload>
      */
-    private function autoclaim(RedisConsume $command, string $lastIdentity): iterable
+    private function autoclaim(Context $ctx, RedisConsume $command, string $lastIdentity): iterable
     {
         /**
          * @psalm-var Closure(RedisConsume, string): iterable<non-empty-string, Payload> $fn
@@ -99,8 +88,8 @@ private function autoclaim(RedisConsume $command, string $lastIdentity): iterabl
         };
 
         /**
-         * @phpstan-var iterable<non-empty-string, Payload>
+         * @var iterable<non-empty-string, Payload>
          */
-        return async($fn(...), $command, $lastIdentity)->await();
+        return $ctx->awaitFuture(async($fn(...), $command, $lastIdentity)) ?? [];
     }
 }
diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php
@@ -5,16 +5,11 @@
 namespace kuaukutsu\poc\queue\stream\internal\workflow;
 
 use Closure;
-use Amp\CancelledException;
-use Amp\TimeoutCancellation;
-use kuaukutsu\poc\queue\stream\event\Event;
-use kuaukutsu\poc\queue\stream\event\SystemExceptionEvent;
 use kuaukutsu\poc\queue\stream\internal\stream\RedisConsume;
 use kuaukutsu\poc\queue\stream\internal\Context;
 use kuaukutsu\poc\queue\stream\internal\Payload;
 
 use function Amp\async;
-use function Amp\Future\await;
 
 /**
  * @psalm-internal kuaukutsu\poc\queue\stream
@@ -29,8 +24,7 @@ public function __construct(
 
     public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch): void
     {
-        $lastAction = time();
-        $action = $this->action->run(...);
+        static $action = $this->action->run(...);
 
         /**
          * @psalm-var Closure(Context, string, Payload, callable, callable): void $workflow
@@ -43,15 +37,17 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc
             callable $catch,
         ): void {
             /** @var non-empty-string $identity */
-            if ($action($catch(...), $ctx, $identity, $payload)) {
+            if ($action($ctx, $identity, $payload, $catch(...))) {
                 $ctx->setAck($identity, $payload->uuid);
             }
         };
 
+        $lastAction = time();
+
         /** @phpstan-ignore while.alwaysTrue */
         while (true) {
             $list = [];
-            foreach ($this->read($this->stream) as $identity => $payload) {
+            foreach ($this->read($ctx, $this->stream) as $identity => $payload) {
                 $list[] = async($workflow(...), $ctx, $identity, $payload, $action, $catch);
             }
 
@@ -60,23 +56,15 @@ public function __invoke(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catc
                 continue;
             }
 
-            try {
-                await($list, new TimeoutCancellation(1800));
-            } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) {
-                $ctx->trigger(
-                    Event::TimeoutCancellation,
-                    new SystemExceptionEvent($exception),
-                );
-            }
-
+            $ctx->awaitFutures($list);
             $ctx->sendAck();
         }
     }
 
     /**
      * @return iterable<non-empty-string, Payload>
      */
-    private function read(RedisConsume $command): iterable
+    private function read(Context $ctx, RedisConsume $command): iterable
     {
         /**
          * @psalm-var Closure(RedisConsume):iterable<non-empty-string, Payload> $fn
@@ -100,9 +88,9 @@ private function read(RedisConsume $command): iterable
         };
 
         /**
-         * @phpstan-var iterable<non-empty-string, Payload>
+         * @var iterable<non-empty-string, Payload>
          */
-        return async($fn(...), $command)->await();
+        return $ctx->awaitFuture(async($fn(...), $command)) ?? [];
     }
 
     private function autoclaim(Context $ctx, WorkflowClaim $claim, WorkflowCatch $catch, int &$lastAction): void