Heartbeat

I job di lunga durata e i passaggi in streaming possono inviare heartbeat per dimostrare che sono ancora attivi. Senza heartbeat, il rilevatore di job obsoleti potrebbe riprendere un job ancora in elaborazione perché il suo timestamp reserved_at non è stato aggiornato.

Come funziona

Quando un worker inizia a elaborare un job, chiama Heartbeat::init() per impostare il contesto del job corrente. Nel gestore, chiamando Heartbeat::beat() si aggiorna la colonna reserved_at del job impostandola su NOW(). Il rilevatore di job obsoleti usa reserved_at per stabilire se un job è bloccato; aggiornarlo lo mantiene al sicuro.

Quando il job termina, il worker chiama Heartbeat::clear() per reimpostare il contesto.

Invio di heartbeat

Chiamate Heartbeat::beat() all'interno di qualsiasi gestore di job o passaggio in streaming:

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

Dati di avanzamento

Il parametro $progress è un array associativo facoltativo. Se fornito, viene codificato in JSON e memorizzato nella colonna heartbeat_data della riga del job. Questo permette di monitorare l'avanzamento dall'esterno del worker:

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

Se chiamato senza argomenti, viene aggiornato solo reserved_at:

Heartbeat::beat();

API degli heartbeat

La classe Queuety\Heartbeat è interamente statica. Non va istanziata.

Uso nei passaggi in streaming

Il worker invia già un heartbeat dopo ogni blocco salvato, con heartbeat_data impostato su {"streaming_chunks": n}, quindi uno stream che produce dati regolarmente resta aggiornato da solo. Chiamate Heartbeat::beat() voi stessi se possono esserci lunghe pause tra un blocco e l'altro, ad esempio mentre aspettate il primo token. I dati passati vengono sovrascritti dall'heartbeat successivo del worker per ogni blocco:

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

Quando usare gli heartbeat

Usate gli heartbeat quando un singolo job o passaggio dura più di QUEUETY_STALE_TIMEOUT secondi (predefinito: 600). Casi comuni:

• Chiamate LLM API con risposte in streaming
• Download o caricamenti di file di grandi dimensioni
• Ripetizioni di elaborazione in batch all'interno di un singolo passaggio
• Qualsiasi chiamata esterna a API che potrebbe richiedere minuti per rispondere

Se il passaggio termina in meno di un minuto, gli heartbeat non sono necessari. Il rilevatore di job obsoleti si attiva solo una volta trascorso il timeout configurato.