Workflow-Export und Wiedergabe

Queuety kann den vollständigen Ausführungsverlauf eines Workflows in ein portables Format JSON exportieren und in einer anderen Umgebung wiedergeben. Das ist nützlich, um Produktionsfehler lokal nachzustellen, Workflows zwischen Umgebungen zu migrieren und Prüfungen durchzuführen.

Workflow exportieren

PHP API

Rufe Queuety::export_workflow() mit der Workflow-ID auf. Die Methode gibt ein JSON-serialisierbares Array zurück:

use Queuety\Queuety;

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

So erhältst du direkt eine JSON-Zeichenfolge:

$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

Workflow wiedergeben

PHP API

Rufe Queuety::replay_workflow() mit den Exportdaten auf. Die Methode gibt die neue Workflow-ID zurück:

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.

Struktur von JSON

Der Export enthält den vollständigen Workflow-Status, zugehörige Jobs, Trace-Ereignisse, Protokolleinträge, gespeicherte Signale, Artefakte, Datenblöcke und alle Zeilen für Wartevorgänge auf Workflow-Abhängigkeiten:

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

Einige Felder sind im obigen Beispiel abgekürzt. Der tatsächliche Export enthält alle Trace-Spalten jeder Ereigniszeile.

Workflow-Exporte erhalten auch Metadaten wie definition_version, definition_hash und idempotency_key, sofern diese beim ursprünglichen Lauf festgelegt wurden.

So funktioniert die Wiedergabe

Bei der Wiedergabe eines exportierten Workflows geschieht Folgendes:

1. Eine neue Workflow-Zeile mit dem exportierten Status und den Schrittdefinitionen wird erstellt
2. Der neue Workflow erhält den Namen {original_name}_replay_{timestamp}
3. Frühere Workflow-Ereignisse und Protokolle werden in die neue Umgebung eingefügt
4. Gespeicherte Signale und Abhängigkeitszeilen für Workflow-Wartevorgänge werden als früherer Status wiederhergestellt
5. War der ursprüngliche Workflow noch running, wird der aktuelle Schritt zur Verarbeitung eingereiht. Repeat- und start_workflows()-Schritte werden nicht erneut eingereiht. Eine Wiedergabe, die bei einem dieser Schritte endet, bleibt daher running, ohne Job (#17)
6. Hatte der ursprüngliche Workflow den Status waiting_for_signal, waiting_for_workflows, paused, failed, cancelled oder completed, bleibt dieser Status beim wiedergegebenen Workflow erhalten

Der wiedergegebene Workflow erhält eine neue ID und läuft unabhängig. Er beeinflusst das Original nicht.

Bei waiting_for_workflows-Wiedergaben werden die exportierten Abhängigkeitszeilen mit denselben Workflow-IDs neu erstellt. Das ist für Prüfungen und Umgebungen nützlich, in denen diese abhängigen Workflows ebenfalls vorhanden sind. Abhängigkeits-IDs werden dabei nicht automatisch umgebungsübergreifend zugeordnet.

Anwendungsfälle

Produktionsfehler nachstellen

Exportiere einen fehlgeschlagenen Workflow aus der Produktion und spiele ihn lokal wieder ab:

# 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

Der wiedergegebene Workflow hat denselben Status und dieselben Schrittdefinitionen. Bei laufenden Workflows werden die aktuellen Schritte erneut eingereiht, damit du die Ausführung an der Stelle fortsetzen kannst, an der sie unterbrochen wurde. Pausierte oder wartende Workflows werden im blockierten Zustand zur Prüfung wiederhergestellt.

Workflows zwischen Umgebungen migrieren

Verschiebe einen Workflow durch Export und Wiedergabe von Staging nach Produktion oder umgekehrt:

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

# On production
wp queuety workflow replay workflow.json

Prüfung und Archivierung

Exportiere abgeschlossene Workflows zur langfristigen Speicherung oder für Compliance-Zwecke. Die Datei JSON enthält den vollständigen Ausführungsverlauf mit Eingaben, Ausgaben, Status davor und danach, Zeitangaben und Fehlerdetails.

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