Heartbeat

Pekerjaan berdurasi panjang dan langkah streaming dapat mengirim heartbeat untuk membuktikan bahwa prosesnya masih berjalan. Tanpa heartbeat, pendeteksi pekerjaan usang dapat mengambil kembali pekerjaan yang masih aktif diproses karena stempel waktu reserved_at belum diperbarui.

Cara kerjanya

Saat worker mulai memproses pekerjaan, worker memanggil Heartbeat::init() untuk menetapkan konteks pekerjaan saat ini. Di dalam handler, pemanggilan Heartbeat::beat() memperbarui kolom reserved_at pekerjaan menjadi NOW(). Pendeteksi pekerjaan usang menggunakan reserved_at untuk menentukan apakah pekerjaan macet, sehingga memperbaruinya menjaga pekerjaan tetap aman.

Saat pekerjaan selesai, worker memanggil Heartbeat::clear() untuk mengatur ulang konteks.

Mengirim heartbeat

Panggil Heartbeat::beat() dari dalam handler pekerjaan atau langkah streaming mana pun:

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.
    }
}

Data progres

Parameter $progress adalah array asosiatif opsional. Jika diberikan, parameter ini dienkode dalam JSON dan disimpan di kolom heartbeat_data pada baris pekerjaan. Dengan begitu, Anda dapat melacak progres dari luar worker:

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

Jika dipanggil tanpa argumen, hanya reserved_at yang diperbarui:

Heartbeat::beat();

API Heartbeat

Kelas Queuety\Heartbeat sepenuhnya statis. Anda tidak perlu membuat instansinya.

Penggunaan dalam langkah streaming

Worker sudah mengirim heartbeat setelah setiap potongan yang disimpan, dengan heartbeat_data diatur ke {"streaming_chunks": n}, sehingga stream yang menghasilkan data secara berkala akan tetap segar dengan sendirinya. Panggil Heartbeat::beat() sendiri jika ada jeda panjang di antara hasil, misalnya saat menunggu token pertama. Data yang Anda kirim akan ditimpa oleh heartbeat worker berikutnya pada tiap potongan:

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 [];
    }
}

Kapan menggunakan heartbeat

Gunakan heartbeat saat satu pekerjaan atau langkah berjalan lebih dari QUEUETY_STALE_TIMEOUT detik (default: 600). Kasus umum:

• Panggilan API LLM dengan respons streaming
• Mengunduh atau mengunggah file besar
• Pengulangan pemrosesan batch dalam satu langkah
• Panggilan API eksternal apa pun yang mungkin membutuhkan waktu beberapa menit untuk merespons

Jika langkah Anda selesai dalam waktu kurang dari satu menit, heartbeat tidak diperlukan. Pendeteksi pekerjaan usang baru aktif setelah batas waktu yang dikonfigurasi berlalu.