Εξαγωγή και επανάληψη ροής εργασίας

Το 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.