message increment attempts

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": "*",

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}

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);
         }

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

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,

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<non-empty-string, Payload>
+         */
         $fn = static function (RedisStreamGroup $command): iterable {
             $batch = $command->autoclaim();
             if ($batch === []) {
@@ -78,7 +81,7 @@ private function autoclaim(RedisStreamGroup $command): iterable
         };
 
         /**
-         * @var iterable<non-empty-string, Payload>
+         * @phpstan-var iterable<non-empty-string, Payload>
          */
         return async($fn(...), $command)->await();
     }

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<non-empty-string, Payload>
+         */
         $fn = static function (RedisStreamGroup $command): iterable {
             $batch = $command->read();
             if ($batch === []) {
@@ -86,7 +89,7 @@ private function read(RedisStreamGroup $command): iterable
         };
 
         /**
-         * @var iterable<non-empty-string, Payload>
+         * @phpstan-var iterable<non-empty-string, Payload>
          */
         return async($fn(...), $command)->await();
     }

tests/simulation/publisher.php

@@ -44,7 +44,6 @@
             ],
         ),
         QueueContext::make($schema)
-            ->withExternal(['requestId' => $item])
             ->withTimeout(300)
     );
 }

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);
     }
 }

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,
             );