Outbox

Dispatch、Commit、Relay、Worker、Retryを一続きで実行し、at-least-once境界を確認する。

Transactional Outboxは、業務MutationとDeferred child Operationの発行を同じNamed ConnectionのTransactionへ記録する公開済みExperimental Stable 1.2.1のSurfaceです。External BrokerやExactly Onceは提供しません。Releasesを確認してから利用します。

登録から配送までの流れ

業務変更とOutbox Rowは、同じNamed ConnectionのTransactionで確定します。 Commit後にRelayがOutbox Rowを取得してPostgreSQLのDurable Transportへ送り、Workerがchild Operationを実行します。 Relayの送信成功はchild Handlerの完了を意味しません。 RelayとWorkerはat-least-onceで動くため、Applicationで副作用の重複に備えます。

Root Operationの業務変更とOutbox登録を同じTransactionでCommitし、Relay、PostgreSQLのDurable Transport、Worker、child Operationの順に配送する。Commit前にchild Handlerは実行されず、重複への対応はApplicationが担う。

DispatchからCommitまで

Root OperationのHandler内でOperations::dispatch()を呼び、childを登録します。業務変更とOutbox Rowは同じTransactionでCommitされ、Commit前にchild Handlerは実行されません。既存のCommunity Boardで動作する最小Recipeを、ApplicationのProject Rootへ次の配置で作ります。BoardService、Repository実装、認証PolicyはApplication所有であり、Frameworkが生成するものではありません。

app/Feature/Comment/AddComment/AddComment.php
app/Feature/Comment/AddComment/AddCommentValue.php
app/Feature/Comment/AddComment/CommentAdded.php
app/Feature/Notification/NotifyPostOwner/NotifyPostOwner.php
app/Feature/Notification/NotifyPostOwner/NotifyPostOwnerValue.php
app/Feature/Notification/NotifyPostOwner/NotificationDelivered.php
app/Domain/Board/BoardRepository.php
app/Infrastructure/Persistence/DoctrineBoardRepository.php

AddCommentValue.phpはHTTP入力、CommentAdded.phpはRoot Outcomeです。

<?php
declare(strict_types=1);

namespace App\Feature\Comment\AddComment;

use BlackOps\Core\OperationValue;
use BlackOps\Core\Validation\Attribute\Length;
use BlackOps\Core\Validation\Attribute\NotBlank;
use BlackOps\Http\Attribute\FromBody;
use BlackOps\Http\Attribute\FromPath;

final readonly class AddCommentValue implements OperationValue
{
    public function __construct(
        #[FromPath]
        public string $postId,
        #[FromBody]
        #[NotBlank]
        #[Length(min: 1, max: 2000)]
        public string $body,
    ) {}
}

<?php
declare(strict_types=1);

namespace App\Feature\Comment\AddComment;

use BlackOps\Core\Outcome;

final readonly class CommentAdded implements Outcome
{
    public function __construct(
        public string $commentId,
        public string $postId,
        public string $createdAt,
    ) {}
}

BoardRepository.phpはDomainが必要とするPersistence境界です。SQLをOperationへ書かず、DoctrineBoardRepository.phpでこのInterfaceを実装し、Application Service ProviderでBoardRepositoryへBindingします。

<?php
declare(strict_types=1);

namespace App\Domain\Board;

use DateTimeImmutable;

interface BoardRepository
{
    public function lockPostAuthorId(string $postId): ?string;

    public function createComment(
        string $commentId,
        string $postId,
        string $authorId,
        string $body,
        DateTimeImmutable $createdAt,
    ): void;
}

AddComment.phpでMutationとDispatchを同じTransactionへ置きます。BoardServiceがRepositoryを呼び、childのValueへ必要なIDだけを渡します。

<?php
declare(strict_types=1);

namespace App\Feature\Comment\AddComment;

use App\Domain\Board\BoardService;
use App\Domain\Board\PostNotFound;
use App\Feature\BoardTime;
use App\Feature\Notification\NotifyPostOwner\NotifyPostOwner;
use App\Feature\Notification\NotifyPostOwner\NotifyPostOwnerValue;
use App\Security\AuthenticatedUser;
use App\Security\AuthenticatedUserPolicy;
use BlackOps\Core\Attribute\Authorize;
use BlackOps\Core\Attribute\OperationType;
use BlackOps\Core\Exception\OperationRejectedException;
use BlackOps\Core\ExecutionContext;
use BlackOps\Core\Operation;
use BlackOps\Database\Attribute\Transactional;
use BlackOps\Execution\Operations;
use BlackOps\Http\Attribute\Route;

#[Route(method: 'POST', path: '/posts/{postId}/comments')]
#[OperationType('board.comment.add')]
#[Authorize(AuthenticatedUserPolicy::class)]
readonly class AddComment implements Operation
{
    public function __construct(
        private BoardService $board,
        private Operations $operations,
    ) {}

    #[Transactional]
    public function handle(AddCommentValue $value, ExecutionContext $context): CommentAdded
    {
        try {
            $comment = $this->board->addComment($value->postId, AuthenticatedUser::id($context), $value->body);
        } catch (PostNotFound) {
            throw OperationRejectedException::notFound('board.post.not_found');
        }

        if ($comment->postOwnerId !== $comment->authorId) {
            $this->operations->dispatch(
                NotifyPostOwner::class,
                new NotifyPostOwnerValue($comment->postOwnerId, $comment->postId, $comment->commentId),
            );
        }

        return new CommentAdded($comment->commentId, $comment->postId, BoardTime::http($comment->createdAt));
    }
}

NotifyPostOwnerValue.php、NotificationDelivered.php、NotifyPostOwner.phpはDeferred childのValue／Outcome／Handlerです。NotifyPostOwnerへ#[OperationType]と#[Deferred]を付け、Application-owned NotificationServiceへ副作用を閉じ込めます。

NotifyPostOwnerValue.php:

<?php
declare(strict_types=1);

namespace App\Feature\Notification\NotifyPostOwner;

use BlackOps\Core\OperationValue;
use BlackOps\Core\Validation\Attribute\NotBlank;

final readonly class NotifyPostOwnerValue implements OperationValue
{
    public function __construct(
        #[NotBlank] public string $recipientUserId,
        #[NotBlank] public string $postId,
        #[NotBlank] public string $commentId,
    ) {}
}

NotificationDelivered.php:

<?php
declare(strict_types=1);

namespace App\Feature\Notification\NotifyPostOwner;

use BlackOps\Core\Outcome;

final readonly class NotificationDelivered implements Outcome
{
    public function __construct(public bool $created) {}
}

NotifyPostOwner.php:

<?php
declare(strict_types=1);

namespace App\Feature\Notification\NotifyPostOwner;

use App\Domain\Notification\NotificationService;
use BlackOps\Core\Attribute\Deferred;
use BlackOps\Core\Attribute\OperationType;
use BlackOps\Core\ExecutionContext;
use BlackOps\Core\Operation;
use BlackOps\Database\Attribute\Transactional;

#[OperationType('board.notification.notify')]
#[Deferred]
readonly class NotifyPostOwner implements Operation
{
    public function __construct(private NotificationService $notifications) {}

    #[Transactional]
    public function handle(NotifyPostOwnerValue $value, ExecutionContext $context): NotificationDelivered
    {
        return new NotificationDelivered($this->notifications->notifyPostOwner(
            $value->recipientUserId,
            $value->postId,
            $value->commentId,
            $context->operationId()->toString(),
        ));
    }
}

Operations::dispatch()の戻りはDispatch Receiptです。Root Operation IDと混同せず、必要な場合だけ$receipt->operationId()->toString()をApplication Outcomeまたは安全なStatus参照へ渡します。Canonical Payload、Outbox Record ID、Credentialは公開しません。Build、Migration、Service Binding後にphp blackops build:compileを実行します。

Commit失敗時は業務変更もOutbox Rowも残りません。Commit後にRelayが停止してもPending Rowは再開可能な状態で残ります。TransactionのNamed Connection設定はTransactionを参照してください。

RelayとWorkerを分けて実行する

Project Rootで、まずRelayを起動します。

php blackops outbox:relay:run --until-empty

有限Batchだけを処理する場合は--batches=1、常駐監督は次を使います。

php blackops outbox:relay:daemon --interval-milliseconds=1000

Relayの出力はclaimed、sent、retried、dead-lettered、staleの件数です。sentはTransportへ渡した結果であり、child Handlerの完了やOutcomeの生成を意味しません。続けてDeferred Workerを別Processで実行します。

php blackops worker:run --iterations=1 --idle-sleep-milliseconds=1

常駐Workerでは--iterationsを省略し、--idle-sleep-milliseconds（既定1000）を設定します。Workerの出力はWorker stopped. Processed claims: Nです。WorkerがchildをClaimし、Attempt、Journal、Status／Outcomeを進めます。

確認とFailure Journey

Root OperationのResponseまたはDiagnosticsからOperation IDを記録します。
Relayの件数でOutbox Rowが配送対象になったことだけを確認します。
Workerを実行し、childのoperation.accepted、attempt.started、Terminal EventをJournalで確認します。
php blackops operation:inspect <operation-id> --jsonでSafe Status／Outcomeを確認します。
Retry Scheduled、Failed、Dead LetterをCompletedと混同しない。

Relay／Workerはat-least-onceです。Lease、Fencing、Retryにより同じchild Identityが複数回配送される可能性があるため、外部副作用はIdempotency Keyまたは重複耐性をApplicationで設計します。outbox:relay:runの終了だけでHandler成功と判断しません。

Dead Letterを再開する場合は、対象Recordと監査理由を確認してからActorとReasonを明示します。

php blackops outbox:dead-letter:retry <record-id> \
  --actor=operations-admin \
  --reason='approved after provider recovery'

成功時はdead-letter retry scheduledだけを表示します。Payload、Context、SQL、Credential、Throwableは表示されません。再開後もRelayとWorkerを順に実行し、Status／Journal／Outcomeを再確認します。

External Message Broker、Exactly Once、Canonical Journalの置換は現行Capabilityではありません。Scheduled Application Operationは別のone-shot入口であり、Outbox Relayの完了とは別に扱います。Scheduled Operation、Deploymentのプロセス一覧、Deferred HTTPが202だがOutcomeがないも併読してください。

次にJournalの正本を読む

DispatchからWorkerまでの事実をどのRecordで追うかは、JournalでCanonicalとObservedを分けて確認します。