Señales de actividad

Los trabajos de larga duración y los pasos de transmisión pueden enviar señales de actividad para indicar que siguen en ejecución. Sin ellas, el detector de trabajos obsoletos podría reclamar un trabajo que aún se está procesando activamente porque no se ha actualizado su marca de tiempo reserved_at.

Cómo funciona

Cuando un proceso de trabajo empieza a procesar un trabajo, llama a Heartbeat::init() para establecer el contexto del trabajo actual. Dentro de su controlador, al llamar a Heartbeat::beat() se actualiza la columna reserved_at del trabajo a NOW(). El detector de trabajos obsoletos usa reserved_at para determinar si un trabajo está atascado, así que actualizarla mantiene el trabajo a salvo.

Cuando termina el trabajo, el proceso de trabajo llama a Heartbeat::clear() para restablecer el contexto.

Enviar señales de actividad

Llame a Heartbeat::beat() desde cualquier controlador de trabajo o paso de transmisión:

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

Datos de progreso

El parámetro $progress es una matriz asociativa opcional. Si se proporciona, se codifica en JSON y se almacena en la columna heartbeat_data de la fila del trabajo. Esto le permite hacer seguimiento del progreso desde fuera del proceso de trabajo:

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

Si se llama sin argumentos, solo se actualiza reserved_at:

Heartbeat::beat();

API de actividad

La clase Queuety\Heartbeat es completamente estática. No se instancia.

Uso en pasos de transmisión

El proceso de trabajo ya envía una señal de actividad después de cada fragmento que guarda, con heartbeat_data establecido en {"streaming_chunks": n}, así que una transmisión que produce fragmentos con regularidad se mantiene actualizada por sí sola. Llame a Heartbeat::beat() si puede haber intervalos largos entre fragmentos, por ejemplo, mientras espera el primer token. Los datos que pase se sobrescribirán con la siguiente señal por fragmento del proceso de trabajo:

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

Cuándo usar señales de actividad

Use señales de actividad cuando un solo trabajo o paso dure más de QUEUETY_STALE_TIMEOUT segundos (predeterminado: 600). Casos habituales:

• LLM API con respuestas en streaming
• Descargas o cargas de archivos grandes
• Repeticiones de procesamiento por lotes dentro de un solo paso
• Cualquier llamada externa a API que pueda tardar minutos en responder

Si el paso se completa en menos de un minuto, no hacen falta señales de actividad. El detector de procesos obsoletos solo se activa cuando ha transcurrido el tiempo de espera configurado.