Introduzione

Visualizza come Markdown Modifica su GitHub Documentazione di 2cc7b2b

Queuety è un plugin WordPress che offre una coda di job veloce, un motore di workflow durevoli e un runtime per macchine a stati durevoli. I worker acquisiscono i job direttamente da MySQL e li elaborano all’interno di un processo WP-CLI a lunga esecuzione. Il runtime PHP che carica WordPress e WP-CLI richiede mysqli o pdo_mysql.

Perché Queuety?

WordPress non dispone di un vero sistema di job in background. wp_cron si attiva solo quando si visitano le pagine. Il runner predefinito di Action Scheduler si basa su WP-Cron e su richieste HTTP di loopback, elaborando batch per un massimo di 30 secondi. Una chiamata LLM API che dura 60 secondi viene interrotta dal timeout di 30 secondi di PHP. Non c’è modo di eseguire processi composti da più passaggi che sopravvivano agli arresti anomali e riprendano da dove si erano interrotti.

Queuety risolve il problema con tre elementi fondamentali:

Job per attività in background da avviare senza attenderne il completamento
Workflow per processi durevoli composti da più passaggi e dotati di stato persistente
Macchine a stati per gestire stati del ciclo di vita duraturi e basati su eventi

Esempio rapido

Invia un job usando la moderna API:

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

readonly class SendEmailJob implements Job {
    use Dispatchable;

    public function __construct(
        public string $to,
        public string $subject,
        public string $body,
    ) {}

    public function handle(): void {
        wp_mail( $this->to, $this->subject, $this->body );
    }
}

SendEmailJob::dispatch( 'user@example.com', 'Welcome', 'Hello from Queuety!' );

Oppure usa il metodo classico con il nome dell’handler:

PHP

use Queuety\Queuety;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

Esegui un workflow durevole:

PHP

Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Modella una sessione di lunga durata con una macchina a stati:

PHP

use Queuety\Enums\StateMachineStatus;

Queuety::machine( 'agent_session' )
    ->state( 'awaiting_user' )
    ->on( 'user_message', 'planning' )
    ->state( 'planning' )
    ->action( PlanSessionAction::class )
    ->on( 'planned', 'completed' )
    ->state( 'completed', StateMachineStatus::Completed )
    ->dispatch( [ 'thread_id' => 42 ] );

Avvia un worker:

Shell

wp queuety work

Se stai creando un sistema basato su agenti, inizia da Orchestrazione degli agenti. Spiega come si integrano i flussi planner/executor, i passaggi di revisione umana e le attese tra workflow.

Se il sistema è basato su eventi e deve restare attivo a lungo, inizia da Macchine a stati insieme ai workflow.

Per esempi di integrazione end-to-end, consulta Esempi, che ora include modelli basati su Neuron per cambiare provider, gestire lo streaming, ripetere le revisioni con memoria e coordinare workflow dipendenti.

Se vuoi avviare i workflow a partire da eventi di dominio WordPress come save_post, consulta Trigger delle azioni.

Per usare il runtime basato su DB al di fuori di WordPress, consulta Uso autonomo.

Elementi distintivi

Esecuzione rapida. I worker accedono direttamente alla coda MySQL ed evitano il modello cron per richiesta, che rallenta il lavoro in background in WordPress.
Flussi di lavoro durevoli. Processi articolati in più passaggi, con stato persistente, che resistono a timeout PHP, arresti anomali e nuovi tentativi. Ogni confine tra passaggi è una transazione atomica MySQL.
Macchine a stati esplicite. Modellate sessioni di chat, approvazioni e altri cicli di vita di lunga durata con stati denominati, transizioni protette, azioni in coda all'ingresso in uno stato e cronologie degli eventi durevoli.
Dispatch moderno API. Classi di job autonome con il trait Dispatchable, proprietà del costruttore tipizzate come payload e $tries/$timeout/$backoff dichiarati nella classe. Una classe readonly non può assegnare valori predefiniti a queste proprietà: dichiaratele quindi in una classe normale con proprietà promosse readonly nel costruttore.
Pipeline di middleware. Middleware in stile onion per limitare la frequenza, gestire le eccezioni, creare job univoci e aggiungere logica personalizzata. Si dichiara per ogni job tramite un metodo middleware().
Passaggi in streaming. L'interfaccia StreamingStep genera blocchi che vengono salvati immediatamente nel database. In caso di nuovo tentativo, vengono forniti i blocchi già salvati, così lo stream può riprendere da dove si era interrotto. Ideale per le risposte LLM e i download di grandi dimensioni.
Livello di cache estensibile. CacheFactory rileva automaticamente APCu oppure ripiega su una cache in memoria. I backend personalizzati implementano Contracts\Cache. Viene usato internamente per le ricerche del limitatore di frequenza e per lo stato della coda.
Elaborazione in batch e concatenamento. Inviate gruppi di job con callback then/catch/finally e monitoraggio dell'avanzamento, oppure concatenate i job in sequenza, in modo che ciascuno dipenda dal completamento del precedente.
Ritardi e attese durevoli. delay() aggiunge ritardi sicuri anche in caso di riavvio del processo. wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() e wait_for_workflow() consentono ai flussi di lavoro di attendere eventi esterni o altri flussi senza mantenere attivo PHP.
Cicli dinamici for-each. for_each() espande i rami di lavoro individuati a runtime in base allo stato del flusso di lavoro e supporta le modalità di completamento al successo di tutti, al primo successo e al raggiungimento del quorum.
Passaggi asincroni tra flussi di lavoro. start_workflows() trasforma gli elementi individuati a runtime in flussi di lavoro indipendenti di primo livello, sui quali è possibile attendere in seguito.
Trigger di azioni WordPress. on_action() associa gli hook delle azioni core o dei plugin a invii durevoli di flussi di lavoro, con controlli e idempotenza.
Artefatti durevoli. Memorizzate gli output nominati dei flussi di lavoro al di fuori del contenitore di stato principale e manteneteli disponibili nello stato, nell'ispezione CLI, nell'esportazione e nella riproduzione.
Misure di protezione dei flussi di lavoro. version(), un hash deterministico della definizione, chiavi di invio idempotenti e limiti del flusso di lavoro rendono più facili da ispezionare le esecuzioni in stile agente e riducono il rischio di duplicazioni o di esecuzioni incontrollate e silenziose.
Gestione delle risorse. I gruppi di concorrenza, le unità di costo, i limiti dei flussi di lavoro e l'ammissione dei worker aiutano a mantenere le attività costose in stile agente entro limiti definiti.
Compensazione dei passaggi. compensate_with() e compensate_on_failure() forniscono hook di rollback per i passaggi completati quando i flussi di lavoro vengono annullati o non riescono.
Heartbeat. I passaggi di lunga durata inviano heartbeat per impedire al sistema di recupero dei job obsoleti di riprendere il controllo delle attività ancora in corso.
Tracciamento dei flussi di lavoro. Ogni transizione di passaggio viene registrata con input, output, stato precedente e successivo, contesto, artefatti, blocchi, errori e debug temporale tramite workflow_state_at().
Passaggi paralleli. Eseguite i passaggi contemporaneamente e attendete che siano tutti completati prima di proseguire.
Diramazioni condizionali. Passate a passaggi denominati in base allo stato precedente usando _next_step.
Esecuzione di flussi di lavoro. Avviate flussi di lavoro secondari che restituiscono i risultati al flusso padre.
Code con priorità. Quattro livelli (Low, Normal, High, Urgent) tramite enum PHP con controllo dei tipi.
Worker con più code. Elaborate più code con un ordinamento rigoroso per priorità in un singolo worker.
Limitazione della frequenza. Limiti di esecuzione per handler con finestra scorrevole.
Job ricorrenti. Pianificazione a intervalli e basata su cron, con criteri configurabili per le sovrapposizioni (Allow, Skip, Buffer).
Annullamento dei flussi di lavoro. Annullate i flussi di lavoro in esecuzione e attivate gli handler di pulizia registrati.
Pulizia dello stato. Rimuovete automaticamente i vecchi output dei passaggi per mantenere gestibile lo stato dei flussi di lavoro di grandi dimensioni.
Concorrenza dei worker. --workers=N avvia più processi con riavvio automatico in caso di arresto anomalo.
Osservabilità completa. Registrazione nel database, metriche per handler e webhook degli eventi.
Strumenti di test. QueueFake registra i job e i batch inviati per consentire asserzioni nei test unitari.
Attributi PHP. #[QueuetyHandler('name')] per registrare gli handler senza configurazione.

Come funziona

I worker vengono eseguiti all'interno di un processo WP-CLI di lunga durata e acquisiscono i job direttamente da MySQL. Le righe delle code, lo stato dei flussi di lavoro, i log, i batch, i segnali e i blocchi dello streaming risiedono tutti in MySQL, quindi il riavvio dei worker non fa perdere lo stato di orchestrazione.

I flussi di lavoro suddividono le attività di lunga durata in passaggi. L'output di ogni passaggio viene salvato in un contenitore di stato condiviso nel database. Se PHP si arresta durante un passaggio, il worker riprova quel passaggio mantenendo intatto tutto lo stato precedente. Il confine del passaggio è una singola transazione MySQL: l'aggiornamento dello stato, il completamento del job e l'inserimento in coda del passaggio successivo avvengono tutti atomicamente.

Codice

Workflow: generate_report (3 steps)

Step 0: fetch_data     -> state: {user_data: {...}}
Step 1: call_llm       -> PHP dies -> retry -> state: {user_data: {...}, llm_response: "..."}
Step 2: format_output  -> state: {user_data: {...}, llm_response: "...", report_url: "/reports/42.pdf"}

Workflow: completed

Requisiti

PHP 8.2 o versioni successive
WordPress 6.4 o versioni successive
MySQL 8.0+ o MariaDB 10.6+, perché i worker acquisiscono i job con SELECT ... FOR UPDATE SKIP LOCKED
mysqli o pdo_mysql nell'ambiente di runtime PHP che carica WordPress e WP-CLI
WP-CLI per eseguire worker dedicati con wp queuety work
Le estensioni pcntl e posix PHP per i pool di worker (--workers, --min-workers, --max-workers)

Per l'installazione, consultate Guida introduttiva.