Exportation et relecture des flux de travail

Queuety peut exporter l’historique complet d’exécution d’un flux de travail dans un format JSON portable et le relire dans un autre environnement. Cela permet de reproduire localement des bogues de production, de migrer des flux entre environnements et de réaliser des audits.

Exporter un flux de travail

PHP API

Appelez Queuety::export_workflow() avec l’identifiant du flux de travail. La méthode renvoie un tableau sérialisable en JSON :

use Queuety\Queuety;

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

Pour obtenir directement une chaîne 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

Relire un flux de travail

PHP API

Appelez Queuety::replay_workflow() avec les données d’exportation. La méthode renvoie le nouvel identifiant du flux de travail :

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.

Structure de JSON

L’exportation contient l’état complet du flux de travail, les tâches associées, les événements de trace, les entrées de journal, les signaux stockés, les artefacts, les blocs et toutes les lignes d’attente des dépendances entre flux :

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

Certains champs sont abrégés dans l’exemple ci-dessus. L’exportation réelle inclut toutes les colonnes de trace de chaque ligne d’événement.

Les exportations de flux de travail conservent également les métadonnées du flux, telles que definition_version, definition_hash et idempotency_key, si elles étaient définies lors de l’exécution d’origine.

Fonctionnement de la relecture

Lorsque vous relisez un flux de travail exporté :

1. Une nouvelle ligne de flux de travail est créée avec l’état exporté et les définitions des étapes
2. Le nom du nouveau flux de travail est {original_name}_replay_{timestamp}
3. Les événements et journaux historiques du flux sont réinsérés dans le nouvel environnement
4. Les signaux stockés et les lignes de dépendance d’attente sont relus comme état historique
5. Si le flux d’origine était toujours running, l’étape actuelle est mise en file pour traitement. Les étapes de répétition et start_workflows() ne sont pas remises en file ; une relecture qui s’arrête sur l’une d’elles reste donc running sans tâche (#17)
6. Si le flux d’origine était waiting_for_signal, waiting_for_workflows, paused, failed, cancelled ou completed, ce statut est conservé sur le flux relu

Le flux relu reçoit un nouvel identifiant et fonctionne indépendamment. Il n’a aucune incidence sur l’original.

Pour les relectures waiting_for_workflows, les lignes de dépendance exportées sont recréées avec les mêmes identifiants de flux. C’est utile pour les audits et les environnements où ces flux de dépendance existent également, mais les identifiants de dépendance ne sont pas automatiquement remappés entre environnements.

Cas d’utilisation

Reproduire les bogues de production

Exportez un flux de travail en échec depuis la production et relisez-le en local :

# 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

Le flux relu possède exactement le même état et les mêmes définitions d’étapes. Pour les flux en cours, l’étape actuelle est remise en file afin que vous puissiez reprendre l’exécution là où elle s’était arrêtée. Les flux en pause ou en attente sont recréés dans cet état bloqué pour inspection.

Migrer des flux entre environnements

Déplacez un flux de travail de la préproduction vers la production, ou inversement, en l’exportant puis en le relisant :

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

# On production
wp queuety workflow replay workflow.json

Audit et archivage

Exportez les flux terminés pour les conserver à long terme ou répondre à des exigences de conformité. Le fichier JSON contient l’historique complet d’exécution, y compris les entrées, les sorties, l’état avant et après, les durées et les détails des erreurs.

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