Экспорт и повторное воспроизведение рабочего процесса

Queuety может экспортировать полную историю выполнения рабочего процесса в переносимый формат JSON и воспроизводить ее в другой среде. Это полезно для локального воспроизведения ошибок в рабочей среде, переноса рабочих процессов между средами и аудита.

Экспорт рабочего процесса

PHP API

Вызовите Queuety::export_workflow(), указав ID рабочего процесса. Метод возвращает массив, сериализуемый в 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() с данными экспорта. Метод возвращает новый ID рабочего процесса:

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, этот статус сохраняется в воспроизведенном рабочем процессе

Повторно воспроизведенный рабочий процесс получает новый ID и работает независимо. Он не влияет на исходный.

При повторном воспроизведении waiting_for_workflows экспортированные строки зависимостей создаются заново с теми же ID рабочих процессов. Это полезно для аудита и для сред, где эти зависимые рабочие процессы также существуют, но ID зависимостей между средами автоматически не переназначаются.

Сценарии использования

Воспроизведение ошибок в рабочей среде

Экспортируйте неудачно завершившийся рабочий процесс из рабочей среды и воспроизведите его локально:

# 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.