Munkafolyamat exportálása és visszajátszása

A Queuety a munkafolyamat teljes végrehajtási előzményeit hordozható JSON formátumba exportálhatja, majd egy másik környezetben visszajátszhatja. Ez hasznos az éles környezet hibáinak helyi reprodukálásához, a munkafolyamatok környezetek közötti áthelyezéséhez és az auditáláshoz.

Munkafolyamat exportálása

PHP API

Hívja meg a Queuety::export_workflow() elemet a munkafolyamat-azonosítóval. Egy JSON formátumban szerializálható tömböt ad vissza:

use Queuety\Queuety;

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

Közvetlenül JSON sztring lekérése:

$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

Munkafolyamat visszajátszása

PHP API

Hívja meg a Queuety::replay_workflow() elemet az exportált adatokkal. Az új munkafolyamat azonosítóját adja vissza:

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.

JSON felépítése

Az export a teljes munkafolyamat-állapotot, a kapcsolódó feladatokat, a nyomvonal-eseményeket, naplóbejegyzéseket, tárolt jelzéseket, mellékleteket, adatdarabokat, valamint a munkafolyamat-függőségi várakozások sorait tartalmazza:

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

A fenti példában néhány mező rövidített. A tényleges export minden eseménysor összes nyomvonaloszlopát tartalmazza.

A munkafolyamat-exportok a munkafolyamat metaadatait is megőrzik, például a definition_version, definition_hash és idempotency_key értékét, ha ezek az eredeti futásban be voltak állítva.

A visszajátszás működése

Exportált munkafolyamat visszajátszásakor:

1. Létrejön egy új munkafolyamatsor az exportált állapottal és lépésdefiníciókkal
2. Az új munkafolyamat neve {original_name}_replay_{timestamp}
3. A korábbi munkafolyamat-események és naplók újra bekerülnek az új környezetbe
4. A tárolt jelzések és munkafolyamat-várakozási függőségi sorok korábbi állapotként kerülnek visszajátszásra
5. Ha az eredeti munkafolyamat még running állapotban volt, az aktuális lépés feldolgozásra kerül a sorba. A Repeat és start_workflows() lépések nem kerülnek újra sorba, így az egyiknél megálló visszajátszás running állapotban marad feladat nélkül (#17)
6. Ha az eredeti munkafolyamat waiting_for_signal, waiting_for_workflows, paused, failed, cancelled vagy completed állapotban volt, ez az állapot megmarad a visszajátszott munkafolyamatban

A visszajátszott munkafolyamat új azonosítót kap, és önállóan működik. Nincs hatással az eredetire.

A waiting_for_workflows visszajátszásoknál az exportált függőségi sorok ugyanazokkal a munkafolyamat-azonosítókkal jönnek létre újra. Ez hasznos auditáláshoz és olyan környezetekben, ahol ezek a függő munkafolyamatok is léteznek, de a függőségi azonosítókat nem rendeli át automatikusan a környezetek között.

Felhasználási esetek

Éles hibák reprodukálása

Exportáljon egy sikertelen munkafolyamatot az éles környezetből, és játssza vissza helyben:

# 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

A visszajátszott munkafolyamat állapota és lépésdefiníciói pontosan megegyeznek az eredetivel. A futó munkafolyamatok ismét sorba állítják az aktuális lépést, így onnan folytathatja a végrehajtást, ahol abbamaradt. A szüneteltetett vagy várakozó munkafolyamatok blokkolt állapotban jönnek létre újra, hogy megvizsgálhatók legyenek.

Munkafolyamatok áthelyezése környezetek között

Helyezzen át egy munkafolyamatot az előkészítő környezetből az élesbe vagy fordítva exportálással és visszajátszással:

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

# On production
wp queuety workflow replay workflow.json

Auditálás és archiválás

Exportálja a befejezett munkafolyamatokat hosszú távú tároláshoz vagy megfelelőségi célokra. A JSON fájl a teljes végrehajtási előzményt tartalmazza, beleértve a bemeneteket, kimeneteket, az előtte és utána lévő állapotot, az időzítési adatokat és a hiba részleteit.

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