Експортиране и повторно изпълнение на работен поток

Queuety може да експортира пълната история на изпълнение на работен поток в преносим формат JSON и да я изпълни отново в друга среда. Това е полезно за локално възпроизвеждане на грешки от продукционна среда, мигриране на работни потоци между среди и одит.

Експортиране на работен поток

PHP API

Извикайте Queuety::export_workflow() с идентификатора на работния поток. Той връща масив, който може да се сериализира като JSON:

use Queuety\Queuety;

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

За да получите директно низ 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

Повторно изпълнение на работен поток

PHP API

Извикайте Queuety::replay_workflow() с данните за експортиране. Той връща новия идентификатор на работния поток:

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

Експортирането съдържа пълното състояние на работния поток, свързаните задачи, събитията от проследяването, записите в регистъра, съхранените сигнали, артефактите, частите и всички редове за изчакване на зависимости между работни потоци:

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

Някои полета в примера по-горе са съкратени. Действителното експортиране включва всички колони за проследяване от всеки ред на събитие.

Експортираните работни потоци запазват и метаданни като definition_version, definition_hash и idempotency_key, когато са били зададени при първоначалното изпълнение.

Как работи повторното изпълнение

Когато изпълните повторно експортиран работен поток:

1. Създава се нов ред за работен поток с експортираното състояние и дефинициите на стъпките
2. Името на новия работен поток е {original_name}_replay_{timestamp}
3. Историческите събития и записи в регистъра на работния поток се добавят отново в новата среда
4. Съхранените сигнали и редовете със зависимости за изчакване на работния поток се възпроизвеждат като историческо състояние
5. Ако първоначалният работен поток все още е бил running, текущата стъпка се добавя в опашката за обработка. Стъпките Repeat и start_workflows() не се добавят отново в опашката, затова повторно изпълнение, което спира на една от тях, остава running без задача (#17)
6. Ако първоначалният работен поток е бил в състояние waiting_for_signal, waiting_for_workflows, paused, failed, cancelled или completed, това състояние се запазва в повторно изпълнения работен поток

Повторно изпълненият работен поток получава нов идентификатор и работи независимо. Той не засяга първоначалния.

При повторни изпълнения на waiting_for_workflows експортираните редове със зависимости се създават отново със същите идентификатори на работни потоци. Това е полезно за одит и за среди, в които съществуват и тези зависими работни потоци, но идентификаторите на зависимостите не се преназначават автоматично между среди.

Случаи на употреба

Възпроизвеждане на грешки от продукционна среда

Експортирайте неуспешен работен поток от продукционна среда и го изпълнете отново локално:

# 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

Повторно изпълненият работен поток има същите състояние и дефиниции на стъпките. При работещите работни потоци текущата стъпка се добавя отново в опашката, за да можете да продължите изпълнението от мястото, на което е спряло. Поставените на пауза или изчакващите работни потоци се създават отново в блокираното си състояние за проверка.

Мигриране на работни потоци между среди

Преместете работен поток от тестова среда в продукционна (или обратно), като го експортирате и изпълните отново:

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

# On production
wp queuety workflow replay workflow.json

Одит и архивиране

Експортирайте приключилите работни потоци за дългосрочно съхранение или съответствие с изискванията. Файлът JSON съдържа пълната история на изпълнение, включително входове, изходи, състояния преди и след изпълнението, времеви данни и подробности за грешките.

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