Heartbeats

Langlopende taken en streamingstappen kunnen heartbeats versturen om aan te geven dat ze nog actief zijn. Zonder heartbeats kan de detector voor verouderde taken een taak die nog wordt verwerkt opnieuw claimen, omdat de tijdstempel reserved_at niet is bijgewerkt.

Werking

Wanneer een worker een taak begint te verwerken, roept deze Heartbeat::init() aan om de huidige taakcontext in te stellen. In uw handler werkt u met Heartbeat::beat() de kolom reserved_at van de taak bij naar NOW(). De detector voor verouderde taken gebruikt reserved_at om te bepalen of een taak vastzit. Door deze bij te werken, blijft de taak dus veilig.

Wanneer de taak is voltooid, roept de worker Heartbeat::clear() aan om de context te resetten.

Heartbeats versturen

Roep Heartbeat::beat() aan vanuit een willekeurige taakhandler of streamingstap:

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

Voortgangsgegevens

De parameter $progress is een optionele associatieve array. Wanneer deze wordt opgegeven, wordt deze gecodeerd als JSON en opgeslagen in de kolom heartbeat_data van de taakrij. Zo kunt u de voortgang van buiten de worker volgen:

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

Wanneer deze zonder argumenten wordt aangeroepen, wordt alleen reserved_at bijgewerkt:

Heartbeat::beat();

Heartbeat API

De klasse Queuety\Heartbeat is volledig statisch. U hoeft deze niet te instantiëren.

Gebruik in streamingstappen

De worker verstuurt al een heartbeat na elk opgeslagen chunk, met heartbeat_data ingesteld op {"streaming_chunks": n}, zodat een stream die regelmatig gegevens oplevert vanzelf actueel blijft. Roep zelf Heartbeat::beat() aan wanneer er lange pauzes tussen opbrengsten kunnen zitten, bijvoorbeeld tijdens het wachten op het eerste token. Gegevens die u meegeeft, worden overschreven door de volgende heartbeat van de worker per chunk:

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

Wanneer heartbeats gebruiken

Gebruik heartbeats wanneer één taak of stap langer duurt dan QUEUETY_STALE_TIMEOUT seconden (standaard: 600). Veelvoorkomende gevallen:

• LLM-API-aanroepen met streamingreacties
• Grote bestanden downloaden of uploaden
• Herhalingen van batchverwerking binnen één stap
• Elke externe API-aanroep die minuten kan duren

Als uw stap binnen een minuut wordt voltooid, zijn heartbeats niet nodig. De detector voor verouderde taken wordt pas actief nadat de geconfigureerde time-out is verstreken.