Esportazione e riproduzione del flusso di lavoro

Queuety può esportare la cronologia completa di esecuzione di un flusso di lavoro in un formato portatile JSON e riprodurla in un ambiente diverso. È utile per riprodurre localmente bug di produzione, migrare flussi di lavoro tra ambienti ed eseguire verifiche.

Esportare un flusso di lavoro

PHP API

Chiamate Queuety::export_workflow() con l'ID del flusso di lavoro. Restituisce un array serializzabile come JSON:

use Queuety\Queuety;

$data = Queuety::export_workflow( $workflow_id );

Per ottenere direttamente una stringa JSON:

$json = json_encode( $data, JSON_PRETTY_PRINT );
file_put_contents( '/tmp/workflow-42.json', $json );

CLI

wp queuety workflow export <id> [--output=<file>]

# Print to stdout
wp queuety workflow export 42

# Write to a file
wp queuety workflow export 42 --output=/tmp/workflow-42.json

Riprodurre un flusso di lavoro

PHP API

Chiamate Queuety::replay_workflow() con i dati esportati. Restituisce il nuovo ID del flusso di lavoro:

use Queuety\Queuety;

$json = file_get_contents( '/tmp/workflow-42.json' );
$data = json_decode( $json, true );

$new_id = Queuety::replay_workflow( $data );

CLI

wp queuety workflow replay <file>

wp queuety workflow replay /tmp/workflow-42.json
# Success: Workflow replayed. New workflow ID: 85.

Struttura di JSON

L'esportazione contiene lo stato completo del flusso di lavoro, i job associati, gli eventi di traccia, le voci di registro, i segnali salvati, gli artefatti, i segmenti e le eventuali righe di attesa per le dipendenze del flusso di lavoro:

{
    "workflow": {
        "id": 42,
        "name": "generate_report",
        "status": "failed",
        "state": { "user_id": 1, "_steps": [...], "_queue": "default" },
        "current_step": 2,
        "total_steps": 4,
        "parent_workflow_id": null,
        "parent_step_index": null,
        "started_at": "2026-03-28 10:00:00",
        "completed_at": null,
        "failed_at": "2026-03-28 10:05:00",
        "error_message": "API timeout",
        "deadline_at": null
    },
    "jobs": [
        {
            "id": 100,
            "queue": "default",
            "handler": "App\\Handlers\\FetchDataHandler",
            "payload": {},
            "status": "completed",
            "attempts": 1,
            "max_attempts": 3,
            "step_index": 0
        }
    ],
    "events": [
        {
            "id": 1,
            "job_id": 100,
            "step_index": 0,
            "step_name": "fetchData",
            "step_type": "single",
            "handler": "App\\Handlers\\FetchDataHandler",
            "event": "step_completed",
            "input": { "user_id": 1 },
            "output": { "records": 150 },
            "state_before": { "user_id": 1 },
            "state_after": { "user_id": 1, "records": 150 },
            "context": { "source": "api" },
            "error": null,
            "duration_ms": 230,
            "created_at": "2026-03-28 10:00:01"
        }
    ],
    "logs": [
        {
            "id": 1,
            "event": "workflow_started",
            "handler": "generate_report",
            "created_at": "2026-03-28 10:00:00"
        }
    ],
    "signals": [],
    "artifacts": [],
    "chunks": [],
    "wait_dependencies": [],
    "exported_at": "2026-03-28T12:00:00+00:00",
    "queuety_version": "0.22.0"
}

Nell'esempio sopra alcuni campi sono abbreviati. L'esportazione effettiva include tutte le colonne di traccia di ogni riga di evento.

Le esportazioni dei flussi di lavoro conservano anche metadati quali definition_version, definition_hash e idempotency_key, se erano stati impostati nell'esecuzione originale.

Come funziona la riproduzione

Quando riproducete un flusso di lavoro esportato:

1. Viene creata una nuova riga del flusso di lavoro con lo stato esportato e le definizioni dei passaggi
2. Il nuovo flusso di lavoro si chiama {original_name}_replay_{timestamp}
3. Gli eventi storici e i registri del flusso di lavoro vengono reinseriti nel nuovo ambiente
4. I segnali salvati e le righe di dipendenza per l'attesa del flusso di lavoro vengono riprodotti come stato storico
5. Se il flusso di lavoro originale era ancora running, il passaggio corrente viene inserito in coda per l'elaborazione. I passaggi Repeat e start_workflows() non vengono reinseriti in coda; pertanto, una riproduzione che si interrompe su uno di questi rimane running senza un job (#17)
6. Se il flusso di lavoro originale era waiting_for_signal, waiting_for_workflows, paused, failed, cancelled o completed, tale stato viene mantenuto nel flusso di lavoro riprodotto

Il flusso di lavoro riprodotto riceve un nuovo ID e opera in modo indipendente. Non influisce sull'originale.

Per le riproduzioni waiting_for_workflows, le righe delle dipendenze esportate vengono ricreate con gli stessi ID dei flussi di lavoro. È utile per le verifiche e negli ambienti in cui esistono anche quei flussi di lavoro dipendenti, ma gli ID delle dipendenze non vengono rimappati automaticamente tra ambienti.

Casi d'uso

Riprodurre bug di produzione

Esportate un flusso di lavoro non riuscito in produzione e riproducetelo localmente:

# On production
wp queuety workflow export 42 --output=/tmp/workflow-42.json

# Copy the file to your local environment

# On local
wp queuety workflow replay /tmp/workflow-42.json

Il flusso di lavoro riprodotto ha esattamente lo stesso stato e le stesse definizioni dei passaggi. I flussi di lavoro in esecuzione reinseriscono in coda il passaggio corrente, così potete riprendere l'esecuzione dal punto in cui si era interrotta; quelli in pausa o in attesa vengono ricreati in quello stato di blocco per poterli esaminare.

Migrare flussi di lavoro tra ambienti

Spostate un flusso di lavoro dallo staging alla produzione, o viceversa, esportandolo e riproducendolo:

# On staging
wp queuety workflow export 15 --output=workflow.json

# On production
wp queuety workflow replay workflow.json

Verifica e archiviazione

Esportate i flussi di lavoro completati per conservarli a lungo termine o per esigenze di conformità. Il file JSON contiene la cronologia completa di esecuzione, inclusi input, output, stato prima e dopo, dati sui tempi e dettagli degli errori.

$data = Queuety::export_workflow( $workflow_id );
// Store $data in your archive system, S3, etc.