Exportul și reluarea fluxurilor de lucru

Queuety poate exporta istoricul complet de execuție al unui flux de lucru într-un format portabil JSON și îl poate relua într-un alt mediu. Acest lucru este util pentru reproducerea locală a erorilor din producție, migrarea fluxurilor de lucru între medii și auditare.

Exportarea unui flux de lucru

PHP API

Apelați Queuety::export_workflow() cu ID-ul fluxului de lucru. Returnează un tablou serializabil JSON:

use Queuety\Queuety;

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

Pentru a obține direct un șir 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

Reluarea unui flux de lucru

PHP API

Apelați Queuety::replay_workflow() cu datele exportului. Returnează ID-ul noului flux de lucru:

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.

Structura JSON

Exportul conține starea completă a fluxului de lucru, joburile asociate, evenimentele de urmărire, intrările în jurnal, semnalele stocate, artefactele, fragmentele și orice rânduri pentru așteptarea dependențelor dintre fluxurile de lucru:

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

Unele câmpuri sunt abreviate în exemplul de mai sus. Exportul real include toate coloanele de urmărire din fiecare rând de eveniment.

Exporturile fluxurilor de lucru păstrează și metadatele acestora, precum definition_version, definition_hash și idempotency_key, dacă acestea au fost setate în execuția inițială.

Cum funcționează reluarea

Când reluați un flux de lucru exportat:

1. Se creează un rând nou pentru fluxul de lucru, cu starea și definițiile pașilor exportate
2. Numele noului flux de lucru este {original_name}_replay_{timestamp}
3. Evenimentele și intrările istorice din jurnalul fluxului de lucru sunt reinserate în noul mediu
4. Semnalele stocate și rândurile dependențelor așteptate dintre fluxurile de lucru sunt reluate ca stare istorică
5. Dacă fluxul de lucru inițial era încă running, pasul curent este pus în coadă pentru procesare. Pașii Repeat și start_workflows() nu sunt repuși în coadă, astfel că o reluare care se oprește la unul dintre ei rămâne running fără job (#17)
6. Dacă fluxul de lucru inițial era waiting_for_signal, waiting_for_workflows, paused, failed, cancelled sau completed, acea stare este păstrată în fluxul de lucru reluat

Fluxul de lucru reluat primește un ID nou și funcționează independent. Nu afectează fluxul original.

Pentru reluările waiting_for_workflows, rândurile de dependență exportate sunt recreate cu aceleași ID-uri de flux de lucru. Acest lucru este util pentru auditare și pentru medii în care există și fluxurile de lucru dependente, dar ID-urile dependențelor nu sunt remapate automat între medii.

Cazuri de utilizare

Reproducerea erorilor din producție

Exportați un flux de lucru eșuat din mediul de producție și reluați-l 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

Fluxul de lucru reluat are exact aceeași stare și aceleași definiții ale pașilor. Fluxurile de lucru aflate în execuție repun pasul curent în coadă, astfel încât să puteți continua de unde s-au oprit, iar cele suspendate sau în așteptare sunt recreate în starea blocată, pentru a fi inspectate.

Migrarea fluxurilor de lucru între medii

Mutați un flux de lucru din staging în producție (sau invers) exportându-l și reluându-l:

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

# On production
wp queuety workflow replay workflow.json

Auditare și arhivare

Exportați fluxurile de lucru încheiate pentru stocare pe termen lung sau conformitate. Fișierul JSON conține istoricul complet de execuție, inclusiv intrările, ieșirile, starea înainte și după, datele de timp și detaliile erorilor.

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