Battements de cœur

Les tâches de longue durée et les étapes de flux peuvent envoyer des battements de cœur pour signaler qu’elles sont toujours actives. Sans ces battements, le détecteur de tâches obsolètes peut récupérer une tâche qui est toujours en cours de traitement, car son horodatage reserved_at n’a pas été mis à jour.

Fonctionnement

Lorsqu’un processus de travail commence à traiter une tâche, il appelle Heartbeat::init() pour définir le contexte de la tâche en cours. Dans votre gestionnaire, l’appel à Heartbeat::beat() met à jour la colonne reserved_at de la tâche avec NOW(). Le détecteur de tâches obsolètes utilise reserved_at pour déterminer si une tâche est bloquée ; son actualisation la protège donc.

Lorsque la tâche se termine, le processus de travail appelle Heartbeat::clear() pour réinitialiser le contexte.

Envoyer des battements de cœur

Appelez Heartbeat::beat() depuis n’importe quel gestionnaire de tâche ou étape de flux :

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

Données de progression

Le paramètre $progress est un tableau associatif facultatif. S’il est fourni, il est encodé en JSON et stocké dans la colonne heartbeat_data de la ligne de la tâche. Cela vous permet de suivre la progression depuis l’extérieur du processus de travail :

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

Appelé sans argument, seul reserved_at est mis à jour :

Heartbeat::beat();

API de battement de cœur

La classe Queuety\Heartbeat est entièrement statique. Vous ne l’instanciez pas.

Utilisation dans les étapes de flux

Le processus de travail envoie déjà un battement de cœur après chaque bloc qu’il enregistre, avec heartbeat_data défini sur {"streaming_chunks": n} ; un flux qui fournit régulièrement des données reste donc à jour de lui-même. Appelez vous-même Heartbeat::beat() lorsqu’il peut y avoir de longues pauses entre les données, par exemple en attendant le premier jeton. Les données que vous transmettez sont écrasées par le battement du processus de travail suivant, après chaque bloc :

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

Quand utiliser les battements de cœur

Utilisez des battements de cœur lorsqu’une seule tâche ou étape dure plus de QUEUETY_STALE_TIMEOUT secondes (par défaut : 600). Cas courants :

• Appels LLM API avec réponses en flux continu
• Téléchargements ou envois de fichiers volumineux
• Traitement par lots répété au sein d’une même étape
• Tout appel externe API pouvant prendre plusieurs minutes avant de répondre

Si votre étape se termine en moins d’une minute, les signaux périodiques sont inutiles. Le détecteur d’inactivité ne se déclenche qu’une fois le délai configuré écoulé.