Exportación y reproducción de flujos de trabajo

Queuety puede exportar el historial de ejecución completo de un flujo de trabajo a un formato JSON portátil y reproducirlo en otro entorno. Esto resulta útil para reproducir localmente errores de producción, migrar flujos de trabajo entre entornos y realizar auditorías.

Exportar un flujo de trabajo

PHP API

Llame a Queuety::export_workflow() con el ID del flujo de trabajo. Devuelve un arreglo serializable en JSON:

use Queuety\Queuety;

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

Para obtener directamente una cadena 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

Reproducir un flujo de trabajo

PHP API

Llame a Queuety::replay_workflow() con los datos de exportación. Devuelve el nuevo ID del flujo de trabajo:

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.

Estructura de JSON

La exportación contiene el estado completo del flujo de trabajo, los trabajos asociados, los eventos de traza, las entradas de log, las señales almacenadas, los artefactos, los fragmentos y las filas de espera de dependencias entre flujos de trabajo:

{
    "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"
}

Algunos campos aparecen abreviados en el ejemplo anterior. La exportación real incluye todas las columnas de traza de cada fila de evento.

Las exportaciones de flujos de trabajo también conservan metadatos como definition_version, definition_hash y idempotency_key cuando se establecieron en la ejecución original.

Cómo funciona la reproducción

Al reproducir un flujo de trabajo exportado:

1. Se crea una nueva fila de flujo de trabajo con el estado exportado y las definiciones de los pasos
2. El nombre del nuevo flujo de trabajo es {original_name}_replay_{timestamp}
3. Los eventos históricos y los logs del flujo de trabajo se vuelven a insertar en el nuevo entorno
4. Las señales almacenadas y las filas de dependencias de espera entre flujos de trabajo se reproducen como estado histórico
5. Si el flujo de trabajo original seguía running, se pone en cola el paso actual para su procesamiento. Los pasos Repeat y start_workflows() no se vuelven a poner en cola, por lo que una reproducción que se detenga en uno de ellos permanece running sin un trabajo (#17)
6. Si el flujo de trabajo original estaba waiting_for_signal, waiting_for_workflows, paused, failed, cancelled o completed, ese estado se conserva en el flujo de trabajo reproducido

El flujo de trabajo reproducido recibe un ID nuevo y funciona de forma independiente. No afecta al original.

En las reproducciones de waiting_for_workflows, las filas de dependencias exportadas se vuelven a crear con los mismos ID de flujo de trabajo. Esto resulta útil para las auditorías y para los entornos donde también existen esos flujos de trabajo dependientes, pero no reasigna automáticamente los ID de dependencias entre entornos.

Casos de uso

Reproducir errores de producción

Exporte un flujo de trabajo fallido de producción y reprodúzcalo 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

El flujo de trabajo reproducido tiene exactamente el mismo estado y las mismas definiciones de pasos. Los flujos de trabajo en ejecución vuelven a poner en cola el paso actual para que pueda continuar la ejecución desde el punto en que se detuvo; los flujos pausados o en espera se recrean en ese estado bloqueado para inspeccionarlos.

Migrar flujos de trabajo entre entornos

Traslade un flujo de trabajo de preparación a producción (o viceversa) mediante la exportación y la reproducción:

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

# On production
wp queuety workflow replay workflow.json

Auditoría y archivo

Exporte los flujos de trabajo completados para almacenarlos a largo plazo o por motivos de cumplimiento. El archivo JSON contiene el historial de ejecución completo, incluidas las entradas, salidas, estados anteriores y posteriores, datos de tiempo y detalles de errores.

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