Сигнали за активност

Продължителните задачи и поточните стъпки могат да изпращат сигнали за активност, за да потвърдят, че все още работят. Без такива сигнали детекторът за остарели задачи може да заяви повторно задача, която все още се обработва активно, защото времевият ѝ печат reserved_at не е актуализиран.

Как работи

Когато работният процес започне да обработва задача, той извиква Heartbeat::init(), за да зададе текущия контекст на задачата. В обработчика си извикайте Heartbeat::beat(), за да актуализирате колоната reserved_at на задачата до NOW(). Детекторът за остарели задачи използва reserved_at, за да определи дали задачата е блокирала, затова обновяването му я предпазва.

Когато задачата приключи, работният процес извиква Heartbeat::clear(), за да нулира контекста.

Изпращане на сигнали за активност

Извикайте Heartbeat::beat() от всеки обработчик на задача или поточна стъпка:

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

Данни за напредъка

Параметърът $progress е незадължителен асоциативен масив. Когато е зададен, той се кодира в JSON и се съхранява в колоната heartbeat_data на реда на задачата. Така можете да проследявате напредъка извън работния процес:

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

Когато бъде извикан без аргументи, се актуализира само reserved_at:

Heartbeat::beat();

API за сигналите за активност

Класът Queuety\Heartbeat е изцяло статичен. Не създавайте негова инстанция.

Употреба в поточни стъпки

Работният процес вече изпраща сигнал за активност след всяка запазена част, като задава heartbeat_data на {"streaming_chunks": n}, така че поток, който редовно генерира данни, остава актуален автоматично. Извикайте сами Heartbeat::beat(), ако между генерирането на данни има дълги паузи, например докато чакате първия токен. Предадените от вас данни се презаписват от следващия сигнал на работния процес за отделна част:

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

Кога да използвате сигнали за активност

Използвайте сигнали за активност, когато една задача или стъпка продължава повече от QUEUETY_STALE_TIMEOUT секунди (по подразбиране: 600). Често срещани случаи:

• Поточни извиквания на LLM API
• Изтегляне или качване на големи файлове
• Повтаряща се пакетна обработка в рамките на една стъпка
• Всяко външно извикване на API, за което може да са нужни минути за отговор

Ако стъпката ви приключва за по-малко от минута, сигналите за активност не са необходими. Детекторът за остарели задачи се задейства едва след изтичане на конфигурирания период на изчакване.