Skip to content

workflow await #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kuaukutsu merged 3 commits into from
Nov 20, 2025
Merged

workflow await #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5796544
workflow await
kuaukutsu Nov 19, 2025
786ccb8
order call
kuaukutsu Nov 19, 2025
a65414a
context timeout
kuaukutsu Nov 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,12 @@ composer require kuaukutsu/poc-queue-stream
- Если сообщения в хранилище нет, то выходим с ошибкой, стрим переносим в dead letter queue (DLQ).
- Если сообщение получено, но выполнение завершилось ошибкой, отправляем на повторный круг — точнее, оставляем в очереди (PEL); свободный консумер получит это сообщение позже через **XAUTOCLAIM** и попробует выполнить ещё раз. Таких попыток по умолчанию будет 2; после третьей попытки сообщение переносим в DLQ.
- Если для консумера задан обработчик исключений (`catch exception`), то обработка ошибок лежит на клиентском ПО, т. е. механизм DLQ работать не будет.

Сомнительно, но окей (?):

- Прочитанные сообщения «акаются» (**ACK**) пачками; также вместе с командой ACK удаляются сообщения из хранилища.
Альтернатива: акаем по завершению таска, с одной стороны убираем await и подтверждаем здесь и сразу,
с другой сокращаем количество запросов.

#### чтиво

Expand Down
2 changes: 1 addition & 1 deletion composer.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": "*",
Expand Down
8 changes: 4 additions & 4 deletions src/internal/stream/RedisStreamGroup.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public function create(): bool
}

/**
* @return ?array<array{0: non-empty-string, 1: array<non-empty-string, string[]>}>
* @return ?array<array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}>
* @see https://redis.io/docs/latest/commands/xreadgroup/
*/
public function read(): ?array
Expand All @@ -82,7 +82,7 @@ public function read(): ?array

if (is_array($result)) {
/**
* @var array<array{0: non-empty-string, 1: array<non-empty-string, string[]>}>
* @var array<array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}>
*/
return $result;
}
Expand All @@ -91,7 +91,7 @@ public function read(): ?array
}

/**
* @return null|array<empty>|array{0: non-empty-string, 1: array<non-empty-string, string[]>}
* @return null|array<empty>|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
Expand All @@ -109,7 +109,7 @@ public function autoclaim(string $start = '0-0'): ?array

if (is_array($result)) {
/**
* @var array<empty>|array{0: non-empty-string, 1: array<non-empty-string, string[]>}
* @var array<empty>|array{0: non-empty-string, 1: array{0: non-empty-string, 1: string[]}}
*/
return $result;
}
Expand Down
39 changes: 33 additions & 6 deletions src/internal/workflow/TaskRunner.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down
67 changes: 41 additions & 26 deletions src/internal/workflow/WorkflowClaim.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -23,45 +26,57 @@ 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();
}
}

/**
* @return iterable<non-empty-string, Payload>
*/
private function autoclaim(RedisStreamGroup $command): iterable
{
$fn = static function (RedisStreamGroup $command): iterable {
$batch = $command->autoclaim();
if ($batch === null) {
return [];
}

/**
* @var array<array<non-empty-string, string[]>> $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<non-empty-string, Payload>
* @var array<array{0: non-empty-string, 1: string[]}> $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 [];
}
}
66 changes: 34 additions & 32 deletions src/internal/workflow/WorkflowMain.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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();
}
}
}
Expand All @@ -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<array<non-empty-string, string[]>> $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<non-empty-string, Payload>
* @var array<array{0: non-empty-string, 1: string[]}> $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);
}
}
}
3 changes: 2 additions & 1 deletion tests/simulation/publisher-with-exception.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -29,4 +30,4 @@
],
);

$publisher->push($schema, $task);
$publisher->push($schema, $task, QueueContext::make($schema)->withTimeout(30));
8 changes: 4 additions & 4 deletions tests/simulation/publisher.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -46,6 +44,8 @@
'name' => 'test range',
],
),
QueueContext::make($schema)->withExternal(['requestId' => $item])
QueueContext::make($schema)
->withExternal(['requestId' => $item])
->withTimeout(300)
);
}