Flux de travail (Sagas)

La fonctionnalité Workflow exécute des processus métier en plusieurs étapes et de longue durée avec des tentatives, des interruptions et des actions compensatoires. Il s'agit du modèle classique de la saga  : un plugin déclare la forme d'un flux de travail (quelles étapes existent, ce que chacune fait, comment chacune doit réessayer), l'hôte est propriétaire de la boucle de conduite et les étapes échouées peuvent annuler les étapes réussies précédentes dans l'ordre inverse.

Quand l'utiliser#

Accédez à Workflow lorsqu'une requête touche plusieurs systèmes externes et que vous avez besoin d'une sémantique tout ou rien sans transaction distribuée :

• Paiement : réserve de stock, carte de paiement, confirmation de commande, envoi du reçu.
• Intégration : créer un utilisateur, mettre à disposition une boîte aux lettres, attacher un compte de facturation.
• Tâches de migration : exporter, transformer, télécharger, vérifier.

Si votre opération s'inscrit dans une seule transaction de base de données, vous n'avez pas besoin de workflow. Utilisez Workflow lorsque les éléments couvrent des services qui ne peuvent pas partager une limite de transaction.

Le trait#

pub trait WorkflowPlugin: Send + Sync {
    fn name(&self) -> &str;
    fn definitions(&self) -> Vec<WorkflowDefinition>;
    fn run_step(
        &self,
        run: &WorkflowRun,
        step_name: &str,
    ) -> Result<StepOutcome, WorkflowError>;
    fn compensate_step(
        &self,
        _run: &WorkflowRun,
        _step_name: &str,
    ) -> Result<(), WorkflowError> { Ok(()) }
    fn on_workflow_complete(
        &self,
        _run: &WorkflowRun,
    ) -> Result<(), WorkflowError> { Ok(()) }
}

Le plugin possède la forme ** et le comportement **. L'hôte est propriétaire de la boucle : quand appeler run_step, quand réessayer, quand commencer la compensation. Les plugins sont sans état entre les appels : l'état passe par [ WorkflowRun ] et context_updates est renvoyé à chaque étape.

Types de clés#

Sémantique de la stratégie de nouvelle tentative#

RetryPolicy::delay_ms_for_attempt(n) renvoie le délai que l'hôte attend avant la tentative n :

• tentative 1 : délai 0 (l'essai initial n'a pas d'attente précédente)
• tentative 2 : délai backoff_ms
• tentative 3 : délai backoff_ms * backoff_multiplier
• tentative N : délai backoff_ms * backoff_multiplier^(N-2)

Tous les délais sont limités à max_backoff_ms. La politique exponentielle par défaut est max_attempts = 3, backoff_ms = 1000, multiplier = 2.0, max_backoff = 30_000.

RetryPolicy {
    max_attempts: 5,
    backoff_ms: 200,
    backoff_multiplier: 2.0,
    max_backoff_ms: 5_000,
}
// attempts: 0ms, 200ms, 400ms, 800ms, 1600ms

Compensation#

Lorsqu'une étape renvoie StepOutcome::Fail { compensate: true, .. }, l'hôte parcourt le préfixe complété dans l'ordre inverse des étapes et appelle compensate_step sur chacun. L'exécution se termine dans WorkflowStatus::Compensated une fois la restauration terminée ; si la compensation elle-même échoue, la course revient à Failed et l'hôte fait apparaître l'erreur.

Le compensate_step par défaut est un non-op. Les graphiques en étapes idempotents qui ne nécessitent pas de restauration laissent simplement non implémentés.

Exemple : une saga de paiement en trois étapes#

# bext.config.toml
[[plugins]]
name = "workflow-pg"
source = "@bext/workflow-pg"

[[plugins.workflows]]
id = "checkout"
version = 1
steps = [
    { name = "reserve_stock", retries = { max_attempts = 3, backoff_ms = 500, backoff_multiplier = 2.0, max_backoff_ms = 5000 }, idempotent = true },
    { name = "charge_card",   retries = { max_attempts = 1, backoff_ms = 0,   backoff_multiplier = 1.0, max_backoff_ms = 0 },    idempotent = false },
    { name = "send_receipt",  retries = { max_attempts = 5, backoff_ms = 200, backoff_multiplier = 2.0, max_backoff_ms = 10000 }, idempotent = true },
]

fn run_step(
    &self,
    run: &WorkflowRun,
    step_name: &str,
) -> Result<StepOutcome, WorkflowError> {
    match step_name {
        "reserve_stock" => match reserve(&run.context) {
            Ok(reservation_id) => {
                let mut updates = HashMap::new();
                updates.insert("reservation_id".into(), json!(reservation_id));
                Ok(StepOutcome::Continue { context_updates: updates })
            }
            Err(e) if e.transient() => Ok(StepOutcome::Retry { after_ms: 500 }),
            Err(e) => Ok(StepOutcome::Fail {
                reason: e.to_string(),
                compensate: false, // nothing to undo yet
            }),
        },
        "charge_card" => match charge(&run.context) {
            Ok(_) => Ok(StepOutcome::Continue { context_updates: HashMap::new() }),
            Err(_) => Ok(StepOutcome::Fail {
                reason: "card declined".into(),
                compensate: true, // roll back the reservation
            }),
        },
        "send_receipt" => { /* ... */ Ok(StepOutcome::Continue { context_updates: HashMap::new() }) }
        other => Err(WorkflowError::step_not_found(other)),
    }
}

Implémentation de référence#

Le monorepo bext est livré aujourd'hui avec @bext/workflow-pg, une implémentation en mémoire. Lorsque @bext/infra-pg atterrit, le même plugin bénéficie d'une persistance soutenue par PostgreSQL afin que les exécutions survivent aux redémarrages de l'hôte. La surface des traits ne change pas, seul le stockage change.

Indicateur de fonctionnalité#

Aucun. Le trait et les types de flux de travail résident dans bext-plugin-api et sont toujours disponibles ; aucune fonction de chargement ne les bloque.

Voir aussi#

• @bext/workflow-pg — l'implémentation de référence.
• Travaux planifiés — utilise la même répartition plug-in-déclarations/hôte-orchestres pour les tâches de style cron.
• Guide des flux durables — modèles pratiques pour les pipelines de longue durée sur bext.
• Aperçu des capacités — la liste complète des fonctionnalités enfichables.