zukucode
主にWEB関連の情報を技術メモとして発信しています。
ASP.NET

ASP.NET Core Transactional OutboxとAzure Queueで処理を確実に実行する

ASP.NET Coreでデータベースを更新したあとに、Webhookの送信などの外部処理を確実に実行する方法を紹介します。

データベースの更新と外部処理を続けて実行すると、データベースの更新だけが成功して、外部処理が実行されない場合があります。

この問題を防ぐために、Transactional OutboxパターンとAzure Queue Storageを使用します。

データベース更新後に外部処理を実行する場合の問題

最初は以下のように、データベースの更新が完了したあとに外部処理を実行する方法が考えられます。

更新処理の例
await database.SaveChangesAsync(cancellationToken);
await webhookClient.SendAsync(message, cancellationToken);

しかし、データベースの更新が完了した直後にサーバーが停止した場合や、通信エラーが発生した場合は、データベースだけが更新された状態になります。

外部処理を先に実行すると、外部処理が成功したあとにデータベースの更新が失敗する可能性があります。

データベースと外部サービスを、同じトランザクションで更新することはできません。

Outboxテーブルにイベントを保存する

業務データを更新するトランザクションの中で、あとから実行する処理をOutboxテーブルにも保存します。

OutboxMessageの保存
await using var transaction = await database.BeginTransactionAsync(cancellationToken);

await orderRepository.UpdateAsync(order, cancellationToken);

var outboxMessage = new OutboxMessage
{
    OutboxMessageId = Guid.NewGuid(),
    EventType = "OrderCompleted",
    Payload = JsonSerializer.Serialize(new { order.OrderId }),
    Status = "Pending",
    AttemptCount = 0,
    NextAttemptAt = DateTimeOffset.UtcNow,
};

await outboxRepository.AddAsync(outboxMessage, cancellationToken);
await transaction.CommitAsync(cancellationToken);

業務データとOutboxの保存は、同じデータベーストランザクションで実行します。

そのため、両方が保存されるか、両方とも保存されないかのどちらかになります。

Outboxに保存した時点では、Webhookなどの外部処理はまだ実行しません。

Azure Queueは処理開始の通知に使用する

トランザクションをコミットしたあとに、Azure Queue Storageへメッセージを送信します。

Queueへの通知
await transaction.CommitAsync(cancellationToken);

try
{
    await queueClient.SendMessageAsync(
        BinaryData.FromObjectAsJson(new { RequestedAt = DateTimeOffset.UtcNow }),
        cancellationToken
    );
}
catch (Exception ex)
{
    logger.LogWarning(ex, "Outboxの処理開始通知に失敗しました。");
}

ここで送信するメッセージには、Webhookの内容などの業務データを含めません。

Azure Queue Storageは、Outboxに未処理のデータがあることをWorkerへ知らせるためだけに使用します。

Queueへの送信が失敗しても、すでにコミットされたデータベースの更新を失敗扱いにはしません。

処理対象のデータはOutboxテーブルに残っているため、あとから再度取得できます。

未処理のOutboxを取得する

複数のWorkerが同時に実行される場合は、同じメッセージを同時に取得しないようにする必要があります。

PostgreSQLでは、FOR UPDATE SKIP LOCKEDを使用して処理対象を取得できます。

Outbox取得SQL
UPDATE outbox_messages
SET
    status = 'Processing',
    attempt_count = attempt_count + 1,
    processing_started_at = @Now,
    lease_expires_at = @LeaseExpiresAt
WHERE outbox_message_id IN (
    SELECT outbox_message_id
    FROM outbox_messages
    WHERE
        (status = 'Pending' AND next_attempt_at <= @Now)
        OR
        (status = 'Processing' AND lease_expires_at <= @Now)
    ORDER BY next_attempt_at, created_at
    FOR UPDATE SKIP LOCKED
    LIMIT 100
)
RETURNING *;

SKIP LOCKEDを指定すると、ほかのWorkerがロックしている行を待たずにスキップします。

また、取得したデータには処理期限を設定します。

Workerが処理中に停止してProcessingのまま残った場合も、処理期限を過ぎたデータを別のWorkerが再取得できます。

処理結果をOutboxに記録する

外部処理が成功した場合は、Processedに更新します。

Outboxの処理
try
{
    await eventPublisher.PublishAsync(outboxMessage, cancellationToken);

    outboxMessage.Status = "Processed";
    outboxMessage.ProcessedAt = DateTimeOffset.UtcNow;
    outboxMessage.LastError = null;
}
catch (HttpRequestException ex)
{
    outboxMessage.Status = "Pending";
    outboxMessage.NextAttemptAt = CalculateNextAttemptAt(
        outboxMessage.AttemptCount,
        DateTimeOffset.UtcNow
    );
    outboxMessage.LastError = ex.Message;
}

await outboxRepository.UpdateAsync(outboxMessage, cancellationToken);

通信エラーやタイムアウトなど、時間を置けば成功する可能性があるエラーは再試行します。

イベントの種類が存在しない場合や、保存したJSONを読み込めない場合など、再試行しても直らないエラーはFailedとして保存します。

失敗したメッセージを削除せずに残すことで、あとから原因を調査できます。

指数バックオフで再試行する

外部サービスで障害が発生しているときに短い間隔で再試行すると、外部サービスへさらに負荷をかけてしまいます。

以下のように、試行回数が増えるごとに再試行までの時間を延ばします。

再試行日時の計算
private static DateTimeOffset CalculateNextAttemptAt(
    int attemptCount,
    DateTimeOffset now
)
{
    var delaySeconds = Math.Min(
        TimeSpan.FromMinutes(30).TotalSeconds,
        Math.Pow(2, attemptCount - 1) * 10
    );

    return now.AddSeconds(delaySeconds);
}

この例では、最初は10秒後に再試行し、20秒、40秒というように間隔を延ばします。

再試行間隔は最大30分とし、試行回数が上限を超えた場合はFailedにします。

Timer Triggerでも定期的に処理する

Queueへの通知が失敗すると、Queue TriggerだけではOutboxが処理されません。

そのため、Queue Triggerに加えてTimer Triggerでも定期的に未処理データを確認します。

OutboxTimerFunction.cs
public sealed class OutboxTimerFunction(OutboxProcessor processor)
{
    [Function(nameof(OutboxTimerFunction))]
    public Task RunAsync(
        [TimerTrigger("0 */5 * * * *")] TimerInfo timer,
        CancellationToken cancellationToken
    )
    {
        return processor.ExecuteAsync(cancellationToken);
    }
}

通常はQueue Triggerによってすぐに処理されます。

Queueへの通知が失敗した場合も、Timer Triggerが5分ごとに未処理データを回収します。

Queueは処理を早く開始するために使用し、処理の確実性はデータベースのOutboxで保証します。

注意点

この方法では、Workerが外部処理に成功した直後、OutboxをProcessedに更新する前に停止すると、同じ処理が再実行される可能性があります。

そのため、Webhookの受信側などは、同じイベントIDを複数回受け取っても問題がないように実装する必要があります。

Transactional Outboxを使用しても、外部処理を必ず1回だけ実行できるわけではありません。

少なくとも1回実行されることを前提にして、イベントIDや冪等性キーを使用して重複処理を防ぎます。

ASP.NET

関連記事