Heartbeats

Als Markdown anzeigen Auf GitHub bearbeiten Dokumentation zu 2cc7b2b

Lang laufende Jobs und Streaming-Schritte können Heartbeats senden, um zu signalisieren, dass sie noch aktiv sind. Ohne Heartbeats kann die Erkennung veralteter Jobs einen Job zurückfordern, der noch verarbeitet wird, weil sein Zeitstempel reserved_at nicht aktualisiert wurde.

So funktioniert es

Wenn ein Worker mit der Verarbeitung eines Jobs beginnt, ruft er Heartbeat::init() auf, um den aktuellen Job-Kontext festzulegen. Innerhalb deines Handlers aktualisiert ein Aufruf von Heartbeat::beat() die Spalte reserved_at des Jobs auf NOW(). Die Erkennung veralteter Jobs verwendet reserved_at, um festzustellen, ob ein Job feststeckt. Eine Aktualisierung hält ihn daher aktiv.

Wenn der Job abgeschlossen ist, setzt der Worker den Kontext mit Heartbeat::clear() zurück.

Heartbeats senden

Rufe Heartbeat::beat() innerhalb eines beliebigen Job-Handlers oder Streaming-Schritts auf:

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

Fortschrittsdaten

Der Parameter $progress ist ein optionales assoziatives Array. Wenn du ihn angibst, wird er mit JSON kodiert und in der Spalte heartbeat_data des Job-Eintrags gespeichert. So kannst du den Fortschritt außerhalb des Workers verfolgen:

PHP

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

Ohne Argumente wird nur reserved_at aktualisiert:

PHP

Heartbeat::beat();

Heartbeat-API

Die Klasse Queuety\Heartbeat besteht vollständig aus statischen Methoden. Du musst sie nicht instanziieren.

Methode	Beschreibung
`Heartbeat::init( int $job_id, Connection $conn )`	Legt den Job-Kontext fest. Wird vor der Verarbeitung automatisch vom Worker aufgerufen.
`Heartbeat::beat( array $progress = [] )`	Sendet einen Heartbeat. Aktualisiert `reserved_at` und speichert optional Fortschrittsdaten.
`Heartbeat::clear()`	Löscht den Job-Kontext. Wird nach der Verarbeitung automatisch vom Worker aufgerufen.
`Heartbeat::current_job_id()`	Gibt die aktuelle Job-ID zurück (nützlich für Tests).

Verwendung in Streaming-Schritten

Der Worker sendet nach jedem gespeicherten Chunk bereits einen Heartbeat mit heartbeat_data auf {"streaming_chunks": n}. Ein Stream, der regelmäßig Daten ausgibt, bleibt dadurch automatisch aktuell. Rufe Heartbeat::beat() selbst auf, wenn zwischen den Ausgaben längere Pausen liegen können, zum Beispiel beim Warten auf das erste Token. Übergebene Daten werden beim nächsten Heartbeat pro Chunk vom Worker überschrieben:

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

Wann Heartbeats sinnvoll sind

Verwende Heartbeats, wenn ein einzelner Job oder Schritt länger als QUEUETY_STALE_TIMEOUT Sekunden läuft (Standard: 600). Häufige Fälle:

LLM-API-Aufrufe mit Streaming-Antworten
Große Datei-Downloads oder -Uploads
Wiederholte Batchverarbeitung innerhalb eines einzelnen Schritts
Alle externen API-Aufrufe, deren Antwort mehrere Minuten dauern kann

Wenn dein Schritt in weniger als einer Minute abgeschlossen ist, brauchst du keine Heartbeats. Die Erkennung veralteter Jobs greift erst, wenn das konfigurierte Zeitlimit abgelaufen ist.