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/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/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..5c5f59a 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. @@ -58,15 +62,35 @@ 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 { - $this->handler->handle($queueMessage); + 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($identity, $payload, $exception->getMessage()); + && $this->isExceededAttempts( + $maxExceededAttempts, + $identity, + $payload, + $exception->getMessage(), + ); } return true; @@ -114,12 +138,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..1b1812f 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,29 +26,37 @@ 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; + $list = []; foreach ($this->read($this->stream) as $identity => $payload) { - if ($this->action->run($ctx, $identity, $payload)) { - $ctx->setAck($identity, $payload->uuid); - $canAutoClaim = false; - } + $list[] = async($fn(...), $this->action, $ctx, $identity, $payload); + } + + try { + await($list, new TimeoutCancellation(1800)); + } /** @noinspection PhpRedundantCatchClauseInspection */ catch (CancelledException) { + // @fixme: logger } $ctx->sendAck(); - // autoclaim - if ($canAutoClaim && $lastQueueClaim < strtotime('-30 seconds')) { + if ($list === [] && $lastAction < strtotime('-30 seconds')) { $ctx->defer( static function () use ($workflowClaim, $ctx): void { $workflowClaim($ctx); } ); - $lastQueueClaim = time(); + $lastAction = time(); } } } @@ -55,31 +66,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-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 ffedbe1..ab099bc 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, @@ -46,6 +44,8 @@ 'name' => 'test range', ], ), - QueueContext::make($schema)->withExternal(['requestId' => $item]) + QueueContext::make($schema) + ->withExternal(['requestId' => $item]) + ->withTimeout(300) ); }