Flux durables

Les flux durables vous permettent de créer des flux de travail en plusieurs étapes qui survivent aux redémarrages, aux pannes et aux déploiements du serveur. Les entrées et sorties de chaque étape sont conservées dans une base de données SQLite à l'aide de la journalisation Write-Ahead (WAL), afin que bext puisse reprendre les flux interrompus exactement là où ils se sont arrêtés. Il s'agit d'une fonctionnalité Pro.

Utilisez des flux durables pour les opérations qui s'étendent sur plusieurs services ou prennent plus de temps qu'une seule demande : traitement des commandes, workflows de paiement, séquences d'intégration, pipelines de données et tâches par lots planifiées.

Comment ça marche#

Un flux est une séquence nommée d’étapes. Lorsque vous démarrez une exécution de flux, bext :

1. Crée un enregistrement d'exécution dans SQLite avec le statut pending
2. Exécute les étapes dans l'ordre, en conservant le résultat de chaque étape avant de passer à la suivante
3. En cas de crash/redémarrage, recherche les exécutions avec le statut running et reprend à partir de la dernière étape terminée.
4. En cas d'échec, applique la stratégie de nouvelle tentative ; en cas d'épuisement, marque l'exécution comme failed et exécute les étapes de compensation

Le mode WAL SQLite garantit que les écritures sont durables même si le processus est interrompu en cours de transaction.

Définition du flux#

Définissez les flux à l'aide du SDK JavaScript / TypeScript :

import { defineFlow, step } from "bext/flows";

export const orderFlow = defineFlow("process-order", {
  maxAttempts: 3,
  timeout: "30m",
  version: "1.0",

  steps: [
    step("validate", async (input, ctx) => {
      const order = await validateOrder(input.orderId);
      if (!order.valid) throw new Error("Invalid order");
      return { order };
    }),

    step("charge-payment", {
      timeout: "60s",
      retries: 2,
      compensate: async (result, ctx) => {
        // Rollback: refund if a later step fails
        await refundPayment(result.chargeId);
      },
      run: async (input, ctx) => {
        const charge = await chargeCard(input.order.paymentMethod, input.order.total);
        return { chargeId: charge.id };
      }
    }),

    step("fulfill", async (input, ctx) => {
      await createShipment(input.order.items, input.order.address);
      return { shipped: true };
    }),

    step("notify", async (input, ctx) => {
      await sendConfirmationEmail(input.order.email, input.order.id);
      return { notified: true };
    })
  ]
});

Démarrage d'une analyse de flux#

Déclenchez l'exécution d'un flux à partir d'un gestionnaire d'API, d'une tâche planifiée ou d'un autre flux :

import { startFlow } from "bext/flows";

// Start a run and get the run ID
const runId = await startFlow("process-order", {
  tenantId: "tenant_abc",
  input: { orderId: "ord_12345" },
  priority: 1,                    // Higher priority runs first
  idempotencyKey: "ord_12345",    // Prevents duplicate runs
});

Le idempotencyKey garantit que les nouveaux appels d'API ne créent pas d'exécutions de flux en double. Si une exécution avec la même clé existe déjà, l’ID d’exécution existant est renvoyé.

Politiques de nouvelle tentative#

Chaque étape et le flux global peuvent définir le comportement des nouvelles tentatives :

step("external-api-call", {
  retries: 5,           // Max retry attempts for this step
  timeout: "30s",       // Per-attempt timeout
  backoff: "exponential", // "fixed", "exponential", or "linear"
  backoffBase: "1s",    // Starting delay (doubles each retry with exponential)
  run: async (input) => {
    return await callExternalService(input);
  }
})

Le niveau de flux maxAttempts contrôle le nombre de fois où l'ensemble du flux redémarre depuis le début après un échec irrécupérable d'étape.

Compensation (annulation)#

Lorsqu'une étape échoue une fois les étapes précédentes terminées, bext exécute les gestionnaires de compensation dans l'ordre inverse. Voici le modèle de la saga :

Step 1: validate      -> OK
Step 2: charge-payment -> OK (chargeId: ch_123)
Step 3: fulfill       -> FAILED

Compensation runs:
  Step 2 compensate: refundPayment(ch_123)

Les gestionnaires de compensation reçoivent le résultat original de l'étape, ils disposent donc des données nécessaires pour annuler l'opération. Si un gestionnaire de compensation lui-même échoue, l'échec est enregistré et bext continue de compenser les étapes précédentes.

Gestion des délais d'attente#

Les flux prennent en charge les délais d'attente à deux niveaux :

• Délai d'expiration du flux -- durée totale autorisée pour une exécution pour toutes les étapes et tentatives (par exemple, "30m" )
• Délai d'expiration de l'étape -- durée pendant laquelle une exécution d'une seule étape est autorisée (par exemple "60s" )

Les chaînes de durée prennent en charge les suffixes ms, s, m, h et d. Lorsqu'un délai d'attente se déclenche, l'étape est marquée comme failed et la logique de nouvelle tentative/compensation s'active.

Statut d'exécution#

Chaque exécution passe par ces statuts :

Les étapes ont un cycle de vie similaire : pending, running, completed, failed, skipped.

Inspection d'une exécution#

Récupérez l’état complet d’une seule exécution, y compris le statut, les tentatives et le résultat de chaque étape, à partir de l’API d’administration. Sondez-le pour obtenir des tableaux de bord et une surveillance en direct :

GET /__bext/admin/api/flows/{id}

La réponse reflète chaque changement de statut au fur et à mesure de la progression de l'exécution :

{
  "id": "01abc...",
  "flow_id": "process-order",
  "status": "completed",
  "attempts": 1,
  "steps": [
    { "step_name": "charge-payment", "step_index": 0, "status": "completed",
      "attempt": 1, "result_json": "{\"chargeId\":\"ch_123\"}" }
  ]
}

Exécutions de requêtes#

Le flux de liste et de filtrage s'exécute via l'API d'administration :

GET /__bext/admin/api/flows?flow_id=process-order&status=failed&limit=50

{
  "flows": [
    {
      "id": "018f3a2b...",
      "flow_id": "process-order",
      "tenant_id": "tenant_abc",
      "status": "failed",
      "error": "Payment declined",
      "created_at": 1712150400000,
      "attempts": 3
    }
  ]
}

Minuteries et retards#

Insérez un délai entre les étapes ou planifiez l'exécution d'une étape à une heure ultérieure :

step("wait-for-confirmation", {
  delay: "24h",  // Wait 24 hours before executing
  run: async (input, ctx) => {
    const confirmed = await checkConfirmation(input.order.id);
    if (!confirmed) throw new Error("Not confirmed in time");
    return { confirmed: true };
  }
})

Les minuteurs sont conservés sur SQLite avec un horodatage wake_at. Au redémarrage, bext recherche les minuteurs déclenchés et reprend les flux correspondants.

Visualisation du flux#

L'application compagnon bext fournit une chronologie visuelle des exécutions de flux, affichant l'état, la durée, les tentatives et les erreurs de chaque étape. Ceci est inestimable pour le débogage des flux de production.

Accédez-y à /__bext/admin/api/flows dans le tableau de bord d'administration ou connectez l'application compagnon autonome à votre instance bext.

En rapport#

• Planificateur de tâches — minuterie fiable pour déclencher des exécutions de débit selon un calendrier
• Capacités — Flux de travail — capacité de flux de travail de niveau supérieur qui encapsule les flux durables
• Actions du serveur — déclenche l'exécution d'un flux à partir d'un appel de soumission de formulaire ou de récupération
• Opérations — Surveillance — métriques d'exécution de flux et alertes d'échec
• Compagnon – Inspecteur de demande — inspecte les demandes de déclenchement de flux de bout en bout