# ハートビート

長時間実行するジョブやストリーミングステップは、ハートビートを送信して、処理が継続中であることを示せます。ハートビートがないと、`reserved_at`タイムスタンプが更新されていないため、まだ処理中のジョブが古いジョブの検出機能によって再取得されることがあります。

## 仕組み

ワーカーはジョブの処理を開始すると、`Heartbeat::init()`を呼び出して現在のジョブコンテキストを設定します。ハンドラー内で`Heartbeat::beat()`を呼び出すと、ジョブの`reserved_at`列が`NOW()`に更新されます。古いジョブの検出機能は`reserved_at`を使ってジョブが停止しているか判断するため、これを更新すればジョブは安全に処理されます。

ジョブが終了すると、ワーカーは`Heartbeat::clear()`を呼び出してコンテキストをリセットします。

## ハートビートの送信

任意のジョブハンドラーまたはストリーミングステップ内から`Heartbeat::beat()`を呼び出します:

```php
use Queuety\Heartbeat;

class LongRunningStep implements \Queuety\Step {

    public function handle( array $state ): array {
        $items = $state['items'];

        foreach ( $items as $i => $item ) {
            $this->process( $item );
            Heartbeat::beat( [ 'processed' => $i + 1, 'total' => count( $items ) ] );
        }

        return [ 'done' => true ];
    }

    public function config(): array {
        return [];
    }

    private function process( $item ): void {
        // Expensive work here.
    }
}
```

## 進捗データ

`$progress`パラメーターは任意の連想配列です。指定するとJSONエンコードされ、ジョブレコードの`heartbeat_data`列に保存されます。これにより、ワーカーの外部から進捗を追跡できます:

```php
Heartbeat::beat( [
    'processed' => 450,
    'total'     => 1000,
    'percent'   => 45,
] );
```

引数なしで呼び出すと、`reserved_at`のみが更新されます:

```php
Heartbeat::beat();
```

## ハートビートAPI

`Queuety\Heartbeat`クラスは完全に静的です。インスタンス化する必要はありません。

| メソッド | 説明 |
|---|---|
| `Heartbeat::init( int $job_id, Connection $conn )` | ジョブコンテキストを設定します。処理の前にワーカーによって自動的に呼び出されます。 |
| `Heartbeat::beat( array $progress = [] )` | ハートビートを送信します。`reserved_at`を更新し、必要に応じて進捗データを保存します。 |
| `Heartbeat::clear()` | ジョブコンテキストをクリアします。処理の後にワーカーによって自動的に呼び出されます。 |
| `Heartbeat::current_job_id()` | 現在のジョブIDを取得します（テストに便利です）。 |

## ストリーミングステップでの使用

ワーカーは、永続化する各チャンクの後に、`heartbeat_data`を`{"streaming_chunks": n}`に設定してすでにハートビートを送信するため、定期的にデータを返すストリームはそのまま最新の状態に保たれます。データが返るまで長い間隔が空く場合（たとえば最初のトークンを待っている間）は、`Heartbeat::beat()`を呼び出してください。渡したデータは、ワーカーが次のチャンクごとのハートビートを送信すると上書きされます。

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamAPIResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $response = $this->client->stream( $state['prompt'] );

        foreach ( $response as $i => $chunk ) {
            Heartbeat::beat( [ 'chunks' => $i + 1 ] );
            yield $chunk;
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [ 'response' => implode( '', $chunks ) ];
    }

    public function config(): array {
        return [];
    }
}
```

## ハートビートを使う場面

1つのジョブまたはステップが`QUEUETY_STALE_TIMEOUT`秒（デフォルト: 600秒）を超えて実行される場合に、ハートビートを使います。よくあるケース:

- LLMのAPI呼び出しでストリーミング応答を使う場合
- 大容量ファイルのダウンロードやアップロード
- 1つのステップ内でのバッチ処理の繰り返し
- 応答に数分かかることがある外部API呼び出し

ステップが1分未満で完了する場合、ハートビートは不要です。古いジョブの検出機能は、設定されたタイムアウトが経過してから動作します。