Waymark: 障害に耐える堅牢なバックグラウンドタスク処理ライブラリ

Waymark は、サーバー再起動やタスククラッシュ、長時間実行ジョブといった障害にも耐えうる堅牢なバックグラウンドタスクを構築することを可能にするライブラリです。 Postgres を基盤としており、追加のインフラストラクチャを展開する必要がありません。現在は Python がサポートされていますが、今後も対応言語を追加していく予定です。

• ドキュメント: waymark.sh
• 特徴:
  • Python で動作する簡易なインストールからワークフロー定義・実行まで一貫性のある体験を提供
  • 再試行ポリシー、スケジュール管理、AI エージェントとの連携など高度な機能を実装

基本要素：ワークフローとアクション

Waymark は制御フロー（ワークフロー）と分散処理（アクション）の 2 つの基本要素で構成されます。

1. アクション (Action)

ネットワーク呼び出しやデータベースクエリなど、失敗しても独立して再試行すべき処理を担当します。

• 定義: デコレータ

  @action

  を使用して単一の非同期関数として記述
• 依存注入:

  Depends

  を使って外部サービス（DB、メール送信用クライアント等）を簡潔に注入可能

例：アクションの定義

from typing import Annotated
from waymark import Depends, action

@action
async def fetch_users(
    user_ids: list[str],
    db: Annotated[Database, Depends(get_db)],
) -> list[User]:
    return await db.get_many(User, user_ids)

@action
async def send_email(
    to: str,
    subject: str,
    emailer: Annotated[EmailClient, Depends(get_email_client)],
) -> EmailResult:
    return await emailer.send(to=to, subject=subject)

2. ワークフロー (Workflow)

アクションをオーケストレーションする堅牢な制御フローを担当します。

• 役割: ループ、条件分岐、並列処理などのロジックを実行
• 実行方法:

  run()

  メソッドで定義された非同期関数として記述

例：ワークフローの実装

import asyncio
from waymark import Workflow, workflow

@workflow
class WelcomeEmailWorkflow(Workflow):
    async def run(self, user_ids: list[str]) -> list[EmailResult]:
        users = await fetch_users(user_ids)
        active_users = [user for user in users if user.active]

        # 並列実行（失敗した場合は例外を返さない設定）
        results = await asyncio.gather(
            *[
                send_email(to=user.email, subject="Welcome")
                for user in active_users
            ],
            return_exceptions=False,
        )

        return results

環境設定と実行方法

1. コンフィグレーション

必要な環境変数を設定します。

export WAYMARK_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/waymark
export WAYMARK_HTTP_ENABLED=true # ウェブアプリを起動 (http://localhost:24119)
uv run waymark-executor

2. ワークフローの起動

任意の Python プロセスからワークフローを実行し、Postgres にキューイングされます。

例：ユーザーにウェルカムメールを送信

async def welcome_users(user_ids: list[str]):
    await WelcomeEmailWorkflow().run(user_ids)

3. 仕組みの概要

• アクション: 分散処理を担当（ネットワーク呼び出し、DB クエリ等）。失敗しても独立して再試行可能。
• ワークフロー: 制御フローを担当（ループ、分岐、並列ブランチ）。重い処理は回避され、Postgres に状態が保存されるためクラッシュ後も回復可能。

機能と実装可能性

ワークフローは単なる非同期 Python コードとして記述でき、以下のような高度な制御が可能です。

呼び出しごとの再試行とタイムアウト

self.run_action(...)

from datetime import timedelta
from waymark import RetryPolicy

async def run(self, order_id: str) -> Receipt:
    return await self.run_action(
        charge_card(order_id),
        retry=RetryPolicy(attempts=5, backoff_seconds=10),
        timeout=timedelta(minutes=2),
    )

ブランチとループ

Python 標準の制御構造はそのまま動作します。

• if/elif/else

• for

  ループ

• while

  ループ

例：配送済みオーダーのフィルタリング

async def run(self, order_ids: list[str]) -> list[str]:
    shipped = []
    for order_id in order_ids:
        status = await fetch_status(order_id)
        if status == "shipped":
            shipped.append(order_id)
    return shipped

並列なファンアウトと堅牢なスリープ

• asyncio.gather

  で複数のアクションを並列実行。

• asyncio.sleep

  でワークフローを一時的に停止し、再起動後も状態を保持可能（1 秒〜1 日の範囲で）。

例：プロフィール取得後の長期待機と推奨送信

import asyncio

async def run(self, user_id: str) -> str:
    profile, history = await asyncio.gather(
        fetch_profile(user_id),
        fetch_purchase_history(user_id),
        return_exceptions=False,
    )

    # 1 日後にフォローアップのため待機（中断・再開に対応）
    await asyncio.sleep(24 * 60 * 60)
    return await send_recommendations(profile, history)

スケジュールされたワークフロー

Cron 表現や固定間隔で自動実行できます。追加のインフラ構築不要です。

await schedule_workflow(DataSyncWorkflow, schedule_name="hourly", schedule="0 * * * *")

組み込みウェブアプリ

waymark-executor

• 可視化: 各インスタンスの呼び出し履歴やノードへの負荷状況を確認可能。
• URL: デフォルト

  http://localhost:24119

フィロソフィーと設計思想

バックグラウンドジョブはフルスタックアプリケーションの基本要素（DB、バックエンド、フロントエンドの横並び）として捉えています。 これにより、ユーザーが API にブロッキングリクエストを送信するか、偶発的なクラッシュでタスクが失われるという状況を回避できます。

設計の 3 つの原則

1. 言語との親和性: 既存の開発言語（現在は Python）で通常のコードとして制御フローが記述できること。
2. 一貫性のある体験: ローカルでのテストとリモートでの展開において同じように動作すること。
3. 合理的なデフォルト: パフォーマンスチューニングなしに、適度なリクエストボリュームまで拡張可能であること。

データフロー処理エンジンとしての位置づけ

• DAG 定義の非制限: 装飾子や独自構文による制限ではなく、通常の Python コード（ループ、条件分岐）がそのまま機能します。
• スケーラビリティの目標: 負荷パフォーマンスが最優先というわけではありませんが、アプリケーションを製品市場適合（PMF）を超えて適切にスケールできるように設計されています。

動作原理：再実行ベースではないアプローチ

Waymark は Temporal や Vercel Workflow のように「コードの再実行」を採用するのではなく、「一度のコンパイルと状態管理」を採用しています。

アプローチの違い

random()

datetime.now()

実行フローの詳細

1. コンパイル: 初回実行時に Python のソースコード（AST）を解析し、中間表現（IR）に変換。さらに堅牢な仮想マシン用のバイトコードへ変換される。
2. 実行: Rust ランタイムがバイトコードを実行。進行状況は Postgres にスナップショットとして保存される。
3. 障害からの回復: サーバーが再起動しても、Postgres の状態から正確な位置で再開する。ワークフロー定義の再実行は行わないため、非決定論的な処理でも安定動作します。

メリット:

random()

や現在時刻などのハッキングが不要。コンパイルエラーを防ぐための複雑な工夫が不要です。

使用すべきか？（適性・不適性）

✅ おすすめの場合

• 既に Postgres とサポート言語（Python など）を使用しており、追加のフレームワーク構築を避けたい場合
• アクションが多い非同期ロジックで、失敗時の自動再試行が必要な場合（3 者 API キャンセル等）
• ローカル開発と本番環境で同じコードをそのまま動かしたい場合
• 既存のユニットテストや静的解析ツールにバックグラウンドジョブのコードを統合したい場合

❌ おすすめではない場合

• 低遅延が必須で、各タスクに対して

  <100ms

  の ACK とハンドリングが必要な場合
• 大規模な並列処理で、同時に数万以上のアクションを調整する必要がある場合
• Temporal.io や RabbitMQ といった専用タスクブローカーを使用し、さらにスケールさせる必要がある場合

補足: パフォーマンスは Rust コアを採用したことで保証されていますが、Postgres の ACID プロパティには限界があります。Postgres を過負荷にすると、より複雑なアーキテクチャへの移行を検討する必要があります。 Temporal や RabbitMQ などの代替案も優秀ですが、専用のブローカーを別サーバーで運用する手間や、イベント課金コスト、あるいはオープンソースの歴史的背景など、考慮すべき要素が異なります。

プロジェクトステータス

重要: Waymark は現在 プレリリース 段階です。

• 本番アプリケーションでの使用は推奨されていません。
• バージョン 1.0.0 以前は後方互換性を保証しておりません。
• ただし、サイドプロジェクト等で試す際は歓迎します。

コンパイルエラーが発生する特定のワークフローについて Issue を作成いただければ、解決に向けた対応を行います。

ソースからのビルドと貢献

ビルド要件

Web アプリ機能（

waymark-executor

make js-deps
cargo build --bin waymark-executor

コントリビューション

より詳細なガイドはコントリビューションガイドラインを参照してください。