Instantanés (snapshots)

Dans Mastra, un snapshot est une représentation sérialisable de l’état d’exécution complet d’un Workflow à un instant donné. Les snapshots capturent toutes les informations nécessaires pour reprendre un Workflow exactement là où il s’était arrêté, notamment :

• L’état actuel de chaque étape du Workflow
• Les sorties des étapes terminées
• Le chemin d’exécution emprunté dans le Workflow
• Les éventuelles étapes suspendues et leurs métadonnées
• Le nombre de nouvelles tentatives restantes pour chaque étape
• Les données contextuelles supplémentaires nécessaires à la reprise de l’exécution

Mastra crée et gère automatiquement les snapshots chaque fois qu’un Workflow est suspendu, puis les conserve dans le système de stockage configuré.

Rôle des snapshots dans la suspension et la reprise

Les snapshots constituent le mécanisme essentiel qui permet à Mastra de suspendre et de reprendre une exécution. Lorsqu’une étape de Workflow appelle await suspend() :

1. L’exécution du Workflow est interrompue à cet endroit précis
2. L’état actuel du Workflow est capturé sous forme de snapshot
3. Le snapshot est conservé dans le stockage
4. L’étape du Workflow est marquée comme suspendue avec l’état 'suspended'
5. Plus tard, lorsque resume() est appelée sur l’étape suspendue, le snapshot est récupéré
6. L’exécution du Workflow reprend exactement là où elle s’était arrêtée

Ce mécanisme offre un moyen puissant de mettre en œuvre des Workflows avec intervention humaine, de gérer la limitation de débit, d’attendre des ressources externes et de créer des Workflows complexes à embranchements susceptibles de rester suspendus pendant de longues périodes.

Anatomie d’un snapshot

Chaque snapshot comprend le runId, l’entrée, l’état des étapes (success, suspended, etc.), les éventuelles charges utiles de suspension et de reprise, ainsi que la sortie finale. Le contexte complet est ainsi disponible lors de la reprise de l’exécution.

{
  "runId": "34904c14-e79e-4a12-9804-9655d4616c50",
  "status": "success",
  "value": {},
  "context": {
    "input": {
      "value": 100,
      "user": "Michael",
      "requiredApprovers": ["manager", "finance"]
    },
    "approval-step": {
      "payload": {
        "value": 100,
        "user": "Michael",
        "requiredApprovers": ["manager", "finance"]
      },
      "startedAt": 1758027577955,
      "status": "success",
      "suspendPayload": {
        "message": "Workflow suspended",
        "requestedBy": "Michael",
        "approvers": ["manager", "finance"]
      },
      "suspendedAt": 1758027578065,
      "resumePayload": { "confirm": true, "approver": "manager" },
      "resumedAt": 1758027578517,
      "output": { "value": 100, "approved": true },
      "endedAt": 1758027578634
    }
  },
  "activePaths": [],
  "serializedStepGraph": [
    {
      "type": "step",
      "step": {
        "id": "approval-step",
        "description": "Accepts a value, waits for confirmation"
      }
    }
  ],
  "suspendedPaths": {},
  "waitingPaths": {},
  "result": { "value": 100, "approved": true },
  "requestContext": {},
  "timestamp": 1758027578740
}

Enregistrement et récupération des snapshots

Les snapshots sont enregistrés dans le système de stockage configuré. Ils utilisent libSQL par défaut, mais vous pouvez choisir Upstash, PostgreSQL ou OracleDB à la place. Chaque snapshot est enregistré dans la table workflow_snapshots et identifié par le runId du Workflow.

Pour en savoir plus :

• Stockage libSQL
• Stockage Upstash
• Stockage PostgreSQL
• Stockage OracleDB

Enregistrer des snapshots

Lorsqu’un Workflow est suspendu, Mastra conserve automatiquement son snapshot en suivant ces étapes :

1. La fonction suspend() appelée pendant l’exécution d’une étape déclenche le processus de création du snapshot
2. La méthode WorkflowInstance.suspend() enregistre la machine suspendue
3. persistWorkflowSnapshot() est appelée pour enregistrer l’état actuel
4. Le snapshot est sérialisé et stocké dans la table workflow_snapshots de la base de données configurée
5. L’enregistrement de stockage comprend le nom du Workflow, l’identifiant de l’exécution et le snapshot sérialisé

Récupérer des snapshots

Lorsqu’un Workflow reprend, Mastra récupère le snapshot conservé en suivant ces étapes :

1. La méthode resume() est appelée avec l’identifiant d’une étape précise
2. Le snapshot est chargé depuis le stockage au moyen de loadWorkflowSnapshot()
3. Le snapshot est analysé et préparé pour la reprise
4. L’exécution du Workflow est recréée avec l’état du snapshot
5. L’étape suspendue reprend, puis l’exécution se poursuit

const storage = mastra.getStorage()
const workflowStore = await storage?.getStore('workflows')

const snapshot = await workflowStore?.loadWorkflowSnapshot({
  runId: '<run-id>',
  workflowName: '<workflow-id>',
})

console.log(snapshot)

Options de stockage des snapshots

Les snapshots sont conservés au moyen d’une instance storage configurée dans la classe Mastra. Cette couche de stockage est partagée par tous les Workflows enregistrés dans cette instance. Mastra prend en charge plusieurs options de stockage afin de s’adapter à différents environnements.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { LibSQLStore } from '@mastra/libsql'
import { approvalWorkflow } from './workflows'

export const mastra = new Mastra({
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: ':memory:',
  }),
  workflows: { approvalWorkflow },
})

• Stockage libSQL
• Stockage PostgreSQL
• Stockage OracleDB
• Stockage MongoDB
• Stockage Upstash
• Cloudflare D1
• DynamoDB
• Autres fournisseurs de stockage

Bonnes pratiques

1. Garantir la sérialisabilité : toutes les données à inclure dans le snapshot doivent être sérialisables (convertibles en JSON).
2. Réduire la taille des snapshots : évitez de stocker directement des objets de données volumineux dans le contexte du Workflow. Stockez plutôt des références vers ces objets (comme des identifiants) et récupérez les données lorsque cela est nécessaire.
3. Gérer soigneusement le contexte de reprise : lors de la reprise d’un Workflow, déterminez avec attention le contexte à fournir. Celui-ci sera fusionné avec les données existantes du snapshot.
4. Mettre en place une surveillance adaptée : surveillez les Workflows suspendus, en particulier ceux de longue durée, et vérifiez qu’ils reprennent correctement.
5. Prévoir la montée en charge du stockage : pour les applications qui comportent de nombreux Workflows suspendus, assurez-vous que votre solution de stockage est correctement dimensionnée.

Métadonnées de snapshot personnalisées

Vous pouvez associer des métadonnées personnalisées lors de la suspension d’un Workflow en définissant un suspendSchema. Ces métadonnées sont stockées dans le snapshot et rendues disponibles lors de la reprise du Workflow.

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import { z } from 'zod'

const approvalStep = createStep({
  id: 'approval-step',
  description: 'Accepts a value, waits for confirmation',
  inputSchema: z.object({
    value: z.number(),
    user: z.string(),
    requiredApprovers: z.array(z.string()),
  }),
  suspendSchema: z.object({
    message: z.string(),
    requestedBy: z.string(),
    approvers: z.array(z.string()),
  }),
  resumeSchema: z.object({
    confirm: z.boolean(),
    approver: z.string(),
  }),
  outputSchema: z.object({
    value: z.number(),
    approved: z.boolean(),
  }),
  execute: async ({ inputData, resumeData, suspend }) => {
    const { value, user, requiredApprovers } = inputData
    const { confirm } = resumeData ?? {}

    if (!confirm) {
      return await suspend({
        message: 'Workflow suspended',
        requestedBy: user,
        approvers: [...requiredApprovers],
      })
    }

    return {
      value,
      approved: confirm,
    }
  },
})

Fournir des données de reprise

Utilisez resumeData pour transmettre une entrée structurée lors de la reprise d’une étape suspendue. Cette entrée doit correspondre au resumeSchema de l’étape.

const workflow = mastra.getWorkflow('approvalWorkflow')

const run = await workflow.createRun()

const result = await run.start({
  inputData: {
    value: 100,
    user: 'Michael',
    requiredApprovers: ['manager', 'finance'],
  },
})

if (result.status === 'suspended') {
  const resumedResult = await run.resume({
    step: 'approval-step',
    resumeData: {
      confirm: true,
      approver: 'manager',
    },
  })
}

Voir aussi

• Flux de contrôle
• Suspendre et reprendre
• Voyage dans le temps
• Intervention humaine