From 5796544ff47030aaf65676f86697f53746419cd7 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Wed, 19 Nov 2025 16:44:21 +0300 Subject: [PATCH 1/3] workflow await --- README.md | 5 ++ src/internal/stream/RedisStreamGroup.php | 8 +-- src/internal/workflow/TaskRunner.php | 34 +++++++++-- src/internal/workflow/WorkflowClaim.php | 67 +++++++++++++-------- src/internal/workflow/WorkflowMain.php | 77 ++++++++++++------------ tests/simulation/publisher.php | 4 +- 6 files changed, 119 insertions(+), 76 deletions(-) diff --git a/README.md b/README.md index 1b614a2..f11664a 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,12 @@ composer require kuaukutsu/poc-queue-stream - Если сообщения в хранилище нет, то выходим с ошибкой, стрим переносим в dead letter queue (DLQ). - Если сообщение получено, но выполнение завершилось ошибкой, отправляем на повторный круг — точнее, оставляем в очереди (PEL); свободный консумер получит это сообщение позже через **XAUTOCLAIM** и попробует выполнить ещё раз. Таких попыток по умолчанию будет 2; после третьей попытки сообщение переносим в DLQ. - Если для консумера задан обработчик исключений (`catch exception`), то обработка ошибок лежит на клиентском ПО, т. е. механизм DLQ работать не будет. + +Сомнительно, но окей (?): + - Прочитанные сообщения «акаются» (**ACK**) пачками; также вместе с командой ACK удаляются сообщения из хранилища. + Альтернатива: акаем по завершению таска, с одной стороны убираем await и подтверждаем здесь и сразу, + с другой сокращаем количество запросов. #### чтиво diff --git a/src/internal/stream/RedisStreamGroup.php b/src/internal/stream/RedisStreamGroup.php index 9bf3963..aba4ae8 100644 --- a/src/internal/stream/RedisStreamGroup.php +++ b/src/internal/stream/RedisStreamGroup.php @@ -61,7 +61,7 @@ public function create(): bool } /** - * @return ?array}> + * @return ?array * @see https://redis.io/docs/latest/commands/xreadgroup/ */ public function read(): ?array @@ -82,7 +82,7 @@ public function read(): ?array if (is_array($result)) { /** - * @var array}> + * @var array */ return $result; } @@ -91,7 +91,7 @@ public function read(): ?array } /** - * @return null|array|array{0: non-empty-string, 1: array} + * @return null|array|array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}} * @see https://redis.io/docs/latest/commands/xautoclaim/ */ public function autoclaim(string $start = '0-0'): ?array @@ -109,7 +109,7 @@ public function autoclaim(string $start = '0-0'): ?array if (is_array($result)) { /** - * @var array|array{0: non-empty-string, 1: array} + * @var array|array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}} */ return $result; } diff --git a/src/internal/workflow/TaskRunner.php b/src/internal/workflow/TaskRunner.php index 9949535..d93a7fe 100644 --- a/src/internal/workflow/TaskRunner.php +++ b/src/internal/workflow/TaskRunner.php @@ -5,6 +5,8 @@ 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; @@ -14,6 +16,8 @@ use kuaukutsu\poc\queue\stream\internal\Context; use kuaukutsu\poc\queue\stream\internal\Payload; +use function Amp\async; + /** * @note если задан обработчик исключений (tryCatch), считаем что отвественность за ошибки лежит на клиенте. * Иначе пробуем несколько раз выполнить, и перекидываем в DLQ. @@ -59,14 +63,29 @@ public function run(Context $ctx, string $identity, Payload $payload, int $maxEx } try { - $this->handler->handle($queueMessage); + async( + $this->handler->handle(...), + $queueMessage + )->await(new TimeoutCancellation(300)); + } /** @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($identity, $payload, $exception->getMessage()); + && $this->isExceededAttempts( + $maxExceededAttempts, + $identity, + $payload, + $exception->getMessage(), + ); } return true; @@ -114,12 +133,15 @@ private function copyPayload(Payload $payload): string|false } /** + * @param positive-int $maxExceededAttempts * @param non-empty-string $identity */ - private function isExceededAttempts(string $identity, Payload $payload, string $previousReason): bool - { - $maxExceededAttempts = 3; - + private function isExceededAttempts( + int $maxExceededAttempts, + string $identity, + Payload $payload, + string $previousReason, + ): bool { try { $pendingState = $this->streamGroup->pending($identity); } catch (Throwable) { diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index 7db8bd3..9e6e733 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/src/internal/workflow/WorkflowClaim.php @@ -4,11 +4,14 @@ namespace kuaukutsu\poc\queue\stream\internal\workflow; +use Amp\CancelledException; +use Amp\TimeoutCancellation; use kuaukutsu\poc\queue\stream\internal\Context; use kuaukutsu\poc\queue\stream\internal\Payload; use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use function Amp\async; +use function Amp\Future\await; /** * @psalm-internal kuaukutsu\poc\queue\stream @@ -23,13 +26,32 @@ public function __construct( public function __invoke(Context $ctx): void { - foreach ($this->autoclaim($this->stream) as $identity => $payload) { - if ($this->action->run($ctx, $identity, $payload, 3)) { + $fn = static function (TaskRunner $action, Context $ctx, string $identity, Payload $payload): void { + /** @var non-empty-string $identity */ + if ($action->run($ctx, $identity, $payload, 3)) { $ctx->setAck($identity, $payload->uuid); } - } + }; + + while (true) { + $list = []; + foreach ($this->autoclaim($this->stream) as $identity => $payload) { + $list[] = async($fn(...), $this->action, $ctx, $identity, $payload); + } + + if ($list === []) { + $ctx->sendAck(); + break; + } + + try { + await($list, new TimeoutCancellation(1800)); + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException) { + // @fixme: logger + } - $ctx->sendAck(); + $ctx->sendAck(); + } } /** @@ -37,31 +59,24 @@ public function __invoke(Context $ctx): void */ private function autoclaim(RedisStreamGroup $command): iterable { - $fn = static function (RedisStreamGroup $command): iterable { - $batch = $command->autoclaim(); - if ($batch === null) { - return []; - } - - /** - * @var array> $src - */ - $src = $batch[1] ?? []; - foreach ($src as [$identity, $payload]) { - $data = []; - foreach (array_chunk($payload, 2) as [$k, $v]) { - $data[$k] = $v; - } - - yield $identity => Payload::fromPayload($data); - } - + $batch = $command->autoclaim(); + if ($batch === null) { return []; - }; + } /** - * @var iterable + * @var array $src */ - return async($fn(...), $command)->await(); + $src = $batch[1] ?? []; + foreach ($src as [$identity, $payload]) { + $data = []; + foreach (array_chunk($payload, 2) as [$k, $v]) { + $data[$k] = $v; + } + + yield $identity => Payload::fromPayload($data); + } + + return []; } } diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 72477a1..63b6c49 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/src/internal/workflow/WorkflowMain.php @@ -4,11 +4,14 @@ namespace kuaukutsu\poc\queue\stream\internal\workflow; +use Amp\CancelledException; +use Amp\TimeoutCancellation; use kuaukutsu\poc\queue\stream\internal\Context; use kuaukutsu\poc\queue\stream\internal\Payload; use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use function Amp\async; +use function Amp\Future\await; /** * @psalm-internal kuaukutsu\poc\queue\stream @@ -23,30 +26,39 @@ public function __construct( public function __invoke(Context $ctx, WorkflowClaim $workflowClaim): void { - $lastQueueClaim = time(); + $lastAction = time(); + $fn = static function (TaskRunner $action, Context $ctx, string $identity, Payload $payload): void { + /** @var non-empty-string $identity */ + if ($action->run($ctx, $identity, $payload)) { + $ctx->setAck($identity, $payload->uuid); + } + }; /** @phpstan-ignore while.alwaysTrue */ while (true) { - $canAutoClaim = true; - foreach ($this->read($this->stream) as $identity => $payload) { - if ($this->action->run($ctx, $identity, $payload)) { - $ctx->setAck($identity, $payload->uuid); - $canAutoClaim = false; - } - } - - $ctx->sendAck(); - - // autoclaim - if ($canAutoClaim && $lastQueueClaim < strtotime('-30 seconds')) { + if ($lastAction < strtotime('-30 seconds')) { $ctx->defer( static function () use ($workflowClaim, $ctx): void { $workflowClaim($ctx); } ); - $lastQueueClaim = time(); + $lastAction = time(); + } + + $list = []; + foreach ($this->read($this->stream) as $identity => $payload) { + $list[] = async($fn(...), $this->action, $ctx, $identity, $payload); + $lastAction = time(); + } + + try { + await($list, new TimeoutCancellation(1800)); + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException) { + // @fixme: logger } + + $ctx->sendAck(); } } @@ -55,31 +67,22 @@ static function () use ($workflowClaim, $ctx): void { */ private function read(RedisStreamGroup $command): iterable { - $fn = static function (RedisStreamGroup $command): iterable { - $batch = $command->read(); - if ($batch === null || $batch === []) { - return []; - } - - /** - * @var array> $src - */ - $src = $batch[0][1] ?? []; - foreach ($src as [$identity, $payload]) { - $data = []; - foreach (array_chunk($payload, 2) as [$k, $v]) { - $data[$k] = $v; - } - - yield $identity => Payload::fromPayload($data); - } - - return []; - }; + $batch = $command->read(); + if ($batch === null || $batch === []) { + return; + } /** - * @var iterable + * @var array $src */ - return async($fn(...), $command)->await(); + $src = $batch[0][1] ?? []; + foreach ($src as [$identity, $payload]) { + $data = []; + foreach (array_chunk($payload, 2) as [$k, $v]) { + $data[$k] = $v; + } + + yield $identity => Payload::fromPayload($data); + } } } diff --git a/tests/simulation/publisher.php b/tests/simulation/publisher.php index ffedbe1..0413d95 100644 --- a/tests/simulation/publisher.php +++ b/tests/simulation/publisher.php @@ -31,11 +31,9 @@ ); $publisher->push($schema, $task); -//$publisher->push($schema, $task); -//$publisher->push($schema, $task); // range -foreach (range(1, 10) as $item) { +foreach (range(1, 100) as $item) { $publisher ->push( $schema, From 786ccb8108e211742720c91b62f61f570e9a0821 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Wed, 19 Nov 2025 17:21:11 +0300 Subject: [PATCH 2/3] order call --- src/internal/workflow/WorkflowMain.php | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 63b6c49..1b1812f 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/src/internal/workflow/WorkflowMain.php @@ -36,20 +36,9 @@ public function __invoke(Context $ctx, WorkflowClaim $workflowClaim): void /** @phpstan-ignore while.alwaysTrue */ while (true) { - if ($lastAction < strtotime('-30 seconds')) { - $ctx->defer( - static function () use ($workflowClaim, $ctx): void { - $workflowClaim($ctx); - } - ); - - $lastAction = time(); - } - $list = []; foreach ($this->read($this->stream) as $identity => $payload) { $list[] = async($fn(...), $this->action, $ctx, $identity, $payload); - $lastAction = time(); } try { @@ -59,6 +48,16 @@ static function () use ($workflowClaim, $ctx): void { } $ctx->sendAck(); + + if ($list === [] && $lastAction < strtotime('-30 seconds')) { + $ctx->defer( + static function () use ($workflowClaim, $ctx): void { + $workflowClaim($ctx); + } + ); + + $lastAction = time(); + } } } From a65414a917e2d1053f405b9fb2d1a72b6886ae02 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Thu, 20 Nov 2025 06:40:15 +0300 Subject: [PATCH 3/3] context timeout --- composer.json | 2 +- src/internal/workflow/TaskRunner.php | 7 ++++++- tests/simulation/publisher-with-exception.php | 3 ++- tests/simulation/publisher.php | 4 +++- 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/composer.json b/composer.json index 52d4049..cdddc00 100644 --- a/composer.json +++ b/composer.json @@ -18,7 +18,7 @@ "php": "^8.3", "ext-redis": "*", "amphp/redis": "^2.0", - "kuaukutsu/queue-core": "^0.3" + "kuaukutsu/queue-core": "^0.4" }, "require-dev": { "ext-pcntl": "*", diff --git a/src/internal/workflow/TaskRunner.php b/src/internal/workflow/TaskRunner.php index d93a7fe..5c5f59a 100644 --- a/src/internal/workflow/TaskRunner.php +++ b/src/internal/workflow/TaskRunner.php @@ -62,11 +62,16 @@ public function run(Context $ctx, string $identity, Payload $payload, int $maxEx 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(new TimeoutCancellation(300)); + )->await($cancellation); } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException $exception) { if ($ctx->tryCatch($message, $exception)) { return true; diff --git a/tests/simulation/publisher-with-exception.php b/tests/simulation/publisher-with-exception.php index db6d9f6..160d122 100644 --- a/tests/simulation/publisher-with-exception.php +++ b/tests/simulation/publisher-with-exception.php @@ -9,6 +9,7 @@ use kuaukutsu\poc\queue\stream\Builder; 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; @@ -29,4 +30,4 @@ ], ); -$publisher->push($schema, $task); +$publisher->push($schema, $task, QueueContext::make($schema)->withTimeout(30)); diff --git a/tests/simulation/publisher.php b/tests/simulation/publisher.php index 0413d95..ab099bc 100644 --- a/tests/simulation/publisher.php +++ b/tests/simulation/publisher.php @@ -44,6 +44,8 @@ 'name' => 'test range', ], ), - QueueContext::make($schema)->withExternal(['requestId' => $item]) + QueueContext::make($schema) + ->withExternal(['requestId' => $item]) + ->withTimeout(300) ); }