From ff0ef08145eed99d5707be2fed0ec450b21b1d50 Mon Sep 17 00:00:00 2001 From: Dmitriy Krivopalov Date: Sat, 22 Nov 2025 13:09:09 +0300 Subject: [PATCH] message increment attempts --- composer.json | 2 +- docker-compose.yml | 22 ++++---- src/interceptor/ExactlyOnceInterceptor.php | 2 +- src/internal/Context.php | 10 ++++ src/internal/workflow/WorkflowCatch.php | 61 +++++++++++++++++----- src/internal/workflow/WorkflowClaim.php | 5 +- src/internal/workflow/WorkflowMain.php | 5 +- tests/simulation/publisher.php | 1 - tests/stub/QueueHandlerStub.php | 5 ++ tests/stub/TaskWriter.php | 3 +- 10 files changed, 87 insertions(+), 29 deletions(-) diff --git a/composer.json b/composer.json index 14d410b..d3c8699 100644 --- a/composer.json +++ b/composer.json @@ -18,7 +18,7 @@ "php": "^8.3", "ext-redis": "*", "amphp/redis": "^2.0", - "kuaukutsu/queue-core": "^0.5.2" + "kuaukutsu/queue-core": "^0.5.3" }, "require-dev": { "ext-pcntl": "*", diff --git a/docker-compose.yml b/docker-compose.yml index 96eb36b..b23b683 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -120,17 +120,6 @@ services: - vredis:/data logging: *default-logging - inside: - container_name: queue_stream_inside - image: redis/redisinsight:latest - profiles: [ "serve" ] - ports: - - "5540:5540" - environment: - - RI_stream_HOST=redis - - RI_stream_PORT=6379 - logging: *default-logging - valkey: container_name: queue_stream_valkey profiles: [ "serve" ] @@ -163,6 +152,17 @@ services: - vvalkey:/data logging: *default-logging + inside: + container_name: queue_stream_inside + image: redis/redisinsight:latest + profiles: [ "serve" ] + ports: + - "5540:5540" + environment: + - RI_stream_HOST=redis + - RI_stream_PORT=6379 + logging: *default-logging + configs: redis: name: conf_${VERSION:-1} diff --git a/src/interceptor/ExactlyOnceInterceptor.php b/src/interceptor/ExactlyOnceInterceptor.php index d347e11..6efea13 100644 --- a/src/interceptor/ExactlyOnceInterceptor.php +++ b/src/interceptor/ExactlyOnceInterceptor.php @@ -29,7 +29,7 @@ public function __construct( 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); + $isSave = $this->redis->set($message->getIdempotencyKey(), '1', $options); if ($isSave) { $handler->handle($message); } diff --git a/src/internal/Context.php b/src/internal/Context.php index 4590212..c85d1d2 100644 --- a/src/internal/Context.php +++ b/src/internal/Context.php @@ -49,6 +49,16 @@ public function getData(string $uuid): ?string return $this->string->get($uuid); } + /** + * @param non-empty-string $uuid + * @param non-empty-string $value + * @param positive-int $ttl + */ + public function setData(string $uuid, string $value, int $ttl = 600): bool + { + return $this->string->set($uuid, $value, $ttl); + } + /** * @param non-empty-string $source * @param non-empty-string $destination diff --git a/src/internal/workflow/WorkflowCatch.php b/src/internal/workflow/WorkflowCatch.php index 45b64a9..6030534 100644 --- a/src/internal/workflow/WorkflowCatch.php +++ b/src/internal/workflow/WorkflowCatch.php @@ -6,6 +6,7 @@ use Throwable; use Amp\CancelledException; +use kuaukutsu\queue\core\QueueMessage; use kuaukutsu\poc\queue\stream\exception\WorkflowException; use kuaukutsu\poc\queue\stream\internal\stream\RedisStreamGroup; use kuaukutsu\poc\queue\stream\internal\Context; @@ -43,7 +44,7 @@ public function __invoke(Context $ctx, string $identity, Payload $payload, Throw } $pending = $this->pending($this->stream, $identity); - $attempts = $pending['deliveryCount'] ?? 0; + $attempts = max(1, $pending['deliveryCount'] ?? 1); if ($attempts >= $ctx->maxExceededAttempts) { $this->dlq( $ctx, @@ -58,31 +59,62 @@ public function __invoke(Context $ctx, string $identity, Payload $payload, Throw $ctx->setAck($identity, $payload->uuid); $ctx->sendAck(); + return; } + + $this->incrAttempt($ctx, $payload, $attempts); } private function dlq(Context $ctx, string $identity, Payload $payload, string $reason): void + { + try { + $this->stream->addDLQ( + [ + 'id' => $identity, + 'reason' => $reason, + 'uuid' => $this->copyData($ctx, $payload), + 'target' => $payload->target, + ], + ); + } catch (Throwable) { + } + } + + /** + * @return non-empty-string + */ + private function copyData(Context $ctx, Payload $payload): string { $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 $ctx->copyData($payload->uuid, $newUuid, 1800) ? $newUuid : $payload->uuid; + } + + /** + * @param positive-int $currentAttempt + */ + private function incrAttempt(Context $ctx, Payload $payload, int $currentAttempt): void + { + $message = $ctx->getData($payload->uuid); + if ($message === null || $message === '') { return; } try { - $this->stream->addDLQ( - [ - 'id' => $identity, - 'reason' => $reason, - 'uuid' => $newUuid, - 'target' => $payload->target, - ], + $queueMessage = QueueMessage::makeFromMessage($message); + $ctx->setData( + $payload->uuid, + QueueMessage::makeMessage( + $queueMessage->task, + $queueMessage->context->incrAttempt(++$currentAttempt), + ) ); + return; } catch (Throwable) { - $ctx->copyData($newUuid, $payload->uuid); + return; } } @@ -93,10 +125,15 @@ private function dlq(Context $ctx, string $identity, Payload $payload, string $r * "deliveryCount": int, * } */ - public function pending(RedisStreamGroup $command, string $identity): array + private function pending(RedisStreamGroup $command, string $identity): array { /** * @param non-empty-string $identity + * @return array{}|array{ + * "consumer": string, + * "elapsedMilliseconds": int, + * "deliveryCount": int, + * } */ $fn = static function (RedisStreamGroup $command, string $identity): array { /** @var non-empty-string $identity */ @@ -108,7 +145,7 @@ public function pending(RedisStreamGroup $command, string $identity): array }; /** - * @var array{}|array{ + * @phpstan-var array{}|array{ * "consumer": string, * "elapsedMilliseconds": int, * "deliveryCount": int, diff --git a/src/internal/workflow/WorkflowClaim.php b/src/internal/workflow/WorkflowClaim.php index 2f98fa5..acc09f3 100644 --- a/src/internal/workflow/WorkflowClaim.php +++ b/src/internal/workflow/WorkflowClaim.php @@ -60,6 +60,9 @@ public function __invoke(Context $ctx, WorkflowCatch $catch): void */ private function autoclaim(RedisStreamGroup $command): iterable { + /** + * @return iterable + */ $fn = static function (RedisStreamGroup $command): iterable { $batch = $command->autoclaim(); if ($batch === []) { @@ -78,7 +81,7 @@ private function autoclaim(RedisStreamGroup $command): iterable }; /** - * @var iterable + * @phpstan-var iterable */ return async($fn(...), $command)->await(); } diff --git a/src/internal/workflow/WorkflowMain.php b/src/internal/workflow/WorkflowMain.php index 6a46dec..5a06858 100644 --- a/src/internal/workflow/WorkflowMain.php +++ b/src/internal/workflow/WorkflowMain.php @@ -68,6 +68,9 @@ static function (string $callbackId) use ($claim, $ctx, $catch): void { */ private function read(RedisStreamGroup $command): iterable { + /** + * @return iterable + */ $fn = static function (RedisStreamGroup $command): iterable { $batch = $command->read(); if ($batch === []) { @@ -86,7 +89,7 @@ private function read(RedisStreamGroup $command): iterable }; /** - * @var iterable + * @phpstan-var iterable */ return async($fn(...), $command)->await(); } diff --git a/tests/simulation/publisher.php b/tests/simulation/publisher.php index a4694e1..790ca4d 100644 --- a/tests/simulation/publisher.php +++ b/tests/simulation/publisher.php @@ -44,7 +44,6 @@ ], ), QueueContext::make($schema) - ->withExternal(['requestId' => $item]) ->withTimeout(300) ); } diff --git a/tests/stub/QueueHandlerStub.php b/tests/stub/QueueHandlerStub.php index 1dd2508..6cfc315 100644 --- a/tests/stub/QueueHandlerStub.php +++ b/tests/stub/QueueHandlerStub.php @@ -5,6 +5,7 @@ namespace kuaukutsu\poc\queue\stream\tests\stub; use Override; +use LogicException; use kuaukutsu\queue\core\QueueContext; use kuaukutsu\queue\core\TaskInterface; @@ -20,6 +21,10 @@ public function __construct( #[Override] public function handle(QueueContext $context): void { + if ($context->attempt === 1 && random_int(0, 1) === 1) { + throw new LogicException('Random exception.'); + } + $this->writer->print($this->id, $this->name, $context); } } diff --git a/tests/stub/TaskWriter.php b/tests/stub/TaskWriter.php index d636cfc..9e00187 100644 --- a/tests/stub/TaskWriter.php +++ b/tests/stub/TaskWriter.php @@ -16,9 +16,10 @@ public function print(int $id, string $name, QueueContext $context): void // check non-blocking EventLoop::delay($delay, static function () use ($id, $name, $context): void { echo sprintf( - "task: %d, %s, route: %s, date: %s\r\n", + "task: %d, %s, attempt: %d, route: %s, date: %s\r\n", $id, $name, + $context->attempt, $context->routingKey, $context->createdAt, );